diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index ca2f4a3..1ba4eab 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -80,6 +80,16 @@ Headless recovery exits with the invalid provider/model and instructions to resume interactively; it sends no model request. The regression tests also cover `/resume` inside the TUI and restarting after a repair. +## Runtime refresh regression + +`bun test test/runtime-context-cache.test.ts test/tui/runtime-refresh.test.ts` +verifies that text/reasoning deltas do not schedule runtime queries and that +tool activity polling reuses a compact, session-scoped token summary. Completed +requests, history changes and changed projection usage invalidate the summary. +In-flight reads from a previous session or revision cannot overwrite newer +statistics. Full message bodies are read for statistics only after invalidation +or when the user opens `/context`; they are not retained in the cache. + ## OAuth login For the OAuth path, run the launcher directly with the login subcommand: diff --git a/package.json b/package.json index 2c47c3c..105896d 100644 --- a/package.json +++ b/package.json @@ -70,7 +70,7 @@ "test:node": "node --test test/node/*.test.cjs", "test:tui": "bun run test:tui:component && bun run test:tui:e2e", "test:tui:component": "bun test test/tui/scenario-http.test.ts test/tui/scenario-runtime.test.ts test/tui/scenario-shell.test.ts test/tui/scenario-workspace.test.ts test/tui/terminal-screen.test.ts", - "test:tui:e2e": "bun test test/tui/allowlisted-shell.test.ts test/tui/http-mock.test.ts test/tui/model-resume.test.ts test/tui/permission-request-queue.test.ts test/tui/run-scenario.test.ts test/tui/session-rename.test.ts test/tui/terminal-session.test.ts test/tui/write-and-diff.test.ts", + "test:tui:e2e": "bun test test/tui/allowlisted-shell.test.ts test/tui/http-mock.test.ts test/tui/model-resume.test.ts test/tui/permission-request-queue.test.ts test/tui/run-scenario.test.ts test/tui/runtime-refresh.test.ts test/tui/session-rename.test.ts test/tui/terminal-session.test.ts test/tui/write-and-diff.test.ts", "test:tui:host": "bun test test/tui/scenario-mountx.test.ts", "test:tui:manual": "bun scripts/tui-scenario.ts --manual", "test:tui-scenario": "bun scripts/tui-scenario.ts", diff --git a/packages/zcode-tui/src/index.ts b/packages/zcode-tui/src/index.ts index 7c0c5f8..24e1c36 100644 --- a/packages/zcode-tui/src/index.ts +++ b/packages/zcode-tui/src/index.ts @@ -167,8 +167,11 @@ import { isVisibleProtocolPart, ProtocolPartView } from "./protocol-part-view.ts import { InputQueue, type QueuedSubmission } from "./input-queue.ts"; import { QueuedInputView } from "./queued-input-view.ts"; import { RuntimeActivityView } from "./runtime-activity-view.ts"; +import { RuntimeContextCache } from "./runtime-context-cache.ts"; import { runtimeActivityActive, + runtimeContextChanged, + runtimeContextRefreshNeeded, runtimePollInterval, runtimeRefreshNeeded, runtimePollStateChanged, @@ -192,7 +195,7 @@ import { import { SkillCatalog } from "./skills.ts"; import { isActiveBackgroundJob, - mergeProjectionContextCache, + mergeProjectionContextSummary, normalizeRuntimeProjection, normalizeTodoGroups, normalizeTodos, @@ -710,6 +713,7 @@ class ZCodeTui { private usageRefreshInFlight = false; private usageRefreshPending = false; private runtimeProjection?: RuntimeProjectionSnapshot; + private readonly runtimeContextCache = new RuntimeContextCache(); private todos: RuntimeTodo[] = []; private todoGroups: RuntimeTodoGroup[] = []; private runtimeRefreshInFlight = false; @@ -2153,7 +2157,9 @@ class ZCodeTui { settingTarget?: SettingTarget ): Promise { if (!isRecord(result)) return; + this.runtimeContextCache.invalidate(); if (result.resetSessionProjection === true) { + this.runtimeContextCache.reset(); this.executionStateRevision++; this.clearTranscriptProjection(); this.workflowView = undefined; @@ -2217,6 +2223,7 @@ class ZCodeTui { this.ui.requestRender(); if (this.sessionModelIssue) await this.recoverSessionModel(); } + this.scheduleRuntimeRefresh(0); } private onEvent(value: unknown, turnEpoch?: number): void { @@ -2224,6 +2231,7 @@ class ZCodeTui { if (turnEpoch !== undefined && turnEpoch !== this.activeTurnEpoch) return; const event = normalizeEvent(value); if (!event || this.isForeignSessionEvent(event)) return; + if (runtimeContextRefreshNeeded(event)) this.runtimeContextCache.invalidate(); const taskScoped = this.backgroundTaskEvents.isTaskScoped(event); this.applyBackgroundTaskEvent(event); if (!taskScoped && event.kind && toolLifecycleEventKinds.has(event.kind)) this.turnHadWorkActivity = true; @@ -2399,8 +2407,9 @@ class ZCodeTui { this.debugEvent("session-subscription", value); const event = normalizeEvent(value); if (!event || this.isForeignSessionEvent(event)) return; + if (runtimeContextRefreshNeeded(event)) this.runtimeContextCache.invalidate(); this.applyBackgroundTaskEvent(event); - this.scheduleRuntimeRefresh(); + if (runtimeRefreshNeeded(event)) this.scheduleRuntimeRefresh(); } private isForeignSessionEvent(event: StreamEvent): boolean { @@ -5515,12 +5524,9 @@ class ZCodeTui { do { this.runtimeRefreshPending = false; const executionStateRevision = this.executionStateRevision; - const [projectionResult, todosResult, contextMessagesResult] = await Promise.allSettled([ + const [projectionResult, todosResult] = await Promise.allSettled([ this.options.readRuntimeProjection?.(), - this.options.readTodos?.(), - this.options.readRuntimeProjection && this.options.loadSessionContextMessages - ? this.options.loadSessionContextMessages() - : Promise.resolve(undefined) + this.options.readTodos?.() ]); if (executionStateRevision !== this.executionStateRevision) { this.runtimeRefreshPending = true; @@ -5533,10 +5539,18 @@ class ZCodeTui { }; if (projectionResult.status === "fulfilled" && projectionResult.value !== undefined) { const projection = normalizeRuntimeProjection(projectionResult.value); - next.projection = contextMessagesResult.status === "fulfilled" - && contextMessagesResult.value !== undefined - ? mergeProjectionContextCache(projection, contextMessagesResult.value) ?? next.projection - : projection ?? next.projection; + if (runtimeContextChanged(this.runtimeProjection, projection)) this.runtimeContextCache.invalidate(); + const cache = this.options.loadSessionContextMessages + ? await this.runtimeContextCache.read( + projection?.sessionId ?? this.sessionId, + this.options.loadSessionContextMessages + ).catch(() => undefined) + : undefined; + if (executionStateRevision !== this.executionStateRevision) { + this.runtimeRefreshPending = true; + continue; + } + next.projection = mergeProjectionContextSummary(projection, cache) ?? next.projection; if (isRecord(projectionResult.value) && Array.isArray(projectionResult.value.todoGroups)) { next.todoGroups = normalizeTodoGroups(projectionResult.value); } diff --git a/packages/zcode-tui/src/runtime-context-cache.ts b/packages/zcode-tui/src/runtime-context-cache.ts new file mode 100644 index 0000000..8fdc3c4 --- /dev/null +++ b/packages/zcode-tui/src/runtime-context-cache.ts @@ -0,0 +1,45 @@ +import { contextCacheUsage, type RuntimeContextUsage } from "./runtime-projection.ts"; + +type CacheUsage = RuntimeContextUsage["cache"]; + +/** A session-scoped, invalidatable summary of persisted request usage. */ +export class RuntimeContextCache { + private sessionId?: string; + private revision = 0; + private loadedRevision = -1; + private usage?: CacheUsage; + private pending?: { revision: number; promise: Promise }; + + invalidate(): void { + this.revision++; + } + + reset(): void { + this.invalidate(); + this.usage = undefined; + this.loadedRevision = -1; + this.pending = undefined; + } + + async read(sessionId: string | undefined, loadMessages: () => Promise): Promise { + if (sessionId !== this.sessionId) { + this.reset(); + this.sessionId = sessionId; + } + if (this.loadedRevision === this.revision) return this.usage; + if (this.pending?.revision === this.revision) return this.pending.promise; + const revision = this.revision; + const promise = Promise.resolve().then(loadMessages).then((messages) => { + // A completed read from the previous session/revision must never replace + // newer usage, including when /resume changes the app during this query. + if (revision !== this.revision) return undefined; + this.usage = contextCacheUsage(messages); + this.loadedRevision = revision; + return this.usage; + }).finally(() => { + if (this.pending?.revision === revision) this.pending = undefined; + }); + this.pending = { revision, promise }; + return promise; + } +} diff --git a/packages/zcode-tui/src/runtime-poll.ts b/packages/zcode-tui/src/runtime-poll.ts index b6909e7..6f21472 100644 --- a/packages/zcode-tui/src/runtime-poll.ts +++ b/packages/zcode-tui/src/runtime-poll.ts @@ -39,3 +39,24 @@ export function runtimeRefreshNeeded( && event.kind !== "reasoning_delta" && event.kind !== "tool_input_delta"; } + +const contextChangeEvents = new Set([ + "model_complete", "model.complete", "turn_complete", "turn.completed", + "turn_error", "turn.failed", "session_created", "session_resumed", "session_forked", + "session_compacted", "compact_boundary", "rewind_triggered", "rewind.triggered" +]); + +export function runtimeContextRefreshNeeded(event: Pick): boolean { + return event.type !== undefined && contextChangeEvents.has(event.type); +} + +/** Polling also notices completed work from runtimes without session events. */ +export function runtimeContextChanged( + current: RuntimeProjectionSnapshot | undefined, + next: RuntimeProjectionSnapshot | undefined +): boolean { + return current?.sessionId !== next?.sessionId + || current?.turnCount !== next?.turnCount + || current?.totalTokenCount !== next?.totalTokenCount + || current?.contextUsage?.used !== next?.contextUsage?.used; +} diff --git a/packages/zcode-tui/src/runtime-projection.ts b/packages/zcode-tui/src/runtime-projection.ts index c6474d6..8c9d4c5 100644 --- a/packages/zcode-tui/src/runtime-projection.ts +++ b/packages/zcode-tui/src/runtime-projection.ts @@ -383,28 +383,40 @@ export function mergeProjectionContextCache( projection: RuntimeProjectionSnapshot | undefined, messages: unknown ): RuntimeProjectionSnapshot | undefined { - const usage = projection?.contextUsage; - if (!projection || !usage) return projection; + return mergeProjectionContextSummary(projection, contextCacheUsage(messages)); +} +/** Retain token statistics only; tool output and message bodies stay in storage. */ +export function contextCacheUsage(messages: unknown): RuntimeContextUsage["cache"] { const trend = extractContextCacheTrend(messages); - if (trend.wholeTree.requests === 0) return projection; + if (trend.wholeTree.requests === 0) return undefined; const latest = trend.turns.findLast((turn) => turn.inputTokens !== undefined); const summary = trend.wholeTree; + return { + inputTokens: latest?.inputTokens, + cacheReadTokens: latest?.cacheReadTokens, + cacheWriteTokens: latest?.cacheWriteTokens, + latestHitRate: latest?.hitRate, + hitRate: summary.hitRate, + hitRateRequestCount: summary.requests, + totalInputTokens: summary.inputTokens, + totalCacheReadTokens: summary.cacheReadTokens, + totalCacheWriteTokens: summary.cacheWriteTokens + }; +} + +export function mergeProjectionContextSummary( + projection: RuntimeProjectionSnapshot | undefined, + cache: RuntimeContextUsage["cache"] +): RuntimeProjectionSnapshot | undefined { + if (!projection?.contextUsage || !cache) return projection; return { ...projection, contextUsage: { - ...usage, + ...projection.contextUsage, cache: { - ...usage.cache, - inputTokens: latest?.inputTokens ?? usage.cache?.inputTokens, - cacheReadTokens: latest?.cacheReadTokens ?? usage.cache?.cacheReadTokens, - cacheWriteTokens: latest?.cacheWriteTokens ?? usage.cache?.cacheWriteTokens, - latestHitRate: latest?.hitRate ?? usage.cache?.latestHitRate, - hitRate: summary.hitRate ?? usage.cache?.hitRate, - hitRateRequestCount: summary.requests, - totalInputTokens: summary.inputTokens, - totalCacheReadTokens: summary.cacheReadTokens, - totalCacheWriteTokens: summary.cacheWriteTokens + ...projection.contextUsage.cache, + ...Object.fromEntries(Object.entries(cache).filter(([, value]) => value !== undefined)) } } }; diff --git a/test/runtime-context-cache.test.ts b/test/runtime-context-cache.test.ts new file mode 100644 index 0000000..2a7775f --- /dev/null +++ b/test/runtime-context-cache.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, test } from "bun:test"; +import { RuntimeContextCache } from "../packages/zcode-tui/src/runtime-context-cache.ts"; +import { runtimeContextChanged, runtimeContextRefreshNeeded } from "../packages/zcode-tui/src/runtime-poll.ts"; +import { normalizeRuntimeProjection } from "../packages/zcode-tui/src/runtime-projection.ts"; + +function messages(input: number, read = 0): unknown[] { + return [{ info: { id: "assistant-1", role: "assistant", tokens: { input, cache: { read, write: 0 } } }, + parts: [{ type: "tool", output: "Large tool output does not belong in the cached summary" }] }]; +} + +describe("runtime context cache", () => { + test("polls reuse a compact summary until usage is invalidated", async () => { + const cache = new RuntimeContextCache(); + let reads = 0; + const load = async () => { reads++; return messages(100, 80); }; + const first = await cache.read("session-1", load); + expect(first?.latestHitRate).toBe(0.8); + for (let index = 0; index < 100; index++) expect(await cache.read("session-1", load)).toBe(first); + expect(reads).toBe(1); + expect(JSON.stringify(first)).not.toContain("Large tool output"); + cache.invalidate(); + await cache.read("session-1", load); + expect(reads).toBe(2); + }); + + test("empty sessions are cached and failed reads can be retried", async () => { + const cache = new RuntimeContextCache(); + let reads = 0; + const empty = async () => { reads++; return []; }; + expect(await cache.read("empty", empty)).toBeUndefined(); + expect(await cache.read("empty", empty)).toBeUndefined(); + expect(reads).toBe(1); + cache.invalidate(); + await expect(cache.read("empty", async () => { throw new Error("Store unavailable"); })).rejects.toThrow("Store unavailable"); + expect((await cache.read("empty", async () => messages(200)))?.inputTokens).toBe(200); + }); + + test.each(["invalidate", "reset", "switch"] as const)("ignores an in-flight read after %s", async (action) => { + const cache = new RuntimeContextCache(); + let resolve!: (value: unknown) => void; + let reads = 0; + const load = () => { reads++; return new Promise((done) => { resolve = done; }); }; + const old = cache.read("old", load); + const duplicate = cache.read("old", load); + await Promise.resolve(); + expect(reads).toBe(1); + if (action === "invalidate") cache.invalidate(); + if (action === "reset") cache.reset(); + const session = action === "switch" ? "new" : "old"; + expect((await cache.read(session, async () => messages(200, 20)))?.latestHitRate).toBe(0.1); + resolve(messages(100, 80)); + expect(await old).toBeUndefined(); + expect(await duplicate).toBeUndefined(); + expect((await cache.read(session, load))?.inputTokens).toBe(200); + }); + + test("invalidates on completed requests and history changes, not presentation events", () => { + for (const type of ["model_complete", "turn_complete", "session_resumed", "session_forked", "session_compacted", "rewind_triggered"]) { + expect(runtimeContextRefreshNeeded({ type })).toBeTrue(); + } + for (const type of ["model_streaming", "part.delta", "tool_call_progress", "background_task_updated"]) { + expect(runtimeContextRefreshNeeded({ type })).toBeFalse(); + } + const projection = (value: Record) => normalizeRuntimeProjection({ sessionId: "s", totalTokenCount: 100, ...value }); + expect(runtimeContextChanged(projection({}), projection({ status: "running" }))).toBeFalse(); + expect(runtimeContextChanged(projection({}), projection({ totalTokenCount: 200 }))).toBeTrue(); + expect(runtimeContextChanged(projection({}), projection({ sessionId: "next" }))).toBeTrue(); + }); +}); diff --git a/test/tui/fixtures/runtime-refresh.ts b/test/tui/fixtures/runtime-refresh.ts new file mode 100644 index 0000000..cd2717f --- /dev/null +++ b/test/tui/fixtures/runtime-refresh.ts @@ -0,0 +1,59 @@ +import { runTui } from "../../../packages/zcode-tui/src/index.ts"; +import { ScenarioRuntimeJournal } from "../runtime/scenario-runtime.ts"; + +const journal = new ScenarioRuntimeJournal(process.env.ZCODE_TUI_SCENARIO_RUNTIME_JOURNAL); +let sessionId = "first-session"; +let contextReads = 0; +let projectionReads = 0; +let completed = false; +let listener: ((event: unknown) => void | Promise) | undefined; + +await runTui({ + initialModel: "scenario/model", + workspaceDirectory: process.cwd(), + subscribeSessionEvents: (sink) => { listener = sink; return () => { listener = undefined; }; }, + readRuntimeProjection: async () => { + projectionReads++; + return { + sessionId, status: "idle", totalTokenCount: completed ? 200 : 100, + contextUsage: { used: 100, size: 1_000 }, activeToolCalls: [], backgroundJobs: [] + }; + }, + loadSessionContextMessages: async () => { + contextReads++; + journal.record("context.read", { sessionId, contextReads }); + return [{ info: { id: "assistant", role: "assistant", tokens: { + input: 100, cache: { read: sessionId === "second-session" ? 20 : completed ? 60 : 80, write: 0 } + } }, parts: [] }]; + }, + submitPrompt: async (input, options) => { + if (String(input) === "/resume second") { + sessionId = "second-session"; + return { response: "Resumed second session", resetSessionProjection: true, restoredMessages: [] }; + } + const before = contextReads; + const projectionsBefore = projectionReads; + const startedAt = performance.now(); + for (let index = 0; index < 40; index++) { + const event = { id: `text-${index}`, sessionId, type: "model_streaming", + payload: { kind: "text_delta", delta: "x", assistantMessageId: "streamed-message" } }; + await options.onEvent?.(event); + await listener?.(event); + await Bun.sleep(20); + } + await Bun.sleep(120); + journal.record("text-deltas.finished", { + before, after: contextReads, projectionReads: projectionReads - projectionsBefore, + elapsedMs: performance.now() - startedAt + }); + for (let index = 0; index < 10; index++) { + await listener?.({ id: `progress-${index}`, sessionId, type: "tool_call_progress", + payload: { toolCallId: "tool", toolName: "Bash", elapsedMs: index * 100 } }); + await Bun.sleep(100); + } + journal.record("tool-progress.finished", { before, after: contextReads }); + completed = true; + await listener?.({ id: "complete", sessionId, type: "model_complete", payload: {} }); + return { response: "Streaming completed" }; + } +}); diff --git a/test/tui/runtime-refresh.test.ts b/test/tui/runtime-refresh.test.ts new file mode 100644 index 0000000..8565bc5 --- /dev/null +++ b/test/tui/runtime-refresh.test.ts @@ -0,0 +1,29 @@ +import { expect, test } from "bun:test"; +import { join } from "node:path"; +import { ScenarioWorkspace } from "./harness/scenario-workspace.ts"; +import { TerminalSession } from "./harness/terminal-session.ts"; + +test("TUI streams and polls tool activity without rereading history, then refreshes completed/resumed usage", async () => { + await using workspace = await ScenarioWorkspace.create({ prefix: "zcode-runtime-refresh-" }); + await using session = TerminalSession.start({ + command: [process.execPath, join(import.meta.dir, "fixtures/runtime-refresh.ts")], workspace + }); + await session.waitForScreen("initial cache", /cache 80% hit/u); + session.send("stream\r"); + await session.waitForHistory("stream finished", /Streaming completed/u); + await session.waitForScreen("completed cache", /cache 60% hit/u); + const entries = (await workspace.readRuntimeJournal()).trim().split("\n").map((line) => JSON.parse(line)); + for (const kind of ["text-deltas.finished", "tool-progress.finished"]) { + const entry = entries.find((entry) => entry.kind === kind); + expect(entry, kind).toBeDefined(); + expect(entry.detail.after, kind).toBe(entry.detail.before); + if (kind === "text-deltas.finished") { + // Allow scheduled activity polls on slow CI; the subscription must not + // introduce the previous 80 ms query loop during presentation-only data. + expect(entry.detail.projectionReads).toBeLessThanOrEqual(Math.ceil(entry.detail.elapsedMs / 1_000) + 1); + } + } + await session.sendAndWait("/resume second\r", "resumed session", /Resumed second session/u); + await session.waitForScreen("resumed cache", /cache 20% hit/u); + await session.exit(); +}, 20_000);