feat(desktop): rebuild WorkHub from Session transcripts

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();
+});

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 () => {

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<never>((_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) => {