fix(runtime-host): unify durable message lifecycle

apps/desktop/e2e/streaming-remount.spec.ts

@@ -77,29 +77,14 @@ test('remounting a live surface leaves accumulated output settled', async ({
     ),
   ).toBe(0);
 
-  await liveBubble.evaluate((element) => {
-    const observed = { texts: [] as string[] };
-    (window as typeof window & { __makaStreamingRemountObserved?: typeof observed })
-      .__makaStreamingRemountObserved = observed;
-    new MutationObserver(() => {
-      observed.texts.push(element.textContent ?? '');
-    }).observe(element, { childList: true, characterData: true, subtree: true });
-  });
-
   const steering = 'trigger rewrite after returning to this conversation';
   await steerActiveTurn(composer, steering);
   const finalText = 'prefix <redacted> NEW streamed after the remount';
   await expect(liveBubble).toContainText(finalText);
-
-  const observed = await page.evaluate(() => (
-    window as typeof window & {
-      __makaStreamingRemountObserved?: {
-        texts: string[];
-      };
-    }
-  ).__makaStreamingRemountObserved);
-  expect(observed?.texts.some((text) => text.includes('<redacted>') && !text.includes(finalText)))
-    .toBe(true);
+  await expect(liveBubble).not.toContainText(accumulatedOutput);
+  // React may batch the one rewrite delta into its final redacted paint. The
+  // product contract is the settled text, not a particular intermediate frame
+  // or DOM node identity across the transcript handoff.
 });
 
 test('keeps a completed reply after an interrupted turn and conversation remount', async ({
@@ -196,7 +181,7 @@ test('returning to a live conversation settles output accumulated while away', a
 
   const accumulatedOutput = 'Fake backend waiting for the test to stop the Turn.';
   const liveBubble = page.locator('.maka-bubble-streaming');
-  await expect(liveBubble).toContainText(accumulatedOutput);
+  await expect(liveBubble).toContainText(accumulatedOutput, { timeout: 20_000 });
 
   const sidebar = page.getByRole('navigation', { name: '任务列表' });
   await page.getByRole('button', { name: '展开侧边栏' }).click();
@@ -231,7 +216,9 @@ test('returning to a live conversation settles output accumulated while away', a
         unsubscribe();
         resolve();
       });
-      void window.maka.sessions.steer(sessionId, steering).catch((error) => {
+      void window.maka.sessions
+        .enqueue(sessionId, 'current_turn', { text: steering })
+        .catch((error) => {
         window.clearTimeout(timeout);
         unsubscribe();
         reject(error);

apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts

@@ -102,7 +102,6 @@ function createActionsDeps() {
     setMessages: () => undefined,
     transcriptRangeRef: { current: undefined },
     setNavSelection: () => undefined,
-    setLiveTurnBySession: () => undefined,
     setInteractionBySession: () => undefined,
     showModelSetupToast: () => undefined,
     toastApi: { error: () => undefined, info: () => undefined },
@@ -119,10 +118,21 @@ function createActionsDeps() {
 const EMPTY_SKILL_INVOCATION = { loaded: [], failed: [], receipts: [] };
 
 describe('busy-raced send settlement', () => {
-  it('a steered send on an existing session disarms its turn and shows no optimistic message', async () => {
+  it('keeps the visible live tail while the Host admits a new message', async () => {
     const activeIdRef = { current: 'session-a' as string | undefined };
     const turnState = createTurnState();
-    const messageState = createMessageState();
+    const visibleTail: LiveTurnProjection = {
+      turnId: 'existing-turn',
+      phase: 'streamed',
+      steps: [
+        {
+          stepId: 'assistant-tail',
+          text: { text: 'still visible', truncated: false, complete: false },
+          tools: [],
+        },
+      ],
+    };
+    turnState.setLiveTurnBySession(() => ({ 'session-a': visibleTail }));
     const restoreWindow = installWindow({
       sessions: {
         send: async (_sessionId: string, command: { turnId: string }) => ({
@@ -136,12 +146,44 @@ describe('busy-raced send settlement', () => {
       },
     });
     try {
-      const actions = createAppShellChatActions({
+      const deps = {
         ...createActionsDeps(),
         activeIdRef,
         setLiveTurnBySession: turnState.setLiveTurnBySession,
+      };
+      const actions = createAppShellChatActions(deps);
+
+      assert.equal(await actions.send('also check the tests'), true);
+      assert.deepEqual(turnState.liveTurnBySession['session-a'], visibleTail);
+    } finally {
+      restoreWindow();
+    }
+  });
+
+  it('a steered send on an existing session shows no optimistic message', async () => {
+    const activeIdRef = { current: 'session-a' as string | undefined };
+    const turnState = createTurnState();
+    const messageState = createMessageState();
+    const restoreWindow = installWindow({
+      sessions: {
+        send: async (_sessionId: string, command: { turnId: string }) => ({
+          ok: true,
+          steered: true,
+          turnId: command.turnId,
+          attachments: [],
+          inlineReferences: [],
+          skillInvocation: EMPTY_SKILL_INVOCATION,
+        }),
+      },
+    });
+    try {
+      const deps = {
+        ...createActionsDeps(),
+        activeIdRef,
         setMessages: messageState.setMessages,
-      });
+        setLiveTurnBySession: turnState.setLiveTurnBySession,
+      };
+      const actions = createAppShellChatActions(deps);
       assert.equal(await actions.send('also check the tests'), true);
       assert.equal(turnState.liveTurnBySession['session-a'], undefined);
       assert.deepEqual(messageState.messages, []);
@@ -150,7 +192,7 @@ describe('busy-raced send settlement', () => {
     }
   });
 
-  it('rebinds the unconfirmed arm onto a Host-chosen turn id', async () => {
+  it('does not invent an empty live turn while the Host admits a message', async () => {
     const activeIdRef = { current: 'session-a' as string | undefined };
     const turnState = createTurnState();
     const messageState = createMessageState();
@@ -166,16 +208,15 @@ describe('busy-raced send settlement', () => {
       },
     });
     try {
-      const actions = createAppShellChatActions({
+      const deps = {
         ...createActionsDeps(),
         activeIdRef,
-        setLiveTurnBySession: turnState.setLiveTurnBySession,
         setMessages: messageState.setMessages,
-      });
+        setLiveTurnBySession: turnState.setLiveTurnBySession,
+      };
+      const actions = createAppShellChatActions(deps);
       assert.equal(await actions.send('also check the tests'), true);
-      const live = turnState.liveTurnBySession['session-a'];
-      assert.equal(live?.turnId, 'host-turn');
-      assert.equal(live?.unconfirmed, true);
+      assert.equal(turnState.liveTurnBySession['session-a'], undefined);
       const optimistic = messageState.messages.filter((message) => message.type === 'user');
       assert.equal(optimistic.length, 1);
       assert.equal(optimistic[0]?.turnId, 'host-turn');
@@ -210,14 +251,12 @@ describe('busy-raced send settlement', () => {
       const actions = createAppShellChatActions({
         ...createActionsDeps(),
         activeIdRef,
-        setLiveTurnBySession: turnState.setLiveTurnBySession,
         setMessages: messageState.setMessages,
       });
       assert.equal(await actions.send('also check the tests'), true);
       const live = turnState.liveTurnBySession['session-a'];
       assert.equal(live?.turnId, 'host-turn');
       assert.equal(live?.phase, 'streamed');
-      assert.equal(live?.unconfirmed, undefined);
     } finally {
       restoreWindow();
     }
@@ -255,7 +294,6 @@ describe('busy-raced send settlement', () => {
           if (sessionId !== undefined) activated.push(sessionId);
           activeIdRef.current = sessionId;
         },
-        setLiveTurnBySession: turnState.setLiveTurnBySession,
         setMessages: messageState.setMessages,
       });
       assert.equal(await actions.send('also check the tests'), true);
@@ -293,11 +331,10 @@ describe('busy-raced send settlement', () => {
         setActiveId: (sessionId: string | undefined) => {
           activeIdRef.current = sessionId;
         },
-        setLiveTurnBySession: turnState.setLiveTurnBySession,
         setMessages: messageState.setMessages,
       });
       assert.equal(await actions.send('also check the tests'), true);
-      assert.equal(turnState.liveTurnBySession['session-new']?.turnId, 'host-turn');
+      assert.equal(turnState.liveTurnBySession['session-new'], undefined);
       const optimistic = messageState.messages.filter((message) => message.type === 'user');
       assert.equal(optimistic.length, 1);
       assert.equal(optimistic[0]?.turnId, 'host-turn');

apps/desktop/src/main/__tests__/app-shell-first-send-cleanup.test.ts

@@ -35,12 +35,8 @@
 import { strict as assert } from 'node:assert';
 import { describe, it } from 'node:test';
 
-import type { SessionSummary } from '@maka/core/session';
-import type { LiveTurnProjection } from '@maka/ui';
 import type { DesktopTranscriptRangeController } from '../../renderer/desktop-transcript-range-store.js';
 import { createAppShellChatActions } from '../../renderer/app-shell-chat-actions.js';
-import { createAppShellSessionUiStateController } from '../../renderer/app-shell-session-ui-state.js';
-import { settledSessionTransientIds } from '../../renderer/settled-session-transients.js';
 
 function installWindow(maka: unknown): () => void {
   const target = globalThis as unknown as { window?: unknown };
@@ -64,23 +60,6 @@ function installWindow(maka: unknown): () => void {
   };
 }
 
-/**
- * The live-turn arm as a real map rather than a black-hole stub: a send that
- * never lands must leave nothing behind, and that cannot be asserted against a
- * no-op setter.
- */
-function createTurnState() {
-  const liveTurnBySession: Record<string, LiveTurnProjection> = {};
-  return {
-    liveTurnBySession,
-    setLiveTurnBySession(updater: (c: Record<string, LiveTurnProjection>) => Record<string, LiveTurnProjection>) {
-      const next = updater({ ...liveTurnBySession });
-      for (const key of Object.keys(liveTurnBySession)) delete liveTurnBySession[key];
-      Object.assign(liveTurnBySession, next);
-    },
-  };
-}
-
 function createActionsDeps() {
   return {
     uiLocale: 'en' as const,
@@ -103,7 +82,6 @@ function createActionsDeps() {
     setMessages: () => undefined,
     transcriptRangeRef: { current: undefined },
     setNavSelection: () => undefined,
-    setLiveTurnBySession: () => undefined,
     setInteractionBySession: () => undefined,
     showModelSetupToast: () => undefined,
     toastApi: { error: () => undefined, info: () => undefined },
@@ -491,28 +469,6 @@ describe('composer send failure feedback', () => {
     assert.deepEqual(setupToasts, [], 'a stale surface must not be navigated to 设置 · 模型');
   });
 
-  // A send that never reaches the runtime must take its arm with it. A leftover
-  // arm still carries its `unconfirmed` claim, which would make
-  // `settledSessionTransientIds` protect a turn that does not exist — leaving a
-  // Stop button nothing can clear.
-  it('leaves no arm behind when the send never lands', async () => {
-    const turnState = createTurnState();
-    const restoreWindow = installWindow(readinessFailure());
-
-    try {
-      const actions = createAppShellChatActions({
-        ...createActionsDeps(),
-        activeIdRef: { current: 'session-a' },
-        setLiveTurnBySession: turnState.setLiveTurnBySession,
-      });
-      assert.equal(await actions.send('hello'), false);
-    } finally {
-      restoreWindow();
-    }
-
-    assert.deepEqual(turnState.liveTurnBySession, {}, 'the arm must be disarmed');
-  });
-
   it('still answers the surface that is actually waiting', async () => {
     const setupToasts: string[] = [];
     const restoreWindow = installWindow(readinessFailure());
@@ -533,91 +489,30 @@ describe('composer send failure feedback', () => {
   });
 });
 
-/**
- * The bug this guards, as the sequence that actually produced it: send arms the
- * turn, a session list that was already in flight lands still carrying the
- * pre-send status, and the settle reconcile runs against it.
- *
- * Nothing in that list is wrong — the runtime writes `status: 'running'` only at
- * the end of `AgentRun.begin` and announces it to nobody until `onRunStarted`.
- * The list simply predates the answer. Reading it as a settle used to drop the
- * arm, so the first content event rebuilt the projection as `'streamed'` and the
- * prominent "正在处理…" silently became the calm "继续中…".
- *
- * Asserted through the real `send`, the real state controller, and the real
- * settle rule, because the defect lived in how those three compose — each one is
- * individually correct.
- */
-describe('a send in flight versus a stale session list', () => {
-  const sessionId = 'session-a';
-
-  function sendingWindow() {
-    return {
-      sessions: {
-        send: async () => ({
-          ok: true,
-          attachments: [],
-          skillInvocation: { loaded: [], failed: [] },
-        }),
+describe('transcript refresh ownership', () => {
+  it('waits for durable transcript data without republishing the subscription snapshot', async () => {
+    let publications = 0;
+    const controller = {
+      ready: async () => undefined,
+      waitForDurableMessage: async () => true,
+      store: {
+        hasDurableMessage: () => true,
+        snapshot: () => ({ sessionId: 'session-a', messages: [] }),
+      },
+    } as unknown as DesktopTranscriptRangeController;
+    const actions = createAppShellChatActions({
+      ...createActionsDeps(),
+      activeIdRef: { current: 'session-a' },
+      transcriptRangeRef: { current: controller },
+      setMessages: () => {
+        publications += 1;
       },
-    };
-  }
-
-  // The list as it reads before the runtime's `running` write — identical to how
-  // it reads after the turn is over, which is exactly why the status alone
-  // cannot settle anything.
-  const preSendList = [{ id: sessionId, status: 'active', statusUpdatedAt: 100 }] as SessionSummary[];
-
-  async function armViaSend(controller: ReturnType<typeof createAppShellSessionUiStateController>) {
-    const restoreWindow = installWindow(sendingWindow());
-    try {
-      const actions = createAppShellChatActions({
-        ...createActionsDeps(),
-        activeIdRef: { current: sessionId },
-        setLiveTurnBySession: controller.setLiveTurnBySession,
-      });
-      assert.equal(await actions.send('hello'), true);
-    } finally {
-      restoreWindow();
-    }
-    const armed = controller.getState().liveTurnBySession[sessionId];
-    assert.equal(armed?.unconfirmed, true, 'the send must arm an unconfirmed turn');
-    return armed!.turnId;
-  }
-
-  function settle(controller: ReturnType<typeof createAppShellSessionUiStateController>) {
-    return settledSessionTransientIds({
-      activeId: sessionId,
-      sessions: preSendList,
-      liveTurnBySession: controller.getState().liveTurnBySession,
     });
-  }
-
-  it('keeps the armed turn, and settles it once the authority names that turn', async () => {
-    const controller = createAppShellSessionUiStateController();
-    const turnId = await armViaSend(controller);
 
-    assert.deepEqual(settle(controller), [], 'a list older than the answer must not settle the turn');
     assert.equal(
-      controller.getState().liveTurnBySession[sessionId]?.phase,
-      'waiting',
-      'the first-token wait must survive the stale refresh',
+      await actions.refreshMessages('session-a', { requiredAssistantMessageId: 'assistant-a' }),
+      true,
     );
-
-    // `sessions:changed` naming this turn — what `onRunStarted` now emits once
-    // the run has begun. This is the same controller entry point the shell
-    // wires that subscription to.
-    controller.confirmLiveTurn(sessionId, turnId);
-
-    assert.deepEqual(settle(controller), [sessionId], 'an answered turn settles under the plain status rules');
-  });
-
-  it('ignores an answer about a turn other than the one in flight', async () => {
-    const controller = createAppShellSessionUiStateController();
-    await armViaSend(controller);
-
-    controller.confirmLiveTurn(sessionId, 'turn-from-another-client');
-
-    assert.deepEqual(settle(controller), [], 'only this send\'s own turn may release its claim');
+    assert.equal(publications, 0, 'the transcript subscription is the only messages publisher');
   });
 });