From 1322af15b02b2f75a57318a5933cf72238fb88c2 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Fri, 17 Jul 2026 11:39:53 -0600 Subject: [PATCH] fix(chat-routes): redact persisted parts, surface run-failure, close manifest hole (audit P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - transformFinalText now applies to every persisted assistant TEXT part, not just the scalar finalText — closes a PII-redaction-at-rest leak where message.parts kept raw PII the /redact transform had scrubbed from the scalar. - onTurnComplete receives failed/failureReason derived from a terminal error/session.run.failed event, so products skip billing an errored turn and render an error row instead of marking an empty turn complete. - browser-safe manifest is enumerated from package.json exports (react family + /lazy splits + pure-mechanism subpaths) instead of a top-level *-react readdir, covering web-react/terminal, design-canvas-react/engine, and every *-react/lazy — the #256/#372 client-bundle-crash blind spot. - mergePersistedPart tool branch overlays only DEFINED incoming fields and never downgrades a settled tool to running, so a later empty update no longer clobbers captured tool output/error. - lifecycle terminal hook is hoisted + idempotent and fires onTurnError on a synchronous pre-stream throw; withStreamHeartbeat clears its timer in finally so a rejected source never orphans a setTimeout. - vertical suite gains a turnLock + lifecycle + contextGate composition scenario. --- AGENTS.md | 2 +- ARCHITECTURE.md | 1 + src/chat-routes/turn-routes.ts | 160 +++++++++++++----- src/stream/stream-normalizer.ts | 34 +++- tests/browser-safe-subpaths.test.ts | 138 ++++++++++----- .../turn-routes-sync-throw.test.ts | 73 ++++++++ tests/chat-routes/turn-routes.test.ts | 55 ++++++ tests/stream-normalizer.test.ts | 32 ++++ tests/vertical/seams-composition.test.ts | 145 ++++++++++++++++ 9 files changed, 544 insertions(+), 96 deletions(-) create mode 100644 tests/chat-routes/turn-routes-sync-throw.test.ts create mode 100644 tests/vertical/seams-composition.test.ts diff --git a/AGENTS.md b/AGENTS.md index 8c4e8b7..373d974 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,7 +29,7 @@ Corollary — **extend, never duplicate.** Before writing anything that complete | `/eval` | `producedFromToolEvents` (side-channel→`RuntimeEventLike` bridge) + `createTokenRecallChecker` | **re-exports** agent-eval's `verifyCompletion`/`extractProducedState`/`weightedComposite`/`createLlmCorrectnessChecker` | | `/integrations` | hub `/exec` client + `resolveIntegrationAction` + `invokeIntegrationHub` (wiring) | peer-dep `@tangle-network/agent-integrations` (the engine/catalog) | | `/interactions` | human-in-the-loop ask channel, both halves: the shared wire/persisted-part contract (`ChatInteraction`, part codecs, composer-as-answer routing, content-signature dedupe) + the server side — structural sidecar `/agents/sessions/{id}/interactions` client and `createInteractionAnswerRoute()` (list/answer endpoint factory: body validation, gone→410 mapping, duplicate-answer safety net, unblock verification, `resolveConnection` product seam) | peer `@tangle-network/agent-interface` (schema types only); connection is structural — no sandbox-SDK import. Server half must never reach client bundles (`tests/interactions-browser-safe.test.ts`) | -| `/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); `createSandboxChatProducer` (raw sandbox events → client vocabulary + persisted parts + usage receipt, non-renderable-ask auto-decline); `createUploadRoute` (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); import-free `./wire` contract (`ChatTurnRequestPayload`, inline-part byte gate) re-exported via `/web-react`'s chat-stream | 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` | +| `/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` (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 → client vocabulary + persisted parts + usage receipt, non-renderable-ask auto-decline); `createUploadRoute` (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); import-free `./wire` contract (`ChatTurnRequestPayload`, inline-part byte gate) re-exported via `/web-react`'s chat-stream | 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` | | `/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`) | diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index dd20cda..ffbbd49 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -88,6 +88,7 @@ consumer of L0/L1 installs none of them): `konva`/`react-konva` → only | Integration-hub `/exec` calls | `integrations` | | Per-workspace key mint/rotate/budget | `billing` | | Resumable chat turns (buffer/replay/coalesce) | `stream` — see [`examples/resumable-turns.md`](./examples/resumable-turns.md) | +| The whole assembled chat turn route (auth → persist → stream → interactions) | `chat-routes` — `createChatTurnRoutes` (peer `agent-runtime`). Product seams: `turnLock` · `contextGate` · `beforeTurn` · `lifecycle` · `heartbeat` · `onRawEvent`, plus `transformFinalText` (pre-persist redaction over the final-text scalar AND every persisted TEXT part) and `onTurnComplete(failed, failureReason)` run-failure surfacing | | 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/src/chat-routes/turn-routes.ts b/src/chat-routes/turn-routes.ts index 8206851..995e5ca 100644 --- a/src/chat-routes/turn-routes.ts +++ b/src/chat-routes/turn-routes.ts @@ -137,6 +137,15 @@ export interface ChatTurnProduceArgs { * injected keepalive. Same shape the engine forwards verbatim. */ type ChatRouteEvent = { type: string; data?: Record } +/** Best-effort human-readable cause from a terminal `error` / + * `session.run.failed` event's `data`. */ +function failureReasonOf(data: Record | undefined): string | undefined { + if (!data) return undefined + const message = data.message ?? data.error ?? data.reason + if (typeof message === 'string' && message.length > 0) return message + return undefined +} + /** Keepalive emitted while the producer is quiet (long tool calls, first-token * wait) so client watchdogs stay re-armed. One is emitted each time * `intervalMs` elapses with no producer event; the window resets on every real @@ -249,9 +258,20 @@ export interface CreateChatTurnRoutesOptions { /** Pre-persist transform of the final text (e.g. `/redact`'s `redactPII`). * Live stream is never altered. */ transformFinalText?(text: string): string | Promise - /** Post-processing after a successful turn (billing, titles, audit). Errors - * are swallowed by the engine — they never fail a streamed turn. */ - onTurnComplete?(input: { identity: ChatTurnIdentity; finalText: string; context: TContext }): Promise + /** Post-processing after a turn settles (billing, titles, audit). Fires with + * `failed:true` + `failureReason` when the turn carried a terminal error + * event (model 402 / rate-limit / server error) instead of a clean + * completion, so products skip the deduct and render an error row rather + * than billing an empty turn and marking it done. A turn that THROWS never + * reaches this hook (the engine skips it on a producer throw). Errors are + * swallowed by the engine — they never fail a streamed turn. */ + onTurnComplete?(input: { + identity: ChatTurnIdentity + finalText: string + context: TContext + failed: boolean + failureReason?: string + }): Promise /** Per-event side channel (product broadcast). The turn-buffer tap is * already wired; this runs in addition. */ onEvent?(event: { type: string; data?: Record }, context: TContext): void | Promise @@ -367,11 +387,18 @@ async function* withStreamHeartbeat( let tick = 0 for (;;) { let timer: ReturnType | undefined - const heartbeat = new Promise<'heartbeat'>((resolve) => { - timer = setTimeout(() => resolve('heartbeat'), intervalMs) - }) - const winner = await Promise.race([pending.then(() => 'event' as const), heartbeat]) - if (timer !== undefined) clearTimeout(timer) + let winner: 'event' | 'heartbeat' + try { + const heartbeat = new Promise<'heartbeat'>((resolve) => { + timer = setTimeout(() => resolve('heartbeat'), intervalMs) + }) + winner = await Promise.race([pending.then(() => 'event' as const), heartbeat]) + } finally { + // Clear the pending timer on EVERY exit — including a `pending` + // rejection — so a rejected source never orphans a setTimeout that + // keeps the runtime alive for up to `intervalMs`. + if (timer !== undefined) clearTimeout(timer) + } if (winner === 'heartbeat') { tick += 1 yield makeEvent({ elapsedMs: Date.now() - windowStart, tick }) @@ -481,6 +508,49 @@ export function createChatTurnRoutes( lockHandle = acquired.handle } + // Turn state, hoisted so the pre-stream `catch` can settle the lifecycle + // (fire `onTurnError`, close the span) even when a seam throws + // synchronously before the drain — the drain would otherwise be the only + // path that runs the terminal hook. + let producer: ChatTurnRouteProducer | undefined + let runFailed = false + // Data of the event that marked the run failed — handed to `onTurnError` + // when no drain throw supplies a richer cause. + let lastFailureData: Record | undefined + let turnStartedAtMs = 0 + let turnStarted = false + let lifecycleSettled = false + + // Exactly one terminal lifecycle hook, after the turn settles (idempotent). + // Failure is this route's own verdict (`runFailed` from error/failed + // events, or a drain/sync throw), not the engine's envelope. + const fireTerminalLifecycle = async (failed: boolean, terminalError: unknown): Promise => { + if (lifecycleSettled) return + lifecycleSettled = true + const lifecycle = options.lifecycle + if (!lifecycle) return + const durationMs = Date.now() - turnStartedAtMs + try { + if (failed) { + await lifecycle.onTurnError?.({ + identity, executionId, turnStreamId, context, durationMs, + error: terminalError ?? lastFailureData ?? new Error('chat turn failed'), + }) + } else { + await lifecycle.onTurnComplete?.({ + identity, executionId, turnStreamId, context, durationMs, + finalText: producer?.finalText() ?? '', + usage: producer?.usage?.() ?? {}, + }) + } + } catch (err) { + log('[chat-routes] lifecycle terminal hook failed', { + turnId: turnStreamId, + error: err instanceof Error ? err.message : String(err), + }) + } + } + try { if (chatTurn.shouldInsertUserMessage) { await options.store.appendMessage({ @@ -520,13 +590,8 @@ export function createChatTurnRoutes( const turnMarker = { type: 'turn', turnId: turnStreamId } await tap.onEvent(turnMarker) - let producer: ChatTurnRouteProducer | undefined - let runFailed = false - // Data of the event that marked the run failed — handed to `onTurnError` - // when no drain throw supplies a richer cause. - let lastFailureData: Record | undefined - - const turnStartedAtMs = Date.now() + turnStartedAtMs = Date.now() + turnStarted = true if (options.lifecycle?.onTurnStart) { try { await options.lifecycle.onTurnStart({ @@ -572,8 +637,23 @@ export function createChatTurnRoutes( ...(options.transformFinalText ? { transformFinalText: options.transformFinalText } : {}), persistAssistantMessage: async ({ finalText }) => { // The typed boundary: stream-normalizer records → stored vocabulary - // (validating projection owned by /chat-store — no cast here). - const parts = producer?.assistantParts ? toChatMessageParts(producer.assistantParts()) : undefined + // (validating projection owned by /chat-store — no cast here). The + // scalar `finalText` arrives already transformed by the engine; the + // producer's text PARTS are raw, so the same transform must run over + // each text segment before persistence or a redaction (legal PII) + // leaks at rest through message.parts. + const rawParts = producer?.assistantParts ? producer.assistantParts() : undefined + const projected = + rawParts && options.transformFinalText + ? await Promise.all( + rawParts.map(async (part) => + String((part as { type?: unknown }).type ?? '') === 'text' + ? { ...part, text: await options.transformFinalText!(String((part as { text?: unknown }).text ?? '')) } + : part, + ), + ) + : rawParts + const parts = projected ? toChatMessageParts(projected) : undefined if (!finalText.trim() && (!parts || parts.length === 0)) return const usage = producer?.usage?.() ?? {} await options.store.appendMessage({ @@ -592,42 +672,25 @@ export function createChatTurnRoutes( }, ...(options.onTurnComplete ? { + // Wired into the engine's completion hook, which fires only when + // the stream ended without throwing. A terminal error EVENT + // (not a throw) still lands here — so surface `runFailed` so the + // product skips billing an errored turn instead of marking it + // complete with empty text. onTurnComplete: ({ identity: turnIdentity, finalText }: { identity: ChatTurnIdentity; finalText: string }) => - options.onTurnComplete!({ identity: turnIdentity, finalText, context }), + options.onTurnComplete!({ + identity: turnIdentity, + finalText, + context, + failed: runFailed, + ...(runFailed ? { failureReason: failureReasonOf(lastFailureData) } : {}), + }), } : {}), ...(options.traceFlush ? { traceFlush: () => options.traceFlush!(context) } : {}), }, }) - // Exactly one terminal lifecycle hook, after the turn settles. Failure is - // this route's own verdict (`runFailed` from error/failed events, or a - // drain throw), not the engine's envelope. - const fireTerminalLifecycle = async (failed: boolean, drainError: unknown): Promise => { - const lifecycle = options.lifecycle - if (!lifecycle) return - const durationMs = Date.now() - turnStartedAtMs - try { - if (failed) { - await lifecycle.onTurnError?.({ - identity, executionId, turnStreamId, context, durationMs, - error: drainError ?? lastFailureData ?? new Error('chat turn failed'), - }) - } else { - await lifecycle.onTurnComplete?.({ - identity, executionId, turnStreamId, context, durationMs, - finalText: producer?.finalText() ?? '', - usage: producer?.usage?.() ?? {}, - }) - } - } catch (err) { - log('[chat-routes] lifecycle terminal hook failed', { - turnId: turnStreamId, - error: err instanceof Error ? err.message : String(err), - }) - } - } - // Tee: one branch to the live client, one drained under waitUntil so the // turn (and its buffering via onEvent) runs to completion after a client // drop — the engine body executes as it is pulled. @@ -680,7 +743,10 @@ export function createChatTurnRoutes( }) } catch (err) { // A throw before the turn began streaming (user-insert, gate, beforeTurn, - // lifecycle-start, tap setup): release the lock, then propagate. + // lifecycle-start, tap setup, engine construction, tee). If the turn had + // already started, settle the lifecycle with `onTurnError` (close the + // span) — the drain never ran to do it. Then release the lock, propagate. + if (turnStarted) await fireTerminalLifecycle(true, err) await releaseLock() throw err } diff --git a/src/stream/stream-normalizer.ts b/src/stream/stream-normalizer.ts index c51d7ac..61c92f2 100644 --- a/src/stream/stream-normalizer.ts +++ b/src/stream/stream-normalizer.ts @@ -208,6 +208,16 @@ export function getPartKey(part: JsonRecord): string { return `${lane}:${String(part.id ?? part.partId ?? part.index ?? 'current')}` } +/** Shallow overlay that skips `undefined` incoming values, so a later partial + * update never erases a field an earlier one captured. */ +function overlayDefined(base: JsonRecord, patch: JsonRecord): JsonRecord { + const out: JsonRecord = { ...base } + for (const [key, value] of Object.entries(patch)) { + if (value !== undefined) out[key] = value + } + return out +} + export function mergePersistedPart(existing: JsonRecord | undefined, incoming: JsonRecord, delta?: string): JsonRecord { const type = String(incoming.type ?? '') if (!existing) { @@ -240,14 +250,24 @@ export function mergePersistedPart(existing: JsonRecord | undefined, incoming: J } if (type === 'tool' && String(existing.type ?? '') === 'tool') { + const existingState = asRecord(existing.state) ?? {} + const incomingState = asRecord(incoming.state) ?? {} + // Overlay only DEFINED incoming fields: a normalized tool part always + // carries `output`/`error` keys (undefined when not captured), so a plain + // spread would clobber a completed tool's output with a later empty update. + const mergedState = overlayDefined(existingState, incomingState) + // A partial update with no captured status/output/error normalizes to + // `running`; never let it downgrade a tool that already settled. + const existingStatus = String(existingState.status ?? '') + if ( + (existingStatus === 'completed' || existingStatus === 'error') && + String(incomingState.status ?? '') === 'running' + ) { + mergedState.status = existingStatus + } return { - ...existing, - ...incoming, - state: { - ...(asRecord(existing.state) ?? {}), - ...(asRecord(incoming.state) ?? {}), - time: asRecord(incoming.state)?.time ?? asRecord(existing.state)?.time, - }, + ...overlayDefined(existing, incoming), + state: mergedState, } } diff --git a/tests/browser-safe-subpaths.test.ts b/tests/browser-safe-subpaths.test.ts index d71a269..9d640cc 100644 --- a/tests/browser-safe-subpaths.test.ts +++ b/tests/browser-safe-subpaths.test.ts @@ -1,36 +1,84 @@ -import { existsSync, readFileSync, readdirSync } from 'node:fs' +import { existsSync, readFileSync } from 'node:fs' import { dirname, resolve } from 'node:path' import { describe, expect, it } from 'vitest' /** - * Browser-safe subpath manifest (generalizes the old catalog-only walker). - * Client components import these subpaths directly; if anything reachable from - * one gains a Node-only or agent-runtime import, every consumer's client - * bundle crashes on module load (legal-agent #256, tax-agent #372). Walk each - * entrypoint's source import graph and fail loud on the first unsafe edge. + * Browser-safe subpath manifest — enumerated from package.json `exports`, not a + * top-level `*-react` readdir. Client components import these subpaths directly; + * if anything reachable from one gains a Node-only or agent-runtime import, + * every consumer's client bundle crashes on module load (legal-agent #256, + * tax-agent #372). The old readdir walked only top-level `*-react` DIRS, so + * NESTED browser entries (`web-react/terminal`, `design-canvas-react/engine`, + * every `*-react/lazy`, `vault/lazy`) were never checked — the exact blind spot + * that reopens the crash class. This derives the set straight from the shipped + * `exports` map instead, so a new nested react/lazy entry is covered the moment + * it is published. * - * Deliberately EXCLUDED (server-side by design, Node imports are fine there): - * runtime, sandbox, billing, tangle, eval, eval-campaign, store, platform, - * teams (non-react), preset-cloudflare, tools, missions, knowledge, - * knowledge-loop, skills, profile, run, harness drizzle/api entries, config, - * composer, assistant, model-resolution, integrations, intakes (non-react), - * sequences (non-react), design-canvas (non-react), vault, brand-extraction. + * Server-side subpaths (Node imports fine there) are simply not browser-intended + * and are excluded by the predicate below: runtime, sandbox, billing, tangle, + * eval(-campaign), store, platform, teams(non-react), preset-cloudflare, tools, + * missions, knowledge(-loop), skills, profile, run, config, composer, assistant, + * model-resolution, integrations, intakes/design-canvas/sequences (non-react), + * vault (non-lazy), brand-extraction, theme-contract (reads fs). */ -const BROWSER_SAFE_ENTRYPOINTS: Record = { - catalog: 'src/catalog/index.ts', - 'web-react': 'src/web-react/index.tsx', - theme: 'src/theme/index.ts', - stream: 'src/stream/index.ts', - trace: 'src/trace/index.ts', - redact: 'src/redact/index.ts', - crypto: 'src/crypto/index.ts', - web: 'src/web/index.ts', - harness: 'src/harness/index.ts', - 'sequences-react': 'src/sequences-react/index.ts', - 'design-canvas-react': 'src/design-canvas-react/index.ts', - 'intakes-react': 'src/intakes-react/index.ts', - 'teams-react': 'src/teams-react/index.ts', - 'studio-react': 'src/studio-react/index.tsx', + +const ROOT = resolve(__dirname, '..') +const pkg = JSON.parse(readFileSync(resolve(ROOT, 'package.json'), 'utf8')) as { + exports: Record +} + +/** Non-react subpaths a client bundle imports directly (pure mechanism / data, + * no engine, no node builtins). Kept explicit — these are stable and few; the + * react/lazy family is derived dynamically below. */ +const BROWSER_NONREACT = new Set([ + 'catalog', + 'theme', + 'stream', + 'trace', + 'redact', + 'crypto', + 'web', + 'harness', +]) + +/** Browser-intended when a client bundle imports it: the whole `*-react` family + * (incl. nested `/engine`, `/lazy`, `/terminal`), any `/lazy` split + * (browser-safe by construction), or a known pure-mechanism subpath. */ +function isBrowserIntended(subpath: string): boolean { + return /-react(\/|$)/.test(subpath) || subpath.endsWith('/lazy') || BROWSER_NONREACT.has(subpath) +} + +function importTarget(entry: { import?: string; default?: string } | string): string | undefined { + return typeof entry === 'string' ? entry : entry.import ?? entry.default +} + +/** `./dist/x/y.js` → the `src/x/y.{ts,tsx}` that produced it, or null. */ +function distToSrc(target: string): string | null { + const m = target.match(/^\.\/dist\/(.+)\.js$/) + if (!m) return null + for (const ext of ['ts', 'tsx']) { + const candidate = resolve(ROOT, 'src', `${m[1]}.${ext}`) + if (existsSync(candidate)) return candidate + } + return null +} + +// { subpath -> src file } for every browser-intended export. +const BROWSER_SAFE_ENTRYPOINTS: Record = {} +/** Browser-intended `.js` exports whose src file could not be resolved — a + * walk blind spot; asserted empty below. */ +const UNRESOLVED: string[] = [] +for (const [key, entry] of Object.entries(pkg.exports)) { + const subpath = key.replace(/^\.\/?/, '') // "./web-react" -> "web-react", "." -> "" + if (!subpath || !isBrowserIntended(subpath)) continue + const target = importTarget(entry) + if (!target || target.endsWith('.css')) continue // CSS assets are not import graphs + const src = distToSrc(target) + if (!src) { + UNRESOLVED.push(`${key} -> ${target}`) + continue + } + BROWSER_SAFE_ENTRYPOINTS[subpath] = src } const FORBIDDEN = [/^node:/, /^@tangle-network\/agent-runtime/, /^child_process$/, /^fs$/, /^util$/] @@ -60,7 +108,7 @@ function resolveLocal(fromFile: string, spec: string): string | null { function walk(entry: string): Set { const seen = new Set() - const queue = [resolve(__dirname, '..', entry)] + const queue = [entry] while (queue.length) { const file = queue.pop()! if (seen.has(file)) continue @@ -82,6 +130,11 @@ function walk(entry: string): Set { } describe('browser-safe subpath manifest', () => { + it('every browser-intended export resolves to a source file (no walk blind spot)', () => { + expect(UNRESOLVED, `unresolved browser-intended exports: ${UNRESOLVED.join(', ')}`).toEqual([]) + expect(Object.keys(BROWSER_SAFE_ENTRYPOINTS).length).toBeGreaterThan(0) + }) + for (const [name, entry] of Object.entries(BROWSER_SAFE_ENTRYPOINTS)) { it(`${name} reaches no node builtins or agent-runtime imports`, () => { const seen = walk(entry) @@ -89,6 +142,22 @@ describe('browser-safe subpath manifest', () => { }) } + it('covers the nested react/lazy entries the old readdir missed (#256/#372 class)', () => { + for (const required of [ + 'web-react/terminal', + 'design-canvas-react/engine', + 'design-canvas-react/lazy', + 'teams-react/lazy', + 'intakes-react/lazy', + 'vault/lazy', + ]) { + expect( + BROWSER_SAFE_ENTRYPOINTS[required], + `${required} exists in exports but is missing from the browser-safe manifest`, + ).toBeDefined() + } + }) + it('the walk traverses into shared files, not just the entry module', () => { // catalog re-exports the model catalogue pipeline shared with /runtime; // if the traversal ever stops at the entry file this canary goes red. @@ -99,17 +168,4 @@ describe('browser-safe subpath manifest', () => { expect([...webReact].some((f) => f.endsWith('web-react/chat-stream.ts'))).toBe(true) expect([...webReact].some((f) => f.endsWith('web-react/chat-interactions.ts'))).toBe(true) }) - - it('every *-react subpath in src/ is covered by the manifest', () => { - const src = resolve(__dirname, '../src') - const reactDirs = readdirSync(src, { withFileTypes: true }) - .filter((d) => d.isDirectory() && d.name.endsWith('-react')) - .map((d) => d.name) - for (const dir of reactDirs) { - expect( - BROWSER_SAFE_ENTRYPOINTS[dir], - `src/${dir} exists but is missing from the browser-safe manifest`, - ).toBeDefined() - } - }) }) diff --git a/tests/chat-routes/turn-routes-sync-throw.test.ts b/tests/chat-routes/turn-routes-sync-throw.test.ts new file mode 100644 index 0000000..026bb18 --- /dev/null +++ b/tests/chat-routes/turn-routes-sync-throw.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from 'vitest' + +// Force the turn engine to throw SYNCHRONOUSLY after the route has already +// fired `onTurnStart` — the pre-drain failure path. The drain (which normally +// settles the lifecycle) never runs here, so this proves the route's own +// `catch` settles it: `onTurnError` fires and the turn lock is released. +import { vi } from 'vitest' +vi.mock('@tangle-network/agent-runtime', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + handleChatTurn: () => { + throw new Error('engine boom') + }, + } +}) + +import { createChatTurnRoutes, type ChatTurnMessageStore } from '../../src/chat-routes/index' +import { createMemoryTurnEventStore } from '../../src/stream/index' + +function memoryStore(): ChatTurnMessageStore { + const rows: Array> = [] + let id = 1 + return { + async listMessages() { + return rows as never + }, + async appendMessage(input) { + rows.push({ id: `m${id++}`, ...input }) + return input + }, + } +} + +describe('createChatTurnRoutes — synchronous engine throw', () => { + it('settles the lifecycle (onTurnError) and releases the lock even when the engine throws after onTurnStart', async () => { + const events: string[] = [] + const lockOrder: string[] = [] + const routes = createChatTurnRoutes({ + projectId: 'sync-throw', + authorize: async () => ({ ok: true, tenantId: 'ws', userId: 'u', context: undefined }), + store: memoryStore(), + turnStore: createMemoryTurnEventStore(), + produce: () => ({ + stream: (async function* () {})(), + finalText: () => '', + }), + lifecycle: { + onTurnStart: () => { events.push('start') }, + onTurnComplete: () => { events.push('complete') }, + onTurnError: () => { events.push('error') }, + }, + turnLock: { + acquire: () => { lockOrder.push('acquire'); return { acquired: true as const, handle: 'h' } }, + release: (handle) => { lockOrder.push(`release:${String(handle)}`) }, + }, + log: () => {}, + }) + + const req = new Request('http://app.test/api/chat', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ threadId: 't-1', content: 'hi' }), + }) + + await expect(routes.turn(req)).rejects.toThrow('engine boom') + // onTurnStart fired, then the sync throw settled with onTurnError — never a + // dangling "started but unsettled" span, and never a false complete. + expect(events).toEqual(['start', 'error']) + // The lock acquired before the turn is released on the throw. + expect(lockOrder).toEqual(['acquire', 'release:h']) + }) +}) diff --git a/tests/chat-routes/turn-routes.test.ts b/tests/chat-routes/turn-routes.test.ts index c098c09..3fb7856 100644 --- a/tests/chat-routes/turn-routes.test.ts +++ b/tests/chat-routes/turn-routes.test.ts @@ -389,6 +389,61 @@ describe('createChatTurnRoutes — product seams', () => { expect(rows).toHaveLength(0) // no user row when the lock is held }) + it('transformFinalText: redacts the persisted text PARTS, not just the scalar finalText', async () => { + // The at-rest leak this closes: legal wires redactPII as transformFinalText; + // the engine redacts the scalar, but message.parts streamed straight off the + // producer kept the raw PII until now. + const redact = (text: string) => text.replaceAll('SSN 123', 'SSN [redacted]') + const { routes, rows, ctx, pending } = makeRoutes({ + transformFinalText: redact, + produce: () => + fakeProducer([{ type: 'text', text: 'Your SSN 123 is on file' }], 'Your SSN 123 is on file', { + assistantParts: () => [ + { type: 'tool', id: 'c1', tool: 'lookup', state: { status: 'completed', output: { ok: true } } }, + { type: 'text', text: 'Your SSN 123 is on file' }, + ], + }), + }) + await readLines((await routes.turn(turnRequest({ threadId: 't-1', content: 'q' }), ctx)).body!) + await Promise.all(pending) + + const assistant = rows.find((r) => r.role === 'assistant')! + // Engine already redacts the scalar column... + expect(assistant.content).toBe('Your SSN [redacted] is on file') + // ...and now the persisted text PART is redacted too. + const textPart = (assistant.parts ?? []).find((p) => p.type === 'text') as { text: string } | undefined + expect(textPart?.text).toBe('Your SSN [redacted] is on file') + // No raw PII survives anywhere in the persisted parts. + expect(JSON.stringify(assistant.parts ?? [])).not.toContain('SSN 123') + }) + + it('onTurnComplete: reports failed:true + failureReason on a terminal error event (not a clean complete)', async () => { + const calls: Array<{ failed: boolean; failureReason?: string; finalText: string }> = [] + const { routes, ctx, pending } = makeRoutes({ + produce: () => fakeProducer([{ type: 'error', data: { message: 'model 402 payment required' } }], ''), + onTurnComplete: async ({ failed, failureReason, finalText }) => { + calls.push({ failed, ...(failureReason ? { failureReason } : {}), finalText }) + }, + }) + await readLines((await routes.turn(turnRequest({ threadId: 't-1', content: 'q' }), ctx)).body!) + await Promise.all(pending) + + expect(calls).toHaveLength(1) + expect(calls[0]!.failed).toBe(true) + expect(calls[0]!.failureReason).toBe('model 402 payment required') + }) + + it('onTurnComplete: reports failed:false on a clean turn', async () => { + const seen: boolean[] = [] + const { routes, ctx, pending } = makeRoutes({ + produce: () => fakeProducer([{ type: 'text', text: 'ok' }], 'ok'), + onTurnComplete: async ({ failed }) => { seen.push(failed) }, + }) + await readLines((await routes.turn(turnRequest({ threadId: 't-1', content: 'q' }), ctx)).body!) + await Promise.all(pending) + expect(seen).toEqual([false]) + }) + it('onRawEvent: observes producer events before the engine frames them', async () => { const raw: string[] = [] const { routes, ctx, pending } = makeRoutes({ diff --git a/tests/stream-normalizer.test.ts b/tests/stream-normalizer.test.ts index 177e9e9..20cdf05 100644 --- a/tests/stream-normalizer.test.ts +++ b/tests/stream-normalizer.test.ts @@ -165,6 +165,38 @@ describe('mergePersistedPart', () => { const merged = mergePersistedPart(first, normalizePersistedPart({ type: 'text', id: 't1', text: '' })!) expect(merged.text).toBe('Hello') }) + + it('keeps a completed tool output when a later empty update arrives (no clobber)', () => { + const completed = normalizePersistedPart({ + type: 'tool', + id: 'call_1', + tool: 'vault_search', + state: { status: 'completed', input: { q: 'lease' }, output: { hits: 2 } }, + })! + // A later partial update for the same tool with no captured output/error — + // its normalized state carries `output: undefined`, `error: undefined`. + const laterEmpty = normalizePersistedPart({ type: 'tool', id: 'call_1', tool: 'vault_search' })! + const merged = mergePersistedPart(completed, laterEmpty) + const state = merged.state as Record + // Captured output survives; the settled status is not downgraded to running. + expect(state.output).toEqual({ hits: 2 }) + expect(state.status).toBe('completed') + expect(state.input).toEqual({ q: 'lease' }) + }) + + it('keeps a tool error message when a later empty update arrives', () => { + const errored = normalizePersistedPart({ + type: 'tool', + id: 'call_2', + tool: 'bash', + state: { status: 'error', input: { cmd: 'exit 1' }, error: 'exit 1' }, + })! + const laterEmpty = normalizePersistedPart({ type: 'tool', id: 'call_2', tool: 'bash' })! + const merged = mergePersistedPart(errored, laterEmpty) + const state = merged.state as Record + expect(state.error).toBe('exit 1') + expect(state.status).toBe('error') + }) }) describe('finalizeAssistantParts', () => { diff --git a/tests/vertical/seams-composition.test.ts b/tests/vertical/seams-composition.test.ts new file mode 100644 index 0000000..5550abf --- /dev/null +++ b/tests/vertical/seams-composition.test.ts @@ -0,0 +1,145 @@ +/** + * Composition scenario for the #200 turn-lifecycle seams the rest of the + * vertical suite doesn't exercise: `turnLock` + `lifecycle` + `contextGate` + * wired together on the assembled `createChatTurnRoutes`, driven through the + * same fake-producer harness (`createSandboxChatProducer` over scripted sidecar + * events) the other vertical tests use. + * + * Two invariants: + * - a `contextGate` reject short-circuits BEFORE the producer/lifecycle run, so + * no assistant row is written (the user row is intentionally kept — a real + * user turn — per the seam's contract); + * - the `turnLock` serializes a duplicate: a second turn fired while the lock + * is held is refused with the product's 409 and writes ZERO rows (the lock is + * acquired before any side effect, so the refused turn never runs `produce`). + */ + +import { describe, expect, it } from 'vitest' + +import { createChatTurnRoutes, createSandboxChatProducer } from '../../src/chat-routes/index' +import { guardResolution } from '../../src/platform/index' +import { createMemoryTurnEventStore } from '../../src/stream/index' +import { createMiniApp, MINI_APP_MODEL, type MiniApp } from './mini-app' + +const BASE = 'http://localhost:3000' + +const RAW_TURN_EVENTS: Array> = [ + { type: 'message.part.updated', data: { part: { type: 'text', id: 'txt1', text: 'Done.' }, delta: 'Done.' } }, + { type: 'result', data: { finalText: 'Done.' } }, +] + +async function* feed(events: Array>, gate?: Promise): AsyncGenerator { + if (gate) await gate + for (const event of events) yield event +} + +function seamRoutes(app: MiniApp, opts: { rejectGate?: boolean; gate?: Promise } = {}) { + const pending: Promise[] = [] + const lifecycleEvents: string[] = [] + const state = { held: false, produceCount: 0 } + const routes = createChatTurnRoutes({ + projectId: 'seams-mini', + authorize: async ({ request }) => { + const auth = await guardResolution(() => app.appAuth.requireApiUser(request)) + if (!auth.ok) return auth + return { ok: true as const, tenantId: 'ws1', userId: auth.value.user.id, context: undefined } + }, + store: app.store, + turnStore: createMemoryTurnEventStore(), + produce: () => { + state.produceCount += 1 + return createSandboxChatProducer({ events: feed(RAW_TURN_EVENTS, opts.gate), model: MINI_APP_MODEL }) + }, + contextGate: async () => + opts.rejectGate + ? { proceed: false, response: Response.json({ needContext: true }, { status: 409 }) } + : { proceed: true }, + turnLock: { + acquire: () => { + if (state.held) return { acquired: false as const, response: Response.json({ code: 'in_flight' }, { status: 409 }) } + state.held = true + return { acquired: true as const, handle: 'lock' } + }, + release: () => { state.held = false }, + }, + lifecycle: { + onTurnStart: () => { lifecycleEvents.push('start') }, + onTurnComplete: () => { lifecycleEvents.push('complete') }, + onTurnError: () => { lifecycleEvents.push('error') }, + }, + log: () => {}, + }) + const ctx = { waitUntil: (p: Promise) => void pending.push(p) } + return { routes, ctx, settle: () => Promise.all(pending), lifecycleEvents, state } +} + +function turnRequest(cookie: string, threadId: string, content: string): Request { + return new Request(`${BASE}/api/chat`, { + method: 'POST', + headers: { 'content-type': 'application/json', cookie }, + body: JSON.stringify({ threadId, content }), + }) +} + +describe('vertical: turnLock + lifecycle + contextGate composition (#200 seams)', () => { + it('contextGate reject writes no assistant row and never runs the producer or lifecycle', async () => { + const app = await createMiniApp() + await app.createWorkspace('ws1') + const cookie = await app.signUp('gate@example.com') + app.grantMembership('gate@example.com', 'ws1') + const thread = await app.store.createThread({ workspaceId: 'ws1', firstMessage: 'need context' }) + + const { routes, ctx, settle, lifecycleEvents, state } = seamRoutes(app, { rejectGate: true }) + const res = await routes.turn(turnRequest(cookie, thread.id, 'do the thing'), ctx) + await settle() + + expect(res.status).toBe(409) + expect(await res.json()).toEqual({ needContext: true }) + expect(state.produceCount).toBe(0) // gate short-circuits before produce + expect(lifecycleEvents).toEqual([]) // gate runs before onTurnStart + const messages = await app.store.listMessages(thread.id) + expect(messages.filter((m) => m.role === 'assistant')).toHaveLength(0) + }) + + it('turnLock serializes a duplicate: the second turn is refused 409 and writes zero rows', async () => { + const app = await createMiniApp() + await app.createWorkspace('ws1') + const cookie = await app.signUp('lock@example.com') + app.grantMembership('lock@example.com', 'ws1') + const thread = await app.store.createThread({ workspaceId: 'ws1', firstMessage: 'serialize me' }) + + // Hold the first turn's producer open so its lock stays held while the + // duplicate fires (a tiny synchronous producer would drain and release + // before the second call, making the race untestable). + let openGate!: () => void + const gate = new Promise((resolve) => { openGate = resolve }) + const { routes, ctx, settle, lifecycleEvents, state } = seamRoutes(app, { gate }) + + // First turn acquires the lock and returns a streaming Response; its lock is + // released only when the drain settles (blocked on the gate here). + const first = await routes.turn(turnRequest(cookie, thread.id, 'first'), ctx) + expect(first.status).toBe(200) + + // Duplicate fired while the lock is held → refused, no side effects at all. + const dup = await routes.turn(turnRequest(cookie, thread.id, 'duplicate'), ctx) + expect(dup.status).toBe(409) + expect(await dup.json()).toEqual({ code: 'in_flight' }) + + // Exactly one turn ran the producer; the refused one is acquired-before- + // any-side-effect, so it wrote no user row either. + const beforeDrain = await app.store.listMessages(thread.id) + expect(state.produceCount).toBe(1) + expect(beforeDrain.filter((m) => m.role === 'user')).toHaveLength(1) + + // Open the gate, drain the winner so its lock releases and its assistant + // row persists. + openGate() + await new Response(first.body).text() + await settle() + + const messages = await app.store.listMessages(thread.id) + expect(messages.map((m) => m.role)).toEqual(['user', 'assistant']) + expect(messages[1]!.content).toBe('Done.') + expect(lifecycleEvents).toEqual(['start', 'complete']) + }) +})