You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/08/26 16:14:37 UTC
(maka) 07/27: fix: reconcile transient messages from Host outcomes
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch fix/canonical-message-projection
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 7073a08e800e63ed8e59f6beebd8a85de50e1aca
Author: AstroHan <le...@outlook.com>
AuthorDate: Wed Aug 26 04:36:53 2026 +0800
fix: reconcile transient messages from Host outcomes
Generated-by: Codex
---
.../app-shell-busy-race-settlement.test.ts | 12 +--
.../__tests__/app-shell-first-send-cleanup.test.ts | 20 ++--
.../main/__tests__/message-queue-ui-state.test.ts | 14 +++
...runtime-host-session-execution-ipc-main.test.ts | 14 ++-
.../__tests__/transient-message-projection.test.ts | 23 +++++
.../runtime-host-session-execution-ipc-main.ts | 10 +-
.../src/renderer/app-shell-session-events.ts | 7 ++
apps/desktop/src/renderer/app-shell.tsx | 1 +
.../src/renderer/transient-message-projection.ts | 11 ++-
packages/cli/src/__tests__/pi-transcript.test.ts | 51 ++++++++++
packages/cli/src/__tests__/pi-tui-runner.test.ts | 108 +++++++++++++++++++--
.../__tests__/runtime-host-session-driver.test.ts | 48 +++++++--
packages/cli/src/pi-transcript.ts | 59 +++++++++--
packages/cli/src/pi-tui-runner.ts | 13 ++-
packages/cli/src/runtime-host-session-driver.ts | 28 ++++--
packages/cli/src/session-driver.ts | 2 +-
16 files changed, 356 insertions(+), 65 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts b/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts
index 867611dbd..d68c639e5 100644
--- a/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts
+++ b/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts
@@ -186,7 +186,7 @@ describe('busy-raced send settlement', () => {
});
const restoreWindow = installWindow({
sessions: {
- send: async (_sessionId: string, command: { messageId: string }) => {
+ submitMessage: async (_sessionId: string, command: { messageId: string }) => {
submittedMessageId = command.messageId;
observeSubmit();
await admission;
@@ -233,7 +233,7 @@ describe('busy-raced send settlement', () => {
const messageState = createMessageState();
const restoreWindow = installWindow({
sessions: {
- send: async (_sessionId: string, command: { messageId: string }) => ({
+ submitMessage: async (_sessionId: string, command: { messageId: string }) => ({
ok: true,
disposition: 'steering',
messageId: command.messageId,
@@ -270,7 +270,7 @@ describe('busy-raced send settlement', () => {
const messageState = createMessageState();
const restoreWindow = installWindow({
sessions: {
- send: async (_sessionId: string, command: { messageId: string }) => ({
+ submitMessage: async (_sessionId: string, command: { messageId: string }) => ({
ok: true,
disposition: 'turn_started',
messageId: command.messageId,
@@ -308,7 +308,7 @@ describe('busy-raced send settlement', () => {
const messageState = createMessageState();
const restoreWindow = installWindow({
sessions: {
- send: async (_sessionId: string, command: { messageId: string }) => {
+ submitMessage: async (_sessionId: string, command: { messageId: string }) => {
// The Host streamed under its own turn id before the IPC response.
turnState.setLiveTurnBySession((current) => ({
...current,
@@ -361,7 +361,7 @@ describe('busy-raced send settlement', () => {
remove: async (sessionId: string) => {
removed.push(sessionId);
},
- send: async (_sessionId: string, command: { messageId: string }) => ({
+ submitMessage: async (_sessionId: string, command: { messageId: string }) => ({
ok: true,
disposition: 'steering',
messageId: command.messageId,
@@ -405,7 +405,7 @@ describe('busy-raced send settlement', () => {
create: async () => ({ id: 'session-new' }),
},
sessions: {
- send: async (_sessionId: string, command: { messageId: string }) => ({
+ submitMessage: async (_sessionId: string, command: { messageId: string }) => ({
ok: true,
disposition: 'turn_started',
messageId: command.messageId,
diff --git a/apps/desktop/src/main/__tests__/app-shell-first-send-cleanup.test.ts b/apps/desktop/src/main/__tests__/app-shell-first-send-cleanup.test.ts
index 9127e1f38..40b6cceb9 100644
--- a/apps/desktop/src/main/__tests__/app-shell-first-send-cleanup.test.ts
+++ b/apps/desktop/src/main/__tests__/app-shell-first-send-cleanup.test.ts
@@ -123,7 +123,7 @@ describe('composer first-send cleanup', () => {
let sends = 0;
const restoreWindow = installWindow({
sessions: {
- send: async () => {
+ submitMessage: async () => {
sends += 1;
return { ok: true, attachments: [], skillInvocation: { loaded: [], failed: [] } };
},
@@ -162,7 +162,7 @@ describe('composer first-send cleanup', () => {
},
},
sessions: {
- send: async () => ({
+ submitMessage: async () => ({
ok: true,
attachments: [],
skillInvocation: { loaded: [], failed: [] },
@@ -212,7 +212,7 @@ describe('composer first-send cleanup', () => {
},
},
sessions: {
- send: async () => ({
+ submitMessage: async () => ({
ok: true,
attachments: [],
skillInvocation: { loaded: [], failed: [] },
@@ -248,7 +248,7 @@ describe('composer first-send cleanup', () => {
},
},
sessions: {
- send: async () => ({
+ submitMessage: async () => ({
ok: true,
attachments: [],
skillInvocation: { loaded: [], failed: [] },
@@ -293,7 +293,7 @@ describe('composer first-send cleanup', () => {
},
},
sessions: {
- send: async () => ({
+ submitMessage: async () => ({
ok: true,
attachments: [],
skillInvocation: { loaded: [], failed: [] },
@@ -329,7 +329,7 @@ describe('composer first-send cleanup', () => {
newTasks: { create: async () => ({ id: 'session-1' }) },
sessions: {
// What `prepareSkillInvocation` does when Skill discovery fails.
- send: async () => Promise.reject(new Error('Skill discovery failed')),
+ submitMessage: async () => Promise.reject(new Error('Skill discovery failed')),
remove: async (sessionId: string) => {
removed.push(sessionId);
},
@@ -350,7 +350,7 @@ describe('composer first-send cleanup', () => {
const restoreWindow = installWindow({
newTasks: { create: async () => ({ id: 'session-1' }) },
sessions: {
- send: async () => ({
+ submitMessage: async () => ({
ok: true,
attachments: [],
skillInvocation: { loaded: [], failed: [] },
@@ -378,7 +378,7 @@ describe('composer first-send cleanup', () => {
const removed: string[] = [];
const restoreWindow = installWindow({
sessions: {
- send: async () => Promise.reject(new Error('Skill discovery failed')),
+ submitMessage: async () => Promise.reject(new Error('Skill discovery failed')),
remove: async (sessionId: string) => {
removed.push(sessionId);
},
@@ -415,7 +415,7 @@ describe('composer first-send cleanup', () => {
const transcriptRangeRef = { current: transcript as DesktopTranscriptRangeController | undefined };
const restoreWindow = installWindow({
sessions: {
- send: async () => {
+ submitMessage: async () => {
order.push('send');
return { ok: true, attachments: [], skillInvocation: { loaded: [], failed: [] } };
},
@@ -462,7 +462,7 @@ function deferred<T>() {
describe('composer send failure feedback', () => {
const readinessFailure = () => ({
sessions: {
- send: async () =>
+ submitMessage: async () =>
Promise.reject(new Error('NO_REAL_CONNECTION:missing_api_key: no ready connection')),
remove: async () => undefined,
},
diff --git a/apps/desktop/src/main/__tests__/message-queue-ui-state.test.ts b/apps/desktop/src/main/__tests__/message-queue-ui-state.test.ts
index fc157aec4..9d1a466c6 100644
--- a/apps/desktop/src/main/__tests__/message-queue-ui-state.test.ts
+++ b/apps/desktop/src/main/__tests__/message-queue-ui-state.test.ts
@@ -25,6 +25,7 @@ import { createAppShellSessionUiStateController } from '../../renderer/app-shell
test('queue_update events drive the independent desktop queue projection', () => {
const controller = createAppShellSessionUiStateController();
const transientMessages: unknown[] = [];
+ const removedTransientMessageIds: string[] = [];
const handlers = createAppShellSessionEventHandlers({
uiLocale: 'zh',
activeIdRef: { current: 'session-1' },
@@ -35,6 +36,8 @@ test('queue_update events drive the independent desktop queue projection', () =>
setInteractionBySession: controller.setInteractionBySession,
setMessageQueueBySession: controller.setMessageQueueBySession,
projectTransientMessage: (_sessionId, message) => transientMessages.push(message),
+ removeTransientMessage: (_sessionId, messageId) =>
+ removedTransientMessageIds.push(messageId),
showModelSetupToast() {},
toastApi: { error() {} },
});
@@ -118,6 +121,17 @@ test('queue_update events drive the independent desktop queue projection', () =>
followup: [],
});
assert.equal(controller.getState().messageQueueBySession['session-1'], undefined);
+ assert.deepEqual(removedTransientMessageIds, []);
+
+ handlers.handleEvent('session-1', {
+ type: 'message_admission',
+ id: 'retracted-message-next',
+ turnId: 'turn-1',
+ ts: 3,
+ messageId: 'message-next',
+ outcome: 'retracted',
+ });
+ assert.deepEqual(removedTransientMessageIds, ['message-next']);
});
test('complete events deliver the durable context compaction outcome to Desktop', () => {
diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts
index 6975dfbc6..717c43ab9 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts
@@ -871,14 +871,18 @@ test("retries a dispatched busy fallback with its original message identity", as
inlineReferences: [],
skillInvocation: { loaded: [], failed: [], receipts: [] },
});
- await assert.rejects(
- ipc.invoke("sessions:send", "session-1", {
+ assert.deepEqual(
+ await ipc.invoke("sessions:send", "session-1", {
type: "send",
- turnId: "turn-unknown",
+ messageId: "turn-unknown",
text: "ordinary chat keeps the existing failure contract",
}),
- (error: unknown) =>
- error instanceof RuntimeHostOperationError && error.code === 'outcome_unknown',
+ {
+ ok: false,
+ reason: "outcome_unknown",
+ messageId: "turn-unknown",
+ skillInvocation: { loaded: [], failed: [], receipts: [] },
+ },
);
assert.deepEqual(
await ipc.invoke("sessions:send", "side-session", {
diff --git a/apps/desktop/src/main/__tests__/transient-message-projection.test.ts b/apps/desktop/src/main/__tests__/transient-message-projection.test.ts
index 4ab798449..9ff7acf6b 100644
--- a/apps/desktop/src/main/__tests__/transient-message-projection.test.ts
+++ b/apps/desktop/src/main/__tests__/transient-message-projection.test.ts
@@ -67,6 +67,29 @@ test('canonicalizing one send does not hide a later transient send', () => {
assert.deepEqual([...pending.keys()], ['message-2']);
});
+test('keeps a transient message in submission order inside a sparse durable tail', () => {
+ const pending = new Map([[transient.id, transient]]);
+ const durable: StoredMessage[] = [
+ { type: 'user', id: 'old-user', turnId: 'old-turn', ts: 1, text: 'before' },
+ {
+ type: 'assistant',
+ id: 'later-assistant',
+ turnId: 'turn-1',
+ ts: 3,
+ text: 'after',
+ modelId: 'model-1',
+ },
+ ];
+
+ const projected = reconcileTransientMessages(pending, durable);
+
+ assert.deepEqual(projected.map((message) => message.id), [
+ 'old-user',
+ 'message-1',
+ 'later-assistant',
+ ]);
+});
+
test('keeps a transient message out of a sparse historical range', () => {
const live = { ...transient, id: 'message-live', turnId: 'message-live', text: 'latest prompt' };
const old = { ...transient, id: 'message-old', turnId: 'turn-old', ts: 1, text: 'old prompt' };
diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
index e6fc574ad..22ec921ba 100644
--- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
+++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
@@ -327,13 +327,21 @@ export function registerRuntimeHostSessionExecutionIpc(
(command.skillIds?.length ?? 0) === 0 &&
command.turnOrchestration === undefined
) {
- const submitted = await deps.client.submitMessage({
+ const submitted = await submitMessageWithReconnect(deps.client, {
sessionId,
messageId: command.messageId,
content: startInput.content,
placement: 'current_turn',
});
const skillInvocation = { loaded: [], failed: [], receipts: [] };
+ if (!submitted) {
+ return {
+ ok: false as const,
+ reason: 'outcome_unknown' as const,
+ messageId: command.messageId,
+ skillInvocation,
+ };
+ }
if (submitted.disposition === 'turn_started') {
deps.emitSessionsChanged('status-change', sessionId, {
turnId: submitted.turnId,
diff --git a/apps/desktop/src/renderer/app-shell-session-events.ts b/apps/desktop/src/renderer/app-shell-session-events.ts
index b8c9f0479..4aa0536b6 100644
--- a/apps/desktop/src/renderer/app-shell-session-events.ts
+++ b/apps/desktop/src/renderer/app-shell-session-events.ts
@@ -85,6 +85,7 @@ export function createAppShellSessionEventHandlers(options: {
setInteractionBySession: StateUpdater<InteractionQueues>;
setMessageQueueBySession?: StateUpdater<Record<string, MessageQueueUiState>>;
projectTransientMessage?: (sessionId: string, message: StoredMessage) => void;
+ removeTransientMessage?: (sessionId: string, messageId: string) => void;
onInteractionChanged?: (sessionId: string) => void;
/** A boundary decision settled: the session's execution boundary may have moved. */
onExecutionBoundaryChanged?: (sessionId: string) => void;
@@ -113,6 +114,7 @@ export function createAppShellSessionEventHandlers(options: {
setInteractionBySession,
setMessageQueueBySession,
projectTransientMessage,
+ removeTransientMessage,
onInteractionChanged,
onExecutionBoundaryChanged,
onContextCompactionOutcome,
@@ -318,6 +320,11 @@ export function createAppShellSessionEventHandlers(options: {
};
});
break;
+ case 'message_admission':
+ if (event.outcome === 'retracted') {
+ removeTransientMessage?.(sessionId, event.messageId);
+ }
+ break;
case 'text_complete':
void refreshMessages(sessionId, { requiredAssistantMessageId: event.messageId }).catch(() => false);
break;
diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx
index dbb6e57bd..9b22a4638 100644
--- a/apps/desktop/src/renderer/app-shell.tsx
+++ b/apps/desktop/src/renderer/app-shell.tsx
@@ -2205,6 +2205,7 @@ function AppShellContent({
setInteractionBySession,
setMessageQueueBySession,
projectTransientMessage: addTransientMessage,
+ removeTransientMessage,
displayBatch: sessionDisplayBatch,
onInteractionChanged: markInteractionChanged,
onExecutionBoundaryChanged: reloadActiveExecutionBoundary,
diff --git a/apps/desktop/src/renderer/transient-message-projection.ts b/apps/desktop/src/renderer/transient-message-projection.ts
index 231687c8c..da2db9c4d 100644
--- a/apps/desktop/src/renderer/transient-message-projection.ts
+++ b/apps/desktop/src/renderer/transient-message-projection.ts
@@ -31,8 +31,11 @@ export function reconcileTransientMessages(
): StoredMessage[] {
for (const message of durable) transient.delete(message.id);
if (transient.size === 0 || options.includeTransient === false) return [...durable];
- return [
- ...durable,
- ...[...transient.values()].sort((left, right) => left.ts - right.ts),
- ];
+ const projected = [...durable];
+ for (const message of [...transient.values()].sort((left, right) => left.ts - right.ts)) {
+ const nextIndex = projected.findIndex((candidate) => candidate.ts > message.ts);
+ if (nextIndex < 0) projected.push(message);
+ else projected.splice(nextIndex, 0, message);
+ }
+ return projected;
}
diff --git a/packages/cli/src/__tests__/pi-transcript.test.ts b/packages/cli/src/__tests__/pi-transcript.test.ts
index 35f237944..c1ee61d3b 100644
--- a/packages/cli/src/__tests__/pi-transcript.test.ts
+++ b/packages/cli/src/__tests__/pi-transcript.test.ts
@@ -517,6 +517,38 @@ describe('Maka Pi TUI transcript', () => {
]);
});
+ test('keeps a transient user row before later durable output in a sparse replacement', () => {
+ const state = createMakaPiTranscriptState();
+ replaceTranscriptWithStoredMessages(state, [
+ { type: 'user', id: 'old-user', turnId: 'old-turn', ts: 1, text: 'before' },
+ ]);
+ appendUserPrompt(state, 'send now', 'message-1', true);
+ state.entries.push({ kind: 'assistant', messageId: 'later-assistant', text: 'after' });
+
+ replaceTranscriptWithStoredMessages(
+ state,
+ [
+ { type: 'user', id: 'old-user', turnId: 'old-turn', ts: 1, text: 'before' },
+ {
+ type: 'assistant',
+ id: 'later-assistant',
+ turnId: 'turn-1',
+ ts: 3,
+ text: 'after',
+ modelId: 'model-1',
+ },
+ ],
+ { preserveTransientMessages: true },
+ );
+
+ assert.deepEqual(
+ state.entries.map((entry) =>
+ entry.kind === 'user' || entry.kind === 'assistant' ? entry.messageId : entry.kind,
+ ),
+ ['old-user', 'message-1', 'later-assistant'],
+ );
+ });
+
test('reconciles a transient user row by messageId when durable history arrives', () => {
const state = createMakaPiTranscriptState();
appendUserPrompt(state, 'send now', 'message-1', true);
@@ -530,6 +562,25 @@ describe('Maka Pi TUI transcript', () => {
assert.deepEqual(state.entries, [{ kind: 'user', messageId: 'message-1', text: 'send now' }]);
});
+ test('removes only the transient row named by a retracted admission', () => {
+ const state = createMakaPiTranscriptState();
+ appendUserPrompt(state, 'keep this', 'message-kept', true);
+ appendUserPrompt(state, 'take this back', 'message-retracted', true);
+
+ applyMakaSessionEventToTranscript(
+ state,
+ event({
+ type: 'message_admission',
+ messageId: 'message-retracted',
+ outcome: 'retracted',
+ }),
+ );
+
+ assert.deepEqual(state.entries, [
+ { kind: 'user', messageId: 'message-kept', text: 'keep this', transient: true },
+ ]);
+ });
+
test('reconciles a live steering event into its transient message position', () => {
const state = createMakaPiTranscriptState();
appendUserPrompt(state, 'send now', 'message-1', true);
diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts
index 82f200057..9cc645f15 100644
--- a/packages/cli/src/__tests__/pi-tui-runner.test.ts
+++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts
@@ -2009,6 +2009,73 @@ describe('Maka Pi TUI runner', () => {
await run;
});
+ test('removes an idle transient message after a definite Host rejection', async () => {
+ const terminal = new FakeTerminal();
+ const driver = new SteeringTurnDriver();
+ driver.nextSubmitError = new Error('Session is archived');
+ const run = runMakaPiTui({
+ title: 'Maka',
+ driver,
+ cwd: '/repo',
+ model: 'm',
+ connectionSlug: 'c',
+ permissionMode: 'bypass',
+ terminal,
+ });
+
+ terminal.input('do not leave a ghost row');
+ terminal.input('\r');
+ await waitFor(() =>
+ plainTerminalOutput(terminal.screenOutput()).includes('Session is archived'),
+ );
+ assert.equal(
+ plainTerminalOutput(terminal.screenOutput()).includes('do not leave a ghost row'),
+ false,
+ );
+
+ terminal.input('/exit');
+ terminal.input('\r');
+ await run;
+ });
+
+ test('keeps an admitted message when a Host-started turn attaches from a sparse tail', async () => {
+ const terminal = new FakeTerminal();
+ const driver = new SteeringTurnDriver();
+ driver.startedTurnMessages = [
+ {
+ type: 'assistant',
+ id: 'later-assistant',
+ turnId: 'turn-started',
+ ts: 2,
+ text: 'Later durable output',
+ modelId: 'model-1',
+ },
+ ];
+ const run = runMakaPiTui({
+ title: 'Maka',
+ driver,
+ cwd: '/repo',
+ model: 'm',
+ connectionSlug: 'c',
+ permissionMode: 'bypass',
+ terminal,
+ });
+
+ terminal.input('keep the accepted identity');
+ terminal.input('\r');
+ await waitFor(() =>
+ plainTerminalOutput(terminal.screenOutput()).includes('Later durable output'),
+ );
+ assert.match(plainTerminalOutput(terminal.screenOutput()), /keep the accepted identity/);
+
+ terminal.input('\x1b');
+ terminal.input('\x1b');
+ await waitFor(() => terminal.progressStates.at(-1) === false);
+ terminal.input('/exit');
+ terminal.input('\r');
+ await run;
+ });
+
test('opens /transcript during a running turn instead of steering it', async () => {
const terminal = new FakeTerminal();
const driver = new SteeringTurnDriver();
@@ -2292,8 +2359,8 @@ describe('Maka Pi TUI runner', () => {
terminal.input('\x1b'); // interrupt
await waitFor(() => terminal.progressStates.at(-1) === false);
// The authoritative queue is cleared and only the followup comes back as
- // a draft. Both already-sent message rows keep their stable identities;
- // interrupting delivery must not make either presentation disappear.
+ // a draft. The consumed steering row remains for canonical reconciliation;
+ // the retracted followup row is removed while its text moves to the editor.
await waitFor(() => {
const screen = plainTerminalOutput(terminal.screenOutput());
return (
@@ -6760,10 +6827,12 @@ class SteeringTurnDriver implements MakaSessionDriver {
readonly steered: string[] = [];
readonly queuedMessages: string[] = [];
readonly turnOrchestrations: Array<MakaPreparePromptOptions['turnOrchestration']> = [];
+ nextSubmitError: Error | undefined;
+ startedTurnMessages: StoredMessage[] = [];
retractCalls = 0;
rewindTargets: RewindTarget[] = [];
- private steering: string[] = [];
- private followup: string[] = [];
+ private steering: Array<{ messageId: string; text: string }> = [];
+ private followup: Array<{ messageId: string; text: string }> = [];
private pendingEvents: SessionEvent[] = [];
private wakeTurn: (() => void) | null = null;
private turnOpen = false;
@@ -6804,8 +6873,8 @@ class SteeringTurnDriver implements MakaSessionDriver {
id: `queue-update-${this.eventSeq}`,
turnId: 'turn-1',
ts: this.eventSeq,
- steering: [...this.steering],
- followup: [...this.followup],
+ steering: this.steering.map((entry) => entry.text),
+ followup: this.followup.map((entry) => entry.text),
});
this.wakeTurn?.();
this.wakeTurn = null;
@@ -6827,12 +6896,17 @@ class SteeringTurnDriver implements MakaSessionDriver {
}
async submitMessage(text: string, options: MakaSubmitMessageOptions) {
+ if (this.nextSubmitError) {
+ const error = this.nextSubmitError;
+ this.nextSubmitError = undefined;
+ throw error;
+ }
if (!this.turnOpen) {
const turn = await this.preparePrompt(text);
queueMicrotask(() =>
this.startedTurnListener?.({
...turn,
- messages: [],
+ messages: this.startedTurnMessages,
summary: fakeSessionSummary(turn.sessionId),
}),
);
@@ -6840,10 +6914,10 @@ class SteeringTurnDriver implements MakaSessionDriver {
}
if (options.placement === 'current_turn') {
this.steered.push(text);
- this.steering.push(text);
+ this.steering.push({ messageId: options.messageId, text });
} else {
this.queuedMessages.push(text);
- this.followup.push(text);
+ this.followup.push({ messageId: options.messageId, text });
}
this.emitQueueUpdate();
return {
@@ -6862,10 +6936,24 @@ class SteeringTurnDriver implements MakaSessionDriver {
async retractQueued(): Promise<string> {
this.retractCalls += 1;
- const joined = [...this.steering, ...this.followup].join('\n\n');
+ const retracted = [...this.steering, ...this.followup];
+ const joined = retracted.map((entry) => entry.text).join('\n\n');
this.steering = [];
this.followup = [];
this.emitQueueUpdate();
+ for (const entry of retracted) {
+ this.eventSeq += 1;
+ this.pendingEvents.push({
+ type: 'message_admission',
+ id: `message-retracted-${this.eventSeq}`,
+ turnId: 'turn-1',
+ ts: this.eventSeq,
+ messageId: entry.messageId,
+ outcome: 'retracted',
+ });
+ }
+ this.wakeTurn?.();
+ this.wakeTurn = null;
return joined;
}
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 51c0fc1cb..5d3944e48 100644
--- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
+++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
@@ -1174,6 +1174,33 @@ describe('Runtime Host Maka Session driver', () => {
});
});
+ test('keeps an unknown message admission available for transcript reconciliation', async () => {
+ const subscription = new FakeSubscription(continuitySnapshot(), Promise.resolve([]));
+ const connection = new FakeConnection([subscription]);
+ connection.messageSubmitOutcomes.push(
+ new RuntimeHostOperationError(
+ 'turn.message.submit',
+ 'outcome_unknown',
+ 'Message disposition cannot be proven in this Host Epoch',
+ ),
+ );
+ const driver = createRuntimeHostMakaSessionDriver({
+ connection: connection.value,
+ cwd: '/tmp',
+ llmConnectionSlug: 'openai-main',
+ model: 'gpt-5',
+ });
+ await driver.switchSession('session-1');
+
+ assert.deepEqual(
+ await driver.submitMessage!('Keep this visible', {
+ messageId: 'message-unknown',
+ placement: 'current_turn',
+ }),
+ { messageId: 'message-unknown', disposition: 'outcome_unknown' },
+ );
+ });
+
test('projects the acknowledgement that releases a question answered through the Host', async () => {
const subscription = new FakeSubscription(
continuitySnapshot({ interactions: { pending: [pendingQuestion()] } }),
@@ -1795,6 +1822,7 @@ class FakeConnection {
readonly goalControlOutcomes: Array<GoalProjection | Error> = [];
/** Scripted goal.query results, shifted per call; defaults to null (no goal). */
readonly goalQueryResults: Array<GoalProjection | null> = [];
+ readonly messageSubmitOutcomes: Array<OperationOutput<'turn.message.submit'> | Error> = [];
readonly value: RuntimeHostMakaSessionDriverInput['connection'];
constructor(
@@ -1899,13 +1927,19 @@ class FakeConnection {
: operation === 'session.execution_boundary.query'
? this.executionBoundary
: operation === 'turn.message.submit'
- ? {
- disposition:
- (input as OperationInput<'turn.message.submit'>).placement === 'next_turn'
- ? 'followup'
- : 'steering',
- queueRevision: 2,
- }
+ ? (() => {
+ const outcome = this.messageSubmitOutcomes.shift();
+ if (outcome instanceof Error) throw outcome;
+ return (
+ outcome ?? {
+ disposition:
+ (input as OperationInput<'turn.message.submit'>).placement === 'next_turn'
+ ? 'followup'
+ : 'steering',
+ queueRevision: 2,
+ }
+ );
+ })()
: operation === 'queue.retract'
? {
hostEpoch: 'host-1',
diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts
index c576831d9..8d5b79b23 100644
--- a/packages/cli/src/pi-transcript.ts
+++ b/packages/cli/src/pi-transcript.ts
@@ -344,18 +344,33 @@ export function replaceTranscriptWithStoredMessages(
options: { preserveTransientMessages?: boolean } = {},
): void {
const durableMessageIds = new Set(messages.map((message) => message.id));
+ const durableEntries = foldStoredShellRunChildren(storedMessagesToTranscriptEntries(messages));
+ const durableEntryIds = new Set(durableEntries.map(transcriptEntryId).filter(Boolean));
const transientEntries = options.preserveTransientMessages
- ? state.entries.filter(
- (entry): entry is Extract<MakaPiTranscriptEntry, { kind: 'user' }> =>
- entry.kind === 'user' &&
- entry.transient === true &&
- !durableMessageIds.has(entry.messageId),
- )
+ ? state.entries.flatMap((entry, index) => {
+ if (
+ entry.kind !== 'user' ||
+ entry.transient !== true ||
+ durableMessageIds.has(entry.messageId)
+ ) {
+ return [];
+ }
+ const nextDurableId = state.entries
+ .slice(index + 1)
+ .map(transcriptEntryId)
+ .find((messageId) => messageId !== undefined && durableEntryIds.has(messageId));
+ return [{ entry, nextDurableId }];
+ })
: [];
- state.entries = [
- ...foldStoredShellRunChildren(storedMessagesToTranscriptEntries(messages)),
- ...transientEntries,
- ];
+ const unanchoredTransientEntries: MakaPiTranscriptEntry[] = [];
+ for (const transient of transientEntries) {
+ const nextIndex = transient.nextDurableId
+ ? durableEntries.findIndex((entry) => transcriptEntryId(entry) === transient.nextDurableId)
+ : -1;
+ if (nextIndex < 0) unanchoredTransientEntries.push(transient.entry);
+ else durableEntries.splice(nextIndex, 0, transient.entry);
+ }
+ state.entries = [...unanchoredTransientEntries, ...durableEntries];
clearPendingInteractions(state);
state.expandAllTools = false;
state.expandAllThinking = false;
@@ -376,6 +391,19 @@ export function replaceTranscriptWithStoredMessages(
}
}
+function transcriptEntryId(entry: MakaPiTranscriptEntry): string | undefined {
+ switch (entry.kind) {
+ case 'user':
+ case 'assistant':
+ case 'thinking':
+ return entry.messageId;
+ case 'tool':
+ return entry.toolUseId;
+ default:
+ return undefined;
+ }
+}
+
/**
* Fill durable tool details that are intentionally absent from Runtime Host
* live events without applying session-switch reset semantics.
@@ -739,6 +767,17 @@ export function applyMakaSessionEventToTranscript(
appendUserPrompt(state, event.content.displayText ?? event.content.text, event.messageId);
break;
+ case 'message_admission':
+ if (event.outcome === 'retracted') {
+ state.entries = state.entries.filter(
+ (entry) =>
+ entry.kind !== 'user' ||
+ entry.transient !== true ||
+ entry.messageId !== event.messageId,
+ );
+ }
+ break;
+
case 'queue_update':
// Authoritative snapshot from the runtime; mirror it for the pending bar.
state.steering = [...event.steering];
diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts
index 1a81fa16e..ad4536a3a 100644
--- a/packages/cli/src/pi-tui-runner.ts
+++ b/packages/cli/src/pi-tui-runner.ts
@@ -1190,7 +1190,9 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
if (superseded()) return;
if (authoritativeAttachedTurn) {
adoptSessionMetadata(authoritativeAttachedTurn.summary);
- replaceTranscript(authoritativeAttachedTurn.messages);
+ replaceTranscript(authoritativeAttachedTurn.messages, {
+ preserveTransientMessages: true,
+ });
shellRunHydration.reset();
if (input.listShellRunUpdates) {
await shellRunHydration.hydrate(authoritativeAttachedTurn.sessionId);
@@ -1258,6 +1260,15 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
// here as "ended without completion" — never report that against the
// adopted Session.
if (superseded()) return;
+ if (
+ request.kind === 'external' &&
+ request.turnOrchestration === undefined &&
+ input.driver.submitMessage &&
+ optimisticUserEntry?.kind === 'user'
+ ) {
+ removeTransientUserMessage(optimisticUserEntry.messageId);
+ optimisticUserEntry = undefined;
+ }
appendTurnFailureToTranscript(state, error);
attention.attentionNeeded();
shellRunElapsedTicker.sync();
diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts
index fda363e6a..562c85060 100644
--- a/packages/cli/src/runtime-host-session-driver.ts
+++ b/packages/cli/src/runtime-host-session-driver.ts
@@ -401,16 +401,24 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver {
this.#assertCurrentSession(sessionId, sessionGeneration);
this.#adoptLoadedConfiguration(configuration);
const modelText = options.modelText ?? text;
- const result = await this.#request('turn.message.submit', {
- originHostEpoch: this.#connection.hostEpoch,
- sessionId,
- messageId: options.messageId,
- content: {
- text: modelText,
- ...(modelText === text ? {} : { displayText: text }),
- },
- placement: options.placement,
- });
+ let result;
+ try {
+ result = await this.#request('turn.message.submit', {
+ originHostEpoch: this.#connection.hostEpoch,
+ sessionId,
+ messageId: options.messageId,
+ content: {
+ text: modelText,
+ ...(modelText === text ? {} : { displayText: text }),
+ },
+ placement: options.placement,
+ });
+ } catch (error) {
+ if (error instanceof RuntimeHostOperationError && error.code === 'outcome_unknown') {
+ return { messageId: options.messageId, disposition: 'outcome_unknown' };
+ }
+ throw error;
+ }
return { messageId: options.messageId, disposition: result.disposition };
}
diff --git a/packages/cli/src/session-driver.ts b/packages/cli/src/session-driver.ts
index ecfd909b6..e55cf7195 100644
--- a/packages/cli/src/session-driver.ts
+++ b/packages/cli/src/session-driver.ts
@@ -106,7 +106,7 @@ export interface MakaSubmitMessageOptions {
export interface MakaMessageAdmission {
messageId: string;
- disposition: 'steering' | 'followup' | 'turn_started';
+ disposition: 'steering' | 'followup' | 'turn_started' | 'outcome_unknown';
}
export class SkillInvocationBlockedError extends Error {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org