Agentic Engineering Lab

Build it in stages

Run the seed’s local stock-tool transcript and inspect its call/result correlation.
Define your own provider-neutral request, call, observation, and final-result types.
Add malformed arguments, unsupported tools, refusal, interrupted stream, and exhausted-budget fixtures.
Use the lab reducer for normalized streaming events and report incomplete output accurately.
Implement a live Responses adapter behind the same interface using environment-configured credentials and model.
Export a redacted run trace with model, prompt version, latency, usage when available, and terminal status.

Your acceptance criteria

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

The offline suite runs without packages, network, or credentials.
Every executed tool has an allowlisted name, validated arguments, and a correlated result ID.
No fixture can cause a tool call after the controller’s budget is exhausted.
An interrupted stream never appears completed.
Live requests are clearly opt-in and use OPENAI_MODEL without an invented default.
A structured extraction with a fabricated evidence quote is rejected.

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

STOCK = {"A7": 8, "B2": 0}
TRANSCRIPT = [
    {"kind":"call", "call_id":"c1", "name":"get_stock", "arguments":'{"sku":"A7"}'},
    {"kind":"final", "text":"A7 has 8 available units."},
]

def dispatch(item):
    if item["name"] != "get_stock":
        raise ValueError("unsupported tool")
    args = json.loads(item["arguments"])
    if not isinstance(args,dict) or set(args) != {"sku"} or not isinstance(args["sku"],str):
        raise ValueError("invalid arguments")
    sku = args["sku"]
    return {"sku":sku, "found":sku in STOCK, "available":STOCK.get(sku)}

def replay(transcript, max_calls=1):
    trace = []
    seen = set()
    final = None
    for item in transcript:
        if final is not None:
            raise ValueError("event after final")
        if item["kind"] == "call":
            if len(seen) >= max_calls or item["call_id"] in seen:
                raise ValueError("call budget or duplicate ID")
            observation = dispatch(item)
            seen.add(item["call_id"])
            trace.append({"call_id":item["call_id"], "observation":observation})
        elif item["kind"] == "final":
            final = item["text"]
        else:
            raise ValueError("unknown transcript item")
    return {"status":"completed" if final is not None else "incomplete",
            "text":final or "", "trace":trace, "tool_calls":len(seen)}

if __name__ == "__main__":
    print(json.dumps(replay(TRANSCRIPT), indent=2))
    print("partial status:", replay(TRANSCRIPT[:1])["status"])

Push it further

Add bounded retries with a fake clock and distinguish retryable transport failures from validation errors; prove a write-like fixture cannot be executed twice after an ambiguous result.