(maka) 12/20: fix(cli): refresh live transcript on durable advance

[as...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

Astro-Han pushed a commit to branch fix/steering-durable-admission
in repository https://gitbox.apache.org/repos/asf/maka.git

commit 8110f831a15f5259822e0412134acacb819b2292
Author: AstroHan <le...@outlook.com>
AuthorDate: Mon Aug 24 00:29:02 2026 +0800

    fix(cli): refresh live transcript on durable advance
    
    Generated-by: Maka
---
 packages/cli/src/__tests__/pi-tui-runner.test.ts   | 39 +++++++++++++++---
 .../__tests__/runtime-host-session-driver.test.ts  | 46 +++++++++++++++++++++-
 packages/cli/src/pi-tui-runner.ts                  | 23 +++++++++--
 packages/cli/src/runtime-host-session-channel.ts   |  5 +++
 4 files changed, 103 insertions(+), 10 deletions(-)

diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts
index 3bb0c4db8..8ed98e01e 100644
--- a/packages/cli/src/__tests__/pi-tui-runner.test.ts
+++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts
@@ -52,6 +52,7 @@ import type {
   MakaSessionRewindResult,
   MakaSessionSwitchOptions,
   MakaSessionSwitchResult,
+  MakaTranscriptReplacementReason,
   RewindTarget,
   SessionResumeAvailability,
 } from '../session-driver.js';
@@ -1882,7 +1883,7 @@ describe('Maka Pi TUI runner', () => {
     terminal.input('also handle Y');
     terminal.input('\r');
     await waitFor(() => driver.steered.length === 1);
-    await delay(0);
+    await waitFor(() => plainTerminalOutput(terminal.screenOutput()).includes('also handle Y'));
     assert.deepEqual(driver.steered, ['also handle Y']);
     assert.equal(
       plainTerminalOutput(terminal.screenOutput()).includes('Steering: also handle Y'),
@@ -2145,10 +2146,7 @@ describe('Maka Pi TUI runner', () => {
     await waitFor(() => terminal.progressStates.at(-1) === false);
     // Only the followup that was still queued comes back into the editor; the
     // consumed steering text must not be resurrected from the stale mirror.
-    await waitFor(() => {
-      const screen = plainTerminalOutput(terminal.screenOutput());
-      return screen.includes('still queued') && !screen.includes('already consumed');
-    });
+    await waitFor(() => editorInputText(terminal) === 'still queued');
 
     terminal.input('\x03'); // clear the refilled draft
     terminal.input('/exit');
@@ -6358,6 +6356,14 @@ class SteeringTurnDriver implements MakaSessionDriver {
   readonly steered: string[] = [];
   readonly queuedMessages: string[] = [];
   readonly turnOrchestrations: Array<MakaPreparePromptOptions['turnOrchestration']> = [];
+  readonly transcriptListeners = new Set<
+    (
+      sessionId: string,
+      turnId: string,
+      messages: StoredMessage[],
+      reason: MakaTranscriptReplacementReason,
+    ) => void
+  >();
   retractCalls = 0;
   rewindTargets: RewindTarget[] = [];
   private steering: string[] = [];
@@ -6420,10 +6426,33 @@ class SteeringTurnDriver implements MakaSessionDriver {
     yield { type: 'complete', id: 'event-complete', turnId, ts: 2, stopReason: 'user_stop' };
   }
 
+  subscribeTranscriptReplacements(
+    listener: (
+      sessionId: string,
+      turnId: string,
+      messages: StoredMessage[],
+      reason: MakaTranscriptReplacementReason,
+    ) => void,
+  ): () => void {
+    this.transcriptListeners.add(listener);
+    return () => this.transcriptListeners.delete(listener);
+  }
+
   async steer(text: string): Promise<QueueEnqueueOutcome> {
     this.steered.push(text);
     this.steering.push(text);
     this.emitQueueUpdate();
+    const message: Extract<StoredMessage, { type: 'user' }> = {
+      type: 'user',
+      id: `steering-${this.steered.length}`,
+      turnId: 'turn-1',
+      ts: this.steered.length,
+      text,
+      steeringEventId: `steering-${this.steered.length}`,
+    };
+    for (const listener of this.transcriptListeners) {
+      listener(this.getSessionId(), 'turn-1', [message], 'reconcile');
+    }
     return { kind: 'queued' };
   }
 
diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
index 1b9dc7abb..bdc92590a 100644
--- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
+++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
@@ -44,7 +44,11 @@ import {
   createRuntimeHostMakaSessionDriver,
   type RuntimeHostMakaSessionDriverInput,
 } from '../runtime-host-session-driver.js';
-import { SkillInvocationBlockedError, type MakaAttachedSessionTurn } from '../session-driver.js';
+import {
+  SkillInvocationBlockedError,
+  type MakaAttachedSessionTurn,
+  type MakaTranscriptReplacementReason,
+} from '../session-driver.js';
 import { WAIT_BUDGET_MS } from './tui-terminal-mock.js';
 
 describe('Runtime Host Maka Session driver', () => {
@@ -551,6 +555,46 @@ describe('Runtime Host Maka Session driver', () => {
     });
   });
 
+  test('publishes a durable transcript advance while the active turn is still running', async () => {
+    const attached = new FakeSubscription(continuitySnapshot(), Promise.resolve([]));
+    const durableMessages = [userMessage('turn-1', 'Steer now')];
+    const refresh = new FakeSubscription(
+      continuitySnapshot(),
+      Promise.resolve(durableMessages),
+      'subscription-2',
+    );
+    const connection = new FakeConnection([attached, refresh]);
+    const driver = createRuntimeHostMakaSessionDriver({
+      connection: connection.value,
+      cwd: '/tmp',
+      llmConnectionSlug: 'openai-main',
+      model: 'gpt-5',
+    });
+    await driver.switchSession('session-1');
+    const replacement = deferred<{
+      messages: readonly StoredMessage[];
+      reason: MakaTranscriptReplacementReason;
+    }>();
+    driver.subscribeTranscriptReplacements!((_sessionId, _turnId, messages, reason) =>
+      replacement.resolve({ messages, reason }),
+    );
+
+    attached.push({
+      kind: 'subscription.transcript_advanced',
+      hostEpoch: 'host-1',
+      subscriptionId: 'subscription-1',
+      sequence: 1,
+      sessionId: 'session-1',
+      throughSequence: 0,
+    });
+
+    await waitFor(() => connection.openedSubscriptions === 2);
+    assert.deepEqual(await replacement.promise, {
+      messages: durableMessages,
+      reason: 'reconcile',
+    });
+  });
+
   test('delivers completed thinking while a later step remains active', async () => {
     const subscription = new FakeSubscription(continuitySnapshot(), Promise.resolve([]));
     const connection = new FakeConnection([subscription]);
diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts
index 5ea8b8f29..8fbfa783a 100644
--- a/packages/cli/src/pi-tui-runner.ts
+++ b/packages/cli/src/pi-tui-runner.ts
@@ -281,11 +281,13 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
   const tui = new TuiMainScreen(terminal);
   const state = createMakaPiTranscriptState();
   let transcriptLastUsedModel: string | undefined;
-  const rememberTranscriptModel = (messages: readonly StoredMessage[]): void => {
+  let transcriptMessageIds = new Set<string>();
+  const rememberTranscript = (messages: readonly StoredMessage[]): void => {
     transcriptLastUsedModel = latestAssistantModelId(messages);
+    transcriptMessageIds = new Set(messages.map((message) => message.id));
   };
   const replaceTranscript = (messages: readonly StoredMessage[]): void => {
-    rememberTranscriptModel(messages);
+    rememberTranscript(messages);
     replaceTranscriptWithStoredMessages(state, messages);
   };
   let cwd = input.cwd;
@@ -540,8 +542,21 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
         requestRender();
         return;
       }
-      rememberTranscriptModel(messages);
-      if (hydrateToolsWithStoredMessages(state, turnId, messages)) {
+      const newSteeringMessages = messages.filter(
+        (message): message is Extract<StoredMessage, { type: 'user' }> =>
+          message.type === 'user' &&
+          message.turnId === turnId &&
+          message.steeringEventId !== undefined &&
+          !transcriptMessageIds.has(message.id),
+      );
+      rememberTranscript(messages);
+      for (const message of newSteeringMessages) {
+        appendUserPrompt(state, message.displayText ?? message.text);
+      }
+      if (
+        newSteeringMessages.length > 0 ||
+        hydrateToolsWithStoredMessages(state, turnId, messages)
+      ) {
         shellRunElapsedTicker.sync();
         requestRender();
       }
diff --git a/packages/cli/src/runtime-host-session-channel.ts b/packages/cli/src/runtime-host-session-channel.ts
index 01714f225..d07666d5b 100644
--- a/packages/cli/src/runtime-host-session-channel.ts
+++ b/packages/cli/src/runtime-host-session-channel.ts
@@ -572,6 +572,11 @@ export class RuntimeHostSessionChannel {
   }
 
   #accept(frame: SubscriptionFrame): void {
+    if (frame.kind === 'subscription.transcript_advanced') {
+      const turnId = this.snapshot.rootTurn?.turnId;
+      if (turnId) this.#onTranscriptSettlement(turnId);
+      return;
+    }
     if (frame.kind === 'subscription.session_domain_changed') {
       if (frame.domain === 'runtime_resource') {
         for (const resource of frame.resources) {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org