diff --git a/AGENTS.md b/AGENTS.md index 065113e..41f380b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,7 +17,32 @@ Corollary — **extend, never duplicate.** Before writing anything that complete 3. **Structural over hard-dep where possible.** `/tangle` and `/billing` take the tcloud client as a structural contract (no tcloud dep). Prefer that to a dep when the surface is small. 4. **Substrate-free is a feature.** `/runtime`, `/web`, `/crypto`, `/redact` import nothing — they're pure mechanism behind callback seams. Keep them that way. 5. **Additive subpaths.** New capability = new `./subpath` (entry in `tsup.config.ts` + `exports` in `package.json` + root barrel). Never a breaking change to an existing export. -6. **Reuse the engine primitive; never hand-roll one the engine already gives you.** Before adding ANY streaming / turn / session / durability primitive, grep the engines (`@tangle-network/sandbox`, `agent-runtime`, `agent-eval`) AND existing agent-app exports for one that already does it. A new primitive requires a written "why every existing one doesn't fit" — no exceptions. The recurring smell is the shell reinventing an engine capability and paying latency + drift for it. **Canonical case: streaming a sandbox agent run to a browser is the sandbox SDK's session gateway — `box.mintScopedToken()` (`@tangle-network/sandbox`) mints the short-lived read-only JWT, and the browser connects with `SessionGatewayClient` (`@tangle-network/sandbox/session-gateway`, WebSocket, replay from `lastEventId`) — durable replay + reconnect, browser-direct, lowest latency.** The server-side leg is a *different* method and belongs to the same package: `box.streamPrompt()` is an SSE `AsyncGenerator` a worker consumes, and its `lastEventId` / `executionId` options are forwarded as `Last-Event-ID` / `X-Execution-ID` headers so the worker can reconnect. A browser never calls `streamPrompt`. Do NOT hand-roll a Durable Object or the `/stream` turn-buffer to re-broadcast a sandbox session's turn events; the `/stream` turn-buffer + `/turn-stream` DO are the **sandbox-FREE path ONLY** (a copilot streaming its own runtime, where no sandbox session exists to attach to). Using them for a sandbox product is the smell (a caught instance: a fleet app streaming a sandbox turn through a hand-rolled `TurnStreamDO` instead of the gateway). +6. **Reuse the engine primitive; never hand-roll one the engine already gives you.** Before adding ANY streaming / turn / session / durability primitive, grep the engines (`@tangle-network/sandbox`, `agent-runtime`, `agent-eval`) AND existing agent-app exports for one that already does it. A new primitive requires a written "why every existing one doesn't fit" — no exceptions. The recurring smell is the shell reinventing an engine capability and paying latency + drift for it. **Canonical case: streaming a sandbox agent run to a browser is the sandbox SDK's session gateway — `box.mintScopedToken()` (`@tangle-network/sandbox`) mints the short-lived read-only JWT, and the browser connects with `SessionGatewayClient` (`@tangle-network/sandbox/session-gateway`, WebSocket, replay from `lastEventId`) — durable replay + reconnect, browser-direct, lowest latency.** The server-side leg is a *different* method and belongs to the same package: `box.streamPrompt()` is an SSE `AsyncGenerator` a worker consumes, and its `lastEventId` / `executionId` options are forwarded as `Last-Event-ID` / `X-Execution-ID` headers so the worker can reconnect. A browser never calls `streamPrompt`. Do NOT hand-roll a Durable Object or the `/stream` turn-buffer to re-broadcast a sandbox session's turn events; the `/stream` turn-buffer + `/turn-stream` DO are the **sandbox-FREE path ONLY** (a copilot streaming its own runtime, where no sandbox session exists to attach to). Using them for a sandbox product is the smell (a caught instance: a fleet app streaming a sandbox turn through a hand-rolled `TurnStreamDO` instead of the gateway). The measured lane split behind this rule, and what is and is not yet true in production, is [Live viewing vs history](#live-viewing-vs-history-the-measured-model) below. + +## Live viewing vs history (the measured model) + +Two different jobs, two different mechanisms. Conflating them is what produces a hand-rolled Durable Object. + +**Live viewing is the platform's session gateway. History is durable storage. Nothing else.** + +| Job | Mechanism | Owner | +|---|---|---| +| Watch a turn as it happens; reconnect within minutes | orchestrator session gateway — Redis sorted set per **viewer connection**, WebSocket fanout, replay by `lastEventId` | platform (`agent-dev-container`) | +| Read a turn after the hot buffer expires | the durable assistant row, written **incrementally** as the turn streams (#250) | agent-app `/chat-routes` | +| Stop two drivers running the same turn at once | `createDurableTurnLock` (`/turn-stream`) | agent-app — **the gateway has no equivalent** | + +**Why the hot buffer stays short, and why "just raise `bufferTtlMs`" is never the answer.** The gateway's buffer is a Redis sorted set with a TTL (`bufferTtlMs`, default 5 min — `apps/orchestrator/src/session-gateway/types.ts`). Its memory is `TTL x concurrent viewers x bytes-per-event`, so raising the TTL to serve late viewers buys history at linear memory cost forever. Measured on real Redis 7.4.9 replaying captured production frames: **814 B/event** on a short turn, **27.3 GiB** at 10,000 concurrent watched sandboxes. That number is dominated by answer length, not viewer count — per-event bytes swing **20x** (484 B to 13,710 B) because every part update re-sends the whole accumulated part (**79.7% snapshot redundancy**), which moves the same 10k figure from 27 GiB to **198 GiB**. The buffer is for *live + reconnect*. History comes from the durable row. **Raising the TTL is not a scaling answer — it is the failure mode.** + +**Both platform lanes now fan out — on `develop`, not yet in production.** A sandbox turn arrives over one of two lanes, and only one of them used to reach the gateway: + +| Lane | Called by | Gateway turn events (measured, production) | +|---|---|---| +| message — `POST /agents/sessions/{id}/messages` | `box.session(id).sendMessage()` | **691 delivered** vs 690 sidecar events; text byte-match `true` | +| run/stream — `POST /agents/run/stream` | `box.streamPrompt()`, and therefore `dispatchPrompt({detach:true})` and `driveTurn` | **0** — measured 0/0/0 across three session-id strategies | + +That second row is why autonomous and detached runs were invisible to a browser. `agent-dev-container` [#4114](https://github.com/tangle-network/agent-dev-container/pull/4114) makes the run/stream lane publish to the same session bus (14 agent-event frames where develop delivered 0). It is **merged to `develop` and not yet on `main`** — production `sandbox.tangle.tools` still delivers 0 on the detach lane. Until it ships, a product that must show a detached run live has to drive it over the message lane. + +**The two lanes do not share text semantics.** run/stream sends explicit deltas; the message lane is snapshot-only — every `message.part.updated` re-sends the whole accumulated part. `createSandboxChatProducer` reconciles both (`textDelta`, `src/chat-routes/sandbox-producer.ts`), pinned at 1.000x on eight lane shapes by `tests/chat-routes/lane-portability.test.ts`; remove the reconciliation and a five-update turn inflates to **3.133x**. Incremental persistence makes this load-bearing: the draft row is written *before* the terminal `result` receipt, so the pre-receipt projection has to be correct on its own — the receipt is no longer a safety net. ## Module map @@ -33,7 +58,7 @@ Corollary — **extend, never duplicate.** Before writing anything that complete | `/plans` | browser-safe durable-plan chat projection: lifecycle event parsing, persisted `type:'plan'` codec, revision-aware transcript keys, and monotonic per-revision transitions | structurally byte-matches the durable plan returned by peer `@tangle-network/sandbox`'s `SandboxSession.plan()`; authority remains SDK-owned | | `/durable-chat` | reusable server workflow around the authoritative plan/question channels: authorization-scoped state/command ports, CAS plan decisions, stable follow-up receipts, independently retryable product effects, interaction answer intent/ack/finalize settlement, producer/route adapters, and a reference process-local test store | structural Sandbox plan authority + `/interactions` sidecar route; products supply auth scope, production storage, Sandbox connection, idempotent effects, and follow-up transport | | `/chat-routes` | the assembled server chat vertical (#188 Phase 1): `createChatTurnRoutes()` — body parse/validate → injected `authorize` seam → producer seam → turn-buffer tap wired BY DEFAULT → NDJSON `Response` + replay endpoint + composed `/interactions` answer endpoints + `/chat-store` persistence (user row on send, assistant row with parts/usage on completion); six optional product seams (`turnLock`/`contextGate`/`beforeTurn`/`onRawEvent` are `@experimental` — single-consumer, proven by a reference consumer, kept FLAT for back-compat; `lifecycle`/`heartbeat` are stable) — `turnLock` (single-flight acquire/release), `contextGate` (pre-producer domain-readiness short-circuit), `beforeTurn` (observe/augment producer input), `lifecycle` (deterministic start/complete/error telemetry, idempotent + settled even on a synchronous pre-stream throw), `heartbeat` (keepalive during silent producer waits), `onRawEvent` (raw producer-event tap) — plus `transformFinalText` (pre-persist redaction applied to BOTH the final-text scalar AND every persisted assistant TEXT part, so `/redact` closes the at-rest PII leak, not just the streamed scalar) and run-failure surfacing (`onTurnComplete` receives `failed`/`failureReason` from a terminal `error`/`session.run.failed` event so an errored turn is skipped for billing + rendered as an error row, never billed/marked-complete with empty text); `createSandboxChatProducer` (raw sandbox events → flattened client vocabulary + persisted parts; authoritative rich usage from `result`/`done`; composed visible raw-error text; severed-stream catch that preserves partial content and ends normally with structured `sandbox.stream_failed` so `onTurnComplete({failed:true})` fires; live dangling-tool terminalization; pending-interaction finalization; non-renderable-ask auto-decline + warning notices, with per-turn plan gating through `isRenderableInteraction`); `runDetachedTurn` (autonomous/detached turn → live buffer bridge: taps the SAME turn-event buffer the interactive lane uses so a browser opening a session MID autonomous-run — mission step, queue job, inbound-email review — tails it token-by-token exactly like an interactive turn; idempotent [an already-`complete` turn returns its cached result without re-streaming a second event sequence] AND crash-safe [a `running` turn — a prior attempt died mid-tap — consults `completedResult` to detect a run that finished server-side, else clears the partial buffer via the `resetBuffer` seam before re-streaming so restarted seqs don't corrupt], marks `running` under `scopeId` for `listRunning` re-attach, surfaces the structured `assistantParts` projection (`result.parts`) so the durable row keeps tool/file/plan parts and not just flat text, forwards the producer's `declineInteraction`/`isRenderableInteraction`/`promoteFilePart` seams — an UNATTENDED run MUST wire `declineInteraction` or a non-renderable ask deadlocks the broker, usage/text/parts fallback via `completedResult` when the stream carries no receipt, terminal-error event → `failed` (including a raw sandbox stream throw normalized by the producer; buffer/tap failures still re-throw); the caller supplies only the raw sandbox stream + store + ids — box/prompt/tooling stay in the product); `createUploadRoute` (the `--chat` scaffold's inline-`data:`-or-ephemeral-sandbox upload path; multipart → inline `data:` part ≤700 KiB, else base64 write through a structural sandbox sink + path-ref part — the >1 MiB gateway cap makes the two-step mandatory); the store-backed attachment vertical (#224) — `resolveChatAttachments` (validates a turn's `attachments` field, re-deriving size from the STORED body, never the wire-reported one), `buildDispatchParts` (assembles attachments + mentions into dispatched `PromptInputPart[]`, inlining under the `DISPATCH_*` byte/parts budget or demoting to a path part), `promoteAgentFilePart` (a harness-emitted file part → a durable store-backed attachment, wired opt-in through `createSandboxChatProducer`'s `promoteFilePart` seam) — all storage-parameterized via `ReadAttachmentFn`/`WriteAttachmentFn` (`./attachment-store`): REQUIRED injection, no default store; `createAttachmentUploadRoute` (#234): hardened durable-store upload factory shared by fleet apps in place of a hand-rolled vault route — two-phase atomic batch, content-sniffed type/kind gate (`sniffBinary`/`checkAttachmentType`, shared client+server), per-kind + aggregate size caps, sanitized filenames; storage via the injected `WriteAttachmentFn`, auth/rate-limit/scope via the injected `authorize`, returns the full `ChatAttachmentInput[]`; import-free `./wire` contract (`ChatTurnRequestPayload`, inline-part byte gate) re-exported via `/web-react`'s chat-stream; INCREMENTAL assistant persistence (`./draft-persistence`) — the assistant row is inserted early and coalesce-patched as the turn streams (time-floored default 2 s + dirty gate + single-flight + best-effort, tuned via `incrementalPersistence`), so a viewer arriving mid-run is served from DURABLE storage instead of the streaming gateway's hot event buffer; that is what lets the buffer stay a short reconnect window (its Redis cost is linear in `ttl x concurrent sessions`, so raising the TTL to cover late viewers is a category error). Idempotency reuses the turn's OWN identity — the row id derives from `deriveExecutionId` (interactive) / `turnId` (detached), so a re-entered turn patches the row a crashed attempt started instead of duplicating parts; the final write is insert-or-patch under that same id and produces the row today's single write produces. Mid-stream parts come from `/stream`'s `draftAssistantParts` (finalize MINUS the dangling-tool terminalizer — a running tool must never persist as a failure). Enabled by default when the store has `updateMessage`; `false` opts out; a store without it is byte-unchanged. `runDetachedTurn`'s `persist` seam transfers row OWNERSHIP to the shell (returns `messageId`; the caller stops inserting) | peer `@tangle-network/agent-runtime` (`handleChatTurn` IS the turn engine — zero loop logic here; subpath-only, not in the root barrel); composes `/stream`, `/chat-store`, `/interactions`, `/web`. Reference assembly: `examples/chat-app.md` | -| `/turn-stream` | shared durable turn replay/broadcast/lock on a Cloudflare DO (#221): pure segment/lock core, `TurnStreamDO` transport shell (thread + workspace channels, WS hibernation fanout, `sync`/`afterSeq` reconnect replay, durable turn-event rows + running-turn index, product extension seams `handleProductRequest`/`shouldDeferLockRelease`/`productSyncEvents`), and the vertical seam adapters — `createDurableObjectTurnEventStore` (the real `turnStore`), `createDurableTurnLock` (dual-scope single-flight `turnLock`, one reconcile-then-retry pass via `reconcileStaleDurableTurnLock`), `createTurnStreamUpgradeHandler`, broadcast helpers, memory harness. Server-only, subpath-only; Cloudflare is structural (bind by re-exporting the class) | composes `/stream`'s `TurnEventStore` contract + `/chat-routes`' `reconcileStaleTurnLock` policy — no new peer | +| `/turn-stream` | shared durable turn replay/broadcast/lock on a Cloudflare DO (#221): pure segment/lock core, `TurnStreamDO` transport shell (thread + workspace channels, WS hibernation fanout, `sync`/`afterSeq` reconnect replay, durable turn-event rows + running-turn index, product extension seams `handleProductRequest`/`shouldDeferLockRelease`/`productSyncEvents`), and the vertical seam adapters — `createDurableObjectTurnEventStore` (the real `turnStore`), `createDurableTurnLock` (dual-scope single-flight `turnLock`, one reconcile-then-retry pass via `reconcileStaleDurableTurnLock`), `createTurnStreamUpgradeHandler`, broadcast helpers, memory harness. Server-only, subpath-only; Cloudflare is structural (bind by re-exporting the class). **Split by [Live viewing vs history](#live-viewing-vs-history-the-measured-model): the broadcast/replay half is superseded by the session gateway for any SANDBOX turn and must not be used for one — but the module is NOT retired, because `createDurableTurnLock` is single-flight mutual exclusion the gateway does not provide** (its only `lock` is `SubscriptionMutex`, an in-process per-instance guard in `sse-bridge.ts` that goes silent across instances — the same defect as the platform's in-process workflow-run emitter). Sole consumer today is gtm-agent's `SessionStreamDO`, which uses the base for the turn lock and product seams (Vault mutation/head endpoints, interaction rendezvous, persistence queue) as well as broadcast; only the last of those has a replacement | composes `/stream`'s `TurnEventStore` contract + `/chat-routes`' `reconcileStaleTurnLock` policy — no new peer | | `/tangle` | app-registration consent URL + cached broker-token provider | structural `TangleAppsClient` (from agent-integrations) | | `/billing` | per-workspace budget-capped key manager (mint/rotate/rollover/usage) | structural tcloud provisioner + store + crypto seams | | `/preflight` | deploy-time secret-liveness probes: `runPreflight(probes)` → per-probe verdict + latency + overall pass/fail (any critical fail → fail); standard builders `routerChatProbe`/`sandboxAuthProbe`/`httpHeadProbe` (explicit config, read nothing global); `formatPreflightReport` table + the `agent-app-preflight` bin reading `preflight.config.mjs`. Binds at DEPLOY time (the one place with real secrets — CI can't hold them); failures name the exact secret to rotate | — (server-only; product declares probes from `process.env`) | @@ -41,7 +66,8 @@ Corollary — **extend, never duplicate.** Before writing anything that complete | `/trace` | flow observability: FlowSpan/FlowTrace + ASCII waterfall/histogram renderers; mission trace bridge (`createMissionTraceContext`/`childSpanContext`/`traceEnv` — 32-hex/16-hex ids + the `TRACE_ID`/`PARENT_SPAN_ID` env pair agent-runtime's `readTraceContextFromEnv` inherits); delegation→FlowSpan converters (`delegationActivityToFlowSpans`, `loopTraceEventsToFlowSpans` over a structural `LoopTraceEventLike`, `composeMissionFlowTrace`, `stepActivityFlowTrace`) | — (pure data; id formats byte-match agent-runtime's OTLP export, no import) | | `/web-react` | shared chat-shell + observability components: ModelPicker/EffortPicker/ChatMessages/RunDrillIn, `MissionActivityLane` (per-step delegated-run sub-rows → web waterfall), `AgentActivityPanel` (cross-context delegation surface over a `fetchActivity(cursor)` data port, missionRef link slot), `FlowWaterfall` + pure `waterfallLayout`/`mergeActivityPages` helpers, `InteractionQuestionCard`/`InteractionPlanCard` + `useChatInteractions`/`createInteractionAnswerSubmitter` (the client half of `/interactions`), `useComposerAttachments` (#234: staged-upload composer hook — pending→uploading→ready/error lifecycle, client-side pre-validation mirroring the server's sniff/type/size gate so a rejection never hits the network, feeds `ChatComposer`'s `pendingFiles` and the wire `attachments` field with the server's pass-through `ChatAttachmentInput[]`), `MessageAttachments` (#234: transcript thumbnails/chips over an injected raw-bytes `resolveFileUrl`, fetch-on-mount for images and fetch-on-click for file chips; also wired through `ChatMessages`' optional `resolveAttachmentUrl` prop) | react peer; renders `/missions` lanes via `/trace` converters and `/interactions` asks via its contract | | `/skills` `/skills-placement` | adoptable skill pipeline (#216): the ONE `SKILL.md` frontmatter parser (`parseSkillFrontmatter` — fail-loud on malformed blocks; kills the ≥5 fleet copies), `skillEntryFromMarkdown`/`parseCorpusSkills`, and BOTH delivery modes as a product choice — `inline` (full bodies → prompt section via `renderInlineSkills`) or `mounted` (`skillRefs` → typed `resources.skills` the platform places per harness + `renderSkillIndex` naming the REAL dir); `composeSkills` mode switch + `assertSkillDeliveryDisjoint` (a skill must never ship both ways). `/skills-placement` is the only materialize-bound code: `resolveSkillDir`/`unsupportedSkillHarnesses`/`composeSkillsForHarness` (auto-inline fallback for harnesses with no cwd skill dir — hermes, amp, …) so neither products nor other agent-app modules ever write a skill path literal. Legacy `skillMountPath` is `@deprecated` (claude-code-only literal). | `/skills` substrate-free (type-only over sandbox shapes); `/skills-placement` composes optional peer `@tangle-network/agent-profile-materialize` (`skillDirForHarness` — the platform's authoritative map) | -| `/crypto` `/web` `/redact` `/stream` | AES-GCM field crypto · web boundary utils (body/context/rate-limit/headers) · PII redaction · SSE normalization + turn identity | — | +| `/crypto` `/web` `/redact` | AES-GCM field crypto · web boundary utils (body/context/rate-limit/headers) · PII redaction | — | +| `/stream` | SSE normalization + turn identity + the assistant-part projection: `finalizeAssistantParts` (terminal) and `draftAssistantParts` (mid-turn — finalize MINUS `terminalizeDanglingToolParts`, so an in-flight tool is not falsely settled into a durable row), `resolveChatTurn`/`messageHasTurnId`/`normalizeClientTurnId` turn identity, and the `TurnEventStore`/`createBufferedTurnTap`/`replayTurnEvents` buffer. That **turn buffer is the SANDBOX-FREE lane only** (a copilot streaming its own runtime); a sandbox turn is watched through the platform session gateway — see [Live viewing vs history](#live-viewing-vs-history-the-measured-model). `draftAssistantParts` is what lets durable history serve late viewers, which is what keeps the gateway's hot buffer short enough to scale | — | | `/theme` `/styles` `/tailwind-preset` | single design-token source for every React surface: `tokens.css` (`:root` + `[data-theme="dark"]`/`.dark`, shadcn channel triples with canvas/sequences `--bg-input`/`--text-primary`/`--border-default`/… aliases resolving to them), typed `AgentAppTheme` + `lightTheme`/`darkTheme`/`themeToCssVars`/`themeColor`, and a Tailwind preset mapping shadcn names (`bg-card`, `text-muted-foreground`…) to the vars. Consumers: `import '@tangle-network/agent-app/styles'` + add the preset. Enforced by `tests/theme/tokens-contract.test.ts` — every `var(--…)` a component references must be defined here, else it ships transparent. | — (pure CSS/data; no peer) | | `/theme-contract` | node-only token-completeness checker consumers run in CI over THEIR OWN source (`checkThemeContract` + the `agent-app-theme-check` bin): a full `var(--…)` reference scan plus a check of the known-dangerous preset utilities (`surface-container*`/`card`/`popover`) against the shipped `tokens.css` + any extra app CSS — catches the invisible-popover/transparent-dropdown class. Split from `/theme` (which stays browser-clean) because it reads `fs`. Single source of truth for the token walk in `tokens-contract.test.ts`. | — (pure `fs` mechanism; no peer) | diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bb299a2..306ed4c 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -93,7 +93,9 @@ consumer of L0/L1 installs none of them): `konva`/`react-konva` → only | `@`-file-mentions end to end | `chat-routes` — `createSandboxFileIndexRoute` (listing, answers `warming` for a cold box OR an unmaterialised root) · `parseFileMentions` (path/charset/count validation) · `fileMentionsToParts`/`buildMentionPromptBlock` (dispatch); `chat-store` — `ChatMentionPart` (persisted vocabulary); `web-react` — `useFileMentions` (picker) · `segmentMentionContent` (transcript pills) | | Store-backed file attachments end to end (upload → dispatch → transcript) | `chat-routes` — `resolveChatAttachments` (validate + re-derive size via `ReadAttachmentFn`) · `buildDispatchParts` (attachments + mentions → dispatched `PromptInputPart[]`, inline-vs-path-demote under the `DISPATCH_*` budget) · `promoteAgentFilePart` (harness-emitted file → store via `WriteAttachmentFn`); `chat-store` — `ChatAttachmentPart` (persisted vocabulary, reuses the `file`/`image` discriminant); `web-react` — `chat-attachments` (read-side re-exports for transcript rendering). Storage is REQUIRED injection (`ReadAttachmentFn`/`WriteAttachmentFn` in `./attachment-store`) — no default store | | Recovering a turn lock whose holder died | `chat-routes` — `reconcileStaleTurnLock`, a policy over injected sandbox/session probes (no SDK); the probes stay in the product | -| Durable turn replay + live-viewer fanout + the single-flight turn lock (production) | `turn-stream` — re-export `TurnStreamDO` from the worker entry, wire `createDurableObjectTurnEventStore` into `turnStore`, `createDurableTurnLock` into `turnLock`, `createTurnStreamUpgradeHandler` before the router | +| Live-viewer fanout for a **sandbox** turn | NOT agent-app. The platform session gateway — `box.mintScopedToken()` + `SessionGatewayClient` (`@tangle-network/sandbox/session-gateway`), browser-direct, replay from `lastEventId`. See [AGENTS.md § Live viewing vs history](./AGENTS.md#live-viewing-vs-history-the-measured-model) | +| Late viewer, past the gateway's hot-buffer TTL | `chat-routes` — `incrementalPersistence` on `createChatTurnRoutes` (`persist` on `runDetachedTurn`), projecting through `stream`'s `draftAssistantParts`. Serve history from the durable row; raising `bufferTtlMs` is not a scaling answer | +| The single-flight turn lock, and durable replay on the **sandbox-free** lane (production) | `turn-stream` — re-export `TurnStreamDO` from the worker entry, wire `createDurableObjectTurnEventStore` into `turnStore`, `createDurableTurnLock` into `turnLock`, `createTurnStreamUpgradeHandler` before the router. The lock has no gateway equivalent; the broadcast half does | | Flow traces / waterfalls | `trace` | | Chat UI + run/observability components | `web-react` | | Agent asks a human mid-run (question/plan cards, answer route) | `interactions` (server + contract) + `web-react` (cards/hook) | diff --git a/tests/chat-routes/lane-portability.test.ts b/tests/chat-routes/lane-portability.test.ts new file mode 100644 index 0000000..9802235 --- /dev/null +++ b/tests/chat-routes/lane-portability.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, it } from 'vitest' +import { createSandboxChatProducer } from '../../src/chat-routes/sandbox-producer' + +/** + * Lane portability of `createSandboxChatProducer`. + * + * A sandbox turn reaches the shell over one of two platform lanes, and they do + * NOT share text semantics: + * + * - run/stream (`POST /agents/run/stream`) carries an explicit `delta`. + * - the message lane (`POST /agents/sessions/{id}/messages`) is SNAPSHOT-only: + * every `message.part.updated` re-sends the whole accumulated part text. + * + * Appending a snapshot as if it were a delta inflates the transcript + * quadratically (measured 3.133x at five updates when the reconciliation in + * `textDelta` is removed), and the inflation reaches the DURABLE row whenever + * the turn is persisted before the terminal `result` receipt arrives — which + * is exactly what incremental draft persistence (#250) does. + * + * These probes pin BOTH the streamed bytes and the persisted projection to + * 1.000x across every lane shape, so the producer stays safe to point at + * either lane. They fail loudly if the reconciliation regresses. + */ +const TRUE_TEXT = 'alpha bravo charlie delta echo' +const CHUNKS = ['alpha ', 'bravo ', 'charlie ', 'delta ', 'echo'] + +/** Message-lane shape: each event re-sends the whole accumulated text. */ +function snapshots(partId?: string): unknown[] { + const out: unknown[] = [] + let acc = '' + for (const chunk of CHUNKS) { + acc += chunk + out.push({ + type: 'message.part.updated', + data: { part: { ...(partId ? { id: partId } : {}), type: 'text', text: acc } }, + }) + } + return out +} + +/** run/stream shape: each event carries only the new suffix. */ +function deltas(partId?: string): unknown[] { + return CHUNKS.map((chunk) => ({ + type: 'message.part.updated', + data: { part: { ...(partId ? { id: partId } : {}), type: 'text', text: chunk }, delta: chunk }, + })) +} + +async function run(events: unknown[]): Promise<{ streamed: string; persisted: string }> { + async function* generate(): AsyncGenerator { + for (const event of events) yield event + } + const producer = createSandboxChatProducer({ events: generate() }) + let streamed = '' + for await (const event of producer.stream) { + const wire = event as { type?: string; text?: string } + if (wire.type === 'text' && typeof wire.text === 'string') streamed += wire.text + } + const parts = (producer.assistantParts?.() ?? []) as Array<{ type?: string; text?: string }> + const persisted = parts + .filter((part) => part.type === 'text') + .map((part) => part.text ?? '') + .join('') + return { streamed, persisted } +} + +const RESULT = { type: 'result', data: { finalText: TRUE_TEXT } } + +describe('createSandboxChatProducer is lane-portable', () => { + const cases: Array<[string, unknown[]]> = [ + ['run/stream: deltas, stable part id', deltas('prt_1')], + ['run/stream: deltas, no part id', deltas()], + ['message lane: snapshots, stable part id', snapshots('prt_1')], + ['message lane: snapshots, no part id', snapshots()], + ['message lane: snapshots + terminal result receipt', [...snapshots('prt_1'), RESULT]], + [ + 'message lane: snapshots + whole-message echo + receipt', + [ + ...snapshots('prt_1'), + { + type: 'message.updated', + data: { message: { parts: [{ id: 'prt_1', type: 'text', text: TRUE_TEXT }] } }, + }, + RESULT, + ], + ], + [ + 'message lane: transcript split across two part ids', + [ + { type: 'message.part.updated', data: { part: { id: 'p1', type: 'text', text: TRUE_TEXT.slice(0, 12) } } }, + { type: 'message.part.updated', data: { part: { id: 'p2', type: 'text', text: TRUE_TEXT.slice(12) } } }, + RESULT, + ], + ], + ] + + for (const [name, events] of cases) { + it(`${name} -> 1.000x streamed and persisted`, async () => { + const { streamed, persisted } = await run(events) + expect(streamed).toBe(TRUE_TEXT) + expect(persisted).toBe(TRUE_TEXT) + }) + } + + it('a mid-stream draft is already correct, without waiting for the receipt', async () => { + // The receipt rescues the persisted row at finalize; incremental + // persistence writes BEFORE it, so the pre-receipt projection must + // already be right on its own. + const withReceipt = await run([...snapshots('prt_1'), RESULT]) + const withoutReceipt = await run(snapshots('prt_1')) + expect(withoutReceipt.persisted).toBe(withReceipt.persisted) + expect(withoutReceipt.persisted).toBe(TRUE_TEXT) + }) +})