diff options
Diffstat (limited to 'scripts')
| -rwxr-xr-x | scripts/fix-dist-perms.sh | 25 | ||||
| -rw-r--r-- | scripts/live-probe-provider-retry.ts | 188 | ||||
| -rw-r--r-- | scripts/live-probe.ts | 479 | ||||
| -rw-r--r-- | scripts/probe-cache-warming.ts | 277 |
4 files changed, 851 insertions, 118 deletions
diff --git a/scripts/fix-dist-perms.sh b/scripts/fix-dist-perms.sh new file mode 100755 index 0000000..471cbdf --- /dev/null +++ b/scripts/fix-dist-perms.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash +# Fix ownership of dist/ so Vite can clean + rebuild it. +# The dist/assets/ dir was created as root (likely a Docker build) and Vite +# can't rmSync it as a non-root user → EACCES on `bun run build`. +# +# Usage: sudo ./scripts/fix-dist-perms.sh +set -euo pipefail + +DIST_DIR="$(cd "$(dirname "$0")/.." && pwd)/dist" + +if [ ! -d "$DIST_DIR" ]; then + echo "No dist/ directory found — nothing to fix." + exit 0 +fi + +OWNER="$(stat -c '%U:%G' "$DIST_DIR")" +echo "dist/ is currently owned by: $OWNER" + +if [ "$OWNER" = "root:root" ]; then + echo "Fixing ownership to $(stat -c '%U:%G' "$(dirname "$DIST_DIR")") ..." +fi + +# chown the whole dist/ tree to the same owner as the repo root +chown -R --reference="$(dirname "$DIST_DIR")" "$DIST_DIR" +echo "Done. dist/ is now owned by: $(stat -c '%U:%G' "$DIST_DIR")" diff --git a/scripts/live-probe-provider-retry.ts b/scripts/live-probe-provider-retry.ts new file mode 100644 index 0000000..952163c --- /dev/null +++ b/scripts/live-probe-provider-retry.ts @@ -0,0 +1,188 @@ +/** + * scripts/live-probe-provider-retry.ts — FOCUSED live probe of the transient + * `provider-retry` AgentEvent seam, run against a RUNNING backend (bin/up). + * NOT part of `bun run test`. + * + * A real `provider-retry` only fires on an upstream 429/5xx, which we can't + * force from here. So this probe verifies the two things unit tests CAN'T: + * + * 1. REGRESSION (real wire): a normal text turn through the REAL WS socket + + * the updated `foldEvent` (provider-retry case + the reduceEvent wrapper) + * seals cleanly and `providerRetry` stays NULL throughout — no spurious + * banner, and the wrapper's re-spread didn't break streaming. + * 2. PARSER + REDUCER SEAM (the new event's effectful boundary): feed a + * synthetic `provider-retry` `chat.delta` JSON string through the REAL + * `parseServerMessage` wire parser (the function that runs on every inbound + * WS frame) → confirm it is ACCEPTED (not rejected as an unknown event) → + * `foldEvent` SETS `providerRetry` (coalesces on a 2nd) and adds NO chunk → + * a subsequent `text-delta` CLEARS it. This proves the new variant survives + * the JSON-parse boundary the unit tests skip (they pass constructed events). + * + * bun scripts/live-probe-provider-retry.ts + * PROBE_MODEL=opencode/glm-5.2 bun scripts/live-probe-provider-retry.ts + */ +import type { ChatDeltaMessage, ChatErrorMessage } from "@dispatch/transport-contract"; +import type { SurfaceServerMessage } from "@dispatch/ui-contract"; +import { createSurfaceSocket } from "../src/adapters/ws/index.ts"; +import { parseServerMessage } from "../src/adapters/ws/logic.ts"; +import { + foldEvent, + initialState, + selectChunks, + selectProviderRetry, +} from "../src/core/chunks/index.ts"; + +const WS_URL = process.env.PROBE_WS ?? "ws://localhost:24205"; +const MODEL = process.env.PROBE_MODEL ?? "opencode/deepseek-v4-flash"; +const PROMPT = process.env.PROBE_PROMPT ?? "Reply with exactly: ok"; + +type ChatMsg = ChatDeltaMessage | ChatErrorMessage; + +const checks: { name: string; ok: boolean; detail?: string }[] = []; +const record = (name: string, ok: boolean, detail?: string) => { + checks.push({ name, ok, ...(detail !== undefined ? { detail } : {}) }); + console.log(` ${ok ? "✅" : "❌"} ${name}${detail ? ` — ${detail}` : ""}`); +}; +const fail = (msg: string): never => { + console.error(`\n[probe] FATAL: ${msg}`); + process.exit(1); +}; + +/** A chat.delta JSON frame carrying the given AgentEvent, exactly as the backend sends. */ +function deltaFrame(event: ChatDeltaMessage["event"]): string { + return JSON.stringify({ type: "chat.delta", event } satisfies ChatDeltaMessage); +} + +async function main() { + console.log(`[probe] provider-retry seam · model=${MODEL} · WS=${WS_URL}\n`); + + // ─── 1. REGRESSION: a real text turn through the updated foldEvent ────────── + // Routed by conversationId via a per-conv handler map (same pattern as live-probe.ts). + const handlers = new Map<string, (msg: ChatMsg) => void>(); + const socket = createSurfaceSocket({ + url: WS_URL, + onMessage: (_m: SurfaceServerMessage) => {}, + onChat: (msg: ChatMsg) => { + const id = msg.type === "chat.error" ? msg.conversationId : msg.event.conversationId; + const h = id !== undefined ? handlers.get(id) : undefined; + h?.(msg); + }, + }); + await new Promise((r) => setTimeout(r, 500)); + + const conversationId = crypto.randomUUID(); + let state = initialState(); + let deltas = 0; + let sealed = false; + let error: string | null = null; + const done = Promise.withResolvers<void>(); + handlers.set(conversationId, (msg) => { + if (msg.type === "chat.error") { + error = msg.message; + done.resolve(); + return; + } + deltas++; + state = foldEvent(state, msg.event); + if (msg.event.type === "turn-sealed") { + sealed = true; + done.resolve(); + } + }); + + socket.send({ type: "chat.send", conversationId, message: PROMPT, model: MODEL }); + const timeout = setTimeout(() => done.resolve(), 90_000); + await done.promise; + clearTimeout(timeout); + handlers.delete(conversationId); + + record( + "regression: a real text turn sealed cleanly", + sealed && error === null, + `${deltas} deltas${error ? ` err=${error}` : ""}`, + ); + record( + "regression: providerRetry stayed NULL through a normal turn (no spurious banner)", + selectProviderRetry(state) === null, + ); + + // ─── 2. PARSER + REDUCER SEAM: synthetic provider-retry through the REAL parser ─ + console.log("\n[probe] parser+reducer seam (synthetic provider-retry)"); + let s = initialState(); + s = foldEvent(s, { type: "turn-start", conversationId, turnId: "t1" }); + + const retryJson = deltaFrame({ + type: "provider-retry", + conversationId, + turnId: "t1", + attempt: 0, + delayMs: 5000, + message: 'HTTP 429: {"error":{"type":"overloaded_error","message":"overloaded"}}', + code: "429", + }); + const parsed1 = parseServerMessage(retryJson); + record( + "REAL parseServerMessage ACCEPTS a provider-retry chat.delta (not rejected as unknown)", + parsed1 !== null && parsed1.type === "chat.delta" && parsed1.event.type === "provider-retry", + parsed1 ? `event.type=${(parsed1 as { event: { type: string } }).event.type}` : "parsed=null", + ); + + if (parsed1 !== null && parsed1.type === "chat.delta") s = foldEvent(s, parsed1.event); + const retry1 = selectProviderRetry(s); + record( + "foldEvent SETS providerRetry from the PARSED event", + retry1 !== null && retry1.attempt === 0 && retry1.delayMs === 5000 && retry1.code === "429", + retry1 ? `attempt=${retry1.attempt} delay=${retry1.delayMs}ms code=${retry1.code}` : "null", + ); + record( + "provider-retry adds NO chunk (never persisted — never pollutes the prompt)", + selectChunks(s).length === 0, + `${selectChunks(s).length} chunk(s)`, + ); + + const retry2Json = deltaFrame({ + type: "provider-retry", + conversationId, + turnId: "t1", + attempt: 1, + delayMs: 10000, + message: "HTTP 429: still overloaded", + code: "429", + }); + const parsed2 = parseServerMessage(retry2Json); + if (parsed2 !== null && parsed2.type === "chat.delta") s = foldEvent(s, parsed2.event); + const retry2 = selectProviderRetry(s); + record( + "a 2nd provider-retry COALESCES (latest attempt + delay replaces previous)", + retry2 !== null && + retry2.attempt === 1 && + retry2.delayMs === 10000 && + retry2.message === "HTTP 429: still overloaded", + retry2 ? `attempt=${retry2.attempt} delay=${retry2.delayMs}ms` : "null", + ); + + const textJson = deltaFrame({ + type: "text-delta", + conversationId, + turnId: "t1", + delta: "here is the reply", + }); + const parsedText = parseServerMessage(textJson); + if (parsedText !== null && parsedText.type === "chat.delta") s = foldEvent(s, parsedText.event); + record( + "a subsequent text-delta CLEARS the banner (retry succeeded → live reply)", + selectProviderRetry(s) === null, + ); + record( + "…and the text-delta content DID land as a chunk (the reply streams normally after retries)", + selectChunks(s).some((c) => c.chunk.type === "text"), + ); + + socket.close(); + const passed = checks.filter((c) => c.ok).length; + const total = checks.length; + console.log(`\n[probe] ${passed}/${total} checks passed`); + process.exit(passed === total ? 0 : 1); +} + +main().catch((e) => fail(String(e))); diff --git a/scripts/live-probe.ts b/scripts/live-probe.ts index bc654dd..6121eac 100644 --- a/scripts/live-probe.ts +++ b/scripts/live-probe.ts @@ -5,157 +5,400 @@ * * bun scripts/live-probe.ts # default model * PROBE_MODEL=opencode/glm-5 bun scripts/live-probe.ts + * PROBE_TOOL_PROMPT="..." bun scripts/live-probe.ts # override the tool turn * * Drives the FE's REAL network-facing modules (the thin live integration test the * methodology calls for — the analogue of the backend's server.bun.test.ts): * - adapters/ws createSurfaceSocket → real WebSocket, one socket multiplexes - * the surface `catalog` AND chat ops. - * - core/chunks foldEvent/applyHistory → fold REAL chat.delta AgentEvents. + * the surface `catalog` AND chat ops (deltas routed by conversationId). + * - core/chunks foldEvent/applyHistory/groupRenderedChunks → fold REAL + * chat.delta AgentEvents and group batched tool calls by stepId. * - features/conversation-cache + adapters/idb (fake-indexeddb) → real cache. * - HTTP GET /conversations/:id?sinceSeq → real ConversationHistoryResponse. * Skips the runes chat store + svelte UI (need the Svelte compiler; thin wrappers). + * + * Turn 1 verifies the text streaming + cache + replay path. + * Turn 2 verifies the tool-call BATCHING path ([email protected] `stepId`): that live + * tool events AND replayed tool chunks carry `stepId`, and that the pure grouping + * selector folds a parallel batch into one group. */ // Provides globalThis.indexedDB + IDBKeyRange etc. for the idb adapter (a real // browser has these natively; Bun does not). The product code is unchanged. import "fake-indexeddb/auto"; import type { - ChatDeltaMessage, - ChatErrorMessage, - ConversationHistoryResponse, + ChatDeltaMessage, + ChatErrorMessage, + ConversationHistoryResponse, + ConversationMetricsResponse, } from "@dispatch/transport-contract"; import type { SurfaceServerMessage } from "@dispatch/ui-contract"; import { createIdbChunkStore } from "../src/adapters/idb/index.ts"; import { createSurfaceSocket } from "../src/adapters/ws/index.ts"; -import { applyHistory, foldEvent, initialState, selectMessages } from "../src/core/chunks/index.ts"; +import { + applyHistory, + foldEvent, + groupRenderedChunks, + initialState, + selectChunks, + selectMessages, + type TranscriptState, +} from "../src/core/chunks/index.ts"; +import { + applyDurableMetrics, + foldMetricsEvent, + initialMetricsState, + type MetricsState, + selectOrderedTurnMetrics, +} from "../src/core/metrics/index.ts"; import { createConversationCache } from "../src/features/conversation-cache/index.ts"; const WS_URL = process.env.PROBE_WS ?? "ws://localhost:24205"; const HTTP_BASE = process.env.PROBE_HTTP ?? "http://localhost:24203"; const MODEL = process.env.PROBE_MODEL ?? "opencode/deepseek-v4-flash"; -const PROMPT = process.env.PROBE_PROMPT ?? "Reply with exactly: hello from dispatch"; -const conversationId = crypto.randomUUID(); +const TEXT_PROMPT = process.env.PROBE_PROMPT ?? "Reply with exactly: hello from dispatch"; +const TOOL_PROMPT = + process.env.PROBE_TOOL_PROMPT ?? + "Make two tool calls AT THE SAME TIME in a single step (parallel tool calls). " + + "For example, run two independent shell commands together: `echo alpha` and `echo beta`. " + + "If you have no shell tool, invoke any two of your available read-only tools simultaneously."; const checks: { name: string; ok: boolean; detail?: string }[] = []; const record = (name: string, ok: boolean, detail?: string) => { - checks.push({ name, ok, ...(detail !== undefined ? { detail } : {}) }); - console.log(` ${ok ? "✅" : "❌"} ${name}${detail ? ` — ${detail}` : ""}`); + checks.push({ name, ok, ...(detail !== undefined ? { detail } : {}) }); + console.log(` ${ok ? "✅" : "❌"} ${name}${detail ? ` — ${detail}` : ""}`); }; +const note = (msg: string) => console.log(` ℹ️ ${msg}`); function fail(msg: string): never { - console.error(`\n[live-probe] FATAL: ${msg}`); - process.exit(1); + console.error(`\n[live-probe] FATAL: ${msg}`); + process.exit(1); +} + +async function historySync( + id: string, + sinceSeq: number, + window?: { limit?: number; beforeSeq?: number }, +): Promise<ConversationHistoryResponse> { + let url = `${HTTP_BASE}/conversations/${encodeURIComponent(id)}?sinceSeq=${sinceSeq}`; + if (window?.limit !== undefined) url += `&limit=${window.limit}`; + if (window?.beforeSeq !== undefined) url += `&beforeSeq=${window.beforeSeq}`; + const res = await fetch(url, { headers: { Origin: "http://localhost:24204" } }); + if (!res.ok) fail(`history fetch ${res.status} for ${url}`); + return (await res.json()) as ConversationHistoryResponse; +} + +/** Raw history GET that returns the status (for the CR-5 validation checks). */ +async function historyStatus(id: string, query: string): Promise<number> { + const url = `${HTTP_BASE}/conversations/${encodeURIComponent(id)}?${query}`; + const res = await fetch(url, { headers: { Origin: "http://localhost:24204" } }); + await res.arrayBuffer(); // drain + return res.status; +} + +/** Durable metrics fetch — returns the response, or the HTTP status when not OK + * (the endpoint is being implemented backend-side; the FE tolerates a 404). */ +async function metricsSync(id: string): Promise<ConversationMetricsResponse | { status: number }> { + const url = `${HTTP_BASE}/conversations/${encodeURIComponent(id)}/metrics`; + const res = await fetch(url, { headers: { Origin: "http://localhost:24204" } }); + if (!res.ok) return { status: res.status }; + return (await res.json()) as ConversationMetricsResponse; } -async function historySync(id: string, sinceSeq: number): Promise<ConversationHistoryResponse> { - const url = `${HTTP_BASE}/conversations/${encodeURIComponent(id)}?sinceSeq=${sinceSeq}`; - const res = await fetch(url, { headers: { Origin: "http://localhost:24204" } }); - if (!res.ok) fail(`history fetch ${res.status} for ${url}`); - return (await res.json()) as ConversationHistoryResponse; +type ChatMsg = ChatDeltaMessage | ChatErrorMessage; +type Socket = ReturnType<typeof createSurfaceSocket>; + +const handlers = new Map<string, (msg: ChatMsg) => void>(); +function convOf(msg: ChatMsg): string | undefined { + return msg.type === "chat.error" ? msg.conversationId : msg.event.conversationId; +} + +/** Drive one turn to turn-sealed (or error), folding events into a fresh state. */ +async function runTurn( + socket: Socket, + conversationId: string, + prompt: string, +): Promise<{ + state: TranscriptState; + metrics: MetricsState; + deltas: number; + sealed: boolean; + error: string | null; +}> { + let state = initialState(); + let metrics = initialMetricsState(); + let deltas = 0; + let sealed = false; + let error: string | null = null; + const done = Promise.withResolvers<void>(); + + handlers.set(conversationId, (msg) => { + if (msg.type === "chat.error") { + error = msg.message; + done.resolve(); + return; + } + deltas++; + state = foldEvent(state, msg.event); + metrics = foldMetricsEvent(metrics, msg.event); + if (msg.event.type === "turn-sealed") { + sealed = true; + done.resolve(); + } + }); + + socket.send({ type: "chat.send", conversationId, message: prompt, model: MODEL }); + const timeout = setTimeout(() => done.resolve(), 90_000); + await done.promise; + clearTimeout(timeout); + handlers.delete(conversationId); + return { state, metrics, deltas, sealed, error }; +} + +function toolChunksOf(state: TranscriptState) { + return selectChunks(state).filter( + (c) => c.chunk.type === "tool-call" || c.chunk.type === "tool-result", + ); } async function main() { - console.log(`[live-probe] conversation=${conversationId} model=${MODEL}`); - console.log(`[live-probe] WS=${WS_URL} HTTP=${HTTP_BASE}\n`); - - const cache = createConversationCache(createIdbChunkStore()); - - let state = initialState(); - let gotCatalog = false; - let deltaCount = 0; - let sawTextDelta = false; - let sawSeal = false; - const done = Promise.withResolvers<void>(); - - const onChat = (msg: ChatDeltaMessage | ChatErrorMessage) => { - if (msg.type === "chat.error") { - record("no chat.error", false, msg.message); - done.resolve(); - return; - } - deltaCount++; - if (msg.event.type === "text-delta") sawTextDelta = true; - state = foldEvent(state, msg.event); - if (msg.event.type === "turn-sealed") { - sawSeal = true; - done.resolve(); - } - }; - - const socket = createSurfaceSocket({ - url: WS_URL, - onMessage: (m: SurfaceServerMessage) => { - if (m.type === "catalog") { - gotCatalog = true; - console.log(` ↳ surface catalog: ${m.catalog.length} surface(s)`); - } - }, - onChat, - }); - - // Give the socket a moment to open + deliver the catalog, then send the turn. - await new Promise((r) => setTimeout(r, 500)); - record("WS connected + surface catalog received", gotCatalog); - - console.log(`\n[live-probe] sending chat.send: "${PROMPT}"`); - socket.send({ type: "chat.send", conversationId, message: PROMPT, model: MODEL }); - - // Wait for turn-sealed (or error), with a hard timeout. - const timeout = setTimeout(() => done.resolve(), 90_000); - await done.promise; - clearTimeout(timeout); - - record("received chat.delta events", deltaCount > 0, `${deltaCount} deltas`); - record("saw text-delta", sawTextDelta); - record("turn reached turn-sealed", sawSeal); - - const provisionalText = selectMessages(state) - .flatMap((m) => m.chunks) - .filter((c) => c.type === "text") - .map((c) => (c as { text: string }).text) - .join(""); - console.log(`\n ↳ streamed assistant text (provisional): ${JSON.stringify(provisionalText)}`); - - // Post-seal: resync authoritative seq'd history + commit to cache (the real path). - const sinceSeq = await cache.sinceSeq(conversationId); - const hist = await historySync(conversationId, sinceSeq); - record( - "history endpoint returned chunks", - hist.chunks.length > 0, - `${hist.chunks.length} chunks, latestSeq=${hist.latestSeq}`, - ); - const monotonic = hist.chunks.every((c, i) => i === 0 || c.seq > (hist.chunks[i - 1]?.seq ?? -1)); - record("history chunks are seq-monotonic", monotonic); - - const merged = await cache.commit(conversationId, hist.chunks); - state = applyHistory(state, merged); - record( - "provisional superseded after applyHistory (sealedTurnId cleared)", - state.sealedTurnId === null, - ); - - const cached = await cache.load(conversationId); - record( - "IndexedDB cache persisted the turn", - cached.length === hist.chunks.length, - `${cached.length} cached`, - ); - - const committedText = selectMessages(state) - .filter((m) => m.role === "assistant") - .flatMap((m) => m.chunks) - .filter((c) => c.type === "text") - .map((c) => (c as { text: string }).text) - .join(""); - console.log(` ↳ committed assistant text (post-sync): ${JSON.stringify(committedText)}`); - record("committed transcript has assistant text", committedText.length > 0); - - socket.close(); - - const passed = checks.filter((c) => c.ok).length; - const total = checks.length; - console.log(`\n[live-probe] ${passed}/${total} checks passed`); - process.exit(passed === total ? 0 : 1); + console.log(`[live-probe] model=${MODEL}`); + console.log(`[live-probe] WS=${WS_URL} HTTP=${HTTP_BASE}\n`); + + const cache = createConversationCache(createIdbChunkStore()); + + let gotCatalog = false; + const socket = createSurfaceSocket({ + url: WS_URL, + onMessage: (m: SurfaceServerMessage) => { + if (m.type === "catalog") { + gotCatalog = true; + console.log(` ↳ surface catalog: ${m.catalog.length} surface(s)`); + } + }, + onChat: (msg: ChatMsg) => { + const id = convOf(msg); + const h = id !== undefined ? handlers.get(id) : undefined; + if (h) h(msg); + }, + }); + + await new Promise((r) => setTimeout(r, 500)); + record("WS connected + surface catalog received", gotCatalog); + + // ─── Turn 1: text streaming + cache + replay ──────────────────────────────── + console.log(`\n[live-probe] TURN 1 (text): "${TEXT_PROMPT}"`); + const textConv = crypto.randomUUID(); + const t1 = await runTurn(socket, textConv, TEXT_PROMPT); + if (t1.error !== null) record("turn 1 had no chat.error", false, t1.error); + record("turn 1 received chat.delta events", t1.deltas > 0, `${t1.deltas} deltas`); + record("turn 1 reached turn-sealed", t1.sealed); + + let state = t1.state; + const sinceSeq = await cache.sinceSeq(textConv); + const hist = await historySync(textConv, sinceSeq); + record( + "turn 1 history endpoint returned chunks", + hist.chunks.length > 0, + `${hist.chunks.length} chunks, latestSeq=${hist.latestSeq}`, + ); + const monotonic = hist.chunks.every((c, i) => i === 0 || c.seq > (hist.chunks[i - 1]?.seq ?? -1)); + record("turn 1 history chunks are seq-monotonic", monotonic); + const merged = await cache.commit(textConv, hist.chunks); + state = applyHistory(state, merged); + record("turn 1 provisional superseded (sealedTurnId cleared)", state.sealedTurnId === null); + const cached = await cache.load(textConv); + record("turn 1 IndexedDB cache persisted the turn", cached.length === hist.chunks.length); + const committedText = selectMessages(state) + .filter((m) => m.role === "assistant") + .flatMap((m) => m.chunks) + .filter((c) => c.type === "text") + .map((c) => (c as { text: string }).text) + .join(""); + record("turn 1 committed transcript has assistant text", committedText.length > 0); + + // ─── CR-5: history windowing (?limit= / ?beforeSeq=, [email protected]) ─────── + const logLen = hist.chunks.length; + record( + "CR-5 seq origin: first chunk is seq 1 (1-based gap-free contract)", + hist.chunks[0]?.seq === 1, + `first seq=${hist.chunks[0]?.seq}`, + ); + const win = await historySync(textConv, 0, { limit: 2 }); + record( + "CR-5 ?limit=2 returns the NEWEST 2, ascending, latestSeq = window tail", + win.chunks.length === Math.min(2, logLen) && + win.chunks[0]?.seq === Math.max(1, logLen - 1) && + win.chunks[win.chunks.length - 1]?.seq === logLen && + win.latestSeq === logLen, + `seqs=[${win.chunks.map((c) => c.seq).join(",")}] latestSeq=${win.latestSeq}`, + ); + const whole = await historySync(textConv, 0, { limit: 200 }); + record( + "CR-5 ?limit= larger than the log returns everything (short-chat flow exact)", + whole.chunks.length === logLen, + `${whole.chunks.length}/${logLen} chunks`, + ); + const oldestLoaded = win.chunks[0]?.seq ?? 0; + if (oldestLoaded > 1) { + const back = await historySync(textConv, 0, { beforeSeq: oldestLoaded, limit: 50 }); + record( + "CR-5 ?beforeSeq= pages the older run (seq < bound, ascending from 1)", + back.chunks.length === oldestLoaded - 1 && + back.chunks[0]?.seq === 1 && + back.chunks.every((c) => c.seq < oldestLoaded), + `seqs=[${back.chunks.map((c) => c.seq).join(",")}]`, + ); + } + record("CR-5 limit=0 rejected with 400", (await historyStatus(textConv, "limit=0")) === 400); + record( + "CR-5 beforeSeq=-1 rejected with 400", + (await historyStatus(textConv, "beforeSeq=-1")) === 400, + ); + + // ─── Metrics: LIVE token + timing ([email protected] usage/step-complete/done) ────── + // (TurnMetricsEntry is `{ turnId, steps, total }` — the turn aggregate lives on + // `total`, present once the live `done` folded.) + const liveTurns = selectOrderedTurnMetrics(t1.metrics); + const m1 = liveTurns[0]; + const m1Total = m1?.total ?? null; + record( + "turn 1 LIVE metrics: a turn with output tokens", + m1Total !== null && m1Total.usage.outputTokens > 0, + m1Total + ? `in=${m1Total.usage.inputTokens} out=${m1Total.usage.outputTokens} steps=${m1?.steps.length}` + : "no finalized turn total", + ); + if (m1 !== undefined) { + const anyGen = m1.steps.some((s) => s.genTotalMs !== undefined); + const anyTtft = m1.steps.some((s) => s.ttftMs !== undefined); + note( + `live timing: durationMs=${m1Total?.durationMs ?? "—"}, ` + + `genTotalMs present=${anyGen}, ttftMs present=${anyTtft}`, + ); + record( + "turn 1 LIVE metrics carries timing (durationMs or step genTotalMs)", + m1Total?.durationMs !== undefined || anyGen, + "requires the backend runtime to have a clock", + ); + } + + // ─── Metrics: DURABLE endpoint (GET /conversations/:id/metrics) ────────────── + const dm = await metricsSync(textConv); + if ("status" in dm) { + note( + `durable /metrics not available yet (HTTP ${dm.status}) — FE degrades to live-only, as designed`, + ); + record( + "durable /metrics is implemented OR gracefully absent (404)", + dm.status === 404 || dm.status === 405, + `HTTP ${dm.status}`, + ); + } else { + record( + "durable /metrics returned TurnMetrics[]", + Array.isArray(dm.turns), + `${dm.turns.length} turn(s)`, + ); + const durableMerged = selectOrderedTurnMetrics( + applyDurableMetrics(initialMetricsState(), dm.turns), + ); + const d1 = durableMerged[0]; + const d1Total = d1?.total ?? null; + record( + "durable /metrics turn has token usage", + d1Total !== null && d1Total.usage.outputTokens > 0, + d1Total ? `out=${d1Total.usage.outputTokens} steps=${d1?.steps.length}` : "no turn total", + ); + } + + // ─── Turn 2: tool-call batching ([email protected] stepId) ───────────────────────── + console.log(`\n[live-probe] TURN 2 (tools): "${TOOL_PROMPT}"`); + const toolConv = crypto.randomUUID(); + const t2 = await runTurn(socket, toolConv, TOOL_PROMPT); + if (t2.error !== null) record("turn 2 had no chat.error", false, t2.error); + record("turn 2 reached turn-sealed", t2.sealed); + + const liveTool = toolChunksOf(t2.state); + const liveCalls = liveTool.filter((c) => c.chunk.type === "tool-call"); + + if (liveCalls.length === 0) { + note( + "INCONCLUSIVE: the model issued no tool calls this run — cannot verify stepId grouping live. " + + "Re-run with a stronger PROBE_TOOL_PROMPT or one tailored to the backend's tool set.", + ); + record("turn 2 tool-call batching (live)", true, "skipped — no tool calls issued"); + } else { + // Every live tool chunk must carry stepId (foldEvent copies it from the event). + const allLiveHaveStep = liveTool.every( + (c) => + (c.chunk.type === "tool-call" || c.chunk.type === "tool-result") && + typeof c.chunk.stepId === "string" && + c.chunk.stepId.length > 0, + ); + record( + "turn 2 every LIVE tool event carries stepId", + allLiveHaveStep, + `${liveCalls.length} call(s), ${liveTool.length - liveCalls.length} result(s)`, + ); + + const liveGroups = groupRenderedChunks(selectChunks(t2.state)); + const liveBatches = liveGroups.filter((g) => g.kind === "tool-batch"); + const distinctSteps = new Set( + liveCalls.map((c) => (c.chunk.type === "tool-call" ? c.chunk.stepId : undefined)), + ); + note( + `live grouping: ${liveCalls.length} call(s) across ${distinctSteps.size} step(s) → ` + + `${liveBatches.length} batch group(s)`, + ); + if (liveBatches.length > 0) { + record( + "turn 2 grouping produced a parallel batch (2+ calls in one step)", + true, + `${liveBatches.length} batch(es)`, + ); + } else { + note( + "the model used tools but did NOT parallelize (each call its own step) — stepId is verified, " + + "but no multi-call batch occurred to render as a list this run.", + ); + } + + // Replay path: persisted tool chunks must also carry chunk.stepId. + const histTool = await historySync(toolConv, 0); + const replayTool = histTool.chunks.filter( + (c) => c.chunk.type === "tool-call" || c.chunk.type === "tool-result", + ); + const allReplayHaveStep = replayTool.every( + (c) => + (c.chunk.type === "tool-call" || c.chunk.type === "tool-result") && + typeof c.chunk.stepId === "string" && + c.chunk.stepId.length > 0, + ); + record( + "turn 2 every REPLAYED tool chunk carries chunk.stepId", + replayTool.length > 0 && allReplayHaveStep, + `${replayTool.length} tool chunk(s) in history`, + ); + + // Grouping on the authoritative replayed history matches the live shape. + const replayState = applyHistory(initialState(), await cache.commit(toolConv, histTool.chunks)); + const replayBatches = groupRenderedChunks(selectChunks(replayState)).filter( + (g) => g.kind === "tool-batch", + ); + record( + "turn 2 replay grouping matches live (batch count)", + replayBatches.length === liveBatches.length, + `live=${liveBatches.length} replay=${replayBatches.length}`, + ); + } + + socket.close(); + + const passed = checks.filter((c) => c.ok).length; + const total = checks.length; + console.log(`\n[live-probe] ${passed}/${total} checks passed`); + process.exit(passed === total ? 0 : 1); } main().catch((e) => fail(String(e))); diff --git a/scripts/probe-cache-warming.ts b/scripts/probe-cache-warming.ts new file mode 100644 index 0000000..1bf1f9b --- /dev/null +++ b/scripts/probe-cache-warming.ts @@ -0,0 +1,277 @@ +/** + * scripts/probe-cache-warming.ts — LIVE probe of the `cache-warming` surface + + * conversation-close lifecycle against a RUNNING backend (bin/up: HTTP :24203 + + * surface WS :24205; override with PROBE_HTTP / PROBE_WS for bin/up2's +1000 + * ports). NOT part of `bun run test`. Verifies the CR-4 handoff end-to-end: + * + * A. draft subscribe (no conversationId) → degenerate "no conversation" spec + * B. fresh conversation → warming defaults OFF, nothing scheduled (CR-4a) + * C. toggle on + 10s interval → repeated automatic warms, each update carrying + * a FUTURE nextWarmAt (CR-4b), initial `surface` echoes conversationId (CR-4d) + * D. POST /conversations/:id/close mid-turn → abortedTurn, done.reason + * "aborted", turn-sealed, warming disabled + unscheduled (CR-4c) + * + * bun scripts/probe-cache-warming.ts + */ +import type { + ChatDeltaMessage, + ChatErrorMessage, + CloseConversationResponse, +} from "@dispatch/transport-contract"; +import type { SurfaceServerMessage, SurfaceSpec } from "@dispatch/ui-contract"; +import { createSurfaceSocket } from "../src/adapters/ws/index.ts"; +import { parseControls } from "../src/features/cache-warming/logic/view-model.ts"; + +const WS_URL = process.env.PROBE_WS ?? "ws://localhost:24205"; +const HTTP_BASE = process.env.PROBE_HTTP ?? "http://localhost:24203"; +const SURFACE_ID = "cache-warming"; + +const checks: { name: string; ok: boolean }[] = []; +const record = (name: string, ok: boolean, detail?: string) => { + checks.push({ name, ok }); + console.log(` ${ok ? "✅" : "❌"} ${name}${detail ? ` — ${detail}` : ""}`); +}; +const log = (msg: string) => console.log(`[${new Date().toISOString().slice(11, 19)}] ${msg}`); +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +function summarize(spec: SurfaceSpec | null): string { + const c = parseControls(spec); + const next = + c.nextWarmAt === null ? "null" : `${Math.round((c.nextWarmAt - Date.now()) / 1000)}s`; + return `enabled=${c.enabled} interval=${c.intervalSeconds}s lastPct=${c.lastPct} next=${next} lastWarmAt=${c.lastWarmAt}`; +} + +let catalog: { id: string; scope?: string }[] = []; +let latestSpec: SurfaceSpec | null = null; +let latestSpecConv: string | undefined; +let specWaiter: (() => void) | null = null; + +const chatHandlers = new Map<string, (msg: ChatDeltaMessage | ChatErrorMessage) => void>(); + +const socket = createSurfaceSocket({ + url: WS_URL, + onMessage: (m: SurfaceServerMessage) => { + if (m.type === "catalog") { + catalog = [...m.catalog]; + log(`catalog: ${m.catalog.map((e) => `${e.id}(scope=${e.scope ?? "—"})`).join(", ")}`); + } else if (m.type === "surface") { + latestSpec = m.spec; + latestSpecConv = m.conversationId; + log(`surface(initial) conv=${m.conversationId ?? "—"}: ${summarize(m.spec)}`); + specWaiter?.(); + } else if (m.type === "update") { + if (m.update.surfaceId !== SURFACE_ID) return; + latestSpec = m.update.spec; + latestSpecConv = m.update.conversationId; + log(`update conv=${m.update.conversationId ?? "—"}: ${summarize(m.update.spec)}`); + specWaiter?.(); + } else if (m.type === "error") { + log(`surface ERROR: ${m.surfaceId ?? "—"}: ${m.message}`); + } + }, + onChat: (msg) => { + const id = msg.type === "chat.error" ? msg.conversationId : msg.event.conversationId; + if (id !== undefined) chatHandlers.get(id)?.(msg); + }, +}); + +/** Wait for the next surface/update message (or time out). */ +function nextSpec(timeoutMs: number): Promise<boolean> { + return new Promise((resolve) => { + const t = setTimeout(() => { + specWaiter = null; + resolve(false); + }, timeoutMs); + specWaiter = () => { + clearTimeout(t); + specWaiter = null; + resolve(true); + }; + }); +} + +async function runTinyTurn(conversationId: string, prompt: string): Promise<boolean> { + const done = Promise.withResolvers<boolean>(); + chatHandlers.set(conversationId, (msg) => { + if (msg.type === "chat.error") { + log(`chat.error: ${msg.message}`); + done.resolve(false); + } else if (msg.event.type === "turn-sealed") { + done.resolve(true); + } + }); + socket.send({ type: "chat.send", conversationId, message: prompt }); + const t = setTimeout(() => done.resolve(false), 90_000); + const ok = await done.promise; + clearTimeout(t); + chatHandlers.delete(conversationId); + return ok; +} + +function invoke(actionId: string, conversationId: string, payload?: unknown): void { + socket.send( + payload === undefined + ? { type: "invoke", surfaceId: SURFACE_ID, actionId, conversationId } + : { type: "invoke", surfaceId: SURFACE_ID, actionId, payload, conversationId }, + ); +} + +async function main() { + await sleep(600); + record( + "catalog includes cache-warming with scope=conversation", + catalog.some((e) => e.id === SURFACE_ID && e.scope === "conversation"), + ); + + // ── A: the DRAFT/new-tab path — subscribe with NO conversationId ─────────── + log("PHASE A: subscribe with NO conversationId (draft / new tab)"); + socket.send({ type: "subscribe", surfaceId: SURFACE_ID }); + await nextSpec(3000); + record( + "draft subscribe → degenerate spec (no toggle parsed)", + !parseControls(latestSpec).enabled, + ); + socket.send({ type: "unsubscribe", surfaceId: SURFACE_ID }); + await sleep(300); + + // ── B: a FRESH conversation defaults OFF (CR-4a) + echo (CR-4d) ──────────── + const conv = crypto.randomUUID(); + log(`PHASE B: creating conversation ${conv}`); + if (!(await runTinyTurn(conv, "Reply with exactly: ok"))) { + log("FATAL: could not create a conversation"); + process.exit(1); + } + socket.send({ type: "subscribe", surfaceId: SURFACE_ID, conversationId: conv }); + await nextSpec(3000); + const fresh = parseControls(latestSpec); + record("CR-4d: initial surface message echoes conversationId", latestSpecConv === conv); + record("CR-4a: fresh conversation defaults to warming OFF", fresh.enabled === false); + record("CR-4a: nothing scheduled while off (nextWarmAt null)", fresh.nextWarmAt === null); + + // ── C: opt in + 10s interval → repeated warms, FUTURE nextWarmAt (CR-4b) ─── + log("PHASE C: toggling warming ON"); + const toggleId = fresh.toggleActionId; + if (toggleId === null) { + record("toggle action present", false); + process.exit(1); + } + invoke(toggleId, conv); + await nextSpec(3000); + let c = parseControls(latestSpec); + record("toggle-on update arrived (enabled)", c.enabled === true); + record( + "CR-4b: enable schedules a FUTURE nextWarmAt", + c.nextWarmAt !== null && c.nextWarmAt > Date.now(), + ); + + const setIntervalId = c.setIntervalActionId; + if (setIntervalId !== null) { + log("PHASE C: set-interval = 10s"); + invoke(setIntervalId, conv, 10); + await nextSpec(3000); + c = parseControls(latestSpec); + record( + "set-interval update: interval=10 + FUTURE nextWarmAt", + c.intervalSeconds === 10 && c.nextWarmAt !== null && c.nextWarmAt > Date.now(), + ); + } + + log("PHASE C: waiting up to 45s for 2 automatic warms…"); + const deadline = Date.now() + 45_000; + let lastSeen = c.lastWarmAt; + let warms = 0; + let allFuture = true; + while (Date.now() < deadline && warms < 2) { + await nextSpec(Math.max(1, deadline - Date.now())); + const now = parseControls(latestSpec); + if (now.lastWarmAt !== null && now.lastWarmAt !== lastSeen) { + lastSeen = now.lastWarmAt; + warms++; + const future = now.nextWarmAt !== null && now.nextWarmAt > Date.now() - 1000; + if (!future) allFuture = false; + log( + ` automatic warm #${warms}: pct=${now.lastPct} retention=${now.retentionPct} ` + + `nextWarmAt ${future ? "FUTURE" : "STALE/PAST"}`, + ); + } + } + record("automatic warms repeat (2 observed @10s)", warms >= 2, `${warms} warm(s)`); + record("CR-4b: every post-warm update carries a FUTURE nextWarmAt", warms >= 2 && allFuture); + + // ── D: close mid-turn → abort + warming disabled (CR-4c) ─────────────────── + log("PHASE D: starting a long turn, then closing the conversation mid-turn…"); + const seenDone = Promise.withResolvers<string>(); // resolves with done.reason + const seenSealed = Promise.withResolvers<void>(); + let turnStarted = false; + const started = Promise.withResolvers<void>(); + chatHandlers.set(conv, (msg) => { + if (msg.type === "chat.error") { + log(`chat.error: ${msg.message}`); + return; + } + const ev = msg.event; + if (ev.type === "turn-start") { + turnStarted = true; + started.resolve(); + } else if (ev.type === "done") { + seenDone.resolve(ev.reason); + } else if (ev.type === "turn-sealed") { + seenSealed.resolve(); + } + }); + socket.send({ + type: "chat.send", + conversationId: conv, + message: + "Write a detailed 1000-word essay about the history of computing. Take your time and be thorough.", + }); + const startTimeout = setTimeout(() => started.resolve(), 15_000); + await started.promise; + clearTimeout(startTimeout); + record("turn started (watcher saw turn-start)", turnStarted); + await sleep(1000); // let it generate a moment + + const res = await fetch(`${HTTP_BASE}/conversations/${encodeURIComponent(conv)}/close`, { + method: "POST", + headers: { Origin: "http://localhost:24204" }, + }); + record("POST /conversations/:id/close → 200", res.ok, `HTTP ${res.status}`); + const body = (await res.json()) as CloseConversationResponse; + record("close aborted the in-flight turn (abortedTurn)", body.abortedTurn === true); + + const doneReason = await Promise.race([seenDone.promise, sleep(15_000).then(() => "(timeout)")]); + record('watcher received done with reason "aborted"', doneReason === "aborted", doneReason); + const sealed = await Promise.race([ + seenSealed.promise.then(() => true), + sleep(15_000).then(() => false), + ]); + record("turn sealed normally after abort", sealed); + chatHandlers.delete(conv); + + // The close also pushed a surface update: warming disabled + unscheduled. + await sleep(1500); + const closed = parseControls(latestSpec); + record( + "CR-4c: close disabled warming + cleared the schedule", + closed.enabled === false && closed.nextWarmAt === null, + summarize(latestSpec), + ); + + // Idempotency: closing again (now idle) succeeds with abortedTurn false. + const res2 = await fetch(`${HTTP_BASE}/conversations/${encodeURIComponent(conv)}/close`, { + method: "POST", + headers: { Origin: "http://localhost:24204" }, + }); + const body2 = (await res2.json()) as CloseConversationResponse; + record("close is idempotent (200 + abortedTurn:false)", res2.ok && body2.abortedTurn === false); + + socket.close(); + const passed = checks.filter((x) => x.ok).length; + console.log(`\n[probe-cache-warming] ${passed}/${checks.length} checks passed`); + process.exit(passed === checks.length ? 0 : 1); +} + +main().catch((e) => { + console.error(`[probe] FATAL: ${e}`); + process.exit(1); +}); |