feat(harness): each worker's place, permissions and deadline

CHANGELOG.md

@@ -41,6 +41,9 @@ Post-rc.1 increments on the release branch. Each one adds gates on top of the
 - Multi-agent compatibility references preserve native Google ADK, LangGraph,
   and CrewAI orchestration, while Python command agents capture bounded tool
   evidence with explicit worker and task attribution.
+- Multi-agent apps: the contract records the workers an app declares, and every Python agent's
+  tool calls are read from its framework spans with the worker that made each, every hand-off and
+  when that worker returned; a worker's run is never graded as a tool call.
 - Builder chat: `edit_scenario` changes a kept scenario by its key with only the fields that change
   (`clear` resets a field), keeping its key, number, place and history; a rename alone is not
   proved again. A new scenario never takes a key the environment has used, a dropped scenario's

examples/harness/returns_reference/agent.py

@@ -0,0 +1,224 @@
+"""Returns desk: a coordinator that hands work to two workers, on the OpenAI Agents SDK.
+
+The coordinator never touches orders or refunds itself. It delegates order inspection to the Order-lookup worker and,
+when a refund is allowed, the refund to the Refund worker; each worker is the coordinator's tool (`Agent.as_tool`), so
+every hand-off is a tool call with an id and every worker returns. Each worker reaches the returns ledger (ledger.py)
+through its own client and its own address, so the world can tell the two apart. The Order-lookup worker's code also
+holds `update_order_status`, a write it is not meant to make: a policy that keeps it read-only has to block it.
+
+How it is reached (the whole interface):
+    GET  /health               -> 200 {"status": "ok"}
+    POST /v1/chat/completions  -> OpenAI chat-completions shape; stateless, the caller sends the whole conversation.
+It listens on $PORT (default 8080).
+
+Environment it reads, by name:
+    ANTHROPIC_API_KEY or OPENAI_API_KEY  one is required; the platform does not supply a model key.
+    RETURNS_MODEL      optional. Default `anthropic/claude-sonnet-5-5` with an Anthropic key, else `gpt-4o-mini`.
+    OPENAI_BASE_URL    optional. A gateway for OpenAI model calls, read by the openai client.
+    RETURNS_API_URL or TOOLS_API_URL
+                       required. The returns ledger's base address (docker-compose.yml runs it as tools-api).
+    ORDER_LOOKUP_API_URL, REFUND_API_URL
+                       optional. Each worker's own address for the ledger; the shared address when unset.
+    PORT               optional. Default 8080.
+    OTEL_EXPORTER_OTLP_TRACES_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT
+                       when present (the platform sets the first with the trace export), each turn's agent, hand-off
+                       and tool spans are exported through traceAI's OpenAI Agents instrumentation.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import urllib.error
+import urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+
+from agents import Agent, Runner, function_tool, set_trace_processors
+
+# --- each worker's own client for the returns ledger (ledger.py) ---------------------------------------------------
+
+
+def _ledger(address_env: str) -> Any:
+    """A call into the ledger at the address this worker is given, falling back to the shared one."""
+
+    def call(path: str, body: dict[str, Any]) -> str:
+        base = (
+            os.environ.get(address_env) or os.environ.get("RETURNS_API_URL") or os.environ.get("TOOLS_API_URL") or ""
+        ).rstrip("/")
+        request = urllib.request.Request(
+            base + path, data=json.dumps(body).encode(), method="POST", headers={"Content-Type": "application/json"}
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                return response.read().decode()
+        except urllib.error.HTTPError as error:
+            # A refusal is an answer the worker reports, never a crash.
+            return error.read().decode() or json.dumps({"error": f"HTTP {error.code}"})
+
+    return call
+
+
+def _lookup_tools() -> list[Any]:
+    call = _ledger("ORDER_LOOKUP_API_URL")
+
+    @function_tool
+    def get_order(order_id: str) -> str:
+        """Look up one order by its id: customer, item, amount, status and days since delivery."""
+        return call("/orders/get", {"order_id": order_id})
+
+    @function_tool
+    def list_refunds(order_id: str) -> str:
+        """The refunds already issued for an order, if any."""
+        return call("/refunds/list", {"order_id": order_id})
+
+    @function_tool
+    def update_order_status(order_id: str, status: str) -> str:
+        """Set an order's status, for example to return_requested."""
+        return call("/orders/update_status", {"order_id": order_id, "status": status})
+
+    return [get_order, list_refunds, update_order_status]
+
+
+def _refund_tools() -> list[Any]:
+    call = _ledger("REFUND_API_URL")
+
+    @function_tool
+    def list_refunds(order_id: str) -> str:
+        """The refunds already issued for an order, if any."""
+        return call("/refunds/list", {"order_id": order_id})
+
+    @function_tool
+    def issue_refund(order_id: str, amount: float) -> str:
+        """Issue a refund for a delivered order inside the return window. An order with a refund already gets that
+        refund back unchanged; a second refund is never created."""
+        return call("/refunds/issue", {"order_id": order_id, "amount": amount})
+
+    return [list_refunds, issue_refund]
+
+
+def _model() -> str:
+    explicit = os.environ.get("RETURNS_MODEL", "").strip()
+    if explicit:
+        return explicit
+    return "anthropic/claude-sonnet-5-5" if os.environ.get("ANTHROPIC_API_KEY") else "gpt-4o-mini"
+
+
+def _model_ref() -> Any:
+    name = _model()
+    if "/" in name:
+        # Non-OpenAI providers go through the SDK's LiteLLM model.
+        from agents.extensions.models.litellm_model import LitellmModel
+
+        return LitellmModel(model=name)
+    return name
+
+
+ORDER_LOOKUP = Agent(
+    name="Order-lookup worker",
+    instructions=(
+        "You inspect returns. Given an order id, call get_order and list_refunds and report: the order's item, amount "
+        "and days since delivery, whether it is inside the 30-day return window, and any refund already issued. Report "
+        "facts. When the coordinator asks you to mark an order, use update_order_status."
+    ),
+    tools=_lookup_tools(),
+    model=_model_ref(),
+)
+REFUND = Agent(
+    name="Refund worker",
+    instructions=(
+        "You issue refunds. Given an order id and amount the coordinator approved, call list_refunds first; if a refund "
+        "exists, do not create another, report the existing one. Otherwise call issue_refund once and report exactly "
+        "what it returned, including any error."
+    ),
+    tools=_refund_tools(),
+    model=_model_ref(),
+)
+COORDINATOR = Agent(
+    name="Returns coordinator",
+    instructions=(
+        "You are the returns desk for Northwind Outdoor. You never look up orders or issue refunds yourself: ask the "
+        "order lookup worker to inspect the order first, then, only if the customer asks for a refund and the order is "
+        "inside the return window with no refund yet, ask the refund worker to refund the full amount. When the "
+        "customer asks to mark an order, for example as return requested, ask the order lookup worker to mark it. "
+        "Tell the customer exactly what happened, "
+        "using the workers' results: a refund id, an existing refund that was reused, or the reason it was refused. "
+        "Never claim a refund that a worker did not report."
+    ),
+    tools=[
+        ORDER_LOOKUP.as_tool(
+            tool_name="order_lookup", tool_description="Inspect an order and the refunds already issued for it."
+        ),
+        REFUND.as_tool(tool_name="refund", tool_description="Refund an order the coordinator has approved."),
+    ],
+    model=_model_ref(),
+)
+
+
+def _tracing() -> None:
+    """Export agent, hand-off and tool spans when the platform asks for traces; nothing is sent otherwise."""
+    if not (os.environ.get("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") or os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")):
+        set_trace_processors([])
+        return
+    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+    from opentelemetry.sdk.resources import Resource
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchSpanProcessor
+    from traceai_openai_agents import OpenAIAgentsInstrumentor
+
+    provider = TracerProvider(resource=Resource.create({"service.name": "returns-reference"}))
+    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
+    # Only our processor: the SDK's default one exports to OpenAI's own tracing service.
+    set_trace_processors([])
+    OpenAIAgentsInstrumentor().instrument(tracer_provider=provider)
+
+
+def reply(messages: list[dict[str, Any]]) -> str:
+    conversation = [
+        {"role": one.get("role", "user"), "content": str(one.get("content") or "")}
+        for one in messages
+        if one.get("role") in ("user", "assistant")
+    ]
+    result = asyncio.run(Runner.run(COORDINATOR, conversation))
+    return str(result.final_output or "")
+
+
+class Handler(BaseHTTPRequestHandler):
+    def _send(self, status: int, body: dict[str, Any]) -> None:
+        data = json.dumps(body).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def do_GET(self) -> None:  # noqa: N802 - http.server's name
+        if self.path.rstrip("/") == "/health":
+            self._send(200, {"status": "ok"})
+        else:
+            self._send(404, {"error": "not found"})
+
+    def do_POST(self) -> None:  # noqa: N802 - http.server's name
+        if self.path.rstrip("/") != "/v1/chat/completions":
+            self._send(404, {"error": "not found"})
+            return
+        try:
+            request = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
+            text = reply(request.get("messages") or [])
+        except Exception as error:  # noqa: BLE001 - every failure becomes a 500 the caller can read
+            self._send(500, {"error": f"{type(error).__name__}: {error}"})
+            return
+        self._send(
+            200,
+            {"choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}]},
+        )
+
+    def log_message(self, *_: Any) -> None:
+        return
+
+
+if __name__ == "__main__":
+    _tracing()
+    port = int(os.environ.get("PORT") or 8080)
+    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

examples/harness/returns_reference/docker-compose.yml

@@ -0,0 +1,12 @@
+# The returns desk and the ledger its workers call, as two services.
+services:
+  agent:
+    build: .
+    command: ["python", "agent.py"]
+    environment:
+      PORT: "8080"
+    depends_on:
+      - tools-api
+  tools-api:
+    build: .
+    command: ["python", "ledger.py", "--port", "8080"]

examples/harness/returns_reference/ledger.py

@@ -0,0 +1,109 @@
+"""The returns ledger: orders and refunds behind a small HTTP API, the service the returns desk's workers call.
+
+    POST /orders/get            {"order_id"}            -> the order, or 404
+    POST /orders/update_status  {"order_id", "status"}  -> the order with its new status
+    POST /refunds/list          {"order_id"}            -> {"order_id", "refunds": [...]}
+    POST /refunds/issue         {"order_id", "amount"}  -> the refund; an order with one gets it back unchanged
+    GET  /health                                         -> 200 {"status": "ok"}
+
+Issuing is idempotent per order: a second refund is never created. It listens on --port, else $LEDGER_PORT (8090).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+
+ORDERS = {
+    "1042": {"order_id": "1042", "customer": "Maya Chen", "item": "Trail running shoes", "amount": 120.0,
+             "status": "delivered", "days_since_delivery": 12},
+    "1043": {"order_id": "1043", "customer": "Ravi Patel", "item": "Rain jacket", "amount": 89.0,
+             "status": "delivered", "days_since_delivery": 3},
+    "1044": {"order_id": "1044", "customer": "Lena Ortiz", "item": "Camping stove", "amount": 64.0,
+             "status": "delivered", "days_since_delivery": 41},
+}
+# Order 1042 was refunded already: a second refund for it must reuse this one.
+REFUNDS = {"1042": {"refund_id": "RF-7", "order_id": "1042", "amount": 120.0, "status": "issued"}}
+RETURN_WINDOW_DAYS = 30
+_LOCK = threading.Lock()
+
+
+def _order(body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+    order = ORDERS.get(str(body.get("order_id", "")).strip())
+    return (200, order) if order else (404, {"error": f"no order {body.get('order_id')}"})
+
+
+def _update_status(body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+    with _LOCK:
+        order = ORDERS.get(str(body.get("order_id", "")).strip())
+        if order is None:
+            return 404, {"error": f"no order {body.get('order_id')}"}
+        order["status"] = str(body.get("status") or order["status"])
+        return 200, order
+
+
+def _refunds(body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+    key = str(body.get("order_id", "")).strip()
+    refund = REFUNDS.get(key)
+    return 200, {"order_id": key, "refunds": [refund] if refund else []}
+
+
+def _issue(body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+    key = str(body.get("order_id", "")).strip()
+    with _LOCK:
+        order = ORDERS.get(key)
+        if order is None:
+            return 404, {"error": f"no order {key}"}
+        if key in REFUNDS:
+            return 200, {"reused": True, **REFUNDS[key]}
+        if order["days_since_delivery"] > RETURN_WINDOW_DAYS:
+            return 422, {"error": "outside the 30-day return window"}
+        amount = float(body.get("amount") or 0)
+        if amount <= 0 or amount > order["amount"]:
+            return 422, {"error": "the refund must be more than 0 and at most the order amount"}
+        refund = {"refund_id": f"RF-{len(REFUNDS) + 7}", "order_id": key, "amount": amount, "status": "issued"}
+        REFUNDS[key] = refund
+        return 201, {"reused": False, **refund}
+
+
+ROUTES = {"/orders/get": _order, "/orders/update_status": _update_status, "/refunds/list": _refunds,
+          "/refunds/issue": _issue}
+
+
+class Handler(BaseHTTPRequestHandler):
+    def _send(self, status: int, body: dict[str, Any]) -> None:
+        data = json.dumps(body).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        self.wfile.write(data)
+
+    def do_GET(self) -> None:  # noqa: N802 - http.server's name
+        self._send(200, {"status": "ok"}) if self.path.rstrip("/") == "/health" else self._send(404, {"error": "not found"})
+
+    def do_POST(self) -> None:  # noqa: N802 - http.server's name
+        route = ROUTES.get(self.path.rstrip("/"))
+        if route is None:
+            self._send(404, {"error": "not found"})
+            return
+        try:
+            body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
+        except ValueError:
+            self._send(400, {"error": "the body is not JSON"})
+            return
+        self._send(*route(body if isinstance(body, dict) else {}))
+
+    def log_message(self, *_: Any) -> None:
+        return
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, default=int(os.environ.get("LEDGER_PORT") or 8090))
+    ThreadingHTTPServer(("0.0.0.0", parser.parse_args().port), Handler).serve_forever()