From 61429ef57a495ff6e7ea4d2a1c5d2cc31f32bdf8 Mon Sep 17 00:00:00 2001 From: SunnyShu Date: Mon, 3 Aug 2026 14:40:16 +0800 Subject: [PATCH] fix(agents): retain durable transcript-repair record when assistant turn persistence fails (#117845) * [AI] fix(agents): retain durable transcript-repair backlog for delivered assistant finals A delivered assistant reply was silently lost when the canonical transcript append failed after a successful send: the finalizer logged the failure and continued delivery, and clearing the transport-replay marker removed the only durable copy of the final (P1 data loss). - finalizeEmbeddedAgentCommand now records a durable pendingTranscriptRepair backlog on the session entry whenever the transcript writer throws, while pendingFinalDelivery cleanup stays unchanged. - The backlog is ordered and appended per turn (deduplicated by per-run lifecycle generation, never by reply text), so consecutive failures and identical reply texts each keep their own recoverable record. - The next CLI/embedded turn best-effort re-appends each missing assistant turn through the real SQLite writer (loaded via the existing lazy attempt-execution runtime boundary) and clears recovered entries; failures keep the backlog and never block the turn. - Replays use exact per-turn idempotency (a per-run idempotency key with scan-assistant lookup) instead of tail-text gap-fill, so a distinct failed turn is re-appended even when an earlier persisted assistant message has identical text, and repeated replays stay idempotent. - Finals owned by another transcript writer (assistantTranscriptOwned) are never queued, avoiding duplicate assistant messages after a user-turn-only writer failure. - Session-entry shape normalization round-trips the backlog and upgrades a legacy single record into the array; slot-key registry updated. - Regression tests: delivery-success + failed write keeps the repair record, next-turn recovery, consecutive failures (distinct and identical texts), recovery after an equal-text earlier assistant message, runtime-owned-final exclusion, and shape normalization; the scoped max-lines exception for the compaction-rotation suite is mirrored in oxlint-config.test.ts. Fixes #117793 Co-Authored-By: Codex * [AI] fix(agents): recover transcript repairs across session rotation; split repair tests assistant-transcript-repair now migrates a pending repair record to the rotated successor session instead of retaining a predecessor-scoped record that can never be written: rotation keeps the same session key while replacing the entry's session id, so the guard previously skipped the record on every later pass and the delivered turn stayed missing from durable history. The recovered turn is appended to the successor transcript under a tested ownership rule. The transcript-repair scenarios moved out of the oversized compaction-rotation suite into a focused test module (assistant-transcript-repair.test.ts), and the new max-lines override plus its oxlint-config assertion were removed per repo policy. An end-to-end regression covers failed predecessor persistence followed by a rotated successor and a successful repair. Co-Authored-By: Codex * refactor(sessions): keep transcript repair array-only * fix(agents): repair transcript before next turn --------- Co-authored-by: Codex Co-authored-by: Dallin Romney Co-authored-by: sallyom --- src/agents/agent-command.ts | 26 + .../assistant-transcript-repair.test.ts | 632 ++++++++++++++++++ .../command/assistant-transcript-repair.ts | 165 +++++ .../command/attempt-execution.runtime.ts | 1 + src/agents/command/attempt-execution.ts | 2 +- src/agents/command/post-run.ts | 23 + src/agents/command/runtime-loaders.ts | 8 + src/agents/command/session.ts | 1 + src/config/sessions/sessions.test.ts | 62 ++ src/config/sessions/store-entry-shape.ts | 49 +- src/config/sessions/types.ts | 22 + src/plugins/session-entry-slot-keys.ts | 1 + 12 files changed, 990 insertions(+), 2 deletions(-) create mode 100644 src/agents/command/assistant-transcript-repair.test.ts create mode 100644 src/agents/command/assistant-transcript-repair.ts diff --git a/src/agents/agent-command.ts b/src/agents/agent-command.ts index 20fe5547f96b..1fe747fcd311 100644 --- a/src/agents/agent-command.ts +++ b/src/agents/agent-command.ts @@ -14,6 +14,7 @@ import { withAgentRunLifecycleGeneration, } from "../infra/agent-events.js"; import { clearAgentRunContext } from "../infra/agent-run-registry.js"; +import { formatErrorMessage } from "../infra/errors.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { isSubagentSessionKey } from "../routing/session-key.js"; import { defaultRuntime, type RuntimeEnv } from "../runtime.js"; @@ -31,6 +32,7 @@ import { } from "./agent-command-restart-recovery.js"; import { resolveAgentRuntimeConfig } from "./agent-runtime-config.js"; import { runAcpAgentCommand } from "./command/acp-execution.js"; +import { repairPendingAssistantTranscriptTurns } from "./command/assistant-transcript-repair.js"; import { emitIngressModelUsageDiagnostic, ingressDiagnosticChannel, @@ -221,6 +223,30 @@ async function agentCommandInternal( }, }); return await sessionWorkAdmission.run(async () => { + if (sessionStore && sessionKey && !suppressVisibleSessionEffects) { + try { + await repairPendingAssistantTranscriptTurns({ + context: { + sessionKey, + sessionEntry, + sessionStore, + storePath, + sessionAgentId, + config: cfg, + }, + }); + sessionEntry = sessionStore[sessionKey] ?? sessionEntry; + } catch (error) { + if (!isNewSession) { + throw error; + } + // A reset starts a fresh transcript. Do not let predecessor repair + // state leak into it when the old transcript remains unavailable. + log.warn( + `Could not repair predecessor transcript before session reset for ${sessionKey}: ${formatErrorMessage(error)}`, + ); + } + } if (opts.deliver === true) { const sendPolicy = resolveSendPolicy({ cfg, diff --git a/src/agents/command/assistant-transcript-repair.test.ts b/src/agents/command/assistant-transcript-repair.test.ts new file mode 100644 index 000000000000..e92005def16b --- /dev/null +++ b/src/agents/command/assistant-transcript-repair.test.ts @@ -0,0 +1,632 @@ +/** Focused tests for durable assistant-transcript repair across turns and session rotation. */ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { setReplyPayloadMetadata } from "../../auto-reply/reply-payload.js"; +import type { SessionEntry } from "../../config/sessions.js"; +import { + listSessionEntries, + loadTranscriptEvents, +} from "../../config/sessions/session-accessor.js"; +import type { appendExactAssistantMessageToSessionTranscript } from "../../config/sessions/transcript.runtime.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import type { EmbeddedAgentRunResult } from "../embedded-agent.js"; +import type { loadManifestModelCatalog } from "../model-catalog.js"; +import type { persistCliTurnTranscript } from "./attempt-execution.js"; +import type { runAgentAttempt } from "./attempt-execution.runtime.js"; +import type { persistSessionEntry } from "./session-helpers.js"; + +type ProviderModelNormalizationParams = { provider: string; context: { modelId: string } }; +type LoadManifestModelCatalogParams = Parameters[0]; +type RunAgentAttempt = typeof runAgentAttempt; +type PersistCliTurnTranscript = typeof persistCliTurnTranscript; +type AppendExactAssistantMessage = typeof appendExactAssistantMessageToSessionTranscript; +type PersistSessionEntry = typeof persistSessionEntry; +type CliCompactionParams = { + sessionEntry?: SessionEntry; + sessionKey: string; + sessionStore?: Record; + storePath?: string; +}; + +const state = vi.hoisted(() => ({ + cfg: undefined as OpenClawConfig | undefined, + workspaceDir: undefined as string | undefined, + agentDir: undefined as string | undefined, + runAgentAttemptMock: vi.fn(), + loadManifestModelCatalogMock: vi.fn((_params: LoadManifestModelCatalogParams) => []), + normalizeProviderModelIdWithRuntimeMock: vi.fn( + (_params: ProviderModelNormalizationParams) => undefined, + ), + runCliTurnCompactionLifecycleMock: vi.fn( + async (params: CliCompactionParams) => params.sessionEntry, + ), + deliverAgentCommandResultMock: vi.fn(), + emitAgentEventMock: vi.fn(), + persistCliTurnTranscriptMock: vi.fn(), + persistCliTurnTranscriptReal: undefined as PersistCliTurnTranscript | undefined, + appendExactAssistantMessageMock: vi.fn(), + appendExactAssistantMessageReal: undefined as AppendExactAssistantMessage | undefined, + persistSessionEntryMock: vi.fn(), + persistSessionEntryReal: undefined as PersistSessionEntry | undefined, + deliveryFreshEntries: [] as Array, +})); + +vi.mock("../../config/io.js", () => ({ + getRuntimeConfig: () => state.cfg, + readConfigFileSnapshotForWrite: async () => ({ snapshot: { valid: false } }), +})); + +vi.mock("../agent-runtime-config.js", () => ({ + resolveAgentRuntimeConfig: async () => ({ + loadedRaw: state.cfg, + sourceConfig: state.cfg, + cfg: state.cfg, + }), +})); + +vi.mock("../../plugins/plugin-metadata-snapshot.js", () => ({ + isPluginMetadataSnapshotCompatible: () => false, + resolvePluginMetadataSnapshot: () => ({ plugins: [] }), +})); + +vi.mock("../agent-scope.js", async () => { + const actual = await vi.importActual("../agent-scope.js"); + return { + ...actual, + clearAutoFallbackPrimaryProbeSelection: vi.fn(), + entryMatchesAutoFallbackPrimaryProbe: () => false, + hasSessionAutoModelFallbackProvenance: () => false, + listAgentIds: () => ["main"], + markAutoFallbackPrimaryProbe: vi.fn(), + resolveAutoFallbackPrimaryProbe: () => undefined, + resolveAgentConfig: () => undefined, + resolveAgentDir: () => state.agentDir ?? "/tmp/openclaw-agent", + resolveDefaultAgentId: () => "main", + resolveEffectiveModelFallbacks: () => undefined, + resolveSessionAgentId: () => "main", + resolveAgentWorkspaceDir: () => state.workspaceDir ?? "/tmp/openclaw-workspace", + }; +}); + +vi.mock("../model-catalog.js", () => ({ + loadManifestModelCatalog: (params: LoadManifestModelCatalogParams) => + state.loadManifestModelCatalogMock(params), +})); + +vi.mock("../model-catalog.runtime.js", () => ({ + loadPreparedModelCatalogSnapshot: vi.fn(async () => ({ + entries: [], + routeVariants: [], + })), +})); + +vi.mock("../provider-model-normalization.runtime.js", () => ({ + normalizeProviderModelIdWithRuntime: (params: { + provider: string; + context: { modelId: string }; + }) => state.normalizeProviderModelIdWithRuntimeMock(params), +})); + +vi.mock("../harness/runtime-plugin.js", () => ({ + ensureSelectedAgentHarnessPlugin: vi.fn(async () => undefined), +})); + +vi.mock("../workspace.js", () => ({ + ensureAgentWorkspace: vi.fn(async () => undefined), +})); + +vi.mock("../auth-profiles/store.js", async () => { + const actual = await vi.importActual( + "../auth-profiles/store.js", + ); + return { + ...actual, + ensureAuthProfileStore: () => ({ profiles: {} }), + saveAuthProfileStore: vi.fn(), + updateAuthProfileStoreWithLock: vi.fn(async () => ({ profiles: {} })), + }; +}); + +vi.mock("../../acp/control-plane/manager.js", () => ({ + getAcpSessionManager: () => ({ + resolveSession: () => null, + }), +})); + +vi.mock("../../skills/runtime/remote.js", () => ({ + getRemoteSkillEligibility: () => ({ enabled: false, reason: "test" }), +})); + +vi.mock("../../skills/runtime/session-snapshot.js", () => ({ + resolveReusableWorkspaceSkillSnapshot: () => ({ + shouldRefresh: true, + snapshot: { + prompt: "", + skills: [], + resolvedSkills: [], + version: 0, + }, + }), +})); + +vi.mock("../exec-defaults.js", () => ({ + resolveNodeExecEligibility: () => ({ canExec: false }), +})); + +vi.mock("../model-fallback-runner.js", () => ({ + runWithModelFallback: async (params: { + provider: string; + model: string; + run: (provider: string, model: string) => Promise; + }) => ({ + result: await params.run(params.provider, params.model), + provider: params.provider, + model: params.model, + attempts: [], + }), +})); + +vi.mock("./attempt-execution.runtime.js", async () => { + const actual = await vi.importActual( + "./attempt-execution.runtime.js", + ); + return { + ...actual, + runAgentAttempt: (...args: Parameters) => state.runAgentAttemptMock(...args), + persistCliTurnTranscript: (...args: Parameters) => { + state.persistCliTurnTranscriptReal = actual.persistCliTurnTranscript; + if (state.persistCliTurnTranscriptMock) { + return state.persistCliTurnTranscriptMock(...args); + } + return actual.persistCliTurnTranscript(...args); + }, + }; +}); + +vi.mock("../../config/sessions/transcript.runtime.js", async () => { + const actual = await vi.importActual< + typeof import("../../config/sessions/transcript.runtime.js") + >("../../config/sessions/transcript.runtime.js"); + return { + ...actual, + appendExactAssistantMessageToSessionTranscript: ( + ...args: Parameters + ) => { + state.appendExactAssistantMessageReal = actual.appendExactAssistantMessageToSessionTranscript; + return state.appendExactAssistantMessageMock(...args); + }, + }; +}); + +vi.mock("./session-helpers.js", async () => { + const actual = + await vi.importActual("./session-helpers.js"); + return { + ...actual, + persistSessionEntry: (...args: Parameters) => { + state.persistSessionEntryReal = actual.persistSessionEntry; + return state.persistSessionEntryMock(...args); + }, + }; +}); + +vi.mock("./cli-compaction.js", () => ({ + runCliTurnCompactionLifecycle: (params: CliCompactionParams) => + state.runCliTurnCompactionLifecycleMock(params), +})); + +vi.mock("../../infra/agent-events.js", async () => { + const actual = await vi.importActual( + "../../infra/agent-events.js", + ); + return { + ...actual, + emitAgentEvent: (...args: Parameters) => { + state.emitAgentEventMock(...args); + return actual.emitAgentEvent(...args); + }, + }; +}); + +vi.mock("./delivery.runtime.js", () => ({ + deliverAgentCommandResult: (params: unknown) => state.deliverAgentCommandResultMock(params), +})); + +let agentCommand: typeof import("../agent-command.js").agentCommand; + +beforeAll(async () => { + agentCommand = (await import("../agent-command.js")).agentCommand; +}); + +beforeEach(async () => { + vi.clearAllMocks(); + state.loadManifestModelCatalogMock.mockReturnValue([]); + state.normalizeProviderModelIdWithRuntimeMock.mockImplementation(() => undefined); + state.runCliTurnCompactionLifecycleMock.mockImplementation( + async (params: CliCompactionParams) => params.sessionEntry, + ); + state.persistCliTurnTranscriptMock.mockImplementation( + async (...args: Parameters) => + state.persistCliTurnTranscriptReal?.(...args), + ); + state.appendExactAssistantMessageMock.mockImplementation( + async (...args: Parameters) => + state.appendExactAssistantMessageReal?.(...args) ?? { + ok: false, + reason: "missing real transcript append", + }, + ); + state.persistSessionEntryMock.mockImplementation( + async (...args: Parameters) => state.persistSessionEntryReal?.(...args), + ); + state.deliveryFreshEntries = []; + state.deliverAgentCommandResultMock.mockImplementation( + async (params: { + resolveFreshSessionEntryForDelivery?: () => Promise; + }) => { + state.deliveryFreshEntries.push(await params.resolveFreshSessionEntryForDelivery?.()); + return { deliverySucceeded: true }; + }, + ); + const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-repair-e2e-")); + state.workspaceDir = path.join(tmpDir, "workspace"); + state.agentDir = path.join(tmpDir, "agent"); + await fs.mkdir(state.workspaceDir, { recursive: true }); + await fs.mkdir(state.agentDir, { recursive: true }); + state.cfg = { + session: { + store: path.join(tmpDir, "sessions.json"), + }, + agents: { + defaults: { + models: { + "openai/gpt-5.5": {}, + }, + }, + }, + } as OpenClawConfig; +}); + +afterEach(async () => { + const storePath = state.cfg?.session?.store; + state.cfg = undefined; + state.workspaceDir = undefined; + state.agentDir = undefined; + if (storePath) { + await fs.rm(path.dirname(storePath), { recursive: true, force: true }); + } +}); + +function makeResult(params: { + sessionId: string; + text?: string; + runner?: "cli" | "embedded"; + payloads?: EmbeddedAgentRunResult["payloads"]; +}): EmbeddedAgentRunResult { + return { + payloads: params.payloads ?? (params.text ? [{ text: params.text }] : []), + meta: { + durationMs: 1, + stopReason: "end_turn", + executionTrace: { + runner: params.runner ?? "embedded", + fallbackUsed: false, + winnerProvider: "openai", + winnerModel: "gpt-5.5", + }, + ...(params.text ? { finalAssistantVisibleText: params.text } : {}), + agentMeta: { + sessionId: params.sessionId, + provider: "openai", + model: "gpt-5.5", + }, + }, + }; +} + +async function readSessionMessages(params: { + agentId: string; + sessionId: string; + storePath: string; +}) { + return (await loadTranscriptEvents(params)) + .filter( + (entry): entry is { message: unknown; type: "message" } => + typeof entry === "object" && + entry !== null && + "message" in entry && + "type" in entry && + entry.type === "message", + ) + .map((entry) => entry.message); +} + +async function readMessageSequence(params: { + agentId: string; + sessionId: string; + storePath: string; +}): Promise> { + return (await readSessionMessages(params)).map((message) => { + const value = message as { + role?: string; + content?: string | Array<{ type?: string; text?: string }>; + }; + const text = Array.isArray(value.content) + ? value.content.map((part) => part.text ?? "").join("") + : (value.content ?? ""); + return { role: value.role, text }; + }); +} + +function requireStorePath(): string { + const storePath = state.cfg?.session?.store; + if (!storePath) { + throw new Error("missing test session store path"); + } + return storePath; +} + +function findStoredSessionEntry(sessionKey: string): SessionEntry | undefined { + return listSessionEntries({ storePath: requireStorePath() }).find( + (candidate) => candidate.sessionKey === sessionKey, + )?.entry; +} + +describe("assistant transcript repair", () => { + it("records the canonical payload fallback when transcript persistence fails", async () => { + const sessionId = "transcript-write-failure"; + const sessionKey = `agent:main:explicit:${sessionId}`; + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ + sessionId, + runner: "cli", + payloads: [{ text: "first payload" }, { text: "second payload" }], + }), + ); + state.persistCliTurnTranscriptMock.mockRejectedValueOnce( + new Error("simulated transcript table corruption"), + ); + + await agentCommand({ message: "first prompt", sessionId, sessionKey, cwd: state.workspaceDir }); + + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toEqual([ + expect.objectContaining({ + id: expect.any(String), + text: "first payload\n\nsecond payload", + }), + ]); + }); + + it("repairs the prior assistant before the next model attempt", async () => { + const sessionId = "transcript-repair-order"; + const sessionKey = `agent:main:explicit:${sessionId}`; + state.runAgentAttemptMock.mockImplementationOnce(async (params) => { + await params.userTurnTranscriptRecorder?.persistApproved(); + return makeResult({ sessionId, text: "assistant one", runner: "cli" }); + }); + state.persistCliTurnTranscriptMock.mockRejectedValueOnce( + new Error("simulated transcript table corruption"), + ); + await agentCommand({ message: "user one", sessionId, sessionKey, cwd: state.workspaceDir }); + + state.runAgentAttemptMock.mockImplementationOnce(async (params) => { + expect( + await readMessageSequence({ agentId: "main", sessionId, storePath: requireStorePath() }), + ).toEqual([ + { role: "user", text: "user one" }, + { role: "assistant", text: "assistant one" }, + ]); + await params.userTurnTranscriptRecorder?.persistApproved(); + return makeResult({ sessionId, text: "assistant two", runner: "cli" }); + }); + await agentCommand({ message: "user two", sessionId, sessionKey, cwd: state.workspaceDir }); + + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toBeUndefined(); + expect( + await readMessageSequence({ agentId: "main", sessionId, storePath: requireStorePath() }), + ).toEqual([ + { role: "user", text: "user one" }, + { role: "assistant", text: "assistant one" }, + { role: "user", text: "user two" }, + { role: "assistant", text: "assistant two" }, + ]); + }); + + it("blocks a continuing turn while transcript repair storage is still unavailable", async () => { + const sessionId = "transcript-repair-barrier"; + const sessionKey = `agent:main:explicit:${sessionId}`; + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: "assistant one", runner: "cli" }), + ); + state.persistCliTurnTranscriptMock.mockRejectedValueOnce( + new Error("simulated transcript table corruption"), + ); + await agentCommand({ message: "user one", sessionId, sessionKey, cwd: state.workspaceDir }); + + state.appendExactAssistantMessageMock.mockResolvedValueOnce({ + ok: false, + reason: "simulated transcript table corruption", + }); + await expect( + agentCommand({ message: "user two", sessionId, sessionKey, cwd: state.workspaceDir }), + ).rejects.toThrow("pending transcript recovery"); + + expect(state.runAgentAttemptMock).toHaveBeenCalledOnce(); + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toHaveLength(1); + }); + + it("does not duplicate a repaired assistant when backlog cleanup retries", async () => { + const sessionId = "transcript-repair-cleanup-retry"; + const sessionKey = `agent:main:explicit:${sessionId}`; + const repairedText = "assistant one"; + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: repairedText, runner: "cli" }), + ); + state.persistCliTurnTranscriptMock.mockRejectedValueOnce( + new Error("simulated transcript table corruption"), + ); + await agentCommand({ message: "user one", sessionId, sessionKey, cwd: state.workspaceDir }); + + state.persistSessionEntryMock.mockRejectedValueOnce(new Error("simulated cleanup failure")); + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: "assistant two", runner: "cli" }), + ); + await agentCommand({ message: "user two", sessionId, sessionKey, cwd: state.workspaceDir }); + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toHaveLength(1); + + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: "assistant three", runner: "cli" }), + ); + await agentCommand({ message: "user three", sessionId, sessionKey, cwd: state.workspaceDir }); + + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toBeUndefined(); + const assistantTexts = ( + await readMessageSequence({ agentId: "main", sessionId, storePath: requireStorePath() }) + ) + .filter((message) => message.role === "assistant") + .map((message) => message.text); + expect(assistantTexts.filter((text) => text === repairedText)).toHaveLength(1); + }); + + it("does not queue a repair for a final owned by another transcript writer", async () => { + const sessionId = "transcript-owner-boundary"; + const sessionKey = `agent:main:explicit:${sessionId}`; + const text = "runtime-owned assistant final"; + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ + sessionId, + text, + runner: "cli", + payloads: [setReplyPayloadMetadata({ text }, { assistantTranscriptOwned: true })], + }), + ); + state.persistCliTurnTranscriptMock.mockRejectedValueOnce( + new Error("simulated transcript table corruption"), + ); + + const result = await agentCommand({ + message: "first prompt", + sessionId, + sessionKey, + cwd: state.workspaceDir, + channel: "discord", + to: "discord:dm:123", + accountId: "main", + deliver: true, + }); + + expect(result).toMatchObject({ deliverySucceeded: true }); + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toBeUndefined(); + }); + + it("re-appends a missing turn whose text matches an earlier assistant message", async () => { + const sessionId = "transcript-repair-equal-tail"; + const sessionKey = `agent:main:explicit:${sessionId}`; + const sameText = "OK"; + let persistFailuresRemaining = 0; + state.persistCliTurnTranscriptMock.mockImplementation( + async (...args: Parameters) => { + if (persistFailuresRemaining > 0) { + persistFailuresRemaining -= 1; + throw new Error("simulated transcript table corruption"); + } + return state.persistCliTurnTranscriptReal?.(...args); + }, + ); + + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: sameText, runner: "cli" }), + ); + await agentCommand({ + message: "first prompt", + sessionId, + sessionKey, + cwd: state.workspaceDir, + }); + + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: sameText, runner: "cli" }), + ); + persistFailuresRemaining = 1; + await agentCommand({ + message: "second prompt", + sessionId, + sessionKey, + cwd: state.workspaceDir, + }); + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toHaveLength(1); + + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ sessionId, text: "third turn reply", runner: "cli" }), + ); + await agentCommand({ + message: "third prompt", + sessionId, + sessionKey, + cwd: state.workspaceDir, + }); + + expect(findStoredSessionEntry(sessionKey)?.pendingTranscriptRepair).toBeUndefined(); + const assistantTexts = ( + await readMessageSequence({ agentId: "main", sessionId, storePath: requireStorePath() }) + ) + .filter((message) => message.role === "assistant") + .map((message) => message.text); + expect(assistantTexts.filter((text) => text === sameText)).toHaveLength(2); + }); + + it("does not carry an unavailable predecessor repair into a reset session", async () => { + const now = Date.now(); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(now); + try { + const predecessorSessionId = "reset-predecessor"; + const sessionKey = `agent:main:explicit:${predecessorSessionId}`; + state.cfg!.session!.reset = { mode: "idle", idleMinutes: 1 }; + state.runAgentAttemptMock.mockResolvedValueOnce( + makeResult({ + sessionId: predecessorSessionId, + text: "missing predecessor reply", + runner: "cli", + }), + ); + state.persistCliTurnTranscriptMock.mockRejectedValueOnce( + new Error("simulated transcript table corruption"), + ); + await agentCommand({ + message: "old user", + sessionId: predecessorSessionId, + sessionKey, + cwd: state.workspaceDir, + }); + + vi.setSystemTime(now + 120_000); + state.appendExactAssistantMessageMock.mockResolvedValueOnce({ + ok: false, + reason: "simulated transcript table corruption", + }); + state.runAgentAttemptMock.mockImplementationOnce(async (params) => + makeResult({ sessionId: params.sessionId, text: "fresh reply", runner: "cli" }), + ); + await agentCommand({ message: "fresh user", sessionKey, cwd: state.workspaceDir }); + + const successor = findStoredSessionEntry(sessionKey); + expect(successor?.sessionId).not.toBe(predecessorSessionId); + expect(successor?.pendingTranscriptRepair).toBeUndefined(); + expect( + await readMessageSequence({ + agentId: "main", + sessionId: successor!.sessionId, + storePath: requireStorePath(), + }), + ).toEqual([ + { role: "user", text: "fresh user" }, + { role: "assistant", text: "fresh reply" }, + ]); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/src/agents/command/assistant-transcript-repair.ts b/src/agents/command/assistant-transcript-repair.ts new file mode 100644 index 000000000000..42df8a89407d --- /dev/null +++ b/src/agents/command/assistant-transcript-repair.ts @@ -0,0 +1,165 @@ +import { randomUUID } from "node:crypto"; +import type { SessionEntry, PendingTranscriptRepairState } from "../../config/sessions/types.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { formatErrorMessage } from "../../infra/errors.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; +import { runAgentHarnessBeforeMessageWriteHook } from "../harness/hook-helpers.js"; +import { loadTranscriptAppendRuntime } from "./runtime-loaders.js"; +import { persistSessionEntry } from "./session-helpers.js"; + +const log = createSubsystemLogger("agents/assistant-transcript-repair"); + +type AssistantTranscriptRepairContext = { + sessionKey: string; + sessionEntry: SessionEntry | undefined; + sessionStore?: Record; + storePath: string; + sessionAgentId: string; + config: OpenClawConfig; +}; + +const EMPTY_USAGE = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + total: 0, + }, +} as const; + +/** Records a final whose canonical transcript append failed. */ +export async function persistAssistantTranscriptRepairRecord(params: { + context: AssistantTranscriptRepairContext; + replyText: string; + provider?: string; + model?: string; + runOwnedSessionId: string; +}): Promise { + const { context, replyText, provider, model, runOwnedSessionId } = params; + if (!replyText.trim() || !context.sessionStore || !context.sessionKey.trim()) { + return; + } + const now = Date.now(); + const existing = context.sessionStore[context.sessionKey] ?? context.sessionEntry; + if (!existing) { + return; + } + const nextRepair: PendingTranscriptRepairState = { + id: randomUUID(), + text: replyText, + ...(provider?.trim() ? { provider: provider.trim() } : {}), + ...(model?.trim() ? { model: model.trim() } : {}), + createdAt: now, + }; + try { + await persistSessionEntry({ + sessionStore: context.sessionStore, + sessionKey: context.sessionKey, + storePath: context.storePath, + initialEntry: existing, + entry: { + ...existing, + pendingTranscriptRepair: [...(existing.pendingTranscriptRepair ?? []), nextRepair], + updatedAt: now, + }, + shouldPersist: (current) => + current?.sessionId === runOwnedSessionId && current.abortedLastRun !== true, + }); + } catch (error) { + log.warn( + `Failed to persist assistant transcript repair record for ${context.sessionKey}: ${formatErrorMessage(error)}`, + ); + } +} + +/** + * Restores missing assistant finals before another turn can observe or extend + * the transcript. Append failures are an admission barrier: continuing would + * give the model incomplete history and make the eventual append out of order. + */ +export async function repairPendingAssistantTranscriptTurns(params: { + context: AssistantTranscriptRepairContext; +}): Promise { + const { context } = params; + if (!context.sessionStore || !context.sessionKey) { + return; + } + const entry = context.sessionStore[context.sessionKey] ?? context.sessionEntry; + const backlog = entry?.pendingTranscriptRepair; + if (!entry || !backlog?.length) { + return; + } + + const { appendExactAssistantMessageToSessionTranscript } = await loadTranscriptAppendRuntime(); + const remaining = [...backlog]; + while (remaining.length > 0) { + const item = remaining[0]!; + let result: Awaited>; + try { + result = await appendExactAssistantMessageToSessionTranscript({ + agentId: context.sessionAgentId, + sessionKey: context.sessionKey, + expectedSessionId: entry.sessionId, + storePath: context.storePath, + config: context.config, + updateMode: "file-only", + idempotencyKey: `transcript-repair:${item.id}`, + beforeMessageWrite: runAgentHarnessBeforeMessageWriteHook, + message: { + role: "assistant", + content: [{ type: "text", text: item.text }], + api: "cli", + provider: item.provider ?? "cli", + model: item.model ?? "default", + usage: EMPTY_USAGE, + stopReason: "stop", + timestamp: item.createdAt, + }, + }); + } catch (error) { + log.warn( + `Assistant transcript repair failed for ${context.sessionKey}: ${formatErrorMessage(error)}`, + ); + throw new Error("Previous assistant reply is still pending transcript recovery; retry.", { + cause: error, + }); + } + + if (!result.ok && result.code !== "blocked") { + log.warn(`Assistant transcript repair failed for ${context.sessionKey}: ${result.reason}`); + throw new Error("Previous assistant reply is still pending transcript recovery; retry."); + } + remaining.shift(); + if (result.ok) { + log.info(`Re-appended missing assistant transcript turn for ${context.sessionKey}`); + } else { + log.warn(`Dropped blocked assistant transcript repair for ${context.sessionKey}`); + } + } + + const current = context.sessionStore[context.sessionKey]; + if (!current || current.sessionId !== entry.sessionId) { + return; + } + try { + await persistSessionEntry({ + sessionStore: context.sessionStore, + sessionKey: context.sessionKey, + storePath: context.storePath, + initialEntry: current, + entry: { ...current, pendingTranscriptRepair: undefined, updatedAt: Date.now() }, + shouldPersist: (latest) => latest?.sessionId === entry.sessionId, + }); + } catch (error) { + // The exact append key makes a later retry safe after cleanup failure. + log.warn( + `Failed to clear assistant transcript repair record for ${context.sessionKey}: ${formatErrorMessage(error)}`, + ); + } +} diff --git a/src/agents/command/attempt-execution.runtime.ts b/src/agents/command/attempt-execution.runtime.ts index 70a77e77a7b8..1832cf9221cc 100644 --- a/src/agents/command/attempt-execution.runtime.ts +++ b/src/agents/command/attempt-execution.runtime.ts @@ -12,6 +12,7 @@ export { emitAcpRuntimeEvent, persistAcpTurnTranscript, persistCliTurnTranscript, + resolveCliTranscriptReplyText, runAgentAttempt, sessionFileHasContent, } from "./attempt-execution.js"; diff --git a/src/agents/command/attempt-execution.ts b/src/agents/command/attempt-execution.ts index 8bbde38ae260..5b3728a84559 100644 --- a/src/agents/command/attempt-execution.ts +++ b/src/agents/command/attempt-execution.ts @@ -384,7 +384,7 @@ async function persistTextTurnTranscript( return { kind: "persisted", sessionEntry: turn.sessionEntry }; } -function resolveCliTranscriptReplyText(result: EmbeddedAgentRunResult): string { +export function resolveCliTranscriptReplyText(result: EmbeddedAgentRunResult): string { const visibleText = result.meta.finalAssistantVisibleText?.trim(); if (visibleText) { return visibleText; diff --git a/src/agents/command/post-run.ts b/src/agents/command/post-run.ts index f4d34854eef5..bcc018863181 100644 --- a/src/agents/command/post-run.ts +++ b/src/agents/command/post-run.ts @@ -17,6 +17,7 @@ import { isHeartbeatLifecycleRunKind } from "../bootstrap-mode.js"; import { persistPendingFinalDeliveryMarker } from "../pending-final-delivery-marker.js"; import type { AgentRunSessionTarget } from "../run-session-target.js"; import { throwAgentRunRestartAbortReason } from "../run-termination.js"; +import { persistAssistantTranscriptRepairRecord } from "./assistant-transcript-repair.js"; import type { PreparedAgentCommandExecution } from "./prepare.js"; import type { EmbeddedAgentAttempt } from "./run-embedded-attempt.js"; import { @@ -212,6 +213,28 @@ export async function finalizeEmbeddedAgentCommand(params: { log.warn( `Turn transcript persistence failed for ${sessionKey ?? sessionId}: ${error instanceof Error ? error.message : String(error)}`, ); + if ( + sessionStore && + sessionKey && + !params.suppressVisibleSessionEffects && + !sessionReboundDuringRun && + !assistantTranscriptOwned + ) { + await persistAssistantTranscriptRepairRecord({ + context: { + sessionKey: internalSessionTarget?.sessionKey ?? sessionKey ?? effectiveSessionId, + sessionEntry: internalSessionTarget?.sessionEntry ?? sessionEntry, + sessionStore, + storePath: internalSessionTarget?.storePath ?? storePath, + sessionAgentId: internalSessionTarget?.agentId ?? sessionAgentId, + config: cfg, + }, + replyText: attemptExecutionRuntime.resolveCliTranscriptReplyText(result), + provider: result.meta.agentMeta?.provider, + model: result.meta.agentMeta?.model, + runOwnedSessionId, + }); + } } } diff --git a/src/agents/command/runtime-loaders.ts b/src/agents/command/runtime-loaders.ts index bccca77f6213..e8cba155254b 100644 --- a/src/agents/command/runtime-loaders.ts +++ b/src/agents/command/runtime-loaders.ts @@ -14,6 +14,7 @@ type SessionStoreRuntime = typeof import("./session-store.runtime.js"); type CliCompactionRuntime = typeof import("./cli-compaction.js"); type TranscriptResolveRuntime = typeof import("../../config/sessions/transcript-resolve.runtime.js"); +type TranscriptAppendRuntime = typeof import("../../config/sessions/transcript.runtime.js"); type CliDepsRuntime = typeof import("../../cli/deps.js"); type ExecDefaultsRuntime = typeof import("../exec-defaults.js"); type SkillsRuntime = { @@ -48,6 +49,9 @@ const cliCompactionRuntimeLoader = createLazyImportLoader( const transcriptResolveRuntimeLoader = createLazyImportLoader( () => import("../../config/sessions/transcript-resolve.runtime.js"), ); +const transcriptAppendRuntimeLoader = createLazyImportLoader( + () => import("../../config/sessions/transcript.runtime.js"), +); const cliDepsRuntimeLoader = createLazyImportLoader( () => import("../../cli/deps.js"), ); @@ -101,6 +105,10 @@ export function loadTranscriptResolveRuntime(): Promise { + return transcriptAppendRuntimeLoader.load(); +} + export function loadExecDefaultsRuntime(): Promise { return execDefaultsRuntimeLoader.load(); } diff --git a/src/agents/command/session.ts b/src/agents/command/session.ts index 81511539fc20..4051b0d8322d 100644 --- a/src/agents/command/session.ts +++ b/src/agents/command/session.ts @@ -92,6 +92,7 @@ export function clearRotatedSessionMetadata(entry: SessionEntry): SessionEntry { sessionStartedAt: undefined, sessionDiffBaseline: undefined, lastInteractionAt: undefined, + pendingTranscriptRepair: undefined, }; transitionMainSessionRecovery(next, { kind: "clear" }); clearAllCliSessions(next); diff --git a/src/config/sessions/sessions.test.ts b/src/config/sessions/sessions.test.ts index d31eeec1388a..9ea149313c3c 100644 --- a/src/config/sessions/sessions.test.ts +++ b/src/config/sessions/sessions.test.ts @@ -81,6 +81,68 @@ it("normalizes boolean-only pending delivery as transport-only", () => { }); }); +it("normalizes and preserves the durable assistant transcript repair backlog", () => { + expect( + normalizePersistedSessionEntryShape({ + sessionId: "session-1", + updatedAt: 42, + pendingTranscriptRepair: [ + { + id: "repair-1", + text: "recoverable assistant final", + provider: "openai", + model: "gpt-5.5", + createdAt: 42, + }, + { + id: "repair-2", + text: "second recoverable assistant final", + createdAt: 43, + }, + ], + }), + ).toMatchObject({ + pendingTranscriptRepair: [ + { + id: "repair-1", + text: "recoverable assistant final", + provider: "openai", + model: "gpt-5.5", + createdAt: 42, + }, + { + id: "repair-2", + text: "second recoverable assistant final", + createdAt: 43, + }, + ], + }); +}); + +it("drops a non-array assistant transcript repair value", () => { + expect( + normalizePersistedSessionEntryShape({ + sessionId: "session-1", + updatedAt: 42, + pendingTranscriptRepair: { + id: "repair-1", + text: "recoverable assistant final", + createdAt: 42, + }, + }), + ).not.toHaveProperty("pendingTranscriptRepair"); +}); + +it("drops malformed assistant transcript repair records", () => { + expect( + normalizePersistedSessionEntryShape({ + sessionId: "session-1", + updatedAt: 42, + pendingTranscriptRepair: [{ kind: "transport-only" }], + }), + ).not.toHaveProperty("pendingTranscriptRepair"); +}); + describe("session path safety", () => { it("rejects unsafe session IDs", () => { const unsafeSessionIds = [ diff --git a/src/config/sessions/store-entry-shape.ts b/src/config/sessions/store-entry-shape.ts index f3d378cd03c8..248506efc12e 100644 --- a/src/config/sessions/store-entry-shape.ts +++ b/src/config/sessions/store-entry-shape.ts @@ -3,7 +3,7 @@ import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { parseAgentSessionKey } from "../../routing/session-key.js"; import { validateSessionId } from "./paths.js"; -import type { SessionEntry } from "./types.js"; +import type { PendingTranscriptRepairState, SessionEntry } from "./types.js"; // Persisted stores may contain old or malformed ids; reject path-like ids before use. function isSafeSessionId(value: unknown): value is string { @@ -87,6 +87,14 @@ export function projectCanonicalSessionEntryShape(value: Record } else { delete canonicalValue.pendingFinalDelivery; } + const pendingTranscriptRepair = normalizePendingTranscriptRepair( + canonicalValue.pendingTranscriptRepair, + ); + if (pendingTranscriptRepair) { + canonicalValue.pendingTranscriptRepair = pendingTranscriptRepair; + } else { + delete canonicalValue.pendingTranscriptRepair; + } const reason = normalizeOptionalString(fallbackNoticeReason); const fallbackNotice = normalizeFallbackNotice(canonicalValue.fallbackNotice) ?? @@ -147,6 +155,45 @@ function normalizePendingFinalDelivery( return value.kind === "replayable" && text ? { kind: "replayable", text, ...base } : undefined; } +function normalizePendingTranscriptRepair( + value: unknown, +): SessionEntry["pendingTranscriptRepair"] | undefined { + if (!Array.isArray(value) || value.length === 0) { + return undefined; + } + const normalized: NonNullable = []; + for (const item of value) { + const record = normalizePendingTranscriptRepairRecord(item); + if (record) { + normalized.push(record); + } + } + return normalized.length > 0 ? normalized : undefined; +} + +function normalizePendingTranscriptRepairRecord( + value: unknown, +): PendingTranscriptRepairState | undefined { + if (!isRecord(value)) { + return undefined; + } + const id = normalizeOptionalString(value.id); + const text = normalizeOptionalString(value.text); + const createdAt = normalizeOptionalTimestamp(value.createdAt); + if (!id || !text || createdAt === undefined) { + return undefined; + } + const provider = normalizeOptionalString(value.provider); + const model = normalizeOptionalString(value.model); + return { + id, + text, + ...(provider ? { provider } : {}), + ...(model ? { model } : {}), + createdAt, + }; +} + function normalizeFallbackNotice(value: unknown): SessionEntry["fallbackNotice"] | undefined { if (!isRecord(value) || value.kind !== "active") { return undefined; diff --git a/src/config/sessions/types.ts b/src/config/sessions/types.ts index 7cda220d67d6..c6d70361f127 100644 --- a/src/config/sessions/types.ts +++ b/src/config/sessions/types.ts @@ -65,6 +65,21 @@ type PendingFinalDeliveryState = { intentId?: string; } & ({ kind: "replayable"; text: string } | { kind: "transport-only" }); +/** + * Durable transcript-repair record: an assistant final that was delivered to + * the user but could not be appended to the canonical transcript. Kept + * separate from `pendingFinalDelivery` so transport-replay cleanup never drops + * the only copy of the missing assistant turn. + */ +export type PendingTranscriptRepairState = { + /** Stable identity for retry-safe transcript insertion. */ + id: string; + text: string; + provider?: string; + model?: string; + createdAt: number; +}; + type FallbackNoticeState = { kind: "active"; selectedModel: string; @@ -514,6 +529,13 @@ type SessionEntryCore = SessionRestartRecoveryState & outputTokens?: number; totalTokens?: number; pendingFinalDelivery?: PendingFinalDeliveryState; + /** + * Ordered durable backlog of delivered assistant finals that failed to + * reach the canonical transcript. Session admission restores each item + * before another turn can extend that transcript. Kept as a list so + * independently admitted writers never overwrite an earlier reply. + */ + pendingTranscriptRepair?: PendingTranscriptRepairState[]; /** * Whether totalTokens reflects a fresh context snapshot for the latest run. * Undefined means legacy/unknown freshness; false forces consumers to treat diff --git a/src/plugins/session-entry-slot-keys.ts b/src/plugins/session-entry-slot-keys.ts index ee00f3755b14..4513a2d2c020 100644 --- a/src/plugins/session-entry-slot-keys.ts +++ b/src/plugins/session-entry-slot-keys.ts @@ -163,6 +163,7 @@ const SESSION_ENTRY_RESERVED_SLOT_KEY_LIST = [ "hookExternalContentSource", "acp", "quotaSuspension", + "pendingTranscriptRepair", "visibility", ] as const satisfies ReadonlyArray< keyof SessionEntry | "__proto__" | "constructor" | "prototype" | "sessionFile" | "transcriptPath"