Agentic Engineering Lab

Build it in stages

Run the seed and inspect the successful replay and payload conflict.
Extract independent envelope, argument, policy, and reservation components.
Add tenant-scoped stock and explicit unknown-outcome reconciliation fixtures.
Add a bounded retry simulator that loses selected responses after commit.
Write a report comparing attempted calls, unique effects, replays, and rejected calls.

Your acceptance criteria

Use these as your project review. Record commands, outputs, and failure cases in your repository.

At least 12 deterministic scenarios cover valid calls, denied calls, malformed input, conflicts, and lost responses.
Replaying an authorized identical operation 20 times changes stock once.
A denied replay reveals no cached result.
The README names the atomicity, persistence, and retention limits of the simulator.

A working starting point

The seed runs as supplied. Extend it to satisfy the full brief. It is a teaching starting point, not a finished portfolio submission.

main.py

python

import json

class Gateway:
    def __init__(self):
        self.stock = {"A": 10}
        self.ledger = {}
        self.audit = []

    def handle(self, request):
        status = "rejected"
        try:
            if request["scope"] != "inventory:reserve":
                raise ValueError("denied")
            args = request["arguments"]
            sku, qty = args["sku"], args["quantity"]
            if type(qty) is not int or not 1 <= qty <= 20:
                raise ValueError("invalid quantity")
            key = (request["tenant"], request["key"])
            fingerprint = (sku, qty)
            if key in self.ledger:
                saved, result = self.ledger[key]
                if saved != fingerprint:
                    raise ValueError("key conflict")
                status = "replayed"
            else:
                if self.stock.get(sku, 0) < qty:
                    raise ValueError("insufficient stock")
                self.stock[sku] -= qty
                result = {"remaining": self.stock[sku]}
                self.ledger[key] = (fingerprint, result)
                status = "committed"
            return {"id": request["id"], "result": dict(result)}
        except (KeyError, ValueError) as exc:
            return {"id": request.get("id"), "error": str(exc)}
        finally:
            self.audit.append({"id": request.get("id"), "status": status})

gateway = Gateway()
base = {"tenant": "north", "scope": "inventory:reserve", "key": "K"}
for attempt, quantity in enumerate([3, 3, 4], 1):
    request = {**base, "id": attempt, "arguments": {"sku": "A", "quantity": quantity}}
    print(json.dumps(gateway.handle(request), sort_keys=True))
print(json.dumps(gateway.audit, sort_keys=True))

Push it further

Implement a SQLite-backed ledger and stock table in one transaction, then demonstrate concurrent same-key requests with separate database connections.