tito: per-turn token record persistence + abridge composition (production capture)

plugins/abridge/README.md

@@ -132,17 +132,22 @@ Claude-speaking agent in front of an OpenAI-shaped recording gateway.
 ### Record the tunnel traffic
 
 `Recorder` wraps any handler client and appends one JSONL line per served
-call — `{ts, path, request, response}` — flushed as it goes, so the file is
-complete up to the last call even if the host dies mid-rollout. It exposes
-the wrapped client's routes and closes it on teardown, so it drops in
-transparently:
+call — `{ts, path, request_id, session_id?, request, response}` — flushed
+as it goes, so the file is complete up to the last call even if the host
+dies mid-rollout. It exposes the wrapped client's routes and closes it on
+teardown, so it drops in transparently:
 
 ```python
 from agentix.bridge import Proxy, Recorder
 
-proxy = Proxy(Recorder(client, "runs/rollout-42.jsonl"))
+proxy = Proxy(Recorder(client, "runs/rollout-42.jsonl", session_id="rollout-42"))
 ```
 
+The `request_id` in each row is the same id the transport stamps as
+`x-request-id` on the upstream hop (bound through a context var), so a
+message-level row joins a downstream token recorder's per-turn record;
+`session_id`, when given, tags every row with the rollout identity.
+
 ## Writing your own handler
 
 Any class with `@on(path)`-decorated methods works. No base class to
@@ -258,6 +263,43 @@ code, so when you expose it to them (`--host`), also set
 `build_session_app`) so only keys your harness minted are served;
 everything else gets a 401.
 
+More serve options:
+
+* `--tito-url http://tito:30000` (mutually exclusive with
+  `--upstream-base-url`) — put the Anthropic shell in front of a
+  token-recording session gateway instead of a plain engine. The gateway
+  keeps one append-only linear conversation per session, while real
+  Anthropic agents multiplex several conversations over one key (helper
+  calls, subagents, reruns), so each caller session **demuxes by
+  conversation**: requests are keyed by the canonicalized
+  `(system, first user message)` pair, and each distinct key gets its own
+  `AnthropicToOpenAI(SessionForward(tito_url).handler())` — its own
+  gateway session. Known boundaries (by design, fail loudly rather than
+  silently): a mid-conversation history rewrite that keeps the opening
+  (deep compaction) still lands in the same gateway session and rides the
+  gateway's rollback / from-scratch paths (rewrites past its rollback
+  window are its documented 400); two genuinely different conversations
+  with a byte-identical opening collide into one session; and a caller
+  key evicted by the serve LRU (`--max-sessions`) mid-rollout continues
+  in fresh gateway sessions — the gateway-side capture splits there
+  (`turn_index` restarts), so size `--max-sessions` above your concurrent
+  rollout-key count.
+* `--tito-delete-on-evict` — reap a caller session's gateway sessions
+  when it closes (LRU eviction or shutdown). Default off: gateway
+  sessions stay alive for harvest and the harvester deletes them.
+* `--record-dir DIR` (either mode) — wrap each session's client in a
+  `Recorder` writing `DIR/<session_id>.jsonl`; rows carry `session_id`,
+  `request_id`, and (in tito mode) `gateway_session_id` — the gateway's
+  own session id, i.e. the `session_id` in its token records. The same
+  `request_id` reaches the upstream as `x-request-id`, so message rows
+  join the gateway's token records per call as well as per session.
+
+`GET /_health` reports `translation_spec_sha` — one SHA-256 over the
+source of the Anthropic↔OpenAI transform module and both client modules
+that shape the upstream body (assistant replay, forced non-streaming,
+model override, operator params) — so downstream data contracts can pin
+the exact translation their captured trajectories were produced under.
+
 Programmatic surface in `agentix.bridge.serve`: `build_app(*clients)`
 (shared session) and `build_session_app(factory)` (one client per
 caller key; LRU-bounded with in-flight-safe eviction — an evicted

plugins/abridge/agentix/bridge/_request_id.py

@@ -0,0 +1,54 @@
+"""Per-call request-id propagation between capture and transport layers.
+
+The tunnel deliberately carries no HTTP metadata (see `proxy.Request`), so a
+request id can't ride the `Request` object. But capture and transport must
+agree on ONE id per call: the `Recorder` writes a `request_id` into its JSONL
+row, and the transport layer (`Forward`, the SDK clients) stamps
+`x-request-id` on the upstream hop — downstream token recorders (the TITO
+gateway) echo that header into their own per-turn records. If each layer
+minted its own id, the message-level row and the token-level record for the
+same call could never be joined.
+
+A `ContextVar` is the seam: the outermost interested layer (the `Recorder`,
+when present) mints the id and binds it for the duration of the handler call;
+inner layers reuse a bound id and only mint their own when nothing upstream
+bound one. Works unchanged across `await` within one handler invocation and
+never leaks across concurrent calls.
+
+`current_upstream_session_id` flows the OTHER way on the same principle: the
+transport layer (`Forward`) publishes the session id it stamped upstream as
+`x-session-id` — for a `SessionForward` that is the gateway-assigned session
+id, which the caller-side capture cannot otherwise know (it exists only
+after the lazy session create). The `Recorder` clears it before each handler
+call and reads it afterwards into the row's `gateway_session_id`, restoring
+the session-level join between caller-side rows and gateway-side records.
+"""
+
+from __future__ import annotations
+
+import uuid
+from contextvars import ContextVar
+
+current_request_id: ContextVar[str | None] = ContextVar("abridge_request_id", default=None)
+
+current_upstream_session_id: ContextVar[str | None] = ContextVar(
+    "abridge_upstream_session_id", default=None
+)
+
+
+def mint_request_id() -> str:
+    return uuid.uuid4().hex
+
+
+def get_or_mint_request_id() -> str:
+    """The id bound by an outer capture layer, or a fresh one."""
+    bound = current_request_id.get()
+    return bound if bound else mint_request_id()
+
+
+__all__ = [
+    "current_request_id",
+    "current_upstream_session_id",
+    "get_or_mint_request_id",
+    "mint_request_id",
+]

plugins/abridge/agentix/bridge/clients/_anthropic_transforms.py

@@ -1,15 +1,24 @@
 """Pure Anthropic ↔ OpenAI shape converters.
 
-Used only by `clients.anthropic_from_openai`. The functions here are
-JSON-in, JSON-out — no I/O, no SDK calls, no spans. Anyone writing a
-custom Anthropic-on-OpenAI client can import these directly.
+Used by `clients.anthropic_from_openai` and `clients.anthropic_to_openai`.
+The functions here are JSON-in, JSON-out — no I/O, no SDK calls, no spans.
+Anyone writing a custom Anthropic-on-OpenAI client can import these directly.
+
+This module IS the translation contract for downstream consumers: whatever
+the agent "actually said" to a recording backend is defined by these
+functions, so a change here changes the byte identity of captured
+trajectories. `TRANSLATION_SPEC_SHA` (bottom of the module) hashes this
+file's source so consumers can pin the exact translation their data was
+produced under; abridge-serve surfaces it on `/_health`.
 """
 
 from __future__ import annotations
 
 import dataclasses
+import hashlib
 import json
 import uuid
+from pathlib import Path
 from typing import Any
 
 
@@ -55,6 +64,10 @@ def anthropic_messages_to_openai(
     if tools:
         out["tools"] = tools
 
+    tool_choice = _tool_choice_anthropic_to_openai(body.get("tool_choice"))
+    if tool_choice is not None:
+        out["tool_choice"] = tool_choice
+
     if extra_body:
         out.update(extra_body)
 
@@ -275,6 +288,7 @@ def _messages_anthropic_to_openai(messages: list[Any]) -> list[dict[str, Any]]:
             continue
 
         text_parts: list[str] = []
+        thinking_parts: list[str] = []
         tool_calls: list[dict[str, Any]] = []
         tool_results: list[dict[str, Any]] = []
         for block in content:
@@ -286,6 +300,13 @@ def _messages_anthropic_to_openai(messages: list[Any]) -> list[dict[str, Any]]:
             block_type = block.get("type")
             if block_type == "text":
                 text_parts.append(str(block.get("text", "")))
+            elif block_type == "thinking":
+                # Assistant thinking history maps to the reasoning_content
+                # key the vLLM/sglang dialect reads (Anthropic's crypto
+                # `signature` has no OpenAI-side meaning and is dropped).
+                # Keeping the reasoning in the echoed history preserves byte
+                # identity with what a session-recording backend stored.
+                thinking_parts.append(str(block.get("thinking", "")))
             elif block_type == "tool_use":
                 tool_calls.append(
                     {
@@ -309,9 +330,15 @@ def _messages_anthropic_to_openai(messages: list[Any]) -> list[dict[str, Any]]:
         text = "\n".join(part for part in text_parts if part)
         if role == "assistant":
             message: dict[str, Any] = {"role": "assistant", "content": text or None}
+            if thinking_parts:
+                message["reasoning_content"] = "\n".join(part for part in thinking_parts if part)
             if tool_calls:
                 message["tool_calls"] = tool_calls
-            if message["content"] is not None or tool_calls:
+            # A thinking-only assistant turn (legal Anthropic shape — e.g.
+            # extended thinking cut off at max_tokens) maps to reasoning_content
+            # and must survive: dropping the whole message would leave two
+            # adjacent user turns and silently lose the forwarded reasoning.
+            if message["content"] is not None or tool_calls or thinking_parts:
                 out.append(message)
         else:
             out.extend(tool_results)
@@ -320,6 +347,29 @@ def _messages_anthropic_to_openai(messages: list[Any]) -> list[dict[str, Any]]:
     return out
 
 
+def _tool_choice_anthropic_to_openai(tool_choice: Any) -> Any:
+    """Map Anthropic `tool_choice` to the OpenAI field.
+
+    `auto` -> "auto", `any` -> "required", `none` -> "none", and
+    `{type: tool, name}` -> a named function choice.
+    `disable_parallel_tool_use` is intentionally not mapped (matching the
+    reference translator): OpenAI's `parallel_tool_calls` is not honored by
+    every OpenAI-compatible engine, and a silently ignored knob is worse
+    than a documented drop. Unknown shapes are dropped, not guessed."""
+    if not isinstance(tool_choice, dict):
+        return None
+    kind = tool_choice.get("type")
+    if kind == "auto":
+        return "auto"
+    if kind == "any":
+        return "required"
+    if kind == "none":
+        return "none"
+    if kind == "tool" and tool_choice.get("name"):
+        return {"type": "function", "function": {"name": str(tool_choice["name"])}}
+    return None
+
+
 def _tools_anthropic_to_openai(tools: Any) -> list[dict[str, Any]]:
     if not isinstance(tools, list):
         return []
@@ -359,7 +409,40 @@ def _sse(event: str, data: dict[str, Any]) -> bytes:
     return f"event: {event}\ndata: {payload}\n\n".encode()
 
 
+# The translation contract version: SHA-256 over the source of EVERY module
+# that shapes what an upstream/recording backend receives — the pure
+# transforms here, plus the two client modules that rewrite the body around
+# them (assistant-replay memory, forced stream=False, model override,
+# operator upstream_params). Hashing only this file would let those rewrites
+# drift without moving the pin. Any edit — however small — changes the byte
+# identity of the OpenAI bodies a recording backend sees, so downstream data
+# contracts pin this value (abridge-serve reports it on `/_health`).
+# Deliberately file bytes, not a semantic hash: comments changing the sha is
+# a false positive we accept; a behavior change slipping through unhashed is
+# not. Sibling sources are read directly (no import) to avoid a cycle with
+# the client modules, which import this one.
+_TRANSLATION_SPEC_FILES = (
+    "_anthropic_transforms.py",
+    "anthropic_to_openai.py",
+    "anthropic_from_openai.py",
+)
+
+
+def _translation_spec_sha() -> str:
+    digest = hashlib.sha256()
+    for name in _TRANSLATION_SPEC_FILES:
+        digest.update(name.encode())
+        digest.update(b"\x00")
+        digest.update((Path(__file__).parent / name).read_bytes())
+        digest.update(b"\x00")
+    return digest.hexdigest()
+
+
+TRANSLATION_SPEC_SHA: str = _translation_spec_sha()
+
+
 __all__ = [
+    "TRANSLATION_SPEC_SHA",
     "AnthropicCountTokens",
     "anthropic_messages_to_openai",
     "anthropic_sse",

plugins/abridge/agentix/bridge/clients/anthropic_from_openai.py

@@ -19,6 +19,7 @@
 
 from agentix.utils import trace
 
+from .._request_id import get_or_mint_request_id
 from ..proxy import (
     AbridgeError,
     ClientResponse,
@@ -104,7 +105,9 @@ async def messages(self, request: Request) -> ClientResponse:
         openai_body = anthropic_messages_to_openai(request.body, upstream_model=self._model)
         openai_body.update(self._upstream_params)
         openai_body["stream"] = False
-        record_id = uuid.uuid4().hex
+        # Reuses the id a wrapping capture layer (Recorder) bound for this
+        # call, so its JSONL row and the upstream header share one id.
+        record_id = get_or_mint_request_id()
         extra_headers = {
             "x-session-id": self.session_id,
             "x-request-id": record_id,

plugins/abridge/agentix/bridge/forward.py

@@ -35,6 +35,7 @@
 
 import httpx
 
+from ._request_id import current_upstream_session_id, get_or_mint_request_id
 from .proxy import AbridgeError, ClientResponse, Handler, Request
 
 logger = logging.getLogger(__name__)
@@ -132,7 +133,14 @@ def handler(self, path: str | None = None) -> Handler:
         return _OwnedHandler(routes[path], self)
 
     async def _forward(self, path: str, request: Request) -> ClientResponse:
-        record_id = uuid.uuid4().hex
+        # Reuses the id a wrapping capture layer (Recorder) bound for this
+        # call, so its JSONL row and the sidecar's token record share one id.
+        record_id = get_or_mint_request_id()
+        # Publish the upstream session identity for the capture layer: for a
+        # SessionForward this is the gateway-assigned session id (known only
+        # here, after the lazy create) — the Recorder reads it back into the
+        # row's `gateway_session_id` join key.
+        current_upstream_session_id.set(self.session_id)
         headers = {
             **self._headers,
             "x-session-id": self.session_id,