feat(session): add durable local delivery and isolated terminal streams

apps/desktop/e2e-budget.json

@@ -37,6 +37,10 @@
       "tests": 1,
       "electron": "needs a second real Session (Host round trip) to switch to; the focus and draft-restore halves alone would not earn a window"
     },
+    "session-local-recovery.spec.ts": {
+      "tests": 3,
+      "electron": "SQLite admission crosses preload and main, survives both renderer reload and complete Electron restart, and cached history must render when the live IPC endpoint is unavailable"
+    },
     "session-workbar.spec.ts": {
       "tests": 6,
       "electron": "Git changes re-read on window focus, terminal PTY ownership across Sessions, Side Chat's fork lifecycle, a first send that has to reach the Host, and per-Session collapse persisted across a renderer reload; the composer-usage test is renderer-only and rides along on those windows until app-shell.tsx's composer-to-workbar wiring has a story host"

apps/desktop/e2e/fixtures.ts

@@ -76,23 +76,9 @@ export async function ensureSidebarExpanded(page: Page): Promise<void> {
 }
 
 /**
- * Wait until the composer is in a state where Enter is a real submission: the
- * connections projection has produced at least one connection, the draft is
- * non-empty, no known blocker is showing and no earlier send is still in
- * flight. 发送 is disabled for all of that, so it is the one signal covering
- * it; intermediate signals (a cleared draft, an updated model label) resolve
- * earlier and mean nothing here.
- *
- * It does NOT cover the submission-readiness probe: an unresolved snapshot is
- * not a hard block, so the button is enabled while the probe is in flight, and
- * `send()` awaits the probe again on its own — a first send inside a barrier
- * that gives up at 30s. The post-send assertions wait 20s, which covers every
- * admission measured here but not that whole barrier. Widening them past it
- * buys nothing: the 60s test budget is the real cap, a send admitted at 25s
- * leaves the multi-send specs unable to finish anyway, and the only change
- * would be trading a named assertion failure for a bare test timeout. A probe
- * that comes back blocked still drops the send with no feedback, which is a
- * product gap, not something a test-side fence can close.
+ * Wait until Enter can submit this non-empty draft to its selected target.
+ * The control reflects local admission readiness, not Host connectivity.
+ * Merely mounting the editor does not mean target selection has finished.
  */
 export async function awaitSendReady(page: Page): Promise<void> {
   await expect(page.getByRole('button', { name: '发送' })).toBeEnabled({
@@ -436,7 +422,7 @@ async function withE2eWindow(
     railRenderSessions?: boolean;
     newTaskProject?: boolean;
   },
-  use: (page: Page, context: { userDataDir: string; app: ElectronApplication }) => Promise<void>,
+  use: (page: Page, context: { userDataDir: string; app: ElectronApplication; restart(): Promise<Page> }) => Promise<void>,
 ): Promise<void> {
   const userDataDir = await mkdtemp(path.join(tmpdir(), 'maka-e2e-'));
   // Lives inside the throwaway userData dir so the existing teardown removes
@@ -505,7 +491,17 @@ async function withE2eWindow(
       const rendererDetail = rendererLogs.length > 0 ? `\nRenderer console:\n${rendererLogs.join('\n')}` : '';
       throw new Error(`${detail}${mainDetail}${rendererDetail}`, { cause: error });
     }
-    await use(page, { userDataDir, app });
+    await use(page, { userDataDir, app, restart: async () => {
+      await closeElectronApplication(app!, 5_000);
+      app = await electron.launch({
+        args: ['.', ...(visibleWindow ? inactiveWindowPlatformArgs() : [])],
+        cwd: DESKTOP_ROOT,
+        env: buildFixtureEnv(userDataDir, homeDir, { scenario: e2eFixtureScenario, locale, platform, showWindow: visibleWindow }),
+      });
+      const restored = await app.firstWindow();
+      await restored.waitForSelector(readinessSelector, { timeout: 20_000 });
+      return restored;
+    } });
   } finally {
     try {
       if (app) await closeElectronApplication(app, 5_000);
@@ -516,6 +512,7 @@ async function withE2eWindow(
 }
 
 type E2eTestFixtures = {
+  sessionLocalWindow: { page: Page; app: ElectronApplication; restart(): Promise<Page> };
   window: Page;
   gitReviewWindow: { page: Page; projectRoot: string };
   invocableSkillsWindow: Page;
@@ -532,6 +529,12 @@ type E2eTestFixtures = {
 };
 
 export const test = base.extend<E2eTestFixtures>({
+  sessionLocalWindow: async ({}, use) => {
+    await withE2eWindow(
+      { seed: true, readinessSelector: COMPOSER_INPUT, locale: 'zh-CN', showWindow: true },
+      async (page, { app, restart }) => use({ page, app, restart }),
+    );
+  },
   directoryReferenceWindow: async ({}, use) => {
     await withE2eWindow(
       { seed: true, readinessSelector: COMPOSER_INPUT, locale: 'zh-CN', showWindow: true },

apps/desktop/e2e/quote-window-boundary.spec.ts

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { expect, test, COMPOSER_INPUT } from './fixtures';
+import { awaitSendReady, expect, test, COMPOSER_INPUT } from './fixtures';
 
 // This stays in Electron: the physical pointer leaves the renderer viewport,
 // and Chromium pointer capture must route its release back to the owning Turn.
@@ -27,6 +27,7 @@ test('a transcript drag releases outside the window through its owning Turn', as
   await page.setViewportSize({ width: 1200, height: 800 });
   const composer = page.locator(COMPOSER_INPUT);
   await composer.fill('pointer capture source');
+  await awaitSendReady(page);
   await composer.press('Enter');
 
   // Select from a settled answer. Selecting from a streaming one is broken for

apps/desktop/e2e/session-local-recovery.spec.ts

@@ -0,0 +1,171 @@
+/*
+ * 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 { resolve } from 'node:path';
+import { COMPOSER_INPUT, awaitSendReady, ensureSidebarExpanded, expect, test } from './fixtures';
+
+test('a locally saved message survives renderer and application restart, then executes once', async ({
+  sessionLocalWindow,
+}, testInfo) => {
+  let { page } = sessionLocalWindow;
+  const { app, restart } = sessionLocalWindow;
+  const first = 'durable history before restart';
+  await page.locator(COMPOSER_INPUT).fill(first);
+  await awaitSendReady(page);
+  await page.locator(COMPOSER_INPUT).press('Enter');
+  await expect(page.getByText(`Fake backend received: ${first}`)).toBeVisible();
+  await ensureSidebarExpanded(page);
+  const sessionId = await page
+    .locator('[data-session-id]:has([aria-current="page"])')
+    .getAttribute('data-session-id');
+  expect(sessionId).toBeTruthy();
+  await expect
+    .poll(() =>
+      page.evaluate(
+        async (id) => !!(await window.maka.sessionLocal.readTranscript(id)),
+        sessionId!,
+      ),
+    )
+    .toBe(true);
+
+  // Pause only the delivery scheduler in this isolated main process. Admission,
+  // SQLite, renderer/preload, and the later Host execution all remain real.
+  await app.evaluate((_electron, modulePath) => {
+    const require = process.getBuiltinModule('module').createRequire(`${process.cwd()}/`);
+    const { DesktopSessionLocalService } = require(modulePath);
+    DesktopSessionLocalService.prototype.wake = () => {};
+  }, resolve('dist/main/session-local-service.js'));
+  const pending = 'saved locally across a complete application restart';
+  await page.locator(COMPOSER_INPUT).fill(pending);
+  await awaitSendReady(page);
+  await page.locator(COMPOSER_INPUT).press('Enter');
+  await expect(page.locator(COMPOSER_INPUT)).toHaveText('');
+  await expect(page.getByText('已本地保存 · 等待发送')).toBeVisible();
+  const before = await page.evaluate((id) => window.maka.sessionLocal.listMessages(id), sessionId!);
+  const message = before.find((item) => item.text === pending)!;
+  expect(message.state).toBe('saved');
+  await page.screenshot({ path: testInfo.outputPath('locally-saved.png') });
+
+  await page.reload();
+  await ensureSidebarExpanded(page);
+  await page.locator(`[data-session-id=${JSON.stringify(sessionId)}]`).click();
+  await expect(page.getByText('已本地保存 · 等待发送')).toBeVisible();
+  expect(
+    (await page.evaluate((id) => window.maka.sessionLocal.listMessages(id), sessionId!)).find(
+      (item) => item.text === pending,
+    )?.messageId,
+  ).toBe(message.messageId);
+
+  page = await restart();
+  await ensureSidebarExpanded(page);
+  await page.locator(`[data-session-id=${JSON.stringify(sessionId)}]`).click();
+  await expect(
+    page.getByLabel('Maka 的回答').getByText(`Fake backend received: ${pending}`),
+  ).toBeVisible({ timeout: 20_000 });
+  await expect(
+    page.getByLabel('Maka 的回答').getByText(`Fake backend received: ${pending}`),
+  ).toHaveCount(1);
+  await expect
+    .poll(() =>
+      page.evaluate(
+        async (id) => (await window.maka.sessionLocal.listMessages(id)).length,
+        sessionId!,
+      ),
+    )
+    .toBe(0);
+  await page.screenshot({ path: testInfo.outputPath('recovered.png') });
+});
+
+test('cached history remains readable when the live transcript endpoint is unavailable', async ({
+  sessionLocalWindow,
+}, testInfo) => {
+  const { page, app } = sessionLocalWindow;
+  const prompt = 'history available without a live transcript';
+  await page.locator(COMPOSER_INPUT).fill(prompt);
+  await awaitSendReady(page);
+  await page.locator(COMPOSER_INPUT).press('Enter');
+  await expect(
+    page.getByLabel('Maka 的回答').getByText(`Fake backend received: ${prompt}`),
+  ).toBeVisible();
+  await ensureSidebarExpanded(page);
+  const sessionId = await page
+    .locator('[data-session-id]:has([aria-current="page"])')
+    .getAttribute('data-session-id');
+  await expect
+    .poll(() =>
+      page.evaluate(
+        async (id) => !!(await window.maka.sessionLocal.readTranscript(id)),
+        sessionId!,
+      ),
+    )
+    .toBe(true);
+  // Fault only the live endpoint, not the cache bridge or renderer projection.
+  await app.evaluate(({ ipcMain }) => {
+    ipcMain.removeHandler('sessions:transcript:open');
+    ipcMain.handle('sessions:transcript:open', () => {
+      throw new Error('E2E live transcript unavailable');
+    });
+  });
+  await page.reload();
+  await ensureSidebarExpanded(page);
+  await page.locator(`[data-session-id=${JSON.stringify(sessionId)}]`).click();
+  await expect(
+    page.getByLabel('Maka 的回答').getByText(`Fake backend received: ${prompt}`),
+  ).toBeVisible();
+  await expect(page.locator('.maka-chat-recovery-notice')).toHaveCount(0);
+  await page.screenshot({ path: testInfo.outputPath('cached-history.png') });
+});
+
+test('a new task is readable locally before the Host session exists', async ({
+  sessionLocalWindow,
+}) => {
+  let { page } = sessionLocalWindow;
+  await sessionLocalWindow.app.evaluate((_electron, modulePath) => {
+    const require = process.getBuiltinModule('module').createRequire(`${process.cwd()}/`);
+    const { DesktopSessionLocalService } = require(modulePath);
+    DesktopSessionLocalService.prototype.wake = () => {};
+  }, resolve('dist/main/session-local-service.js'));
+  const prompt = 'first message saved before Host creation';
+  await page.locator(COMPOSER_INPUT).fill(prompt);
+  await awaitSendReady(page);
+  await page.locator(COMPOSER_INPUT).press('Enter');
+  await expect(page.locator(COMPOSER_INPUT)).toHaveText('');
+  await expect(page.getByText('已本地保存 · 等待发送')).toBeVisible();
+  await ensureSidebarExpanded(page);
+  const sessionId = await page
+    .locator('[data-session-id]:has([aria-current="page"])')
+    .getAttribute('data-session-id');
+  expect(sessionId).toBeTruthy();
+  await expect(page.getByText('读取任务失败', { exact: true })).toHaveCount(0);
+  await expect(page.getByText('Graph 状态刷新失败。', { exact: true })).toHaveCount(0);
+  await page.reload();
+  await ensureSidebarExpanded(page);
+  await page.locator(`[data-session-id=${JSON.stringify(sessionId)}]`).click();
+  await expect(page.getByText('已本地保存 · 等待发送')).toBeVisible();
+  await expect(page.getByText('读取任务失败', { exact: true })).toHaveCount(0);
+  page = await sessionLocalWindow.restart();
+  await ensureSidebarExpanded(page);
+  await page.locator(`[data-session-id=${JSON.stringify(sessionId)}]`).click();
+  await expect(
+    page.getByLabel('Maka 的回答').getByText(`Fake backend received: ${prompt}`),
+  ).toBeVisible({ timeout: 20_000 });
+  await expect(
+    page.getByLabel('Maka 的回答').getByText(`Fake backend received: ${prompt}`),
+  ).toHaveCount(1);
+});

apps/desktop/e2e/session-workbar.spec.ts

@@ -17,14 +17,15 @@
  * under the License.
  */
 
-import { COMPOSER_INPUT, test, expect } from './fixtures';
+import { awaitSendReady, COMPOSER_INPUT, test, expect } from './fixtures';
 import type { Page } from '@playwright/test';
 import { writeFile } from 'node:fs/promises';
 import { join } from 'node:path';
 
 async function openGitChanges(page: Page) {
   const composer = page.locator(COMPOSER_INPUT);
   await composer.fill('create review session');
+  await awaitSendReady(page);
   await composer.press('Enter');
   await expect(page.getByText(/Fake backend received: create review session/)).toBeVisible();
   await page.getByRole('button', { name: '展开任务工作栏' }).click();
@@ -36,6 +37,7 @@ async function openGitChanges(page: Page) {
 async function createSession(page: Page, prompt: string) {
   const composer = page.locator(COMPOSER_INPUT);
   await composer.fill(prompt);
+  await awaitSendReady(page);
   await composer.press('Enter');
   await expect(page.getByText(`Fake backend received: ${prompt}`)).toBeVisible();
   await expect(page.getByRole('button', { name: '重新生成' })).toHaveCount(1, {
@@ -219,6 +221,7 @@ test('Terminal ownership follows the active Session and stops the old resource',
     .not.toBe('running');
 
   await composer.fill('create replacement session');
+  await awaitSendReady(page);
   await composer.press('Enter');
   await expect(page.getByText('Fake backend received: create replacement session')).toBeVisible();
   await page.getByRole('button', { name: '展开任务工作栏' }).click();

apps/desktop/e2e/settings.spec.ts

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { COMPOSER_INPUT, ensureSidebarExpanded, expect, test } from './fixtures';
+import { awaitSendReady, COMPOSER_INPUT, ensureSidebarExpanded, expect, test } from './fixtures';
 
 interface SettingsChunkLatchWindow extends Window {
   makaE2eLatch?: {
@@ -78,6 +78,7 @@ test('Settings loading surface owns unmodified Escape', async ({ window: page })
 test('opening settings commits an active titlebar rename', async ({ window: page }) => {
   const composer = page.locator(COMPOSER_INPUT);
   await composer.fill('create a session for settings rename');
+  await awaitSendReady(page);
   await composer.press('Enter');
 
   const identity = page.locator('[data-maka-contract="titlebar-identity"]');
@@ -100,6 +101,7 @@ test('settings hides expanded workbar chrome and restores it on close', async ({
 }) => {
   const composer = page.locator(COMPOSER_INPUT);
   await composer.fill('create a session with an expanded workbar');
+  await awaitSendReady(page);
   await composer.press('Enter');
 
   await page.getByRole('button', { name: '展开任务工作栏' }).click();