From 205722b0e8eafe841f5de4802919d1cf9ee6c128 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 16 Oct 2026 16:32:01 +1100 Subject: [PATCH 1/6] fix(ci): record marked in the packaged dependency ledger Generated-by: Codex --- apps/desktop/artifact-budget.json | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/desktop/artifact-budget.json b/apps/desktop/artifact-budget.json index fc70f8ca4b0..9d279a6508a 100644 --- a/apps/desktop/artifact-budget.json +++ b/apps/desktop/artifact-budget.json @@ -162,6 +162,7 @@ "node_modules/lodash.merge@4.6.2", "node_modules/lodash.pickby@4.6.0", "node_modules/long@5.3.2", + "node_modules/marked@18.0.11", "node_modules/math-intrinsics@1.1.0", "node_modules/minimatch@10.2.5", "node_modules/minipass@7.1.3", From 95c69b030a8b90295577874103f554188d596664 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 16 Oct 2026 16:44:59 +1100 Subject: [PATCH 2/6] fix(ci): refresh the reduced AppShell architecture debt Generated-by: Codex --- apps/desktop/renderer-architecture.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/desktop/renderer-architecture.json b/apps/desktop/renderer-architecture.json index 16d40288db0..f3a9c66c71b 100644 --- a/apps/desktop/renderer-architecture.json +++ b/apps/desktop/renderer-architecture.json @@ -743,7 +743,7 @@ "react": 1 }, "importSpecifiers": 43, - "nonTriviaTokens": 7285 + "nonTriviaTokens": 7280 }, "src/renderer/use-app-shell-session-list.ts": { "importDeclarations": 3, From c1192304f0d88f885b1c30431c5b80367340feb5 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 16 Oct 2026 16:51:31 +1100 Subject: [PATCH 3/6] test(desktop): wait for WorkHub bridge reconciliation Generated-by: Codex --- .../__tests__/workhub-remote-bridge.test.ts | 124 ++++++++---------- 1 file changed, 55 insertions(+), 69 deletions(-) diff --git a/apps/desktop/src/main/__tests__/workhub-remote-bridge.test.ts b/apps/desktop/src/main/__tests__/workhub-remote-bridge.test.ts index 33e5bc0a511..ccf2eef64ab 100644 --- a/apps/desktop/src/main/__tests__/workhub-remote-bridge.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-remote-bridge.test.ts @@ -18,7 +18,7 @@ */ import assert from 'node:assert/strict'; -import test from 'node:test'; +import test, { type TestContext } from 'node:test'; import { mkdtemp, readFile, writeFile, rm } from 'node:fs/promises'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; @@ -31,7 +31,7 @@ import { AsyncFrameQueue, runtimeHostSessionFixture } from './runtime-host-sessi type Deps = Parameters[0]; const message = (chatId: string, text: string, sourceMessageId = text): BotIncomingMessage => ({ platform: 'qq', chatId, userId: 'owner', userName: 'Owner', isGroup: false, text, sourceMessageId, receivedAt: 1 }); -async function fixture(workName: string | ((id: string) => string) = 'SQL 入门') { +async function fixture(t: TestContext, workName: string | ((id: string) => string) = 'SQL 入门') { const directory = await mkdtemp(join(tmpdir(), 'workhub-remote-')); const sends: { chat: string; text: string }[] = []; const starts: { turnId: string; text: string }[] = []; @@ -66,13 +66,17 @@ async function fixture(workName: string | ((id: string) => string) = 'SQL 入门 const deps: Deps = { client, statePath: join(directory, 'routes.json'), readMode: async () => mode, botRegistry: { sendMessage: async (_platform, chat, text) => { sends.push({ chat, text }); return 'sent'; } }, onError: (error) => { errors.push(error); } }; let bridge = createWorkHubRemoteBridge(deps); - const tick = async () => { for (let i = 0; i < 30; i++) await new Promise((resolve) => setTimeout(resolve, 2)); }; + const projectFrame = async (frame: SubscriptionFrame) => { + const nextRead = Math.max(1, events.nextCount) + 1; + events.push(frame); + await t.waitFor(() => assert.ok(events.nextCount >= nextRead)); + }; return { sends, starts, answers, errors, get bridge() { return bridge; }, setBusy(value: boolean) { busy = value; }, setMode(value: typeof mode) { mode = value; }, - async project(value: InteractionPendingSnapshot[]) { pending = value; events.push({ kind: 'subscription.session_projection', snapshot: snapshot() } as SubscriptionFrame); await tick(); }, - async transcript(value: StoredMessage[], done = true) { rows = value; completed = done; events.push({ kind: 'subscription.transcript_advanced', throughSequence: rows.length - 1 } as SubscriptionFrame); await tick(); }, + async project(value: InteractionPendingSnapshot[]) { pending = value; await projectFrame({ kind: 'subscription.session_projection', snapshot: snapshot() } as SubscriptionFrame); }, + async transcript(value: StoredMessage[], done = true) { rows = value; completed = done; await projectFrame({ kind: 'subscription.transcript_advanced', throughSequence: rows.length - 1 } as SubscriptionFrame); }, async recoverLegacy(value: StoredMessage[]) { await bridge.close(); rows = value; @@ -84,17 +88,17 @@ async function fixture(workName: string | ((id: string) => string) = 'SQL 入门 await writeFile(deps.statePath, JSON.stringify(state)); events = new AsyncFrameQueue(); bridge = createWorkHubRemoteBridge(deps); - await tick(); + await projectFrame({ kind: 'subscription.session_projection', snapshot: snapshot() } as SubscriptionFrame); }, - async restart() { await bridge.close(); events = new AsyncFrameQueue(); bridge = createWorkHubRemoteBridge(deps); await tick(); }, + async restart() { await bridge.close(); events = new AsyncFrameQueue(); bridge = createWorkHubRemoteBridge(deps); await projectFrame({ kind: 'subscription.session_projection', snapshot: snapshot() } as SubscriptionFrame); }, async close() { await bridge.close(); await rm(directory, { recursive: true, force: true }); assert.deepEqual(errors, []); }, }; } function question(turnId: string): InteractionPendingSnapshot { return { schemaVersion: 1, revision: 1, status: 'pending', outcome: null, sessionId, turnId, runId: 'run', interactionId: 'question-1', request: { kind: 'question', toolUseId: 'ask', questions: ['项目', '范围', '测试'].map((question) => ({ question, options: ['甲', '乙', '丙'].map((label) => ({ label })) })) } }; } -test('all chats use one WorkHub; questions route to their source and survive restart', async () => { - const f = await fixture(); +test('all chats use one WorkHub; questions route to their source and survive restart', async (t) => { + const f = await fixture(t); try { assert.equal(await f.bridge.handle(message('one', '修改 Maka')), true); const turnId = f.starts[0]!.turnId; @@ -114,8 +118,8 @@ test('all chats use one WorkHub; questions route to their source and survive res assert.equal(f.starts.length, 2, 'answers resume the interaction instead of starting a new Turn'); } finally { await f.close(); } }); -test('delegated asynchronous result returns to original chat after restart, never the last chat', async () => { - const f = await fixture(); +test('delegated asynchronous result returns to original chat after restart, never the last chat', async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', '实现功能')); const firstTurn = f.starts[0]!.turnId; @@ -132,14 +136,14 @@ test('delegated asynchronous result returns to original chat after restart, neve assert.equal(f.sends.length, 1); } finally { await f.close(); } }); -test('task setting leaves the existing Bot Session route in charge', async () => { - const f = await fixture(); +test('task setting leaves the existing Bot Session route in charge', async (t) => { + const f = await fixture(t); try { f.setMode('task'); assert.equal(await f.bridge.handle(message('one', 'hello')), false); assert.equal(f.starts.length, 0); } finally { await f.close(); } }); -test('busy WorkHub queues messages durably and dispatches after restart without changing their source', async () => { - const f = await fixture(); +test('busy WorkHub queues messages durably and dispatches after restart without changing their source', async (t) => { + const f = await fixture(t); try { f.setBusy(true); await f.bridge.handle(message('one', '先做这个')); @@ -148,8 +152,6 @@ test('busy WorkHub queues messages durably and dispatches after restart without assert.equal(f.sends.length, 2); f.setBusy(false); await f.restart(); - assert.equal(f.starts.length, 1); - await f.project([]); assert.equal(f.starts.length, 2); assert.ok(f.starts[0]!.text.includes('先做这个')); assert.ok(f.starts[1]!.text.includes('再做那个')); @@ -157,8 +159,8 @@ test('busy WorkHub queues messages durably and dispatches after restart without assert.equal(f.starts.length, 2, 'same incoming source message is not admitted twice'); } finally { await f.close(); } }); -test('nxt resumes the recent-work picker in its source chat without queuing a new turn', async () => { - const f = await fixture(); +test('nxt resumes the recent-work picker in its source chat without queuing a new turn', async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', 'wn')); const pending: InteractionPendingSnapshot = { @@ -172,8 +174,8 @@ test('nxt resumes the recent-work picker in its source chat without queuing a ne } finally { await f.close(); } }); -test('runtime transcript tool calls route delegated feedback without legacy assignment rows', async () => { - const f = await fixture(); +test('runtime transcript tool calls route delegated feedback without legacy assignment rows', async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', '检查 SQL')); const firstTurn = f.starts[0]!.turnId; @@ -192,8 +194,8 @@ test('runtime transcript tool calls route delegated feedback without legacy assi } finally { await f.close(); } }); -test('legacy cursor recovery delivers missed feedback without repeating the original receipt', async () => { - const f = await fixture(); +test('legacy cursor recovery delivers missed feedback without repeating the original receipt', async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', '检查 SQL')); const turnId = f.starts[0]!.turnId; @@ -211,8 +213,8 @@ test('legacy cursor recovery delivers missed feedback without repeating the orig } finally { await f.close(); } }); -test('delayed feedback keeps its own work heading after a newer request in the same chat', async () => { - const f = await fixture(); +test('delayed feedback keeps its own work heading after a newer request in the same chat', async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', '检查 SQL')); const turnId = f.starts[0]!.turnId; @@ -228,42 +230,26 @@ test('delayed feedback keeps its own work heading after a newer request in the s } finally { await f.close(); } }); -test('direct-answer topic receipts survive paging and label a delayed answer without delegation', async () => { - const f = await fixture(); - try { - await f.bridge.handle(message('one', 'SQL 第一题是什么？')); - const turnId = f.starts[0]!.turnId; - const call: StoredMessage = { type: 'tool_call', id: 'associate', turnId, ts: 1, toolName: 'mcp__desktop_workhub__tasks', args: { request: { operation: 'associate', targetSessionId: 'sql' } } }; - await f.transcript([call]); - await f.bridge.handle(message('one', '再看看 CS336')); - await f.restart(); - await f.transcript([call, - { type: 'tool_result', id: 'topic', turnId, ts: 2, toolUseId: call.id, isError: false, content: { kind: 'json', value: { structuredContent: { operation: 'associate', targetSessionId: 'sql', targetSessionKey: JSON.stringify(['host', 'sql']), targetSessionName: 'SQL 入门' } } } }, - { type: 'assistant', id: 'answer', turnId, ts: 3, text: '查询评分不低于 9 的电影。', modelId: 'test' }, - ]); - assert.deepEqual(f.sends, [{ chat: 'one', text: '[SQL 入门]\n查询评分不低于 9 的电影。' }]); - assert.equal(f.starts.length, 2); - } finally { await f.close(); } -}); - - -test('the latest WeChat follow-up carries its associated work heading without a newer request', async () => { - const f = await fixture('CS336'); - try { - await f.bridge.handle({ ...message('wechat-owner', '刚才到第几讲内容了?'), platform: 'wechat' }); - const turnId = f.starts[0]!.turnId; - const call: StoredMessage = { type: 'tool_call', id: 'associate-current', turnId, ts: 1, toolName: 'mcp__desktop_workhub__tasks', args: { request: { operation: 'associate', targetSessionId: 'cs336' } } }; - await f.transcript([call], false); - await f.restart(); - await f.transcript([call, - { type: 'tool_result', id: 'topic-current', turnId, ts: 2, toolUseId: call.id, isError: false, content: { kind: 'json', value: { structuredContent: { operation: 'associate', targetSessionId: 'cs336', targetSessionKey: JSON.stringify(['host', 'cs336']), targetSessionName: 'CS336' } } } }, - { type: 'assistant', id: 'answer-current', turnId, ts: 3, text: '刚到第 2 讲。', modelId: 'test' }, - ]); - assert.deepEqual(f.sends, [{ chat: 'wechat-owner', text: '[CS336]\n刚到第 2 讲。' }]); - assert.equal(f.starts.length, 1, 'association does not dispatch work'); - await f.restart(); - assert.equal(f.sends.length, 1, 'already delivered answers are not resent'); - } finally { await f.close(); } +test('direct-answer topic receipts survive paging and label a delayed answer without delegation', async (t) => { + for (const newerRequest of [false, true]) { + const f = await fixture(t); + try { + await f.bridge.handle(message('one', 'SQL 第一题是什么？')); + const turnId = f.starts[0]!.turnId; + const call: StoredMessage = { type: 'tool_call', id: 'associate', turnId, ts: 1, toolName: 'mcp__desktop_workhub__tasks', args: { request: { operation: 'associate', targetSessionId: 'sql' } } }; + await f.transcript([call], false); + if (newerRequest) await f.bridge.handle(message('one', '再看看 CS336')); + await f.restart(); + await f.transcript([call, + { type: 'tool_result', id: 'topic', turnId, ts: 2, toolUseId: call.id, isError: false, content: { kind: 'json', value: { structuredContent: { operation: 'associate', targetSessionId: 'sql', targetSessionKey: JSON.stringify(['host', 'sql']), targetSessionName: 'SQL 入门' } } } }, + { type: 'assistant', id: 'answer', turnId, ts: 3, text: '查询评分不低于 9 的电影。', modelId: 'test' }, + ]); + assert.deepEqual(f.sends, [{ chat: 'one', text: '[SQL 入门]\n查询评分不低于 9 的电影。' }]); + assert.equal(f.starts.length, newerRequest ? 2 : 1); + await f.restart(); + assert.equal(f.sends.length, 1, 'already delivered answers are not resent'); + } finally { await f.close(); } + } }); function association(turnId: string, id: string, targetSessionId: string | null, reason = 'unrelated'): [StoredMessage, StoredMessage] { @@ -276,8 +262,8 @@ function association(turnId: string, id: string, targetSessionId: string | null, } for (const reason of ['unrelated', 'overview', 'ambiguous']) { - test(`a same-turn ${reason} correction clears the remote heading across restart`, async () => { - const f = await fixture(); + test(`a same-turn ${reason} correction clears the remote heading across restart`, async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', '这不是 SQL 的问题')); const turnId = f.starts[0]!.turnId; @@ -300,8 +286,8 @@ for (const reason of ['unrelated', 'overview', 'ambiguous']) { } for (const invalid of ['error', 'wrapped-error', 'target', 'turn', 'call', 'reason', 'missing-reason']) { - test(`a ${invalid} clearing receipt cannot erase the current remote heading`, async () => { - const f = await fixture(); + test(`a ${invalid} clearing receipt cannot erase the current remote heading`, async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', 'SQL 问题')); const turnId = f.starts[0]!.turnId; @@ -325,8 +311,8 @@ for (const invalid of ['error', 'wrapped-error', 'target', 'turn', 'call', 'reas } for (const chat of ['one', 'two']) { - test(`a clearing receipt cannot cross turns in chat ${chat}`, async () => { - const f = await fixture(); + test(`a clearing receipt cannot cross turns in chat ${chat}`, async (t) => { + const f = await fixture(t); try { await f.bridge.handle(message('one', 'SQL 问题')); await f.bridge.handle(message(chat, '另一个 SQL 问题')); @@ -347,13 +333,13 @@ for (const chat of ['one', 'two']) { } test('a same-turn topic switch uses the corrected work heading', async (t) => { - const f = await fixture((id) => id === 'cs336' ? 'CS336' : 'SQL 入门'); + const f = await fixture(t, (id) => id === 'cs336' ? 'CS336' : 'SQL 入门'); try { await f.bridge.handle(message('one', '其实是 CS336')); const turnId = f.starts[0]!.turnId; await f.transcript([...association(turnId, 'initial', 'sql'), ...association(turnId, 'switch', 'cs336'), { type: 'assistant', id: 'answer', turnId, ts: 3, text: '课程回答。', modelId: 'test' }, ]); - await t.waitFor(() => assert.deepEqual(f.sends, [{ chat: 'one', text: '[CS336]\n课程回答。' }])); + assert.deepEqual(f.sends, [{ chat: 'one', text: '[CS336]\n课程回答。' }]); } finally { await f.close(); } }); From 40fb1cc3cd10a8a2cd74593719eb3b8f336736de Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 16 Oct 2026 16:52:13 +1100 Subject: [PATCH 4/6] test(plan): exercise paging without repeated model turns Generated-by: Codex --- .../acp-goal-plan-child-process.test.ts | 37 ++-- .../src/__tests__/plan-two-client-uds.test.ts | 158 +++++++++--------- 2 files changed, 97 insertions(+), 98 deletions(-) diff --git a/packages/cli/src/__tests__/acp-goal-plan-child-process.test.ts b/packages/cli/src/__tests__/acp-goal-plan-child-process.test.ts index e8efe418b51..db245c13de0 100644 --- a/packages/cli/src/__tests__/acp-goal-plan-child-process.test.ts +++ b/packages/cli/src/__tests__/acp-goal-plan-child-process.test.ts @@ -279,7 +279,6 @@ describe('ACP Goal/Plan real Host routes', () => { test('Plan paging preserves storeVersion and returns revision_changed after an external write', { timeout: 45_000, }, async () => { - let call = 0; const model = createServer((request, response) => { void readRequest(request) .then((body) => { @@ -306,16 +305,17 @@ describe('ACP Goal/Plan real Host routes', () => { ); return; } - call += 1; const names = (input.tools ?? []).flatMap((tool) => tool.function?.name ? [tool.function.name] : [], ); if (names.includes('SubmitPlan')) respondTool(response, 'SubmitPlan', { - title: `Fixture plan ${call}`, - steps: [ - { id: 'step-1', title: 'Do the work', description: 'Finish the fixture task' }, - ], + title: 'Fixture plan', + steps: Array.from({ length: 3 }, (_, index) => ({ + id: `step-${index + 1}`, + title: 'Do the work', + description: 'x'.repeat(12 * 1024), + })), }); else respondTool(response, 'tool_search', { query: 'SubmitPlan' }); }) @@ -342,7 +342,7 @@ describe('ACP Goal/Plan real Host routes', () => { value: 'plan', }); let latestProposalId = ''; - for (let index = 0; index < 17; index += 1) { + for (let index = 0; index < 2; index += 1) { await context.request(methods.agent.session.prompt, { sessionId, prompt: [{ type: 'text', text: `Prepare fixture plan ${index}` }], @@ -356,20 +356,13 @@ describe('ACP Goal/Plan real Host routes', () => { }; assert.equal(latest.kind, 'page'); latestProposalId = latest.latestProposalId; - if (index < 16) { - try { - await context.request('_maka/plan/control', { - kind: 'request_revision', - sessionId, - proposalId: latestProposalId, - operationId: randomUUID(), - }); - } catch (error) { - throw new Error( - `revision index=${index} proposal=${latestProposalId} modelCalls=${call}`, - { cause: error }, - ); - } + if (index === 0) { + await context.request('_maka/plan/control', { + kind: 'request_revision', + sessionId, + proposalId: latestProposalId, + operationId: randomUUID(), + }); } } const first = (await context.request('_maka/plan/query', { @@ -382,7 +375,7 @@ describe('ACP Goal/Plan real Host routes', () => { items: unknown[]; }; assert.equal(first.kind, 'page'); - assert.equal(first.items.length, 16); + assert.equal(first.items.length, 1); assert.ok(first.nextCursor); const second = (await context.request('_maka/plan/query', { kind: 'list_continue', diff --git a/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts index 2f46b3cf469..5c25fb4a452 100644 --- a/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/plan-two-client-uds.test.ts @@ -411,81 +411,87 @@ function controllableBackendComposition(control: PlanTurnControl): RuntimeHostCo ); } -test('Plan queries include their state header when selecting byte-limited continuation pages', async () => { - const root = await mkdtemp(join(tmpdir(), 'maka-plan-pages-')); - const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); - const owner = await tryAcquireInteractiveRootOwner(capability); - assert.ok(owner); - const store = await openInteractivePlanStoreForWrite(owner.lease); - let sessions: Awaited> | undefined; - try { - sessions = await openInteractiveExecutionStoresForWrite(owner.lease); - const session = await sessions.sessionStore.create({ - cwd: root, - llmConnectionSlug: 'test', - model: 'test-model', - permissionMode: 'explore', - collaborationMode: 'plan', - }); - for (let index = 0; index < 17; index += 1) { - await store.submitProposal({ - operationId: `submit-${index}`, - sessionId: session.id, - turnId: `turn-${index}`, - title: `Proposal ${index}`, - steps: [{ id: 'step-1', title: 'Review', description: '文"\\🙂'.repeat(1000) }], +for (const [limit, description] of [ + ['item', 'Review the task'], + ['byte', '文"\\🙂'.repeat(1000)], +] as const) { + test(`Plan queries include their state header when selecting ${limit}-limited continuation pages`, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-plan-pages-')); + const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + const store = await openInteractivePlanStoreForWrite(owner.lease); + let sessions: Awaited> | undefined; + try { + sessions = await openInteractiveExecutionStoresForWrite(owner.lease); + const session = await sessions.sessionStore.create({ + cwd: root, + llmConnectionSlug: 'test', + model: 'test-model', + permissionMode: 'explore', + collaborationMode: 'plan', + }); + for (let index = 0; index < 17; index += 1) { + await store.submitProposal({ + operationId: `submit-${index}`, + sessionId: session.id, + turnId: `turn-${index}`, + title: `Proposal ${index}`, + steps: [{ id: 'step-1', title: 'Review', description }], + }); + } + const state = await store.readState(session.id); + const expected = state.proposals.map((proposal) => ({ kind: 'proposal' as const, proposal })); + const coordinator = new HostPlanCoordinator({ + store, + sessions: sessions.sessionStore, + sessionAdmission: new SessionAdmissionGate(), + runtime: null as never, + root: null as never, + isSessionActive: () => false, + refreshContinuity: async () => {}, + requestDrain: () => assert.fail('query must not drain'), + }); + const pages: Extract[] = []; + let input: PlanQueryInput = { kind: 'list_start', sessionId: session.id }; + let end = 0; + do { + const outcome = await coordinator.handlers['plan.query'](input, null as never); + assert.ok(outcome.ok && outcome.result.kind === 'page'); + const page = outcome.result; + assert.deepEqual(decodePlanQueryResult(page), page); + assert.equal(page.latestProposalId, state.latestProposalId); + assert.equal(page.storeVersion, state.storeVersion); + assert.ok(page.items.length > 0); + pages.push(page); + end += page.items.length; + assert.equal(page.nextCursor, end < expected.length ? String(end) : null); + if (page.nextCursor === null) break; + input = { + kind: 'list_continue', + sessionId: session.id, + storeVersion: page.storeVersion, + cursor: page.nextCursor, + }; + } while (end < expected.length); + assert.ok(pages.length > 1); + if (limit === 'item') assert.equal(pages[0]!.items.length, PLAN_PAGE_MAX_ITEMS); + else assert.ok(pages[0]!.items.length < PLAN_PAGE_MAX_ITEMS); + assertMaximalJsonPages(pages, expected, { + maxBytes: PLAN_RESULT_MAX_BYTES, + maxItems: PLAN_PAGE_MAX_ITEMS, + items: (page) => page.items, + candidate: (page, items, end) => ({ + ...page, + items, + nextCursor: end < expected.length ? String(end) : null, + }), }); + } finally { + store.close(); + await sessions?.sessionStore.close?.(); + await owner.close(); + await rm(root, { recursive: true, force: true }); } - const state = await store.readState(session.id); - const expected = state.proposals.map((proposal) => ({ kind: 'proposal' as const, proposal })); - const coordinator = new HostPlanCoordinator({ - store, - sessions: sessions.sessionStore, - sessionAdmission: new SessionAdmissionGate(), - runtime: null as never, - root: null as never, - isSessionActive: () => false, - refreshContinuity: async () => {}, - requestDrain: () => assert.fail('query must not drain'), - }); - const pages: Extract[] = []; - let input: PlanQueryInput = { kind: 'list_start', sessionId: session.id }; - let end = 0; - do { - const outcome = await coordinator.handlers['plan.query'](input, null as never); - assert.ok(outcome.ok && outcome.result.kind === 'page'); - const page = outcome.result; - assert.deepEqual(decodePlanQueryResult(page), page); - assert.equal(page.latestProposalId, state.latestProposalId); - assert.equal(page.storeVersion, state.storeVersion); - assert.ok(page.items.length > 0); - pages.push(page); - end += page.items.length; - assert.equal(page.nextCursor, end < expected.length ? String(end) : null); - if (page.nextCursor === null) break; - input = { - kind: 'list_continue', - sessionId: session.id, - storeVersion: page.storeVersion, - cursor: page.nextCursor, - }; - } while (end < expected.length); - assert.ok(pages.length > 1); - assert.ok(pages[0]!.items.length < PLAN_PAGE_MAX_ITEMS); - assertMaximalJsonPages(pages, expected, { - maxBytes: PLAN_RESULT_MAX_BYTES, - maxItems: PLAN_PAGE_MAX_ITEMS, - items: (page) => page.items, - candidate: (page, items, end) => ({ - ...page, - items, - nextCursor: end < expected.length ? String(end) : null, - }), - }); - } finally { - store.close(); - await sessions?.sessionStore.close?.(); - await owner.close(); - await rm(root, { recursive: true, force: true }); - } -}); + }); +} From be580ce2257608de310748aff95f272fd7ded5dc Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 16 Oct 2026 17:15:25 +1100 Subject: [PATCH 5/6] fix(desktop): preserve transcript cache across catalog updates Generated-by: Codex --- .../src/main/__tests__/session-local.test.ts | 49 ++++++++++++++++++- apps/desktop/src/main/runtime-host-boot.ts | 2 +- .../desktop/src/main/session-local-service.ts | 10 ++-- 3 files changed, 55 insertions(+), 6 deletions(-) diff --git a/apps/desktop/src/main/__tests__/session-local.test.ts b/apps/desktop/src/main/__tests__/session-local.test.ts index afeb8aade61..0e6ddbbb9ed 100644 --- a/apps/desktop/src/main/__tests__/session-local.test.ts +++ b/apps/desktop/src/main/__tests__/session-local.test.ts @@ -1493,7 +1493,7 @@ test('cache restoration expires independently of the outbox', async (t) => { test('durable Host evidence retires delivery independently of cache admission and submit ACK order', async (t) => { for (const ackFirst of [false, true]) { - for (const cacheLoss of ['quota', 'revision', 'coalescing']) { + for (const cacheLoss of ['quota', 'coalescing']) { await t.test(`ackFirst=${ackFirst}, cacheLoss=${cacheLoss}`, async (t) => { const db = await database(t); const response = deferred(); @@ -1538,7 +1538,6 @@ test('durable Host evidence retires delivery independently of cache admission an beginsAtTurnBoundary: true, }; service.cacheTranscript(target.scope, snapshot); - if (cacheLoss === 'revision') db.store.enqueue('other-authority', intent('other-message')); if (cacheLoss === 'coalescing') service.cacheTranscript(target.scope, { ...snapshot, durable: [], hasOlder: true }); await nextTurn(); @@ -2008,6 +2007,52 @@ test('durable cancellation proof retires an old Host epoch without crossing auth assert.ok(db.store.get('other-authority', 'cancelled')); }); +test('catalog updates preserve the latest coalesced transcript cache write', async (t) => { + const db = await database(t); + const target: DesktopSessionLocalTarget = { + partition: 'authority', profileId: 'profile', scope: { hostId: 'root', targetEpoch: 'target' }, + }; + const service = new DesktopSessionLocalService(db.store, { targets: () => [target], changed() {}, onError: assert.fail }); + db.beforeClose.push(() => service.close()); + const snapshot: DesktopTranscriptReplicaSnapshot = { + sessionId: 'session-1', generation: 'generation', hostEpoch: 'epoch', durableThrough: 1, + durable: [{ sequence: 1, message: { type: 'user', id: 'user', turnId: 'turn-1', ts: 1, text: 'question' } }], + hasOlder: false, beginsAtTurnBoundary: true, + }; + service.cacheTranscript(target.scope, snapshot); + service.cacheTranscript(target.scope, { + ...snapshot, durableThrough: 2, + durable: [...snapshot.durable, { sequence: 2, message: { type: 'assistant', id: 'answer', turnId: 'turn-1', ts: 2, text: 'cached answer', modelId: 'test' } }], + }); + db.store.saveSession('authority', { id: 'session-1', name: 'Renamed' } as DesktopSessionSummaryInput); + await nextTurn(); + service.close(); + db.reopen(); + assert.equal(db.store.transcript('authority', 'session-1')?.snapshot.durable.at(-1)?.message.id, 'answer'); +}); + +test('Session removal and authority purge cancel pending transcript cache writes', async (t) => { + for (const removal of ['session', 'authority']) { + const db = await database(t); + const target: DesktopSessionLocalTarget = { + partition: 'authority', profileId: 'profile', scope: { hostId: 'root', targetEpoch: 'target' }, + }; + const service = new DesktopSessionLocalService(db.store, { targets: () => [target], changed() {}, onError: assert.fail }); + db.beforeClose.push(() => service.close()); + service.cacheTranscript(target.scope, { + sessionId: 'session-1', generation: 'generation', hostEpoch: 'epoch', durableThrough: 1, + durable: [{ sequence: 1, message: { type: 'user', id: 'user', turnId: 'turn-1', ts: 1, text: 'removed history' } }], + hasOlder: false, beginsAtTurnBoundary: true, + }); + if (removal === 'session') service.removeSession(target, 'session-1'); + else service.purge(target); + await nextTurn(); + service.close(); + db.reopen(); + assert.equal(db.store.transcript('authority', 'session-1'), undefined); + } +}); + test('cancellation cleanup preserves an already scheduled canonical transcript cache write', async (t) => { for (const epoch of ['epoch', 'old-epoch']) { const db = await database(t); diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index 8dbc10b90c2..3df1a5d7727 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -2108,7 +2108,7 @@ function emitSessionsChanged( ): void { sessionLocal.changed(scope); if (reason === 'deleted' && sessionId) { - try { sessionLocalStore.removeSession(sessionLocal.target(scope).partition, sessionId); } catch { /* A retired target cannot repopulate its cache. */ } + try { sessionLocal.removeSession(sessionLocal.target(scope), sessionId); } catch { /* A retired target cannot repopulate its cache. */ } } const event: SessionChangedEvent = { reason, diff --git a/apps/desktop/src/main/session-local-service.ts b/apps/desktop/src/main/session-local-service.ts index e135e7d40b9..c5e650478ed 100644 --- a/apps/desktop/src/main/session-local-service.ts +++ b/apps/desktop/src/main/session-local-service.ts @@ -506,6 +506,11 @@ export class DesktopSessionLocalService { this.#drainCatalogRefreshes(); } + removeSession(target: DesktopSessionLocalTarget, sessionId: string): void { + this.#snapshots.delete(`${target.partition}:${sessionId}`); + this.store.removeSession(target.partition, sessionId); + } + cacheTranscript(scope: DesktopTargetScope, snapshot: DesktopTranscriptReplicaSnapshot): void { if (this.#closed) return; let target: DesktopSessionLocalTarget; @@ -529,11 +534,10 @@ export class DesktopSessionLocalService { const pending = this.#snapshots.has(key); this.#snapshots.set(key, { target, snapshot }); if (pending) return; - const revision = this.store.revision; setImmediate(() => { const latest = this.#snapshots.get(key); this.#snapshots.delete(key); - if (!latest || !this.#current(latest.target) || revision !== this.store.revision) return; + if (!latest || !this.#current(latest.target)) return; try { this.store.saveTranscript(latest.target.partition, latest.snapshot); } catch (error) { @@ -954,7 +958,7 @@ export function registerDesktopSessionLocalIpc(deps: { .some((record) => record.intent.originHostEpoch) ) throw new Error('Session has a dispatched message'); - service.store.removeSession(target.partition, sessionId); + service.removeSession(target, sessionId); deps.changed(target.scope, sessionId); return true; }); From fe1172397e1c208c0e2a918af2b1e2a07d40b6c6 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 16 Oct 2026 21:40:41 +1100 Subject: [PATCH 6/6] fix(desktop): cancel catalog-removed transcript writes Generated-by: Codex --- .../src/main/__tests__/session-local.test.ts | 30 +++++++++++++++++++ .../desktop/src/main/session-local-service.ts | 3 +- apps/desktop/src/main/session-local-store.ts | 10 +++++-- 3 files changed, 39 insertions(+), 4 deletions(-) diff --git a/apps/desktop/src/main/__tests__/session-local.test.ts b/apps/desktop/src/main/__tests__/session-local.test.ts index 0e6ddbbb9ed..7b08b39d976 100644 --- a/apps/desktop/src/main/__tests__/session-local.test.ts +++ b/apps/desktop/src/main/__tests__/session-local.test.ts @@ -2031,6 +2031,36 @@ test('catalog updates preserve the latest coalesced transcript cache write', asy assert.equal(db.store.transcript('authority', 'session-1')?.snapshot.durable.at(-1)?.message.id, 'answer'); }); +test('catalog reconciliation cancels only removed Sessions pending cache writes across restart', async (t) => { + const db = await database(t); + const listed = deferred(); + const target: DesktopSessionLocalTarget = { + partition: 'authority', profileId: 'profile', scope: { hostId: 'root', targetEpoch: 'target' }, + client: { ...client('epoch'), listSessions: () => listed.promise }, + }; + const service = new DesktopSessionLocalService(db.store, { targets: () => [target], changed() {}, onError: assert.fail }); + db.beforeClose.push(() => service.close()); + for (const id of ['removed', 'retained']) + db.store.saveSession('authority', { id, name: id } as DesktopSessionSummaryInput); + service.catalog(); + for (const id of ['removed', 'retained']) { + service.cacheTranscript(target.scope, { + sessionId: id, generation: 'generation', hostEpoch: 'epoch', durableThrough: 1, + durable: [{ sequence: 1, message: { type: 'assistant', id: `answer-${id}`, turnId: 'turn-1', ts: 1, text: `cached ${id}`, modelId: 'test' } }], + hasOlder: false, beginsAtTurnBoundary: true, + }); + } + listed.resolve([{ ...swarmCatalogSession(), id: 'retained', name: 'Retained' }]); + await nextTurn(); + assert.deepEqual(db.store.sessions('authority').map(({ id }) => id), ['retained']); + assert.equal(service.readTranscript(target, 'removed'), null); + assert.ok(service.readTranscript(target, 'retained')); + service.close(); + db.reopen(); + assert.equal(db.store.transcript('authority', 'removed'), undefined); + assert.equal(db.store.transcript('authority', 'retained')?.snapshot.durable[0]?.message.id, 'answer-retained'); +}); + test('Session removal and authority purge cancel pending transcript cache writes', async (t) => { for (const removal of ['session', 'authority']) { const db = await database(t); diff --git a/apps/desktop/src/main/session-local-service.ts b/apps/desktop/src/main/session-local-service.ts index c5e650478ed..7d1eed1943a 100644 --- a/apps/desktop/src/main/session-local-service.ts +++ b/apps/desktop/src/main/session-local-service.ts @@ -452,7 +452,8 @@ export class DesktopSessionLocalService { locallyInvalidated = true; return; } - this.store.saveCatalog(target.partition, sessions.map(toDesktopHostSessionSummary)); + const removed = this.store.saveCatalog(target.partition, sessions.map(toDesktopHostSessionSummary)); + for (const sessionId of removed) this.#snapshots.delete(`${target.partition}:${sessionId}`); // Publish successful observations even under continuous Host events. // The captured version keeps this observation dirty if another change // arrived while reading, so one trailing refresh can converge. diff --git a/apps/desktop/src/main/session-local-store.ts b/apps/desktop/src/main/session-local-store.ts index 3cb3e91f78b..f036dfa0264 100644 --- a/apps/desktop/src/main/session-local-store.ts +++ b/apps/desktop/src/main/session-local-store.ts @@ -338,14 +338,18 @@ export class DesktopSessionLocalStore { return row?.creation ? (JSON.parse(String(row.creation)) as SessionCreateInput) : undefined; } - saveCatalog(partition: string, summaries: readonly DesktopSessionSummaryInput[]): void { - this.#transaction(() => { + saveCatalog(partition: string, summaries: readonly DesktopSessionSummaryInput[]): string[] { + return this.#transaction(() => { + const removed: string[] = []; const seen = new Set(summaries.map((summary) => summary.id)); for (const summary of summaries) this.saveSession(partition, summary); for (const previous of this.sessions(partition)) { - if (!seen.has(previous.id) && !this.creation(partition, previous.id)) + if (!seen.has(previous.id) && !this.creation(partition, previous.id)) { this.removeSession(partition, previous.id); + removed.push(previous.id); + } } + return removed; }); }