WIP: Add multi-route manual AI proxy fixture and evaluator contract

apps/ai-observability/README.md

@@ -24,6 +24,7 @@ Each app exists to test one thing the others don't:
 - `openai-agents/python-travel-triage` — tracing processor; the SDK emits the tree
 - `vercel-ai/nextjs-support-chat` — per-request identity; framework bootstrap
 - `manual-capture/node-http-chat` — hand-built tree; must reuse the existing client
+- `manual-capture/python-multi-route-proxy` — raw proxy routes, streams, failures, and direct calls
 - `google-adk/node-weather` — framework plugin; identity comes from ADK's own ids
 - `opentelemetry/go-weather` — Go; no wrapper SDK exists, so the posthog-go OTel bridge
 
@@ -47,8 +48,9 @@ conversation structure.
   together. A fresh id per call groups nothing and is worse than none.
 - **Spans come from tool registration** (or explicit `$ai_span` capture on the
   manual path). Never from hand-authored wrappers around helper functions.
-- Static fixtures in the style of [`../mcp-analytics`](../mcp-analytics): not
-  executed, no lockfiles, no keys. Node type-checks with `tsc --noEmit`.
+- Static Wizard fixtures in the style of [`../mcp-analytics`](../mcp-analytics):
+  no lockfiles or keys. Node type-checks with `tsc --noEmit`. The multi-route
+  proxy has a local fake-provider contract test; Wizard CI evaluates its diff.
 - Each README states the tree the app must produce and grades **emitted
   events, not the diff** — every observed failure mode so far produced
   plausible code and a broken tree.

apps/ai-observability/manual-capture/python-multi-route-proxy/.env.example

@@ -0,0 +1,4 @@
+OPENAI_BASE_URL=http://127.0.0.1:8001
+OLLAMA_BASE_URL=http://127.0.0.1:11434
+PORT=8000
+# OPENAI_API_KEY=your-provider-key

apps/ai-observability/manual-capture/python-multi-route-proxy/.gitignore

@@ -0,0 +1,3 @@
+.venv/
+__pycache__/
+*.pyc

apps/ai-observability/manual-capture/python-multi-route-proxy/README.md

@@ -0,0 +1,76 @@
+# wb-aio-manual-python-multi-route-proxy
+
+An aiohttp chat gateway forwards raw provider HTTP responses. It has no vendor
+LLM SDK and no PostHog code. The Wizard must use manual AI capture without
+changing the proxy's request or response behavior.
+
+The gateway supplies `X-User-Id` and `X-Conversation-Id` on every request. A
+conversation can make several requests. Each request or WebSocket message is
+one trace in that conversation. These headers stand in for identity resolved by
+an upstream gateway. They are not provider credentials.
+
+## Inference paths
+
+| Entry point | Upstream call | Response shape | Capture target |
+| --- | --- | --- | --- |
+| `POST /v1/chat/completions` | OpenAI-compatible `/v1/chat/completions` | JSON or SSE | One generation per request |
+| `POST /v1/responses` | OpenAI Responses `/v1/responses` | JSON or SSE | One generation per request |
+| `POST /api/chat` | Ollama native `/api/chat` | JSON or NDJSON, streaming by default | One generation per request |
+| `POST /api/generate` | Ollama native `/api/generate` | JSON or NDJSON, streaming by default | One generation per request |
+| `POST /assistant/ask` | Direct OpenAI-compatible calls in `complete()` | Two JSON calls when the order tool runs | Two generations and one tool span |
+| `GET /ws/generate` | Direct Ollama `/api/generate` for each WebSocket message | NDJSON relayed as WebSocket frames | One generation per message |
+
+The two direct paths bypass the four HTTP proxy handlers. Instrumenting only
+`forward()` leaves them dark. Instrumenting only `complete()` leaves the public
+proxy routes dark.
+
+## Expected outcome
+
+- **Coverage.** Every path in the table emits its capture target. A successful
+  stream emits once when it finishes, not once per chunk. A WebSocket with two
+  prompt messages emits two generations.
+- **Grouping.** Each request or WebSocket message gets a new `$ai_trace_id`.
+  Calls and the `lookup_order` span within one `/assistant/ask` share that ID.
+  `$ai_session_id` equals `X-Conversation-Id` across related traces, and the
+  event `distinct_id` equals `X-User-Id`.
+- **Tool flow.** An order question produces `generation → span(lookup_order) →
+  generation`. The span uses `$ai_parent_id` and records the actual tool
+  execution, not the mere presence of a tool schema in the model request.
+- **Provider formats.** OpenAI-compatible chat usage maps `prompt_tokens` and
+  `completion_tokens`. OpenAI Responses JSON and `response.completed` SSE map
+  `input_tokens` and `output_tokens`, with text from `response.output_text.delta`
+  on streams. Ollama's final JSON or NDJSON object maps top-level
+  `prompt_eval_count` and `eval_count`. Missing usage stays unknown, not zero.
+- **Failures.** A provider HTTP error or connection failure produces an errored
+  generation with `$ai_is_error` and a safe `$ai_error`. An interrupted SSE,
+  NDJSON, or WebSocket stream is marked incomplete or errored. No partial stream
+  is reported as a successful completed generation.
+- **Behavior.** HTTP status, body, content type, stream order, and cancellation
+  still reach the caller. The order tool still runs between its two model calls.
+
+Fail the evaluation if any listed inference path is missing, token counts use
+another provider's field names, a stream is captured before its final outcome,
+an error is labeled successful, or the tool span has a separate trace.
+
+## Run locally
+
+Install `requirements.txt`, then point `OPENAI_BASE_URL` and `OLLAMA_BASE_URL`
+at compatible servers. `OPENAI_API_KEY` is optional for local compatible
+servers. The app binds to `127.0.0.1:8000` by default.
+
+```bash
+python -m pip install -r requirements.txt
+python app.py
+```
+
+The contract test uses a fake provider, so it needs no credentials or model:
+
+```bash
+python -m unittest discover -s tests -v
+```
+
+Run the Wizard evaluation from the workbench root:
+
+```bash
+pnpm wizard-ci --app ai-observability/manual-capture/python-multi-route-proxy --evaluate --local
+```

apps/ai-observability/manual-capture/python-multi-route-proxy/app.py

@@ -0,0 +1,233 @@
+"""A chat service that forwards several raw LLM protocols without an SDK."""
+
+import json
+import os
+from collections.abc import Sequence
+from typing import Any
+
+from aiohttp import ClientError, ClientSession, ClientTimeout, WSMsgType, web
+
+from tools import lookup_order
+
+
+CLIENT = web.AppKey("client", ClientSession)
+OPENAI_BASE = web.AppKey("openai_base", str)
+OLLAMA_BASE = web.AppKey("ollama_base", str)
+USER_ID = web.RequestKey("user_id", str)
+CONVERSATION_ID = web.RequestKey("conversation_id", str)
+
+TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "lookup_order",
+            "description": "Look up the current user's latest order.",
+            "parameters": {"type": "object", "properties": {}},
+        },
+    }
+]
+
+
+@web.middleware
+async def resolve_user(request: web.Request, handler):
+    """Treat headers as identity already resolved by this fixture's gateway."""
+    user_id = request.headers.get("X-User-Id")
+    conversation_id = request.headers.get("X-Conversation-Id")
+    if not user_id or not conversation_id:
+        raise web.HTTPUnauthorized(text="user and conversation required")
+    request[USER_ID] = user_id
+    request[CONVERSATION_ID] = conversation_id
+    return await handler(request)
+
+
+def provider_headers(openai: bool) -> dict[str, str]:
+    headers = {"Content-Type": "application/json"}
+    if openai and (api_key := os.environ.get("OPENAI_API_KEY")):
+        headers["Authorization"] = f"Bearer {api_key}"
+    return headers
+
+
+async def forward(
+    request: web.Request, base: str, default_stream: bool, openai: bool
+) -> web.StreamResponse:
+    """Forward one public inference route, including its native stream format."""
+    raw = await request.read()
+    try:
+        payload = json.loads(raw)
+    except (ValueError, UnicodeDecodeError) as exc:
+        raise web.HTTPBadRequest(text="JSON request required") from exc
+    if not isinstance(payload, dict):
+        raise web.HTTPBadRequest(text="JSON object required")
+
+    url = f"{base}{request.path}"
+    try:
+        async with request.app[CLIENT].post(
+            url, data=raw, headers=provider_headers(openai)
+        ) as upstream:
+            content_type = upstream.headers.get("Content-Type", "application/json")
+            response_headers = {"Content-Type": content_type}
+            streamed = payload.get("stream", default_stream) is True
+            if upstream.status >= 400 or not streamed:
+                return web.Response(
+                    status=upstream.status,
+                    body=await upstream.read(),
+                    headers=response_headers,
+                )
+
+            downstream = web.StreamResponse(
+                status=upstream.status, headers=response_headers
+            )
+            await downstream.prepare(request)
+            async for chunk in upstream.content.iter_chunked(4096):
+                await downstream.write(chunk)
+            await downstream.write_eof()
+            return downstream
+    except ClientError as exc:
+        raise web.HTTPBadGateway(text="model provider unavailable") from exc
+
+
+async def openai_chat(request: web.Request) -> web.StreamResponse:
+    return await forward(request, request.app[OPENAI_BASE], False, True)
+
+
+async def openai_responses(request: web.Request) -> web.StreamResponse:
+    return await forward(request, request.app[OPENAI_BASE], False, True)
+
+
+async def ollama_chat(request: web.Request) -> web.StreamResponse:
+    return await forward(request, request.app[OLLAMA_BASE], True, False)
+
+
+async def ollama_generate(request: web.Request) -> web.StreamResponse:
+    return await forward(request, request.app[OLLAMA_BASE], True, False)
+
+
+async def complete(
+    request: web.Request, messages: Sequence[dict[str, Any]]
+) -> dict[str, Any]:
+    """Make the assistant's direct model call, separate from public proxy routes."""
+    payload = {
+        "model": "local-chat",
+        "messages": list(messages),
+        "tools": TOOLS,
+        "stream": False,
+    }
+    try:
+        async with request.app[CLIENT].post(
+            f"{request.app[OPENAI_BASE]}/v1/chat/completions",
+            json=payload,
+            headers=provider_headers(True),
+        ) as upstream:
+            if upstream.status >= 400:
+                raise web.HTTPBadGateway(text=f"model returned {upstream.status}")
+            return await upstream.json()
+    except ClientError as exc:
+        raise web.HTTPBadGateway(text="model provider unavailable") from exc
+
+
+async def assistant_ask(request: web.Request) -> web.Response:
+    """Answer one turn, including the registered order tool when requested."""
+    body = await request.json()
+    question = body.get("question") if isinstance(body, dict) else None
+    if not isinstance(question, str) or not question.strip():
+        raise web.HTTPBadRequest(text="question required")
+
+    messages: list[dict[str, Any]] = [
+        {
+            "role": "system",
+            "content": "Answer briefly. Use lookup_order for order questions.",
+        },
+        {"role": "user", "content": question},
+    ]
+    first = await complete(request, messages)
+    message = first["choices"][0]["message"]
+    tool_calls = message.get("tool_calls", [])
+    if tool_calls:
+        messages.append(message | {"role": "assistant"})
+        for call in tool_calls:
+            if call["function"]["name"] != "lookup_order":
+                raise web.HTTPBadGateway(text="unknown model tool")
+            result = lookup_order(request[USER_ID])
+            messages.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": call["id"],
+                    "content": json.dumps(result),
+                }
+            )
+        final = await complete(request, messages)
+        message = final["choices"][0]["message"]
+
+    return web.json_response({"answer": message.get("content") or ""})
+
+
+async def websocket_generate(request: web.Request) -> web.WebSocketResponse:
+    """Run native Ollama generation directly for each WebSocket message."""
+    socket = web.WebSocketResponse()
+    await socket.prepare(request)
+    async for message in socket:
+        if message.type != WSMsgType.TEXT:
+            continue
+        try:
+            body = json.loads(message.data)
+            prompt = body.get("prompt") if isinstance(body, dict) else None
+        except ValueError:
+            prompt = None
+        if not isinstance(prompt, str) or not prompt.strip():
+            await socket.send_json({"error": "prompt required"})
+            continue
+
+        payload = {"model": "local-ollama", "prompt": prompt, "stream": True}
+        try:
+            async with request.app[CLIENT].post(
+                f"{request.app[OLLAMA_BASE]}/api/generate",
+                json=payload,
+                headers=provider_headers(False),
+            ) as upstream:
+                if upstream.status >= 400:
+                    await socket.send_json(
+                        {"error": "model provider failed", "status": upstream.status}
+                    )
+                    continue
+                async for line in upstream.content:
+                    if line.strip():
+                        await socket.send_str(line.decode().strip())
+        except ClientError:
+            await socket.send_json({"error": "model provider unavailable"})
+    return socket
+
+
+async def start_client(app: web.Application) -> None:
+    app[CLIENT] = ClientSession(timeout=ClientTimeout(total=120))
+
+
+async def close_client(app: web.Application) -> None:
+    await app[CLIENT].close()
+
+
+def create_app(openai_base_url: str, ollama_base_url: str) -> web.Application:
+    """Build the proxy for the two upstream provider bases."""
+    app = web.Application(middlewares=[resolve_user])
+    app[OPENAI_BASE] = openai_base_url.rstrip("/")
+    app[OLLAMA_BASE] = ollama_base_url.rstrip("/")
+    app.on_startup.append(start_client)
+    app.on_cleanup.append(close_client)
+    app.router.add_post("/v1/chat/completions", openai_chat)
+    app.router.add_post("/v1/responses", openai_responses)
+    app.router.add_post("/api/chat", ollama_chat)
+    app.router.add_post("/api/generate", ollama_generate)
+    app.router.add_post("/assistant/ask", assistant_ask)
+    app.router.add_get("/ws/generate", websocket_generate)
+    return app
+
+
+if __name__ == "__main__":
+    web.run_app(
+        create_app(
+            os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:8001"),
+            os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434"),
+        ),
+        host="127.0.0.1",
+        port=int(os.environ.get("PORT", "8000")),
+        handler_cancellation=True,
+    )

apps/ai-observability/manual-capture/python-multi-route-proxy/requirements.txt

@@ -0,0 +1 @@
+aiohttp>=3.14,<4