diff --git a/.maka-shots/workhub-reconstruction-after.png b/.maka-shots/workhub-reconstruction-after.png new file mode 100644 index 0000000000..894851a282 Binary files /dev/null and b/.maka-shots/workhub-reconstruction-after.png differ diff --git a/.maka-shots/workhub-reconstruction-before.png b/.maka-shots/workhub-reconstruction-before.png new file mode 100644 index 0000000000..4df6745f8a Binary files /dev/null and b/.maka-shots/workhub-reconstruction-before.png differ diff --git a/apps/desktop/e2e/workhub-reconstruction.spec.ts b/apps/desktop/e2e/workhub-reconstruction.spec.ts new file mode 100644 index 0000000000..d81dec3861 --- /dev/null +++ b/apps/desktop/e2e/workhub-reconstruction.spec.ts @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { expect, test, COMPOSER_INPUT } from './fixtures'; + +test('WorkHub rebuilds Session conversation after navigating away and back', async ({ + window: page, +}) => { + const initialPrompt = '检查支付回调重复投递时的幂等性'; + const composer = page.locator(COMPOSER_INPUT); + await composer.fill(initialPrompt); + await composer.press('Enter'); + await expect(page.getByRole('button', { name: '重新生成' })).toHaveCount(1, { + timeout: 20_000, + }); + + const sessionName = await page.evaluate(async () => + (await window.maka.sessions.list())[0]?.name, + ); + expect(sessionName).toBeTruthy(); + await page.evaluate(async () => { + await window.maka.settings.updateClient({ workHub: { enabled: true } }); + }); + await expect(page.getByRole('main', { name: 'WorkHub' })).toBeVisible(); + await expect( + page.locator('.workhub-projected-turn .workhub-user-bubble > p', { + hasText: initialPrompt, + }), + ).toBeVisible(); + + const routedPrompt = `继续${sessionName}，补充重复投递测试点。`; + const workHubComposer = page.locator( + '.workhub-surface .maka-composer-editor [contenteditable="true"]', + ); + await workHubComposer.fill(routedPrompt); + await workHubComposer.press('Enter'); + await expect(page.locator('.workhub-submitted').last()).toBeVisible(); + await page.locator('.workhub-submitted > button').last().click(); + await expect(page.getByRole('main', { name: 'WorkHub' })).toBeHidden(); + + await page.getByRole('button', { name: 'WorkHub', exact: true }).click(); + await expect(page.getByRole('main', { name: 'WorkHub' })).toBeVisible(); + await expect( + page.locator('.workhub-projected-turn .workhub-user-bubble > p', { + hasText: routedPrompt, + }), + ).toBeVisible(); +}); diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index fc8122cb1e..509c131797 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -49,6 +49,7 @@ function session( function port(sessions: WorkHubSessionFacts[]): WorkHubSessionPort { return { list: async () => sessions, + recentTurns: async () => [], routingEvidence: async () => [], create: async () => { throw new Error('create is not used by this read test'); @@ -104,6 +105,40 @@ test('read exposes existing ordinary Sessions as factual Work summaries', async updatedAt: 20, }, ]); + assert.deepEqual(projection.turns, []); +}); + +test('read rebuilds a bounded conversation projection from ordinary Session turns', async () => { + const sessions = port([ + session('login', { sessionName: '登录刷新令牌', updatedAt: 30 }), + session('internal', { kind: 'internal', updatedAt: 40 }), + ]); + const requestedTargets: string[][] = []; + sessions.recentTurns = async (targets) => { + requestedTargets.push(targets.map((target) => target.sessionId)); + return [{ + messageId: 'user-1', + target: { sessionId: 'login' }, + turnId: 'turn-login', + text: '检查刷新令牌竞争条件', + state: 'completed', + result: '已定位到并发刷新窗口', + updatedAt: 20, + }]; + }; + + const projection = await createWorkHubController({ sessions }).read(); + + assert.deepEqual(requestedTargets, [['login']]); + assert.deepEqual(projection.turns, [{ + messageId: 'user-1', + target: { sessionId: 'login' }, + turnId: 'turn-login', + text: '检查刷新令牌竞争条件', + state: 'completed', + result: '已定位到并发刷新窗口', + updatedAt: 20, + }]); }); test('archived Sessions stay inspectable but are excluded from routing targets', async () => { diff --git a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts index bb31829e9c..e96bebb817 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -19,8 +19,12 @@ import assert from 'node:assert/strict'; import test from 'node:test'; +import type { StoredMessage } from '@maka/core/session'; +import type { DesktopTranscriptBatch } from '../../preload/transcript-contract.js'; +import { desktopSessionKey } from '../../shared/runtime-host-identity.js'; import { createDesktopWorkHubSessionPort, + projectWorkHubSessionTurns, type WorkHubDesktopSession, } from '../../renderer/workhub-session-port.js'; @@ -41,6 +45,222 @@ function desktopSession( }; } +const unusedTranscripts = { + open: async () => { + throw new Error('transcript is not used by this test'); + }, +}; + +test('projects durable Session messages into an ordered WorkHub conversation', () => { + const turns = projectWorkHubSessionTurns({ + target: { sessionId: 'payment' }, + messages: [ + { type: 'user', id: 'user-1', turnId: 'turn-1', ts: 10, text: '检查重复投递' }, + { + type: 'assistant', + id: 'assistant-1', + turnId: 'turn-1', + ts: 11, + text: '已定位风险', + modelId: 'test-model', + }, + { type: 'user', id: 'user-2', turnId: 'turn-1', ts: 12, text: '再补充测试点' }, + { + type: 'assistant', + id: 'assistant-2', + turnId: 'turn-1', + ts: 13, + text: '已补充测试点', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'state-1', + turnId: 'turn-1', + ts: 14, + status: 'completed', + partialOutputRetained: true, + }, + ], + }); + + assert.deepEqual(turns, [ + { + messageId: 'user-1', + target: { sessionId: 'payment' }, + turnId: 'turn-1', + text: '检查重复投递', + state: 'completed', + result: '已定位风险', + updatedAt: 10, + }, + { + messageId: 'user-2', + target: { sessionId: 'payment' }, + turnId: 'turn-1', + text: '再补充测试点', + state: 'completed', + result: '已补充测试点', + updatedAt: 12, + }, + ]); +}); + +test('desktop adapter rebuilds recent turns from the Session transcript and closes the read', async () => { + const sessionId = desktopSessionKey({ hostId: 'local-host', sessionId: 'payment' }); + const messages: StoredMessage[] = [ + { type: 'user', id: 'user-1', turnId: 'turn-1', ts: 10, text: '检查重复投递' }, + { + type: 'assistant', + id: 'assistant-1', + turnId: 'turn-1', + ts: 11, + text: '已定位风险', + modelId: 'test-model', + }, + { + type: 'turn_state', + id: 'state-1', + turnId: 'turn-1', + ts: 12, + status: 'completed', + partialOutputRetained: true, + }, + ]; + let closes = 0; + const adapter = createDesktopWorkHubSessionPort({ + sessions: { + list: async () => [], + listTurns: async () => [], + create: async () => { + throw new Error('not used'); + }, + send: async () => { + throw new Error('not used'); + }, + stop: async () => {}, + subscribeChanges: () => () => {}, + }, + transcripts: { + open: async (requestedSessionId, handler) => { + assert.equal(requestedSessionId, sessionId); + const fragments = messages.map((message, sequence) => { + const data = new TextEncoder().encode(JSON.stringify(message)); + return { + source: 'durable' as const, + identity: sequence, + order: null, + byteOffset: 0, + totalBytes: data.byteLength, + data, + }; + }); + handler({ + sessionId: 'payment', + deliverySequence: 1, + generation: 'generation-1', + hostEpoch: 'epoch-1', + durableThrough: 2, + fragments, + evictedDurableSequences: [], + completedOverlayMessageIds: [], + hasOlder: false, + hasNewer: false, + reset: true, + ready: true, + } satisfies DesktopTranscriptBatch); + return { + sessionId, + generation: 'generation-1', + hostEpoch: 'epoch-1', + readThroughMessageId: null, + loadBefore: async () => {}, + loadAround: async () => {}, + close: async () => { + closes += 1; + }, + }; + }, + }, + projectName: () => 'Maka', + newTurnId: () => 'unused', + }); + + assert.deepEqual(await adapter.recentTurns([{ sessionId }]), [{ + messageId: 'user-1', + target: { sessionId }, + turnId: 'turn-1', + text: '检查重复投递', + state: 'completed', + result: '已定位风险', + updatedAt: 10, + }]); + assert.equal(closes, 1); +}); + +test('desktop adapter cancels an unavailable transcript without hiding ready Sessions', async (t) => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + const unavailableId = desktopSessionKey({ hostId: 'local-host', sessionId: 'unavailable' }); + const readyId = desktopSessionKey({ hostId: 'local-host', sessionId: 'ready' }); + let cancellations = 0; + const adapter = createDesktopWorkHubSessionPort({ + sessions: { + list: async () => [], + listTurns: async () => [], + create: async () => { throw new Error('not used'); }, + send: async () => { throw new Error('not used'); }, + stop: async () => {}, + subscribeChanges: () => () => {}, + }, + transcripts: { + open: async (sessionId, handler, registerCancellation) => { + if (sessionId === unavailableId) { + return await new Promise((_resolve, reject) => { + registerCancellation?.(() => { + cancellations += 1; + reject(new Error('cancelled unavailable transcript')); + }); + }); + } + const message: StoredMessage = { + type: 'user', id: 'user-ready', turnId: 'turn-ready', ts: 10, text: '可用工作', + }; + const data = new TextEncoder().encode(JSON.stringify(message)); + handler({ + sessionId: 'ready', deliverySequence: 1, generation: 'generation-ready', + hostEpoch: 'epoch-ready', durableThrough: 0, + fragments: [{ + source: 'durable', identity: 0, order: null, byteOffset: 0, + totalBytes: data.byteLength, data, + }], + evictedDurableSequences: [], completedOverlayMessageIds: [], + hasOlder: false, hasNewer: false, reset: true, ready: true, + }); + return { + sessionId: readyId, generation: 'generation-ready', hostEpoch: 'epoch-ready', + readThroughMessageId: null, loadBefore: async () => {}, loadAround: async () => {}, + close: async () => {}, + }; + }, + }, + projectName: () => 'Maka', + newTurnId: () => 'unused', + }); + + const turns = adapter.recentTurns([ + { sessionId: unavailableId }, + { sessionId: readyId }, + ]); + await Promise.resolve(); + t.mock.timers.tick(5_000); + + assert.deepEqual(await turns, [{ + messageId: 'user-ready', target: { sessionId: readyId }, turnId: 'turn-ready', + text: '可用工作', state: 'completed', updatedAt: 10, + }]); + assert.equal(cancellations, 1); +}); + test('desktop adapter projects Session catalog facts without owning copies', async () => { const source = [ desktopSession('ordinary', { @@ -65,6 +285,7 @@ test('desktop adapter projects Session catalog facts without owning copies', asy }), ]; const adapter = createDesktopWorkHubSessionPort({ + transcripts: unusedTranscripts, sessions: { list: async () => source, listTurns: async () => [], @@ -126,6 +347,7 @@ test('desktop adapter delegates create, send, and invalidation to Session APIs', const calls: unknown[] = []; let onChanged: (() => void) | undefined; const adapter = createDesktopWorkHubSessionPort({ + transcripts: unusedTranscripts, sessions: { list: async () => [desktopSession('created', { status: 'running', @@ -175,6 +397,7 @@ test('desktop adapter delegates create, send, and invalidation to Session APIs', test('desktop adapter preserves when Session delivery steered an existing root Turn', async () => { const adapter = createDesktopWorkHubSessionPort({ + transcripts: unusedTranscripts, sessions: { list: async () => [], listTurns: async () => [], @@ -202,6 +425,7 @@ test('desktop adapter preserves when Session delivery steered an existing root T test('desktop adapter binds stop to the root Turn owned by the WorkHub submission', async () => { const stopped: unknown[] = []; const adapter = createDesktopWorkHubSessionPort({ + transcripts: unusedTranscripts, sessions: { list: async () => [], listTurns: async () => [], @@ -231,6 +455,7 @@ test('desktop adapter binds stop to the root Turn owned by the WorkHub submissio test('desktop adapter derives stable origin evidence from the existing Session log', async () => { let reads = 0; const adapter = createDesktopWorkHubSessionPort({ + transcripts: unusedTranscripts, sessions: { list: async () => [], listTurns: async (sessionId) => { diff --git a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts index f58d44c029..dc6479c082 100644 --- a/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-surface-flow.test.ts @@ -20,12 +20,16 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { + WorkHubProjectionRefreshGate, WorkHubSurfaceRouteGate, + projectedWorkHubTurnPresentation, submitWorkHubSurfaceInput, + visibleWorkHubProjectedTurns, workHubSubmissionCanCorrect, workHubSubmissionClearsDraft, } from '../../renderer/workhub-surface.js'; import { + boundedWorkHubTimelineText, createWorkHubController, type WorkHubController, type WorkHubSubmitInput, @@ -53,6 +57,24 @@ test('surface route gate rejects same-frame duplicate operations and reopens aft assert.equal(await gate.run(async () => 'next'), 'next'); }); +test('surface projection refresh gate rejects older reads after a newer refresh starts', () => { + const gate = new WorkHubProjectionRefreshGate(); + const first = gate.begin(); + const second = gate.begin(); + + assert.equal(first(), false); + assert.equal(second(), true); + gate.invalidate(); + assert.equal(second(), false); +}); + +test('projected archived Session keeps the actual turn state visible', () => { + assert.deepEqual(projectedWorkHubTurnPresentation('failed', true, 'zh'), { + heading: '来自已归档 Session：', + state: '失败', + }); +}); + test('surface keeps the Composer draft when routing fails or the target is waiting', () => { assert.equal(workHubSubmissionClearsDraft(undefined), false); assert.equal(workHubSubmissionClearsDraft({ @@ -84,10 +106,108 @@ test('surface disables correction after a request was steered into existing work assert.equal(workHubSubmissionCanCorrect({ ...submission, steered: true }), false); }); +test('surface hides a rebuilt Session turn while the matching local turn is still mounted', () => { + assert.deepEqual( + visibleWorkHubProjectedTurns( + [{ + messageId: 'user-0', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + text: '检查支付回调风险', + state: 'running', + updatedAt: 9, + }, { + messageId: 'user-1', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + text: '补充重复投递测试', + state: 'completed', + updatedAt: 10, + }], + [{ + requestId: 'request-payment', + text: '补充重复投递测试', + state: 'settled', + outcome: { + kind: 'submitted', + strategyId: 'wh-r2.3-session-core-evidence', + requestId: 'request-payment', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + evidence: 'explicit_target', + }, + }], + ), + [{ + messageId: 'user-0', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + text: '检查支付回调风险', + state: 'running', + updatedAt: 9, + }], + ); +}); + +test('surface canonicalizes bounded text before suppressing a local duplicate', () => { + const text = '长'.repeat(700); + assert.deepEqual(visibleWorkHubProjectedTurns([{ + messageId: 'user-long', + target: { sessionId: 'payment' }, + turnId: 'turn-long', + text: boundedWorkHubTimelineText(text), + state: 'running', + updatedAt: 10, + }], [{ + requestId: 'request-long', + text, + state: 'settled', + outcome: { + kind: 'submitted', + strategyId: 'wh-r2.3-session-core-evidence', + requestId: 'request-long', + target: { sessionId: 'payment' }, + turnId: 'turn-long', + evidence: 'explicit_target', + }, + }]), []); +}); + +test('surface suppresses the newest matching projected steering turn', () => { + const projected = [{ + messageId: 'user-old', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + text: '继续检查', + state: 'running' as const, + updatedAt: 9, + }, { + messageId: 'user-new', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + text: '继续检查', + state: 'running' as const, + updatedAt: 10, + }]; + assert.deepEqual(visibleWorkHubProjectedTurns(projected, [{ + requestId: 'request-new', + text: '继续检查', + state: 'settled', + outcome: { + kind: 'submitted', + strategyId: 'wh-r2.3-session-core-evidence', + requestId: 'request-new', + target: { sessionId: 'payment' }, + turnId: 'turn-payment', + evidence: 'explicit_target', + }, + }]), [projected[0]]); +}); + test('surface keeps clarification and successful routing in WorkHub', async () => { const submissions: WorkHubSubmitInput[] = []; const controller: WorkHubController = { - read: async () => ({ sessions: [] }), + read: async () => ({ sessions: [], turns: [] }), subscribe: () => () => {}, submit: async (input) => { submissions.push(input); @@ -135,7 +255,7 @@ test('surface keeps clarification and successful routing in WorkHub', async () = test('surface leaves discussion in WorkHub instead of creating a task view', async () => { const controller: WorkHubController = { - read: async () => ({ sessions: [] }), + read: async () => ({ sessions: [], turns: [] }), subscribe: () => () => {}, submit: async (input) => ({ kind: 'discussion', @@ -171,6 +291,11 @@ test('real Session projection creates new guide topics and preserves origin ambi ]]); const created: string[] = []; const port = createDesktopWorkHubSessionPort({ + transcripts: { + open: async () => { + throw new Error('transcript is not used by this routing test'); + }, + }, sessions: { list: async () => sessions, listTurns: async (sessionId) => (prompts.get(sessionId) ?? []) diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index e1059c625e..e3509f8409 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -1648,6 +1648,7 @@ function AppShellContent({ const workHubController = useMemo(() => createWorkHubController({ sessions: createDesktopWorkHubSessionPort({ sessions: window.maka.sessions, + transcripts: window.maka.transcripts, projectName: (projectId) => projects.find((project) => project.id === projectId)?.name, newTurnId: () => crypto.randomUUID(), }), diff --git a/apps/desktop/src/renderer/workhub-controller.ts b/apps/desktop/src/renderer/workhub-controller.ts index 781410f710..514d47794d 100644 --- a/apps/desktop/src/renderer/workhub-controller.ts +++ b/apps/desktop/src/renderer/workhub-controller.ts @@ -53,8 +53,31 @@ export interface WorkHubSessionFacts { export type WorkHubSessionSummary = Omit; +export type WorkHubProjectedTurnState = 'running' | 'completed' | 'aborted' | 'failed'; + +export interface WorkHubProjectedTurn { + messageId: string; + target: WorkHubSessionTarget; + turnId: string; + text: string; + state: WorkHubProjectedTurnState; + result?: string; + updatedAt: number; +} + +const WORKHUB_TIMELINE_TEXT_LIMIT = 600; + +export function boundedWorkHubTimelineText(value: string): string { + const text = value.trim(); + const chars = Array.from(text); + return chars.length <= WORKHUB_TIMELINE_TEXT_LIMIT + ? text + : `${chars.slice(0, WORKHUB_TIMELINE_TEXT_LIMIT - 1).join('')}…`; +} + export interface WorkHubProjection { sessions: WorkHubSessionSummary[]; + turns: WorkHubProjectedTurn[]; } export interface WorkHubSubmitInput { @@ -106,6 +129,11 @@ export type WorkHubSubmission = ( */ export interface WorkHubSessionPort { list(): Promise; + /** + * Rebuilds a bounded recent conversation from the authoritative Session + * transcripts. Missing transcripts are omitted rather than copied elsewhere. + */ + recentTurns(targets: readonly WorkHubSessionTarget[]): Promise; /** * Returns rebuildable routing evidence read from the authoritative Session * log. Implementations must not persist a second writable copy of it. @@ -138,11 +166,13 @@ export function createWorkHubController(deps: { }, async read() { const facts = await deps.sessions.list(); + const ordinary = facts + .filter((session) => session.kind === 'ordinary') + .sort((left, right) => right.updatedAt - left.updatedAt); return { - sessions: facts - .filter((session) => session.kind === 'ordinary') - .sort((left, right) => right.updatedAt - left.updatedAt) + sessions: ordinary .map(({ kind: _kind, ...session }) => session), + turns: await deps.sessions.recentTurns(ordinary.map((session) => session.target)), }; }, async submit(input) { diff --git a/apps/desktop/src/renderer/workhub-session-port.ts b/apps/desktop/src/renderer/workhub-session-port.ts index 2a1218524c..47234d0b37 100644 --- a/apps/desktop/src/renderer/workhub-session-port.ts +++ b/apps/desktop/src/renderer/workhub-session-port.ts @@ -17,12 +17,20 @@ * under the License. */ +import { deriveTurnRecords, type StoredMessage } from '@maka/core/session'; import type { + DesktopTranscriptBatch, + DesktopTranscriptHandle, +} from '../preload/transcript-contract.js'; +import { DesktopTranscriptRangeStore } from './desktop-transcript-range-store.js'; +import type { + WorkHubProjectedTurn, WorkHubSessionFacts, WorkHubSessionPort, WorkHubSessionState, WorkHubSessionTarget, } from './workhub-controller.js'; +import { boundedWorkHubTimelineText } from './workhub-controller.js'; export interface WorkHubDesktopSession { id: string; @@ -56,8 +64,21 @@ export interface WorkHubDesktopSessionBridge { subscribeChanges(handler: () => void): () => void; } +export interface WorkHubDesktopTranscriptBridge { + open( + sessionId: string, + handler: (batch: DesktopTranscriptBatch) => void, + registerCancellation?: (cancel: () => void) => void, + ): Promise; +} + +const WORKHUB_TIMELINE_SESSION_LIMIT = 10; +const WORKHUB_TIMELINE_TURN_LIMIT = 40; +const WORKHUB_TRANSCRIPT_READY_TIMEOUT_MS = 5_000; + export function createDesktopWorkHubSessionPort(deps: { sessions: WorkHubDesktopSessionBridge; + transcripts: WorkHubDesktopTranscriptBridge; projectName(projectId: string): string | undefined; newTurnId(): string; }): WorkHubSessionPort { @@ -93,6 +114,28 @@ export function createDesktopWorkHubSessionPort(deps: { async list() { return (await deps.sessions.list()).map(projectSession); }, + async recentTurns(targets) { + const turnsBySession = await Promise.all( + targets.slice(0, WORKHUB_TIMELINE_SESSION_LIMIT).map(async (target) => { + try { + const messages = await readWorkHubSessionMessages(deps.transcripts, target); + return projectWorkHubSessionTurns({ target, messages }); + } catch { + // One unavailable transcript must not hide the other Sessions or + // turn WorkHub into a second recovery authority. + return []; + } + }), + ); + return turnsBySession + .flat() + .sort((left, right) => + left.updatedAt - right.updatedAt || + left.target.sessionId.localeCompare(right.target.sessionId) || + left.messageId.localeCompare(right.messageId), + ) + .slice(-WORKHUB_TIMELINE_TURN_LIMIT); + }, async routingEvidence(targets) { return Promise.all(targets.map(async (target) => { const cached = originPromptCache.get(target.sessionId); @@ -139,6 +182,86 @@ export function createDesktopWorkHubSessionPort(deps: { }; } +export function projectWorkHubSessionTurns(input: { + target: WorkHubSessionTarget; + messages: readonly StoredMessage[]; +}): WorkHubProjectedTurn[] { + const stateByTurnId = new Map( + deriveTurnRecords(input.messages).map((turn) => [turn.turnId, turn.status]), + ); + const turns: WorkHubProjectedTurn[] = []; + const latestUserIndexByTurnId = new Map(); + + for (const message of input.messages) { + if (message.type === 'user') { + const text = boundedWorkHubTimelineText(message.displayText ?? message.text); + if (!text) continue; + const state = stateByTurnId.get(message.turnId) ?? 'completed'; + turns.push({ + messageId: message.id, + target: input.target, + turnId: message.turnId, + text, + state, + updatedAt: message.ts, + }); + latestUserIndexByTurnId.set(message.turnId, turns.length - 1); + continue; + } + if (message.type !== 'assistant') continue; + const result = boundedWorkHubTimelineText(message.text); + if (!result) continue; + const userIndex = latestUserIndexByTurnId.get(message.turnId); + if (userIndex === undefined) continue; + turns[userIndex] = { ...turns[userIndex]!, result }; + } + + return turns; +} + +async function readWorkHubSessionMessages( + transcripts: WorkHubDesktopTranscriptBridge, + target: WorkHubSessionTarget, +): Promise { + const store = new DesktopTranscriptRangeStore(target.sessionId); + let resolveReady: ((messages: readonly StoredMessage[]) => void) | undefined; + const ready = new Promise((resolve) => { + resolveReady = resolve; + }); + let cancelOpen = () => {}; + let timedOut = false; + let handle: DesktopTranscriptHandle | undefined; + let rejectTimeout!: (error: Error) => void; + const timeoutFailure = new Promise((_resolve, reject) => { + rejectTimeout = reject; + }); + void timeoutFailure.catch(() => undefined); + const timeout = globalThis.setTimeout(() => { + timedOut = true; + cancelOpen(); + rejectTimeout(new Error('WorkHub Session transcript did not become ready')); + }, WORKHUB_TRANSCRIPT_READY_TIMEOUT_MS); + const opening = transcripts.open( + target.sessionId, + (batch) => { + store.accept(batch); + if (batch.ready) resolveReady?.(store.snapshot().messages); + }, + (cancel) => { + cancelOpen = cancel; + if (timedOut) cancel(); + }, + ); + void opening.catch(() => undefined); + try { + handle = await Promise.race([opening, timeoutFailure]); + return await Promise.race([ready, timeoutFailure]); + } finally { + globalThis.clearTimeout(timeout); + await handle?.close().catch(() => undefined); + } +} + function projectState(session: WorkHubDesktopSession): WorkHubSessionState { // A root Turn can remain live while it is blocked on a user interaction. // WorkHub must surface that interaction boundary before the broader running diff --git a/apps/desktop/src/renderer/workhub-surface.tsx b/apps/desktop/src/renderer/workhub-surface.tsx index d569784c7d..de24b0a3f1 100644 --- a/apps/desktop/src/renderer/workhub-surface.tsx +++ b/apps/desktop/src/renderer/workhub-surface.tsx @@ -17,7 +17,7 @@ * under the License. */ -import { useCallback, useEffect, useRef, useState } from 'react'; +import { useCallback, useEffect, useRef, useState, type ReactNode } from 'react'; import { ChatMessage, ChatMessageBubble, ChatMessageList } from '@astryxdesign/core'; import { Button } from '@astryxdesign/core/Button'; import type { UiLocale } from '@maka/core/ui-locale'; @@ -25,12 +25,14 @@ import { ChatSurfaceLayout, Composer } from '@maka/ui'; import type { WorkHubController, WorkHubProjection, + WorkHubProjectedTurn, WorkHubSessionSummary, WorkHubSubmission, WorkHubSubmitInput, } from './workhub-controller.js'; +import { boundedWorkHubTimelineText } from './workhub-controller.js'; -interface WorkHubConversationTurn { +export interface WorkHubConversationTurn { requestId: string; text: string; state: 'routing' | 'settled' | 'failed'; @@ -55,6 +57,19 @@ export class WorkHubSurfaceRouteGate { } } +export class WorkHubProjectionRefreshGate { + #generation = 0; + + begin(): () => boolean { + const generation = ++this.#generation; + return () => generation === this.#generation; + } + + invalidate(): void { + this.#generation += 1; + } +} + export function workHubSubmissionClearsDraft( result: WorkHubSubmission | undefined, ): boolean { @@ -74,6 +89,54 @@ export async function submitWorkHubSurfaceInput(input: { return input.controller.submit(input.input); } +export function visibleWorkHubProjectedTurns( + projected: readonly WorkHubProjectedTurn[], + local: readonly WorkHubConversationTurn[], +): WorkHubProjectedTurn[] { + const localTurnCounts = new Map(); + for (const turn of local) { + if (turn.outcome?.kind !== 'submitted') continue; + const key = projectedTurnKey(turn.outcome.target, turn.outcome.turnId, turn.text); + localTurnCounts.set(key, (localTurnCounts.get(key) ?? 0) + 1); + } + const visible: WorkHubProjectedTurn[] = []; + for (let index = projected.length - 1; index >= 0; index -= 1) { + const turn = projected[index]!; + const key = projectedTurnKey(turn.target, turn.turnId, turn.text); + const remaining = localTurnCounts.get(key) ?? 0; + if (remaining > 0) { + localTurnCounts.set(key, remaining - 1); + } else { + visible.push(turn); + } + } + return visible.reverse(); +} + +function projectedTurnKey( + target: { sessionId: string }, + turnId: string, + text: string, +): string { + return JSON.stringify([ + target.sessionId, + turnId, + boundedWorkHubTimelineText(text), + ]); +} + +export function projectedWorkHubTurnPresentation( + state: WorkHubProjectedTurn['state'], + archived: boolean, + locale: UiLocale, +): { heading: string; state: string } { + const copy = workHubCopy(locale); + return { + heading: archived ? copy.archivedSessionRecord : copy.sessionRecord, + state: copy.turnStates[state], + }; +} + /** * A transient conversation projection over ordinary Sessions. * @@ -87,26 +150,35 @@ export function WorkHubSurface(props: { onOpenSession(sessionId: string): void; }) { const copy = workHubCopy(props.locale); - const [projection, setProjection] = useState({ sessions: [] }); + const [projection, setProjection] = useState({ sessions: [], turns: [] }); const [turns, setTurns] = useState([]); const [pending, setPending] = useState(false); // React state paints the lock; the gate closes the same-frame window before // a rerender can disable Composer and clarification controls. const routeGate = useRef(new WorkHubSurfaceRouteGate()).current; + const refreshGate = useRef(new WorkHubProjectionRefreshGate()).current; const [loadError, setLoadError] = useState(false); const refresh = useCallback(async () => { + const isLatest = refreshGate.begin(); try { - setProjection(await props.controller.read()); + const next = await props.controller.read(); + if (!isLatest()) return; + setProjection(next); setLoadError(false); } catch { + if (!isLatest()) return; setLoadError(true); } - }, [props.controller]); + }, [props.controller, refreshGate]); useEffect(() => { void refresh(); - return props.controller.subscribe(() => void refresh()); - }, [props.controller, refresh]); + const unsubscribe = props.controller.subscribe(() => void refresh()); + return () => { + refreshGate.invalidate(); + unsubscribe(); + }; + }, [props.controller, refresh, refreshGate]); const route = useCallback(async ( input: WorkHubSubmitInput, @@ -153,6 +225,8 @@ export function WorkHubSurface(props: { // ref-blocked duplicate keep the exact text available for retry. return workHubSubmissionClearsDraft(result); }, [route, routeGate]); + const projectedTurns = visibleWorkHubProjectedTurns(projection.turns, turns); + const conversationEmpty = projectedTurns.length === 0 && turns.length === 0; return ( {loadError ? (
{copy.loadFailed}
- ) : turns.length === 0 ? ( + ) : conversationEmpty ? (

{copy.emptyTitle}

{copy.emptyBody(projection.sessions.length)}

) : (
+ {projectedTurns.map((turn) => ( + + ))} {turns.map((turn) => ( ; + onOpenSession(sessionId: string): void; +}) { + const session = props.projection.sessions.find( + (candidate) => candidate.target.sessionId === props.turn.target.sessionId, + ); + const presentation = projectedWorkHubTurnPresentation( + props.turn.state, + session?.archived ?? false, + props.copy.locale, + ); + return ( + + + + ); +} + function WorkHubTurnView(props: { turn: WorkHubConversationTurn; projection: WorkHubProjection; @@ -247,14 +362,7 @@ function WorkHubTurnView(props: { : undefined; return ( -
- - -

{turn.text}

- - - - + {turn.state === 'routing' ? (

{copy.routing}

) : turn.state === 'failed' ? ( @@ -299,6 +407,11 @@ function WorkHubTurnView(props: { ) : undefined} targetSessionId={submitted.target.sessionId} + heading={copy.sentTo} + state={target + ? (target.archived ? copy.archived : copy.states[target.state]) + : copy.accepted} + result={target?.latestResult} copy={copy} correctionOptions={workHubSubmissionCanCorrect(submitted) ? props.projection.sessions.filter( @@ -312,6 +425,29 @@ function WorkHubTurnView(props: { onOpenSession={props.onOpenSession} /> ) : null} + + ); +} + +function WorkHubMessageFrame(props: { + text: string; + state: string; + projected?: boolean; + children: ReactNode; +}) { + return ( +
+ + +

{props.text}

+ + + + + {props.children}
@@ -322,23 +458,25 @@ function SubmittedWorkView(props: { session: WorkHubSessionSummary | undefined; correctedFrom: WorkHubSessionSummary | undefined; targetSessionId: string; + heading: string; + state: string; + result: string | undefined; copy: ReturnType; correctionOptions: WorkHubSessionSummary[]; pending: boolean; - onCorrect(target: { sessionId: string }): void; + onCorrect?(target: { sessionId: string }): void; onOpenSession(sessionId: string): void; }) { const { session, copy } = props; - const state = session ? (session.archived ? copy.archived : copy.states[session.state]) : copy.accepted; return (
-

{copy.sentTo}

+

{props.heading}