Skip to content

fix(chat-routes): redact persisted parts, surface run-failure, close manifest hole (audit P1) #201

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
drewstone merged 1 commit into from
Jul 17, 2026
Merged

fix(chat-routes): redact persisted parts, surface run-failure, close manifest hole (audit P1) #201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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`) |
Expand Down
1 change: 1 addition & 0 deletions ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) |
Expand Down
160 changes: 113 additions & 47 deletions src/chat-routes/turn-routes.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,15 @@ export interface ChatTurnProduceArgs<TContext> {
* injected keepalive. Same shape the engine forwards verbatim. */
type ChatRouteEvent = { type: string; data?: Record<string, unknown> }

/** Best-effort human-readable cause from a terminal `error` /
* `session.run.failed` event's `data`. */
function failureReasonOf(data: Record<string, unknown> | 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
Expand Down Expand Up @@ -249,9 +258,20 @@ export interface CreateChatTurnRoutesOptions<TContext = void> {
/** Pre-persist transform of the final text (e.g. `/redact`'s `redactPII`).
* Live stream is never altered. */
transformFinalText?(text: string): string | Promise<string>
/** 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<void>
/** 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<void>
/** Per-event side channel (product broadcast). The turn-buffer tap is
* already wired; this runs in addition. */
onEvent?(event: { type: string; data?: Record<string, unknown> }, context: TContext): void | Promise<void>
Expand Down Expand Up @@ -367,11 +387,18 @@ async function* withStreamHeartbeat(
let tick = 0
for (;;) {
let timer: ReturnType<typeof setTimeout> | 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 })
Expand Down Expand Up @@ -481,6 +508,49 @@ export function createChatTurnRoutes<TContext = void>(
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<string, unknown> | 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<void> => {
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({
Expand Down Expand Up @@ -520,13 +590,8 @@ export function createChatTurnRoutes<TContext = void>(
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<string, unknown> | undefined

const turnStartedAtMs = Date.now()
turnStartedAtMs = Date.now()
turnStarted = true
if (options.lifecycle?.onTurnStart) {
try {
await options.lifecycle.onTurnStart({
Expand Down Expand Up @@ -572,8 +637,23 @@ export function createChatTurnRoutes<TContext = void>(
...(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({
Expand All @@ -592,42 +672,25 @@ export function createChatTurnRoutes<TContext = void>(
},
...(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<void> => {
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.
Expand Down Expand Up @@ -680,7 +743,10 @@ export function createChatTurnRoutes<TContext = void>(
})
} 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
}
Expand Down
34 changes: 27 additions & 7 deletions src/stream/stream-normalizer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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,
}
}

Expand Down
Loading