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/09/07 00:26:24 UTC
(maka) 09/15: fix(runtime): resume transcript migration safely after interruption
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch refactor/4791-single-transcript-authority
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 354145761e48646a8c6bcb5e1b8add683bd3e220
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Sun Sep 6 20:03:57 2026 +0800
fix(runtime): resume transcript migration safely after interruption
---
.../src/__tests__/runtime-ledger-repair.test.ts | 118 +++++++++++++++++++--
packages/runtime/src/runtime-ledger-repair.ts | 12 ++-
packages/runtime/src/session-manager.ts | 9 +-
3 files changed, 124 insertions(+), 15 deletions(-)
diff --git a/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts b/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts
index 3a0c41ce3c..6d967b05e9 100644
--- a/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts
+++ b/packages/runtime/src/__tests__/runtime-ledger-repair.test.ts
@@ -22,6 +22,7 @@ import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';
+import { DatabaseSync } from 'node:sqlite';
import type { StoredMessage } from '@maka/core/session';
import { createSqliteAgentRunStore } from '@maka/storage/agent-run-store';
import { createExternalSessionAdapterRegistry } from '@maka/storage/external-sessions';
@@ -36,6 +37,7 @@ import {
import { runtimeInvocationFailureClass } from '../runtime-event-read-model.js';
import { backfillRuntimeEventsFromStoredMessages } from '../runtime-event-backfill.js';
import { RuntimeLedgerRepair } from '../runtime-ledger-repair.js';
+import { BackendRegistry, SessionManager } from '../session-manager.js';
test('repairs imported transcript turns into provider-neutral canonical history', async () => {
const root = await mkdtemp(join(tmpdir(), 'maka-transcript-ledger-repair-'));
@@ -104,7 +106,6 @@ test('repairs imported transcript turns into provider-neutral canonical history'
const repair = new RuntimeLedgerRepair({
runtimeEventStore: runtimeEvents,
readMessages: (sessionId) => sessions.readMessages(sessionId),
- now: () => 100,
});
await repair.materializeTranscriptLedger(session);
@@ -284,7 +285,6 @@ test('an imported snapshot cutoff survives materialization as aborted', async ()
const repair = new RuntimeLedgerRepair({
runtimeEventStore: runtimeEvents,
readMessages: (sessionId) => sessions.readMessages(sessionId),
- now: () => 100,
});
await repair.materializeTranscriptLedger(session);
@@ -339,7 +339,6 @@ test('does not import Host-handed-off transcript messages as synthetic runs', as
const repair = new RuntimeLedgerRepair({
runtimeEventStore: runtimeEvents,
readMessages: (sessionId) => sessions.readMessages(sessionId),
- now: () => 100,
});
await repair.materializeTranscriptLedger(session);
@@ -389,7 +388,6 @@ test('an imported turn with no terminal state is repaired to failed', async () =
const repair = new RuntimeLedgerRepair({
runtimeEventStore: runtimeEvents,
readMessages: (sessionId) => sessions.readMessages(sessionId),
- now: () => 100,
});
await repair.materializeTranscriptLedger(session);
@@ -464,18 +462,33 @@ test("converts Maka's own legacy transcript whole, and resumes an interrupted co
const repair = new RuntimeLedgerRepair({
runtimeEventStore: runtimeEvents,
readMessages: (sessionId) => sessions.readMessages(sessionId),
- now: () => 100,
});
- await repair.materializeTranscriptLedger(await sessions.readHeader(session.id));
- // The same transcript converts to the same events, so a second pass — the
- // retry after an interrupted one — adds nothing.
- await repair.materializeTranscriptLedger(await sessions.readHeader(session.id));
+ const append = runtimeEvents.appendRuntimeEvent.bind(runtimeEvents);
+ runtimeEvents.appendRuntimeEvent = async (sessionId, runId, event) => {
+ await append(sessionId, runId, event);
+ if (event.role === 'user') throw new Error('interrupted conversion');
+ };
+ await assert.rejects(
+ repair.materializeTranscriptLedger(await sessions.readHeader(session.id)),
+ /interrupted conversion/,
+ );
+ runtimeEvents.appendRuntimeEvent = append;
+ const [interrupted] = await runtimeEvents.listSessionInvocations(session.id);
+ assert.ok(interrupted);
+ const prefix = await runtimeEvents.readRuntimeEvents(session.id, interrupted.runId);
+ assert.equal(interrupted.terminalEvent, undefined);
+ const resumed = new RuntimeLedgerRepair({
+ runtimeEventStore: runtimeEvents,
+ readMessages: (sessionId) => sessions.readMessages(sessionId),
+ });
+ await resumed.materializeTranscriptLedger(await sessions.readHeader(session.id));
const [run] = await runtimeEvents.listSessionInvocations(session.id);
assert.ok(run);
assert.equal(runtimeInvocationOutcome(run), 'completed');
const events = await runtimeEvents.readRuntimeEvents(session.id, run.runId);
+ assert.deepEqual(events.slice(0, prefix.length), prefix);
assert.deepEqual(
events.flatMap((event) => (event.content ? [event.content.kind] : [])),
['invocation_opened', 'text', 'function_call', 'function_response', 'system_note', 'text'],
@@ -486,6 +499,92 @@ test("converts Maka's own legacy transcript whole, and resumes an interrupted co
}
});
+test('startup recovery leaves an interrupted legacy conversion for the importer to finish', async () => {
+ const root = await mkdtemp(join(tmpdir(), 'maka-transcript-restart-'));
+ const sessions = createSessionStore(root);
+ const runs = createSqliteAgentRunStore(root);
+ const runtimeEvents = createSqliteRuntimeStore(join(root, 'runtime.sqlite'));
+ try {
+ const session = await sessions.create({
+ cwd: '/repo',
+ llmConnectionSlug: 'fake',
+ model: 'fake-model',
+ permissionMode: 'ask',
+ });
+ const db = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ db.prepare(
+ "UPDATE session_metadata SET payload_json = json_remove(payload_json, '$.transcriptLedgerVersion') WHERE session_id = ?",
+ ).run(session.id);
+ } finally {
+ db.close();
+ }
+ await sessions.appendMessages(session.id, [
+ {
+ type: 'user',
+ id: 'legacy-user',
+ turnId: 'legacy-turn',
+ ts: 10,
+ text: 'Keep this conversation',
+ },
+ {
+ type: 'assistant',
+ id: 'legacy-answer',
+ turnId: 'legacy-turn',
+ ts: 20,
+ text: 'The complete original answer',
+ modelId: 'fake-model',
+ },
+ {
+ type: 'turn_state',
+ id: 'legacy-end',
+ turnId: 'legacy-turn',
+ ts: 30,
+ status: 'completed',
+ partialOutputRetained: true,
+ },
+ ]);
+ const append = runtimeEvents.appendRuntimeEvent.bind(runtimeEvents);
+ runtimeEvents.appendRuntimeEvent = async (sessionId, runId, event) => {
+ await append(sessionId, runId, event);
+ if (event.role === 'user') throw new Error('interrupted conversion');
+ };
+ const repair = new RuntimeLedgerRepair({
+ runtimeEventStore: runtimeEvents,
+ readMessages: (id) => sessions.readMessages(id),
+ });
+ await assert.rejects(
+ repair.materializeTranscriptLedger(await sessions.readHeader(session.id)),
+ /interrupted conversion/,
+ );
+ runtimeEvents.appendRuntimeEvent = append;
+ let id = 0;
+ const manager = new SessionManager({
+ store: sessions,
+ runStore: runs,
+ runtimeEventStore: runtimeEvents,
+ backends: new BackendRegistry(),
+ now: () => 100,
+ newId: () => `recovery-${++id}`,
+ });
+ await manager.recoverInterruptedSessionsStrict({ sessionStore: sessions, agentRunStore: runs });
+ const [pending] = await runtimeEvents.listSessionInvocations(session.id);
+ assert.ok(pending);
+ assert.equal(pending.terminalEvent, undefined);
+ const messages = await manager.getMessages(session.id);
+ assert.deepEqual(
+ messages.map((message) => message.id),
+ ['legacy-user', 'legacy-answer', 'legacy-end'],
+ );
+ assert.equal((await sessions.readHeader(session.id)).transcriptLedgerVersion, 1);
+ } finally {
+ runtimeEvents.close();
+ await runs.close?.();
+ await sessions.close?.();
+ await rm(root, { recursive: true, force: true });
+ }
+});
+
test('a resolved Claude transcript replays as the conversation the user kept', async () => {
// The whole path, end to end: raw records → lineage resolution → conversion
// → Ledger materialization → the replay a continuation would be given.
@@ -635,7 +734,6 @@ test('a resolved Claude transcript replays as the conversation the user kept', a
const repair = new RuntimeLedgerRepair({
runtimeEventStore: runtimeEvents,
readMessages: (sessionId) => sessions.readMessages(sessionId),
- now: () => 100,
});
await repair.materializeTranscriptLedger(session);
diff --git a/packages/runtime/src/runtime-ledger-repair.ts b/packages/runtime/src/runtime-ledger-repair.ts
index 569ccaaf71..f8c7125e98 100644
--- a/packages/runtime/src/runtime-ledger-repair.ts
+++ b/packages/runtime/src/runtime-ledger-repair.ts
@@ -39,7 +39,6 @@ export interface RuntimeLedgerRepairDeps {
runtimeEventStore: RuntimeEventStore;
/** The legacy transcript this converter reads; nothing writes back to it. */
readMessages(sessionId: string): Promise<StoredMessage[]>;
- now: () => number;
}
export class RuntimeLedgerRepair {
@@ -107,7 +106,9 @@ export class RuntimeLedgerRepair {
// whole: its tool calls are the ones it would replay.
modelHistory: header.externalOrigin ? 'conversation_text' : 'full',
newId: transcriptEventIds(runId),
- now: this.deps.now,
+ // The payload must be as repeatable as its id: SQLite dedupes
+ // complete events, including the backfill provenance timestamps.
+ now: () => openedAt,
}).events,
];
for (const event of events) {
@@ -141,6 +142,13 @@ export class RuntimeLedgerRepair {
}
}
+/** Synthetic conversion runs belong to the importer, never execution recovery. */
+export function isTranscriptLedgerInvocation(
+ invocation: Pick<RuntimeInvocationRecord, 'sessionId' | 'turnId' | 'runId'>,
+): boolean {
+ return invocation.runId === transcriptRunId(invocation.sessionId, invocation.turnId);
+}
+
function transcriptRunId(sessionId: string, turnId: string): string {
const digest = createHash('sha256').update(sessionId).update('\0').update(turnId).digest('hex');
return `transcript-${digest.slice(0, 48)}`;
diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts
index 18d05f6cdd..26d98f39f1 100644
--- a/packages/runtime/src/session-manager.ts
+++ b/packages/runtime/src/session-manager.ts
@@ -153,7 +153,7 @@ import {
type RuntimeReadModelSessionView,
} from './runtime-read-model.js';
import { inspectAgentRunReadModel, type AgentRunInspectModel } from './agent-run-inspect.js';
-import { RuntimeLedgerRepair } from './runtime-ledger-repair.js';
+import { isTranscriptLedgerInvocation, RuntimeLedgerRepair } from './runtime-ledger-repair.js';
import {
buildRecoveredTerminalRuntimeEvent,
classifyTerminalRuntimeLedger,
@@ -903,7 +903,6 @@ export class SessionManager {
this.runtimeLedgerRepair = new RuntimeLedgerRepair({
runtimeEventStore: deps.runtimeEventStore,
readMessages: (sessionId) => deps.store.readMessages(sessionId),
- now: deps.now,
});
}
this.runtimeKernel = deps.runtimeKernel ?? new RuntimeKernel({ ...deps });
@@ -4383,7 +4382,11 @@ export class SessionManager {
): Promise<{ hasLedger: boolean; recovered: boolean }> {
if (!this.deps.runStore || !this.deps.runtimeEventStore)
return { hasLedger: false, recovered: false };
- const runs = await this.listInvocations(sessionId);
+ // The importer may have committed only a prefix before a restart. Sealing
+ // that prefix here would make the next read skip the unconverted history.
+ const runs = (await this.listInvocations(sessionId)).filter(
+ (run) => !isTranscriptLedgerInvocation(run),
+ );
if (runs.length === 0) return { hasLedger: false, recovered: false };
const continuationAuthority = runtimeContinuationAuthority(this.deps.runtimeEventStore);
const claimOwnedUnsettledRunIds = new Set<string>();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org