fix(copilot): keep tool transcript groups structurally complete (#114403)

* refactor(copilot): make SQLite own transcripts

* fix(copilot): harden transcript event projection

* test(copilot): tighten transcript journal types

* fix(copilot): retain unmatched usage metadata

* refactor(copilot): commit tool groups atomically

* fix(copilot): break journal bridge import cycle

* fix(copilot): preserve grouped transcript replay

* test(copilot): cover opaque replay state

* fix(copilot): invalidate lossy transcript replay

* fix(copilot): guard transcript rewrite topology

* fix(copilot): drain transcript finalization

* fix(copilot): preserve resumed session history

* fix(copilot): dedupe replayed transcript snapshots

* fix(copilot): isolate transcript write hooks

* fix(copilot): track staged transcript identities

* fix(copilot): isolate ephemeral transcript events

* fix(copilot): coalesce assistant snapshots

* fix(copilot): invalidate unprojected SDK context

* fix(copilot): guard resumed transcript replay

* fix(plugin-sdk): expose strict transcript append

* fix(copilot): persist journal provenance safely

* fix(copilot): separate live and durable deltas

* fix(copilot): require durable projection ownership

* fix(copilot): preserve strict append typing

* fix(copilot): gate deferred session deletion

* fix(copilot): preserve transcript replay identity

* fix(copilot): retain cited replay history

* fix(copilot): require complete replay provenance

* fix(copilot): reject ambiguous assistant snapshots

* fix(copilot): return latest assistant snapshot

* chore(copilot): align transcript lint gates

* test(lint): cover Copilot line budgets

* test(ci): dedupe Codex prewarm shard
This commit is contained in:
Peter Steinberger
2026-07-27 15:42:00 -04:00
committed by GitHub
parent bbc73c36bd
commit 3dc3370d9e
26 changed files with 3894 additions and 1287 deletions

View File

@@ -362,6 +362,18 @@
"rules": {
"max-lines": ["error", { "max": 1000, "skipBlankLines": true, "skipComments": true }]
}
},
{
"files": ["extensions/copilot/src/event-bridge.ts"],
"rules": {
"max-lines": ["error", { "max": 950, "skipBlankLines": true, "skipComments": true }]
}
},
{
"files": ["extensions/copilot/src/attempt-transcript-journal.test.ts"],
"rules": {
"max-lines": ["error", { "max": 1200, "skipBlankLines": true, "skipComments": true }]
}
}
]
}

View File

@@ -947,7 +947,7 @@ describe("createCopilotAgentHarness", () => {
sdkSessionId: "sdk-sess-warm",
pooledClient: { key: TEST_POOL_KEY, client },
});
return ATTEMPT_RESULT;
return { ...ATTEMPT_RESULT, journalValidated: true, sdkSessionId: "sdk-sess-warm" };
});
const harness = createCopilotAgentHarness({ pool });
@@ -956,13 +956,46 @@ describe("createCopilotAgentHarness", () => {
expect(mocks.runCopilotAttempt).toHaveBeenCalledTimes(2);
const secondCallParams = mocks.runCopilotAttempt.mock.calls[1]?.[0] as {
initialReplayState?: { sdkSessionId?: string; replayInvalid?: boolean };
initialReplayState?: {
journalValidated?: boolean;
sdkSessionId?: string;
replayInvalid?: boolean;
};
};
expect(secondCallParams.initialReplayState?.sdkSessionId).toBe("sdk-sess-warm");
expect(secondCallParams.initialReplayState?.journalValidated).toBe(true);
// Must not synthesize a replayInvalid signal: undefined → resumable.
expect(secondCallParams.initialReplayState?.replayInvalid).toBeUndefined();
});
it("clears journal provenance before a resumed attempt and restores it after validation", async () => {
const pool = makePoolMock();
const client = createMockCopilotClient({ deleteSession: vi.fn() });
const sessionStore = makeSessionStoreMock();
let call = 0;
mocks.runCopilotAttempt.mockImplementation(async (_params, deps) => {
call += 1;
deps.onSessionEstablished?.({
sdkSessionId: "sdk-sess-provenance",
pooledClient: { key: TEST_POOL_KEY, client },
});
if (call === 2) {
expect(sessionStore.entries.get("oc-sess-reuse")?.journalVersion).toBeUndefined();
}
return {
...ATTEMPT_RESULT,
journalValidated: true,
sdkSessionId: "sdk-sess-provenance",
};
});
const harness = createCopilotAgentHarness({ pool, sessionStore: sessionStore.store });
await harness.runAttempt(makeAttemptParams({ runId: "t1" }));
await harness.runAttempt(makeAttemptParams({ runId: "t2" }));
expect(sessionStore.entries.get("oc-sess-reuse")?.journalVersion).toBe(1);
});
it("blocks reuse while timed-out compaction is pending, then resumes after completion", async () => {
const pool = makePoolMock();
const sessionStore = makeSessionStoreMock();
@@ -1409,9 +1442,10 @@ describe("createCopilotAgentHarness", () => {
}),
);
const secondCallParams = mocks.runCopilotAttempt.mock.calls[1]?.[0] as {
initialReplayState?: { sdkSessionId?: string };
initialReplayState?: { journalValidated?: boolean; sdkSessionId?: string };
};
expect(secondCallParams.initialReplayState?.sdkSessionId).toBe("sdk-sess-sqlite");
expect(secondCallParams.initialReplayState?.journalValidated).toBeUndefined();
});
it("persists BYOK session compatibility with endpoint fingerprints instead of raw URLs", async () => {

View File

@@ -44,6 +44,7 @@ interface CreateCopilotAgentHarnessOptions {
}
interface TrackedSession {
journalVersion?: 1;
sdkSessionId: string;
client: CopilotClient;
clientOptions: ClientCreateOptions;
@@ -91,6 +92,7 @@ interface CopilotHistoryCompactSession {
export type CopilotSessionBinding = {
schemaVersion: 2;
journalVersion?: 1;
sdkSessionId: string;
compatKey: string;
compactKey: string;
@@ -107,7 +109,10 @@ type LegacyCopilotSessionBinding = {
updatedAt: number;
};
type CopilotAttemptSessionBinding = Pick<CopilotSessionBinding, "compatKey" | "sdkSessionId">;
type CopilotAttemptSessionBinding = Pick<
CopilotSessionBinding,
"compatKey" | "journalVersion" | "sdkSessionId"
>;
type DeferredCompactionCleanupOutcome = "aborted" | "completed" | "deadline";
type DeferredCompactionCleanup = {
abort: () => void;
@@ -160,6 +165,7 @@ function normalizeBinding(
value.compatKey.trim() === "" ||
typeof value.compactKey !== "string" ||
value.compactKey.trim() === "" ||
(value.journalVersion !== undefined && value.journalVersion !== 1) ||
(value.authMode !== "gitHubToken" &&
value.authMode !== "byok" &&
value.authMode !== "useLoggedInUser") ||
@@ -175,6 +181,7 @@ function normalizeBinding(
}
return {
schemaVersion: 2,
...(value.journalVersion === 1 ? { journalVersion: 1 as const } : {}),
sdkSessionId: value.sdkSessionId.trim(),
compatKey: value.compatKey,
compactKey: value.compactKey,
@@ -704,12 +711,13 @@ export function createCopilotAgentHarness(
? undefined
: lookupStoredBinding(options?.sessionStore, openclawSessionId)
: undefined;
const resumableSessionId =
const resumableBinding =
tracked && tracked.compatKey === currentCompatKey
? tracked.sdkSessionId
? tracked
: !tracked && stored && stored.compatKey === currentCompatKey
? stored.sdkSessionId
? stored
: undefined;
const resumableSessionId = resumableBinding?.sdkSessionId;
if (operation === "settled-tool-finalization" && !resumableSessionId) {
throw new Error(
"[copilot] cannot safely finalize a settled tool turn without its compatible SDK session",
@@ -739,15 +747,19 @@ export function createCopilotAgentHarness(
// still requiring the exact compatible native session above.
initialReplayState:
operation === "settled-tool-finalization"
? { sdkSessionId: resumableSessionId }
? {
...(resumableBinding?.journalVersion === 1 ? { journalValidated: true } : {}),
sdkSessionId: resumableSessionId,
}
: {
...params.initialReplayState,
...(resumableBinding?.journalVersion === 1 ? { journalValidated: true } : {}),
sdkSessionId: resumableSessionId,
},
} as AgentHarnessAttemptParams)
: params;
return runCopilotAttempt(effectiveParams, {
const result = await runCopilotAttempt(effectiveParams, {
pool,
...(operation === "settled-tool-finalization" ? { operation } : {}),
onSessionEstablished:
@@ -835,6 +847,32 @@ export function createCopilotAgentHarness(
}
: undefined,
});
if (operation === "attempt" && openclawSessionId) {
const attemptResult = result as AgentHarnessAttemptResult & {
journalValidated?: boolean;
sdkSessionId?: string;
};
const sdkSessionId = attemptResult.sdkSessionId;
const trackedSession = trackedSessions.get(openclawSessionId);
if (sdkSessionId && trackedSession?.sdkSessionId === sdkSessionId) {
const { journalVersion: _journalVersion, ...baseTracked } = trackedSession;
const nextTracked: TrackedSession = {
...baseTracked,
...(attemptResult.journalValidated ? { journalVersion: 1 } : {}),
};
trackedSessions.set(openclawSessionId, nextTracked);
registerStoredBinding(options?.sessionStore, openclawSessionId, {
schemaVersion: 2,
...(attemptResult.journalValidated ? { journalVersion: 1 } : {}),
sdkSessionId,
compatKey: nextTracked.compatKey,
compactKey: nextTracked.compactKey,
...sessionAuthFields(nextTracked),
updatedAt: Date.now(),
});
}
}
return result;
})();
inFlight.add(attemptPromise);
try {

View File

@@ -54,6 +54,7 @@ export function deferBackgroundCompactionCleanup(params: {
pool: CopilotClientPool;
cleanupByokProxy?: () => Promise<void>;
cleanupToolBridge?: () => void;
deleteSessionOnIncompleteCleanup: boolean;
finalizeNativeSubagents?: () => void;
sdkSessionId?: string;
session: SessionLike;
@@ -81,7 +82,11 @@ export function deferBackgroundCompactionCleanup(params: {
} catch {}
params.cleanupToolBridge?.();
await params.cleanupByokProxy?.();
if (outcome !== "completed" && params.sdkSessionId) {
if (
outcome !== "completed" &&
params.deleteSessionOnIncompleteCleanup &&
params.sdkSessionId
) {
try {
await params.handle.client.deleteSession(params.sdkSessionId);
} catch {}

View File

@@ -30,6 +30,8 @@ export function createResult(
params: AttemptParamsLike,
state: {
aborted?: boolean;
assistantTranscriptOwned?: boolean;
assistantTranscriptIdempotencyKey?: string;
assistantTexts?: string[];
currentAttemptAssistant?: AssistantMessage;
currentAttemptCompletedAssistant?: AssistantMessage;
@@ -37,8 +39,10 @@ export function createResult(
externalAbort?: boolean;
itemLifecycle?: { activeCount: number; completedCount: number; startedCount: number };
lastAssistant?: AssistantMessage;
journalValidated?: boolean;
lastToolError?: AgentHarnessAttemptResult["lastToolError"];
messagesSnapshot: AgentMessage[];
nativeReplayInvalid?: boolean;
now: () => number;
promptError: Error | undefined;
resumeFailureRecovered?: boolean;
@@ -52,13 +56,21 @@ export function createResult(
},
): AttemptResultWithSdkSessionId {
const promptError = state.promptError;
const transcriptPersistenceFailed =
(promptError as PromptErrorWithCode | undefined)?.code === "transcript_persistence_failed";
const timedOut = state.timedOut === true;
const toolMetas = state.toolMetas ?? [];
const replayMetadata =
params.operation === "settled-tool-finalization"
? { hadPotentialSideEffects: false, replaySafe: true }
? {
hadPotentialSideEffects: false,
replaySafe: state.nativeReplayInvalid !== true && !transcriptPersistenceFailed,
}
: computeReplayMetadata({
priorReplayInvalid: params.initialReplayState?.replayInvalid,
priorReplayInvalid:
params.initialReplayState?.replayInvalid === true ||
state.nativeReplayInvalid === true ||
transcriptPersistenceFailed,
priorHadPotentialSideEffects: params.initialReplayState?.hadPotentialSideEffects,
thisAttemptTimedOut: timedOut,
thisAttemptHadPotentialSideEffects: copilotToolMetasHavePotentialSideEffects(toolMetas),
@@ -82,7 +94,16 @@ export function createResult(
promptError !== undefined ? withPromptFailure(interruption, promptError) : interruption;
return {
terminal,
...(state.assistantTranscriptOwned
? {
assistantTranscriptOwned: true,
...(state.assistantTranscriptIdempotencyKey
? { assistantTranscriptIdempotencyKey: state.assistantTranscriptIdempotencyKey }
: {}),
}
: {}),
...(state.sdkSessionId ? { sdkSessionId: state.sdkSessionId } : {}),
...(state.journalValidated !== undefined ? { journalValidated: state.journalValidated } : {}),
assistantTexts: state.assistantTexts ?? [],
attemptUsage: state.usage,
cloudCodeAssistFormatError: false,
@@ -334,36 +355,6 @@ export function createSystemMessageContent(
export function isRawCopilotModelRun(params: AttemptParamsLike): boolean {
return params.modelRun === true || params.promptMode === "none";
}
export function readTailUserText(messages: AgentMessage[]): string | undefined {
const tail = messages[messages.length - 1];
if (!tail || tail.role !== "user") {
return undefined;
}
const content = (tail as { content?: unknown }).content;
if (typeof content === "string") {
return content;
}
if (Array.isArray(content)) {
for (const part of content) {
if (part && typeof part === "object" && (part as { type?: unknown }).type === "text") {
const text = (part as { text?: unknown }).text;
if (typeof text === "string" && text.length > 0) {
return text;
}
}
}
}
return undefined;
}
export function hasMirrorIdentity(message: AgentMessage): boolean {
const record = message as unknown as { __openclaw?: unknown };
const meta = record["__openclaw"];
if (!meta || typeof meta !== "object" || Array.isArray(meta)) {
return false;
}
const id = (meta as Record<string, unknown>).mirrorIdentity;
return typeof id === "string" && id.length > 0;
}
export function readString(value: unknown): string | undefined {
return typeof value === "string" && value.length > 0 ? value : undefined;
}

View File

@@ -28,6 +28,10 @@ import {
import { completeCopilotAttempt } from "./attempt-finalize.js";
import { resolveCopilotAttemptSandbox } from "./attempt-prepare.js";
import { createCopilotSessionSetup } from "./attempt-session-setup.js";
import {
createAttemptTranscriptJournal,
type AttemptTranscriptJournal,
} from "./attempt-transcript-journal.js";
import type {
AgentHarnessAttemptResult,
AttemptParamsLike,
@@ -94,10 +98,15 @@ export async function runCopilotExecution(context: {
let promptError: Error | undefined;
let sdkSessionId: string | undefined;
let sessionIdUsed = input.sessionId;
// Resumed sessions may predate the atomic journal or survive a crash. Only a
// session created under this journal can be deleted after incomplete cleanup.
let nativeSessionCreatedFresh = false;
let nativeSessionHistoryValidated = false;
let disconnectError: Error | undefined;
let handle: PooledClient | undefined;
let session: SessionLike | undefined;
let bridge: ReturnType<typeof attachEventBridge> | undefined;
let transcriptJournal: AttemptTranscriptJournal | undefined;
const nativeSubagentTaskMirror = createCopilotNativeSubagentTaskMirror({
agentId: sessionAgentId,
now,
@@ -337,6 +346,7 @@ export async function runCopilotExecution(context: {
...sessionConfig,
continuePendingWork: false,
})) as unknown as SessionLike;
nativeSessionHistoryValidated = input.initialReplayState?.journalValidated === true;
} catch (error: unknown) {
if (settledToolFinalization) {
throw createPromptError(
@@ -351,15 +361,25 @@ export async function runCopilotExecution(context: {
}
resumeFailureRecovered = true;
session = (await client.createSession(sessionConfig)) as unknown as SessionLike;
nativeSessionCreatedFresh = true;
nativeSessionHistoryValidated = true;
}
} else {
session = (await client.createSession(sessionConfig)) as unknown as SessionLike;
nativeSessionCreatedFresh = true;
nativeSessionHistoryValidated = true;
}
sessionRef.current = session;
sdkSessionId =
readString(session.sessionId) ??
readString(session.id) ??
(resumeFailureRecovered ? undefined : resumeSessionId);
if (!sdkSessionId) {
throw createPromptError(
"transcript_persistence_failed",
"[copilot-attempt] canonical transcript persistence requires the Copilot SDK session id",
);
}
sessionIdUsed = sdkSessionId ?? input.sessionId;
if (sdkSessionId && deps.onSessionEstablished && !settledToolFinalization) {
try {
@@ -408,7 +428,17 @@ export async function runCopilotExecution(context: {
});
},
getSdkSessionId: () => sdkSessionId,
isAborted: () => aborted,
isAborted: () => aborted || transcriptJournal?.hasFailed() === true,
transcriptProjection: {
journal: (transcriptJournal = createAttemptTranscriptJournal({
abortSession: () => session?.abort() ?? Promise.resolve(),
attempt: input,
messages,
sdkSessionId,
})),
modelRef,
now,
},
});
activeRunHandleRef = registerCopilotActiveRun({
abortActiveSession,
@@ -426,11 +456,15 @@ export async function runCopilotExecution(context: {
workspaceOnly: effectiveFsWorkspaceOnly,
});
sessionSetup.setPromptImagesCount(messageOptions.attachments?.length ?? 0);
if (!settledToolFinalization) {
await transcriptJournal.persistInitialUser();
}
if (abortRequested || params.abortSignal?.aborted) {
aborted = true;
externalAbort = true;
} else {
sentTurnStarted = true;
input.userTurnTranscriptRecorder?.markSentToProvider?.();
if (!hasNativePromptHook) {
emitLlmInput(attemptInput.prompt);
}
@@ -438,6 +472,7 @@ export async function runCopilotExecution(context: {
await bridge.awaitDeltaChain();
await bridge.awaitAgentEventChain();
const assistantCompleted = bridge.recordSendResult(result);
await transcriptJournal.barrier("sendAndWait");
settledFinalizationAssistantCompleted = settledToolFinalization && assistantCompleted;
if (!assistantCompleted && !aborted) {
timedOut = true;
@@ -457,12 +492,30 @@ export async function runCopilotExecution(context: {
await bridge?.awaitDeltaChain();
} catch {}
await bridge?.awaitAgentEventChain();
bridge?.flushTranscriptProjection();
try {
await transcriptJournal?.barrier("timeout");
} catch (transcriptError) {
promptError = toError(transcriptError);
}
} else {
promptError = toError(error);
try {
bridge?.flushTranscriptProjection();
await transcriptJournal?.barrier("attempt error");
promptError = toError(error);
} catch (transcriptError) {
promptError = toError(transcriptError);
}
}
}
} finally {
settled = true;
try {
bridge?.flushTranscriptProjection();
await transcriptJournal?.barrier("bridge detach");
} catch (transcriptError) {
promptError = toError(transcriptError);
}
userInputBridgeRef?.cancelPending();
if (activeRunHandleRef) {
clearActiveEmbeddedRun(
@@ -472,8 +525,14 @@ export async function runCopilotExecution(context: {
input.sessionFile,
);
}
const journalSnapshot = transcriptJournal?.snapshot();
const initialUserValidated =
!sentTurnStarted ||
settledToolFinalization ||
journalSnapshot?.initialSdkUserValidated === true;
const retainSessionForDeferredCleanup =
bridge?.hasObservedCompaction() || (timedOut && bridge?.hasObservedSessionIdle() === false);
journalSnapshot?.replayInvalid !== true &&
(bridge?.hasObservedCompaction() || (timedOut && bridge?.hasObservedSessionIdle() === false));
if (retainSessionForDeferredCleanup && bridge && session && handle) {
const cleanupAbort = new AbortController();
const abortCleanup = () => cleanupAbort.abort();
@@ -488,6 +547,7 @@ export async function runCopilotExecution(context: {
bridge,
cleanupToolBridge,
cleanupByokProxy,
deleteSessionOnIncompleteCleanup: nativeSessionCreatedFresh && initialUserValidated,
finalizeNativeSubagents: () => nativeSubagentTaskMirror?.finalizeActiveRuns(),
handle,
pool: deps.pool,
@@ -556,6 +616,8 @@ export async function runCopilotExecution(context: {
input,
lastToolError,
messages,
nativeSessionHistoryUnvalidated: !nativeSessionHistoryValidated,
transcriptJournal,
modelRef,
now,
promptError,

View File

@@ -4,7 +4,8 @@ import {
runAgentHarnessLlmOutputHook,
} from "openclaw/plugin-sdk/agent-harness-runtime";
import { finalizeCopilotAttempt } from "./attempt-cleanup.js";
import { createResult, hasMirrorIdentity, readString, readTailUserText } from "./attempt-config.js";
import { createResult } from "./attempt-config.js";
import type { AttemptTranscriptJournal } from "./attempt-transcript-journal.js";
import { withPromptFailure } from "./attempt-types.js";
import type {
AgentHarnessAttemptResult,
@@ -12,10 +13,6 @@ import type {
CopilotAgentEndHookParams,
ModelRef,
} from "./attempt-types.js";
import {
attachCopilotMirrorIdentity,
dualWriteCopilotTranscriptBestEffort,
} from "./dual-write-transcripts.js";
import { attachEventBridge } from "./event-bridge.js";
export async function completeCopilotAttempt(params: {
aborted: boolean;
@@ -32,6 +29,8 @@ export async function completeCopilotAttempt(params: {
input: AttemptParamsLike;
lastToolError: AgentHarnessAttemptResult["lastToolError"];
messages: AgentMessage[];
nativeSessionHistoryUnvalidated: boolean;
transcriptJournal: AttemptTranscriptJournal | undefined;
modelRef: ModelRef;
now: () => number;
promptError: Error | undefined;
@@ -57,6 +56,8 @@ export async function completeCopilotAttempt(params: {
input,
lastToolError,
messages,
nativeSessionHistoryUnvalidated,
transcriptJournal,
modelRef,
now,
promptError,
@@ -74,86 +75,21 @@ export async function completeCopilotAttempt(params: {
const snap = bridge?.snapshot();
const assistantTexts = bridge?.finalizeAssistantTexts() ?? [];
const lastAssistant = bridge?.buildAssistantMessage({ modelRef, now });
const syntheticUserText = readString(input.transcriptPrompt) ?? readString(input.prompt);
const tailUserText = readTailUserText(messages);
const tailUserIndex = messages.findLastIndex((message) => message.role === "user");
const currentTurnMessages = messages.map((message, index) => {
if (syntheticUserText !== tailUserText || index !== tailUserIndex) {
return message;
}
return projectAgentHarnessTranscriptMessageForDisplay({
hidden: input.trigger === "memory",
message: attachCopilotMirrorIdentity(
{ ...message, idempotencyKey: `${input.runId}:user` } as unknown as AgentMessage,
`${input.runId}:prompt`,
),
});
});
const syntheticUser: AgentMessage | undefined =
syntheticUserText && syntheticUserText !== tailUserText
? projectAgentHarnessTranscriptMessageForDisplay({
hidden: input.trigger === "memory",
message: attachCopilotMirrorIdentity(
{
role: "user",
content: syntheticUserText,
timestamp: now(),
idempotencyKey: `${input.runId}:user`,
} as unknown as AgentMessage,
`${input.runId}:prompt`,
),
})
: undefined;
const taggedLastAssistant = lastAssistant
? projectAgentHarnessTranscriptMessageForDisplay({
hidden: input.trigger === "memory",
message: attachCopilotMirrorIdentity(lastAssistant, `${input.runId}:assistant:final`),
})
: undefined;
const messagesSnapshot: AgentMessage[] = [
...currentTurnMessages,
...(syntheticUser ? [syntheticUser] : []),
...(taggedLastAssistant ? [taggedLastAssistant] : []),
];
const openClawSessionIdForMirror = readString(input.sessionId);
const sessionKeyForMirror = readString((input as { sessionKey?: unknown }).sessionKey);
const openClawStorePathForMirror = readString(input.sessionTarget?.storePath);
const mirrorScopeSessionId = sessionIdUsed ?? openClawSessionIdForMirror;
if (
openClawSessionIdForMirror &&
sessionKeyForMirror &&
openClawStorePathForMirror &&
messagesSnapshot.length > 0
) {
const taggedMessages = messagesSnapshot.map((message, index) => {
if (
message.role !== "user" &&
message.role !== "assistant" &&
message.role !== "toolResult"
) {
return message;
}
if (hasMirrorIdentity(message)) {
return message;
}
const identityScope = sdkSessionId ?? mirrorScopeSessionId ?? "attempt";
return attachCopilotMirrorIdentity(message, `${identityScope}:${message.role}:${index}`);
});
await dualWriteCopilotTranscriptBestEffort({
sessionId: openClawSessionIdForMirror,
sessionKey: sessionKeyForMirror,
agentId: readString(input.agentId),
storePath: openClawStorePathForMirror,
messages: taggedMessages,
idempotencyScope: mirrorScopeSessionId ? `copilot:${mirrorScopeSessionId}` : undefined,
config: (input as { config?: unknown }).config as never,
}).catch((mirrorError: unknown) => {
console.warn(
"[copilot-attempt] dual-write transcript wrapper rejected unexpectedly",
mirrorError,
);
});
}
const transcript = transcriptJournal?.snapshot();
// Pre-journal failures keep the prepared input snapshot. Reconstructing a
// user/assistant mirror here would restore the deleted dual-write owner.
const recorder = input.userTurnTranscriptRecorder;
const currentRunUserKey = `${input.runId}:user`;
const messagesSnapshot =
transcript?.messagesSnapshot ??
(recorder?.isBlocked()
? removePreparedUser(messages, recorder.message, currentRunUserKey)
: includePreparedUser(
messages,
recorder?.message,
input.trigger === "memory",
currentRunUserKey,
));
const result = createResult(input, {
aborted,
assistantTexts,
@@ -170,7 +106,19 @@ export async function completeCopilotAttempt(params: {
},
lastAssistant,
lastToolError,
journalValidated:
transcript !== undefined &&
!aborted &&
!timedOut &&
promptError === undefined &&
!nativeSessionHistoryUnvalidated &&
transcriptJournal?.hasFailed() !== true &&
!transcript?.replayInvalid &&
(!sentTurnStarted || settledToolFinalization || transcript.initialSdkUserValidated),
messagesSnapshot,
assistantTranscriptOwned: transcript?.assistantTranscriptOwned,
assistantTranscriptIdempotencyKey: transcript?.assistantTranscriptIdempotencyKey,
nativeReplayInvalid: transcript?.replayInvalid === true || nativeSessionHistoryUnvalidated,
now,
promptError,
resumeFailureRecovered,
@@ -182,7 +130,7 @@ export async function completeCopilotAttempt(params: {
usage: snap?.usage,
yieldDetected,
});
if (sentTurnStarted && !settledToolFinalization) {
if (sentTurnStarted && !settledToolFinalization && !transcriptJournal?.hasFailed()) {
runAgentHarnessLlmOutputHook({
event: {
runId: input.runId,
@@ -222,3 +170,77 @@ export async function completeCopilotAttempt(params: {
? result
: finalizeCopilotAttempt(input, result, hookContext, attemptStartedAt, now);
}
function includePreparedUser(
messages: AgentMessage[],
prepared: Extract<AgentMessage, { role: "user" }> | undefined,
hidden: boolean,
currentRunUserKey: string,
): AgentMessage[] {
if (!prepared) {
return messages;
}
const projected = projectAgentHarnessTranscriptMessageForDisplay({
hidden: hidden || (prepared as { display?: boolean }).display === false,
message: prepared,
}) as Extract<AgentMessage, { role: "user" }>;
const tail = messages.at(-1);
if (isSamePreparedUser(tail, projected, currentRunUserKey)) {
return [...messages.slice(0, -1), projected];
}
return [...messages, projected];
}
function removePreparedUser(
messages: AgentMessage[],
prepared: Extract<AgentMessage, { role: "user" }> | undefined,
currentRunUserKey: string,
): AgentMessage[] {
return prepared && isSamePreparedUser(messages.at(-1), prepared, currentRunUserKey)
? messages.slice(0, -1)
: messages;
}
function isSamePreparedUser(
candidate: AgentMessage | undefined,
prepared: Extract<AgentMessage, { role: "user" }>,
currentRunUserKey: string,
): boolean {
if (candidate?.role !== "user") {
return false;
}
if (candidate === prepared) {
return true;
}
const candidateKey = (candidate as { idempotencyKey?: unknown }).idempotencyKey;
const preparedKey = (prepared as { idempotencyKey?: unknown }).idempotencyKey;
if (typeof candidateKey === "string" || typeof preparedKey === "string") {
if (typeof candidateKey === "string" && typeof preparedKey === "string") {
return candidateKey === preparedKey;
}
if (
typeof candidateKey !== "string" ||
typeof preparedKey === "string" ||
(!candidateKey.startsWith("copilot:") && candidateKey !== currentRunUserKey)
) {
return false;
}
}
return (
candidate.timestamp === prepared.timestamp &&
userText(candidate.content) === userText(prepared.content)
);
}
function userText(content: unknown): string {
if (typeof content === "string") {
return content;
}
if (Array.isArray(content) && content.length === 1) {
const part = content[0] as { text?: unknown; type?: unknown };
if (part?.type === "text" && typeof part.text === "string") {
return part.text;
}
}
return JSON.stringify(content) ?? "";
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,620 @@
import { isDeepStrictEqual } from "node:util";
import type { AgentMessage } from "openclaw/plugin-sdk/agent-harness-runtime";
import {
projectAgentHarnessTranscriptMessageForDisplay,
runAgentHarnessBeforeMessageWriteHook,
} from "openclaw/plugin-sdk/agent-harness-runtime";
import {
appendSessionTranscriptMessageByIdentityStrict,
appendSessionTranscriptMessagesByIdentity,
publishSessionTranscriptUpdateByIdentity,
readVisibleSessionTranscriptMessageEntries,
type SessionTranscriptTargetParams,
} from "openclaw/plugin-sdk/session-transcript-runtime";
import type { AttemptParamsLike } from "./attempt-types.js";
type TranscriptMessage = Extract<AgentMessage, { role: "user" | "assistant" | "toolResult" }>;
type AppendResult =
| { appended: boolean; message: TranscriptMessage; messageId: string }
| undefined;
type PendingWrite = { eventId?: string; message: TranscriptMessage };
type ToolGroup = {
assistant: PendingWrite;
assistantKey: string;
order: string[];
results: Map<string, PendingWrite>;
};
export type AttemptTranscriptJournal = ReturnType<typeof createAttemptTranscriptJournal>;
export function createAttemptTranscriptJournal(params: {
abortSession: () => Promise<void>;
attempt: AttemptParamsLike;
messages: AgentMessage[];
sdkSessionId: string;
}) {
const hiddenTurn = params.attempt.trigger === "memory";
const projectDisplay = (message: AgentMessage) =>
projectAgentHarnessTranscriptMessageForDisplay({
hidden: hiddenTurn || (message as { display?: boolean }).display === false,
message,
});
const messagesSnapshot = [...params.messages];
const snapshotIdempotencyKeys = new Set(
messagesSnapshot.flatMap((message) => {
const key = readIdempotencyKey(message);
return key && isCurrentJournalIdentity(key, params) ? [key] : [];
}),
);
const replaceTailUser = (
current: Extract<AgentMessage, { role: "user" }> | undefined,
next?: AgentMessage,
) => {
if (isSameUserTurn(messagesSnapshot.at(-1), current, `${params.attempt.runId}:user`)) {
const removed = messagesSnapshot.pop();
const removedKey = removed ? readIdempotencyKey(removed) : undefined;
if (removedKey && isCurrentJournalIdentity(removedKey, params)) {
snapshotIdempotencyKeys.delete(removedKey);
}
}
if (next) {
messagesSnapshot.push(next);
const nextKey = readIdempotencyKey(next);
if (nextKey && isCurrentJournalIdentity(nextKey, params)) {
snapshotIdempotencyKeys.add(nextKey);
}
}
};
// The host recorder owns prompt construction. Missing recorders fail closed
// before dispatch below; the journal never reconstructs a prompt string.
const currentUser = params.attempt.userTurnTranscriptRecorder?.message;
if (currentUser) {
replaceTailUser(currentUser, projectDisplay(currentUser));
}
const target = resolveTranscriptTarget(params.attempt);
const config = params.attempt.config;
const seenEventIds = new Set<string>();
const deferredUserWrites: PendingWrite[] = [];
let pendingTools: ToolGroup | undefined;
let queue = Promise.resolve();
let firstFailure: Error | undefined;
let abortPromise: Promise<void> | undefined;
let replayInvalid = false;
let initialSdkUserObserved = false;
let initialSdkUserValidated = false;
let persistedInitialUser: Extract<AgentMessage, { role: "user" }> | undefined;
let latestAssistantKey: string | undefined;
let assistantTranscriptOwned = false;
let assistantTranscriptIdempotencyKey: string | undefined;
const captureFailure = (error: unknown) => {
if (firstFailure) {
return;
}
firstFailure = error instanceof Error ? error : new Error(String(error));
replayInvalid = true;
pendingTools = undefined;
abortPromise = params.abortSession().catch(() => undefined);
};
const claim = (eventId: string) =>
!firstFailure && !seenEventIds.has(eventId) && Boolean(seenEventIds.add(eventId));
const schedule = (task: () => Promise<void> | void) => {
if (firstFailure) {
return;
}
// The SDK checkpoint can advance before this queue. SQLite closes the window inside a
// complete group; a crash between groups leaves a structurally valid prefix.
queue = queue.then(() => (firstFailure ? undefined : task())).catch(captureFailure);
};
const prepare = (
write: PendingWrite,
options: { singleton?: boolean } = {},
): TranscriptMessage | undefined => {
const message = structuredClone(write.message) as TranscriptMessage;
const originalReplayPayload = structuredClone(projectReplayPayload(message));
const hooked = runAgentHarnessBeforeMessageWriteHook({
message: structuredClone(message) as TranscriptMessage,
agentId: target.agentId,
sessionKey: target.sessionKey,
});
if (!hooked) {
return undefined;
}
if (
!isDeepStrictEqual(originalReplayPayload, projectReplayPayload(hooked as TranscriptMessage))
) {
replayInvalid = true;
}
const idempotencyKey = (message as { idempotencyKey?: string }).idempotencyKey;
const toolIdentity =
message.role === "toolResult"
? { toolCallId: message.toolCallId, toolName: message.toolName }
: {};
const prepared = projectDisplay({
...hooked,
...toolIdentity,
...(idempotencyKey ? { idempotencyKey } : {}),
...((message as { display?: boolean }).display === false ? { display: false } : {}),
}) as TranscriptMessage;
return options.singleton && !isCompatibleSingletonRewrite(message, prepared)
? undefined
: prepared;
};
const append = async (write: PendingWrite): Promise<AppendResult> => {
const outcome = await appendSessionTranscriptMessageByIdentityStrict({
...target,
...(config ? { config } : {}),
...(write.eventId ? { eventId: write.eventId } : {}),
idempotencyLookup: "scan",
message: write.message,
prepareMessageAfterIdempotencyCheck: () => prepare(write, { singleton: true }),
});
if (outcome.kind === "suppressed") {
return undefined;
}
if (outcome.kind === "rejected") {
throw new Error("Transcript session changed before singleton append");
}
if (
!isDeepStrictEqual(
projectReplayPayload(write.message),
projectReplayPayload(outcome.result.message as TranscriptMessage),
)
) {
replayInvalid = true;
}
return outcome.result as AppendResult;
};
const appendToolGroup = async (group: ToolGroup) => {
const writes = [group.assistant, ...group.order.map((id) => group.results.get(id)!)];
const keys = writes.map((write) => readIdempotencyKey(write.message));
const persistedKeys = new Set(
(await readVisibleSessionTranscriptMessageEntries(target)).flatMap((entry) =>
entry.idempotencyKey ? [entry.idempotencyKey] : [],
),
);
const persistedCount = keys.filter((key) => key && persistedKeys.has(key)).length;
if (persistedCount > 0 && persistedCount < writes.length) {
// The pre-atomic journal was never shipped. Partial identity is corruption,
// not a runtime compatibility shape; recovery stays fail-closed.
throw new Error("Copilot transcript found a partial persisted tool group");
}
// Hooks must finish before BEGIN. This skips steady-state replay hooks; the
// transaction still revalidates all identities against cross-process races.
const messages =
persistedCount === writes.length
? writes.map((write) => write.message)
: writes.map((write) => prepare(write));
// Hook omission belongs to policy, but the journal owns structure: one block or
// structurally destructive rewrite suppresses the complete assistant/result group.
if (
messages.some((message) => !message) ||
!isCompleteToolGroup(messages as TranscriptMessage[], group.order)
) {
return undefined;
}
const results = await appendSessionTranscriptMessagesByIdentity({
...target,
...(config ? { config } : {}),
messages: writes.map((write, index) => ({
eventId: write.eventId!,
idempotencyLookup: "scan" as const,
message: messages[index]!,
})),
});
if (
!isCompleteToolGroup(
results.map((result) => result.message),
group.order,
)
) {
throw new Error("Copilot transcript replayed an invalid tool group");
}
if (
results.some(
(result, index) =>
!isDeepStrictEqual(
projectReplayPayload(writes[index]!.message),
projectReplayPayload(result.message as TranscriptMessage),
),
)
) {
replayInvalid = true;
}
return results;
};
const publish = async (appended: boolean) => {
if (appended) {
await publishSessionTranscriptUpdateByIdentity({ ...target }).catch((error: unknown) => {
console.warn("[copilot-attempt] transcript update notification failed", error);
});
}
};
const accept = (result: AppendResult): boolean => {
if (!result) {
return false;
}
const key = readIdempotencyKey(result.message);
const snapshotKey = key && isCurrentJournalIdentity(key, params) ? key : undefined;
if (!snapshotKey || !snapshotIdempotencyKeys.has(snapshotKey)) {
messagesSnapshot.push(result.message);
if (snapshotKey) {
snapshotIdempotencyKeys.add(snapshotKey);
}
}
return result.appended;
};
const ownAssistant = (key: string, persisted: boolean) => {
if (latestAssistantKey === key) {
assistantTranscriptOwned = true;
assistantTranscriptIdempotencyKey = persisted ? key : undefined;
}
};
const drainQueue = async () => {
// SDK events can append work while an earlier write is pending. Drain until
// the observed tail stays stable so finalization cannot outrun persistence.
while (true) {
const tail = queue;
await tail;
if (tail === queue) {
break;
}
}
};
const barrier = async (boundary: string) => {
await drainQueue();
if (!firstFailure) {
// Give an already-delivered SDK event one microtask turn to reach the
// bridge, then re-drain without another yield before returning.
await Promise.resolve();
await drainQueue();
}
if (!firstFailure && pendingTools) {
captureFailure(
new Error(
`Copilot transcript reached ${boundary} with unresolved tool results: ${pendingTools.order.join(", ")}`,
),
);
}
if (abortPromise) {
await abortPromise;
}
if (firstFailure) {
const error = new Error(
`[copilot-attempt] canonical transcript persistence failed: ${firstFailure.message}`,
{ cause: firstFailure },
) as Error & { code?: string };
error.code = "transcript_persistence_failed";
throw error;
}
};
return {
markReplayIncomplete() {
replayInvalid = true;
},
recordAssistantProjectionGap() {
replayInvalid = true;
latestAssistantKey = undefined;
assistantTranscriptOwned = false;
assistantTranscriptIdempotencyKey = undefined;
},
async persistInitialUser() {
const recorder = params.attempt.userTurnTranscriptRecorder;
if (!recorder) {
captureFailure(new Error("Copilot transcript requires a user-turn recorder"));
return await barrier("user prompt");
}
if (recorder.isBlocked()) {
replayInvalid = true;
replaceTailUser(recorder.message);
return;
}
const persistence = (async () => {
const resolved = await recorder.resolveMessage();
if (!resolved) {
throw new Error("Copilot transcript user turn resolved without a message");
}
const outcome = await append({
message: {
...resolved,
idempotencyKey: `${params.attempt.runId}:user`,
} as TranscriptMessage,
});
replaceTailUser(currentUser);
if (!outcome) {
replayInvalid = true;
recorder.markBlocked();
return;
}
const persisted = outcome.message as Extract<AgentMessage, { role: "user" }>;
accept(outcome);
persistedInitialUser = persisted;
recorder.markRuntimePersisted(persisted);
params.attempt.onUserMessagePersisted?.(persisted);
await publish(outcome.appended);
})();
recorder.markRuntimePersistencePending(persistence);
await persistence.catch(captureFailure);
await barrier("user prompt");
},
recordSdkUser(input: {
eventId: string;
message: Extract<AgentMessage, { role: "user" }>;
autopilotContinuation: boolean;
replayIncomplete?: boolean;
}) {
if (!claim(input.eventId)) {
return;
}
replayInvalid ||= input.replayIncomplete === true;
if (!initialSdkUserObserved && !input.autopilotContinuation) {
initialSdkUserObserved = true;
if (
!persistedInitialUser ||
userText(persistedInitialUser.content) !== userText(input.message.content)
) {
replayInvalid = true;
} else {
initialSdkUserValidated = true;
}
return;
}
initialSdkUserObserved = true;
schedule(async () => {
const write = { eventId: input.eventId, message: input.message };
if (pendingTools) {
deferredUserWrites.push(write);
return;
}
const outcome = await append(write);
if (!outcome) {
replayInvalid = true;
}
await publish(accept(outcome));
});
},
recordAssistant(input: {
eventId: string;
message: Extract<AgentMessage, { role: "assistant" }>;
replayIncomplete?: boolean;
toolCallIds: string[];
}) {
if (!claim(input.eventId)) {
return;
}
replayInvalid ||= input.replayIncomplete === true;
const key = `copilot-sdk:${params.sdkSessionId}:${input.eventId}`;
latestAssistantKey = key;
assistantTranscriptOwned = false;
assistantTranscriptIdempotencyKey = undefined;
schedule(async () => {
if (pendingTools) {
throw new Error("Copilot emitted an assistant message before tool results settled");
}
const write = {
eventId: input.eventId,
message: { ...input.message, idempotencyKey: key } as TranscriptMessage,
};
if (input.toolCallIds.length > 0) {
pendingTools = {
assistant: write,
assistantKey: key,
order: input.toolCallIds,
results: new Map(),
};
return;
}
const outcome = await append(write);
if (!outcome) {
replayInvalid = true;
}
ownAssistant(key, Boolean(outcome));
await publish(accept(outcome));
});
},
recordToolResult(input: {
eventId: string;
message: Extract<AgentMessage, { role: "toolResult" }>;
replayIncomplete?: boolean;
}) {
if (!claim(input.eventId)) {
return;
}
schedule(async () => {
const group = pendingTools;
if (!group || !group.order.includes(input.message.toolCallId)) {
throw new Error(`Copilot emitted an unmatched tool result: ${input.message.toolCallId}`);
}
group.results.set(input.message.toolCallId, {
eventId: input.eventId,
message: {
...input.message,
idempotencyKey: `copilot-sdk:${params.sdkSessionId}:${input.eventId}`,
} as TranscriptMessage,
});
replayInvalid ||= input.replayIncomplete === true;
if (!group.order.every((toolCallId) => group.results.has(toolCallId))) {
return;
}
const results = await appendToolGroup(group);
let appended = false;
if (!results) {
replayInvalid = true;
ownAssistant(group.assistantKey, false);
} else {
for (const result of results) {
const didAppend = accept(result as AppendResult);
appended ||= didAppend;
}
ownAssistant(group.assistantKey, true);
}
pendingTools = undefined;
for (const write of deferredUserWrites.splice(0)) {
const outcome = await append(write);
if (!outcome) {
replayInvalid = true;
}
const didAppend = accept(outcome);
appended ||= didAppend;
}
await publish(appended);
});
},
barrier,
hasFailed: () => firstFailure !== undefined,
snapshot: () => ({
assistantTranscriptOwned,
assistantTranscriptIdempotencyKey,
initialSdkUserValidated,
messagesSnapshot: [...messagesSnapshot],
replayInvalid,
}),
};
}
function resolveTranscriptTarget(attempt: AttemptParamsLike): SessionTranscriptTargetParams {
const sessionId = readString(attempt.sessionTarget?.sessionId);
const sessionKey = readString(attempt.sessionTarget?.sessionKey);
const storePath = readString(attempt.sessionTarget?.storePath);
if (!sessionId || !sessionKey || !storePath) {
const error = new Error(
"[copilot-attempt] canonical transcript persistence requires an exact runtime session target",
) as Error & { code?: string };
error.code = "transcript_persistence_failed";
throw error;
}
const agentId = readString(attempt.sessionTarget?.agentId ?? attempt.agentId);
return { sessionId, sessionKey, storePath, ...(agentId ? { agentId } : {}) };
}
function readAssistantToolCallIds(message: TranscriptMessage): string[] {
return message.role === "assistant"
? message.content.flatMap((part) => (part.type === "toolCall" ? [part.id] : []))
: [];
}
function isCompatibleSingletonRewrite(
original: TranscriptMessage,
prepared: TranscriptMessage,
): boolean {
// Hooks may redact content, but role and tool topology are journal-owned;
// accepting either rewrite would make the canonical replay structurally false.
return (
original.role === prepared.role &&
(original.role !== "assistant" ||
JSON.stringify(readAssistantToolCallIds(original)) ===
JSON.stringify(readAssistantToolCallIds(prepared)))
);
}
function projectReplayPayload(message: TranscriptMessage): unknown {
switch (message.role) {
case "user":
return { role: message.role, content: message.content };
case "assistant":
return {
role: message.role,
content: message.content,
api: message.api,
model: message.model,
provider: message.provider,
stopReason: message.stopReason,
};
case "toolResult":
return {
role: message.role,
content: message.content,
isError: message.isError,
toolCallId: message.toolCallId,
toolName: message.toolName,
};
}
return undefined;
}
function readIdempotencyKey(message: AgentMessage): string | undefined {
const key = (message as { idempotencyKey?: unknown }).idempotencyKey;
return typeof key === "string" && key ? key : undefined;
}
function isCurrentJournalIdentity(
key: string,
params: { attempt: AttemptParamsLike; sdkSessionId: string },
): boolean {
// Old mirror keys can be content fingerprints and are not turn identity.
// Current journal keys use a run id or the SDK's unique event id.
return (
key === `${params.attempt.runId}:user` || key.startsWith(`copilot-sdk:${params.sdkSessionId}:`)
);
}
function isCompleteToolGroup(messages: TranscriptMessage[], order: string[]): boolean {
const [assistant, ...results] = messages;
return (
assistant?.role === "assistant" &&
JSON.stringify(readAssistantToolCallIds(assistant)) === JSON.stringify(order) &&
results.length === order.length &&
results.every(
(message, index) => message.role === "toolResult" && message.toolCallId === order[index],
)
);
}
function isSameUserTurn(
candidate: AgentMessage | undefined,
current: Extract<AgentMessage, { role: "user" }> | undefined,
currentRunUserKey: string,
): boolean {
if (candidate?.role !== "user" || !current) {
return false;
}
if (candidate === current) {
return true;
}
const candidateKey = (candidate as { idempotencyKey?: unknown }).idempotencyKey;
const currentKey = (current as { idempotencyKey?: unknown }).idempotencyKey;
if (typeof candidateKey === "string" || typeof currentKey === "string") {
if (typeof candidateKey === "string" && typeof currentKey === "string") {
return candidateKey === currentKey;
}
if (
typeof candidateKey !== "string" ||
typeof currentKey === "string" ||
(!candidateKey.startsWith("copilot:") && candidateKey !== currentRunUserKey)
) {
return false;
}
}
// The embedded-runner boundary identifies the active user as the last user
// and stamps it with this recorder timestamp; historical turns are ineligible.
return (
candidate.timestamp === current.timestamp &&
userText(candidate.content) === userText(current.content)
);
}
function userText(content: unknown): string {
if (typeof content === "string") {
return content;
}
if (Array.isArray(content) && content.length === 1) {
const part = content[0] as { text?: unknown; type?: unknown };
if (part?.type === "text" && typeof part.text === "string") {
return part.text;
}
}
return JSON.stringify(content) ?? "";
}
function readString(value: unknown): string | undefined {
return typeof value === "string" && value.trim() ? value.trim() : undefined;
}

View File

@@ -33,7 +33,10 @@ export type AgentHarnessAttemptResult = Extract<
{ terminal: unknown }
>;
type AttemptTerminal = AgentHarnessAttemptResult["terminal"];
export type AttemptResultWithSdkSessionId = AgentHarnessAttemptResult & { sdkSessionId?: string };
export type AttemptResultWithSdkSessionId = AgentHarnessAttemptResult & {
journalValidated?: boolean;
sdkSessionId?: string;
};
export function withPromptFailure(terminal: AttemptTerminal, error: unknown): AttemptTerminal {
return terminal.kind === "aborted" || terminal.kind === "timeout"
? { ...terminal, failure: { source: "prompt", error } }
@@ -93,7 +96,10 @@ export type AttemptParamsLike = AgentHarnessAttemptParams & {
enableSessionTelemetry?: boolean;
hooksConfig?: CopilotHooksConfig;
infiniteSessionConfig?: SessionConfig["infiniteSessions"];
initialReplayState?: AgentHarnessAttemptParams["initialReplayState"] & { sdkSessionId?: string };
initialReplayState?: AgentHarnessAttemptParams["initialReplayState"] & {
journalValidated?: boolean;
sdkSessionId?: string;
};
messages?: AgentMessage[];
model?: string | { api?: string; id?: string; input?: string[]; provider?: string };
onAssistantDelta?: (payload: OnAssistantDeltaPayload) => void | Promise<void>;

File diff suppressed because it is too large Load Diff

View File

@@ -1,502 +0,0 @@
// Copilot tests cover dual write transcripts plugin behavior.
import { createHash } from "node:crypto";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import type { AgentMessage } from "openclaw/plugin-sdk/agent-harness-runtime";
import {
initializeGlobalHookRunner,
resetGlobalHookRunner,
} from "openclaw/plugin-sdk/hook-runtime";
import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime";
import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime";
import { readSessionTranscriptEvents } from "openclaw/plugin-sdk/session-transcript-runtime";
import {
castAgentMessage,
makeAgentAssistantMessage,
makeAgentUserMessage,
} from "openclaw/plugin-sdk/test-fixtures";
import { afterEach, describe, expect, it } from "vitest";
import {
attachCopilotMirrorIdentity,
dualWriteCopilotTranscriptBestEffort,
} from "./dual-write-transcripts.js";
const mirrorCopilotTranscript = dualWriteCopilotTranscriptBestEffort;
type MirroredAgentMessage = Extract<AgentMessage, { role: "user" | "assistant" | "toolResult" }>;
function expectedFingerprint(message: MirroredAgentMessage): string {
const payload = JSON.stringify({ role: message.role, content: message.content });
return createHash("sha256").update(payload).digest("hex").slice(0, 16);
}
const tempDirs: string[] = [];
afterEach(async () => {
resetGlobalHookRunner();
for (const dir of tempDirs.splice(0)) {
await fs.rm(dir, { recursive: true, force: true });
}
});
async function makeRoot(prefix: string): Promise<string> {
const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix));
tempDirs.push(root);
return root;
}
function readEventMessages(events: unknown[]): Array<{ role?: string; text?: string }> {
return events
.map((event) =>
event && typeof event === "object" ? (event as { message?: unknown }).message : undefined,
)
.filter((message): message is { role?: string; content?: unknown } =>
Boolean(message && typeof message === "object"),
)
.map((message) => {
const content = Array.isArray(message.content)
? message.content.find((part): part is { text: string } =>
Boolean(part && typeof part === "object" && typeof part.text === "string"),
)?.text
: typeof message.content === "string"
? message.content
: undefined;
return { role: message.role, text: content };
});
}
async function createSqliteMirrorTarget(prefix: string, options: { sessionId?: string } = {}) {
const root = await makeRoot(prefix);
const agentId = "main";
const sessionId = options.sessionId ?? "session-1";
const sessionKey = `agent:${agentId}:${sessionId}`;
const storePath = path.join(root, "openclaw-agent.sqlite");
await upsertSessionEntry({
agentId,
sessionKey,
storePath,
entry: {
sessionFile: `sqlite:${agentId}:${sessionId}:${storePath}`,
sessionId,
updatedAt: 1,
},
});
return {
agentId,
sessionId,
sessionKey,
storePath,
bogusSessionFile: path.join(root, "should-not-be-created.jsonl"),
};
}
async function readMirrorEvents(target: {
agentId: string;
sessionId: string;
sessionKey: string;
storePath: string;
}): Promise<unknown[]> {
return await readSessionTranscriptEvents(target);
}
async function readMirrorRaw(target: {
agentId: string;
sessionId: string;
sessionKey: string;
storePath: string;
}): Promise<string> {
return (await readMirrorEvents(target)).map((event) => JSON.stringify(event)).join("\n");
}
async function readMirrorMessages(target: {
agentId: string;
sessionId: string;
sessionKey: string;
storePath: string;
}): Promise<Array<{ role?: string; text?: string }>> {
return readEventMessages(await readMirrorEvents(target));
}
describe("mirrorCopilotTranscript", () => {
it("hides current memory-maintenance messages without hiding replayed turns", async () => {
initializeGlobalHookRunner(
createMockPluginRegistry([
{
hookName: "before_message_write",
handler: (event) => {
const { display: _display, ...message } = (
event as { message: Record<string, unknown> }
).message;
return { message: castAgentMessage(message) };
},
},
]),
);
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-memory-");
const messages = [
attachCopilotMirrorIdentity(
makeAgentAssistantMessage({
content: [{ type: "text", text: "ordinary prior reply" }],
timestamp: Date.now(),
}),
"run-prior:assistant:final",
),
attachCopilotMirrorIdentity(
makeAgentUserMessage({
content: [{ type: "text", text: "Pre-compaction memory flush" }],
timestamp: Date.now() + 1,
}),
"run-memory:prompt",
),
attachCopilotMirrorIdentity(
makeAgentAssistantMessage({
content: [{ type: "toolCall", id: "call-1", name: "write", arguments: {} }],
timestamp: Date.now() + 2,
}),
"run-memory:tool-call:call-1",
),
attachCopilotMirrorIdentity(
castAgentMessage({
role: "toolResult",
toolCallId: "call-1",
toolName: "write",
content: [{ type: "toolResult", toolCallId: "call-1", content: "saved" }],
timestamp: Date.now() + 3,
}),
"run-memory:tool-result:call-1",
),
attachCopilotMirrorIdentity(
makeAgentAssistantMessage({
content: [{ type: "text", text: "NO_REPLY" }],
timestamp: Date.now() + 4,
}),
"run-memory:assistant:final",
),
];
for (const message of messages.slice(1)) {
Object.assign(message, { display: false });
}
await mirrorCopilotTranscript({
...target,
messages,
idempotencyScope: "copilot:memory",
});
const persistedMessages = (await readMirrorEvents(target))
.map((event) =>
event && typeof event === "object" ? (event as { message?: unknown }).message : undefined,
)
.filter((message): message is Record<string, unknown> =>
Boolean(message && typeof message === "object"),
);
expect(persistedMessages).toHaveLength(messages.length);
expect(persistedMessages[0]).not.toHaveProperty("display", false);
expect(persistedMessages.slice(1).every((message) => message.display === false)).toBe(true);
});
it("mirrors user, assistant, and tool result messages by SQLite identity", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-basic-");
const userMessage = makeAgentUserMessage({
content: [{ type: "text", text: "hello" }],
timestamp: Date.now(),
});
const assistantMessage = makeAgentAssistantMessage({
content: [{ type: "text", text: "hi there" }],
timestamp: Date.now() + 1,
});
const toolResultMessage = castAgentMessage({
role: "toolResult",
toolCallId: "call-1",
toolName: "read",
content: [{ type: "toolResult", toolCallId: "call-1", content: "read output" }],
timestamp: Date.now() + 2,
}) as MirroredAgentMessage;
await mirrorCopilotTranscript({
...target,
messages: [userMessage, assistantMessage, toolResultMessage],
idempotencyScope: "copilot:session-1",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain('"role":"user"');
expect(raw).toContain('"role":"assistant"');
expect(raw).toContain('"role":"toolResult"');
expect(raw).toContain('"toolCallId":"call-1"');
expect(raw).toContain(
`"idempotencyKey":"copilot:session-1:user:${expectedFingerprint(userMessage)}"`,
);
expect(raw).toContain(
`"idempotencyKey":"copilot:session-1:assistant:${expectedFingerprint(assistantMessage)}"`,
);
expect(raw).toContain(
`"idempotencyKey":"copilot:session-1:toolResult:${expectedFingerprint(toolResultMessage)}"`,
);
await expect(fs.readFile(target.bogusSessionFile, "utf8")).rejects.toHaveProperty(
"code",
"ENOENT",
);
});
it("preserves gateway user-turn identity across Copilot transcript mirroring", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-user-identity-");
const userMessage = castAgentMessage({
...makeAgentUserMessage({
content: [{ type: "text", text: "client prompt" }],
timestamp: Date.now(),
}),
idempotencyKey: "client-run:user",
});
await mirrorCopilotTranscript({
...target,
messages: [userMessage],
idempotencyScope: "copilot:session-1",
});
await mirrorCopilotTranscript({
...target,
messages: [userMessage],
idempotencyScope: "copilot:session-1",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain('"idempotencyKey":"client-run:user"');
expect(raw).not.toContain('"idempotencyKey":"copilot:session-1:user:');
expect(
(await readMirrorMessages(target)).filter((message) => message.role === "user"),
).toHaveLength(1);
});
it("deduplicates re-emits by idempotency scope", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-dedupe-");
const messages = [
makeAgentUserMessage({
content: [{ type: "text", text: "hello" }],
timestamp: Date.now(),
}),
makeAgentAssistantMessage({
content: [{ type: "text", text: "hi there" }],
timestamp: Date.now() + 1,
}),
] as const;
await mirrorCopilotTranscript({
...target,
messages: [...messages],
idempotencyScope: "copilot:session-1",
});
await mirrorCopilotTranscript({
...target,
messages: [...messages],
idempotencyScope: "copilot:session-1",
});
expect((await readMirrorMessages(target)).filter((message) => message.role)).toHaveLength(2);
});
it("runs before_message_write before appending mirrored messages", async () => {
initializeGlobalHookRunner(
createMockPluginRegistry([
{
hookName: "before_message_write",
handler: (event) => ({
message: castAgentMessage({
...((event as { message: unknown }).message as Record<string, unknown>),
content: [{ type: "text", text: "hello [hooked]" }],
}),
}),
},
]),
);
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-hook-");
const sourceMessage = makeAgentAssistantMessage({
content: [{ type: "text", text: "hello" }],
timestamp: Date.now(),
});
await mirrorCopilotTranscript({
...target,
messages: [sourceMessage],
idempotencyScope: "copilot:session-1",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain('"content":[{"type":"text","text":"hello [hooked]"}]');
expect(raw).toContain(
`"idempotencyKey":"copilot:session-1:assistant:${expectedFingerprint(sourceMessage)}"`,
);
});
it("respects before_message_write blocking decisions", async () => {
initializeGlobalHookRunner(
createMockPluginRegistry([
{ hookName: "before_message_write", handler: () => ({ block: true }) },
]),
);
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-block-");
await mirrorCopilotTranscript({
...target,
messages: [
makeAgentAssistantMessage({
content: [{ type: "text", text: "should not persist" }],
timestamp: Date.now(),
}),
],
idempotencyScope: "copilot:session-1",
});
expect(await readMirrorMessages(target)).toEqual([]);
});
it("is a no-op when no mirrorable messages are present", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-empty-");
await mirrorCopilotTranscript({
...target,
messages: [],
idempotencyScope: "copilot:session-1",
});
expect(await readMirrorMessages(target)).toEqual([]);
});
it("uses content fingerprint when no explicit mirror identity is attached", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-fingerprint-");
const message = makeAgentAssistantMessage({
content: [{ type: "text", text: "fp" }],
timestamp: Date.now(),
});
await mirrorCopilotTranscript({
...target,
messages: [message],
idempotencyScope: "scope-fp",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain(`"idempotencyKey":"scope-fp:assistant:${expectedFingerprint(message)}"`);
});
it("uses attached identity instead of content fingerprint when provided", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-identity-");
const baseMessage = makeAgentAssistantMessage({
content: [{ type: "text", text: "explicit" }],
timestamp: Date.now(),
});
const tagged = attachCopilotMirrorIdentity(baseMessage, "sdk-session-1:assistant:0");
await mirrorCopilotTranscript({
...target,
messages: [tagged],
idempotencyScope: "copilot:openclaw-session-1",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain(
'"idempotencyKey":"copilot:openclaw-session-1:sdk-session-1:assistant:0"',
);
expect(raw).not.toContain(expectedFingerprint(baseMessage));
});
it("omits idempotencyKey when no idempotencyScope is provided", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-no-scope-");
await mirrorCopilotTranscript({
...target,
messages: [
makeAgentAssistantMessage({
content: [{ type: "text", text: "no scope" }],
timestamp: Date.now(),
}),
],
});
const raw = await readMirrorRaw(target);
expect(raw).toContain('"content":[{"type":"text","text":"no scope"}]');
expect(raw).not.toContain("idempotencyKey");
});
it("filters out non-mirrorable roles", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-filter-");
const userMessage = makeAgentUserMessage({
content: [{ type: "text", text: "u" }],
timestamp: Date.now(),
});
const systemLike = castAgentMessage({
role: "system" as never,
content: [{ type: "text", text: "system note" }],
timestamp: Date.now() + 1,
});
await mirrorCopilotTranscript({
...target,
messages: [userMessage, systemLike],
idempotencyScope: "scope",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain('"role":"user"');
expect(raw).not.toContain("system note");
});
it("preserves explicit identity across attachCopilotMirrorIdentity overrides", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-override-");
const base = makeAgentAssistantMessage({
content: [{ type: "text", text: "x" }],
timestamp: Date.now(),
});
const first = attachCopilotMirrorIdentity(base, "id-1");
const second = attachCopilotMirrorIdentity(first, "id-2");
await mirrorCopilotTranscript({
...target,
messages: [second],
idempotencyScope: "scope",
});
const raw = await readMirrorRaw(target);
expect(raw).toContain('"idempotencyKey":"scope:id-2"');
expect(raw).not.toContain('"idempotencyKey":"scope:id-1"');
});
});
describe("dualWriteCopilotTranscriptBestEffort", () => {
it("returns normally when mirror succeeds", async () => {
const target = await createSqliteMirrorTarget("openclaw-copilot-mirror-best-effort-");
await expect(
dualWriteCopilotTranscriptBestEffort({
...target,
messages: [
makeAgentAssistantMessage({
content: [{ type: "text", text: "ok" }],
timestamp: Date.now(),
}),
],
idempotencyScope: "scope",
}),
).resolves.toBeUndefined();
expect(await readMirrorMessages(target)).toContainEqual({ role: "assistant", text: "ok" });
});
it("swallows missing runtime identity and does not write JSONL", async () => {
const root = await makeRoot("openclaw-copilot-mirror-invalid-");
const sessionFile = path.join(root, "agents", "main", "sessions", "session-1.jsonl");
await expect(
dualWriteCopilotTranscriptBestEffort({
agentId: "main",
sessionId: "session-1",
messages: [
makeAgentAssistantMessage({
content: [{ type: "text", text: "should-not-throw" }],
timestamp: Date.now(),
}),
],
idempotencyScope: "scope",
}),
).resolves.toBeUndefined();
await expect(fs.access(sessionFile)).rejects.toHaveProperty("code", "ENOENT");
});
});

View File

@@ -1,247 +0,0 @@
/**
* Mirrors the AgentMessages produced by the copilot agent runtime into the
* OpenClaw audit transcript that sits next to (but is distinct from) the
* SDK's own session storage.
*
* The OpenClaw shell (src/agents/command/attempt-execution.ts) already
* writes the user prompt and the terminal assistant text into the
* transcript at the end of each attempt. That is the bare minimum to
* keep `/history` working. It does NOT capture tool calls, tool
* results, or intermediate assistant turns — those live only in the
* SDK's own session file.
*
* For audit/compliance and for the codex-parity guarantees we promised
* in the proposal, we mirror the full `messagesSnapshot` (user +
* assistant + toolResult) into the OpenClaw transcript via the same
* plugin-sdk primitives that the codex extension uses
* (extensions/codex/src/app-server/transcript-mirror.ts). Both writers
* cooperate via idempotency-key dedupe: user entries retain a gateway
* turn key when present; other entries use `${idempotencyScope}:${identity}`.
* We skip any key already present in the transcript on disk. Both
* attempt-execution's untagged entries (no idempotencyKey) and our
* tagged mirror entries can coexist; attempt-execution dedupes its own
* final-assistant append via `embeddedAssistantGapFill` content match.
*
* Failures (lock contention, fs errors, etc.) are swallowed by the
* caller-side `dualWriteCopilotTranscriptBestEffort` wrapper used
* in attempt.ts so they cannot break the attempt; this module itself
* throws on infrastructure failure so callers can choose policy.
*/
import { createHash } from "node:crypto";
import {
projectAgentHarnessTranscriptMessageForDisplay,
runAgentHarnessBeforeMessageWriteHook,
type AgentMessage,
} from "openclaw/plugin-sdk/agent-harness-runtime";
import {
withSessionTranscriptWriteLock,
type SessionTranscriptTargetParams,
type SessionTranscriptWriteLockContext,
type SessionTranscriptWriteLockParams,
} from "openclaw/plugin-sdk/session-transcript-runtime";
type MirroredAgentMessage = Extract<AgentMessage, { role: "user" | "assistant" | "toolResult" }>;
const MIRROR_IDENTITY_META_KEY = "mirrorIdentity" as const;
/**
* Tag a message with a stable logical identity for mirror dedupe.
* Callers should use a value that is invariant for the same logical
* message across re-emits (e.g. `${sdkSessionId}:assistant:${turnIndex}`)
* but distinct for genuinely-distinct messages. When present this
* identity replaces the role/content fingerprint in the idempotency
* key, so the dedupe survives caller-scope rotation without collapsing
* distinct same-content turns. Symmetric to
* `attachCodexMirrorIdentity` in the codex extension.
*/
export function attachCopilotMirrorIdentity<T extends AgentMessage>(
message: T,
identity: string,
): T {
const record = message as unknown as Record<string, unknown>;
const existing = record["__openclaw"];
const baseMeta =
existing && typeof existing === "object" && !Array.isArray(existing)
? (existing as Record<string, unknown>)
: {};
return {
...record,
__openclaw: { ...baseMeta, [MIRROR_IDENTITY_META_KEY]: identity },
} as unknown as T;
}
function readMirrorIdentity(message: MirroredAgentMessage): string | undefined {
const record = message as unknown as { __openclaw?: unknown };
const meta = record["__openclaw"];
if (!meta || typeof meta !== "object" || Array.isArray(meta)) {
return undefined;
}
const id = (meta as Record<string, unknown>)[MIRROR_IDENTITY_META_KEY];
return typeof id === "string" && id.length > 0 ? id : undefined;
}
function fingerprintMirrorMessageContent(message: MirroredAgentMessage): string {
const payload = JSON.stringify({ role: message.role, content: message.content });
return createHash("sha256").update(payload).digest("hex").slice(0, 16);
}
function buildMirrorDedupeIdentity(message: MirroredAgentMessage): string {
const explicit = readMirrorIdentity(message);
if (explicit) {
return explicit;
}
return `${message.role}:${fingerprintMirrorMessageContent(message)}`;
}
interface MirrorCopilotTranscriptParams {
sessionId: string;
sessionKey?: string;
agentId?: string;
storePath?: string;
messages: AgentMessage[];
/**
* Stable per-harness/per-thread scope. The codex equivalent uses
* `codex-app-server:${threadId}`; we use `copilot:${sessionId}`
* by convention (see attempt.ts call site). Keeping the scope
* thread-stable (not per-turn) is what lets a re-emitted prior-turn
* entry collide with its existing on-disk key and be a true no-op.
*/
idempotencyScope?: string;
config?: SessionTranscriptWriteLockParams["config"];
}
async function mirrorCopilotTranscript(params: MirrorCopilotTranscriptParams): Promise<void> {
const messages = params.messages.filter(
(message): message is MirroredAgentMessage =>
message.role === "user" || message.role === "assistant" || message.role === "toolResult",
);
if (messages.length === 0) {
return;
}
const transcriptTarget = resolveCopilotMirrorTranscriptTarget(params);
await withCopilotMirrorTranscriptWriteLock(
{ ...transcriptTarget, config: params.config },
async (transcript) => {
let didAppendMessage = false;
const existingIdempotencyKeys = readTranscriptIdempotencyKeys(await transcript.readEvents());
for (const message of messages) {
const dedupeIdentity = buildMirrorDedupeIdentity(message);
const sourceIdempotencyKey = (message as unknown as { idempotencyKey?: unknown })
.idempotencyKey;
const sourceUserIdempotencyKey =
message.role === "user" &&
typeof sourceIdempotencyKey === "string" &&
sourceIdempotencyKey.trim()
? sourceIdempotencyKey.trim()
: undefined;
const idempotencyKey =
sourceUserIdempotencyKey ??
(params.idempotencyScope ? `${params.idempotencyScope}:${dedupeIdentity}` : undefined);
if (idempotencyKey && existingIdempotencyKeys.has(idempotencyKey)) {
continue;
}
const transcriptMessage = {
...message,
...(idempotencyKey ? { idempotencyKey } : {}),
} as AgentMessage;
const nextMessage = runAgentHarnessBeforeMessageWriteHook({
message: transcriptMessage,
agentId: params.agentId,
sessionKey: params.sessionKey,
});
if (!nextMessage) {
continue;
}
const messageWithIdentity = (
idempotencyKey
? {
...(nextMessage as unknown as Record<string, unknown>),
idempotencyKey,
}
: nextMessage
) as AgentMessage;
const messageToAppend = projectAgentHarnessTranscriptMessageForDisplay({
hidden: (message as { display?: boolean }).display === false,
message: messageWithIdentity,
});
const appended = await transcript.appendMessage({
message: messageToAppend,
idempotencyLookup: idempotencyKey ? "caller-checked" : "scan",
});
if (!appended) {
continue;
}
didAppendMessage = true;
if (idempotencyKey) {
existingIdempotencyKeys.add(idempotencyKey);
}
}
if (didAppendMessage) {
await transcript.publishUpdate(
params.sessionKey ? { sessionKey: params.sessionKey } : undefined,
);
}
return didAppendMessage;
},
);
}
function resolveCopilotMirrorTranscriptTarget(params: {
agentId?: string;
sessionId: string;
sessionKey?: string;
storePath?: string;
}): SessionTranscriptTargetParams {
const sessionKey = params.sessionKey?.trim();
const storePath = params.storePath?.trim();
if (!sessionKey || !storePath) {
throw new Error("Copilot transcript mirror requires a runtime session identity");
}
return {
...(params.agentId ? { agentId: params.agentId } : {}),
sessionId: params.sessionId,
sessionKey,
storePath,
};
}
function withCopilotMirrorTranscriptWriteLock<T>(
params: SessionTranscriptTargetParams & { config?: SessionTranscriptWriteLockParams["config"] },
run: (context: SessionTranscriptWriteLockContext) => Promise<T> | T,
): Promise<T> {
return withSessionTranscriptWriteLock(params, run);
}
function readTranscriptIdempotencyKeys(events: unknown[]): Set<string> {
const keys = new Set<string>();
for (const event of events) {
if (!event || typeof event !== "object" || Array.isArray(event)) {
continue;
}
const parsed = event as { message?: { idempotencyKey?: unknown } };
if (typeof parsed.message?.idempotencyKey === "string") {
keys.add(parsed.message.idempotencyKey);
}
}
return keys;
}
/**
* Caller-side wrapper that swallows mirror failures. attempt.ts uses
* this so that a transient transcript-mirror failure (lock contention,
* disk full, etc.) never breaks an otherwise-successful attempt. The
* SDK's own session file remains the source of truth in that case;
* the OpenClaw audit trail just misses the intermediate messages for
* this turn.
*/
export async function dualWriteCopilotTranscriptBestEffort(
params: MirrorCopilotTranscriptParams,
): Promise<void> {
try {
await mirrorCopilotTranscript(params);
} catch (error) {
console.warn("[copilot-attempt] dual-write transcript mirror failed", error);
}
}

View File

@@ -0,0 +1,167 @@
import type { Attachment, SessionEvent } from "@github/copilot-sdk";
import type { AgentMessage } from "openclaw/plugin-sdk/agent-harness-runtime";
import { sanitizeToolResult } from "openclaw/plugin-sdk/agent-harness-runtime";
import { buildCopilotAssistantUsage, type CopilotUsageSnapshot } from "./usage-bridge.js";
export type AssistantMessage = Extract<AgentMessage, { role: "assistant" }>;
export type AssistantUsageSnapshot = CopilotUsageSnapshot;
export interface AttemptTranscriptJournalProjection {
markReplayIncomplete(): void;
recordAssistantProjectionGap(): void;
recordAssistant(input: {
eventId: string;
message: AssistantMessage;
replayIncomplete?: boolean;
toolCallIds: string[];
}): void;
recordSdkUser(input: {
autopilotContinuation: boolean;
eventId: string;
message: Extract<AgentMessage, { role: "user" }>;
replayIncomplete?: boolean;
}): void;
recordToolResult(input: {
eventId: string;
message: Extract<AgentMessage, { role: "toolResult" }>;
replayIncomplete?: boolean;
}): void;
}
export function buildAssistantMessage(params: {
assistantTexts: string[];
event?: Extract<SessionEvent, { type: "assistant.message" }>;
modelRef: { api?: string; id: string; provider: string };
now: () => number;
reasoningText?: string;
usage?: AssistantUsageSnapshot;
}): AssistantMessage | undefined {
const event = params.event;
const text = event ? event.data.content || params.assistantTexts.at(-1) || "" : "";
const reasoningText = event?.data.reasoningText ?? params.reasoningText;
const toolRequests = event?.data.toolRequests ?? [];
if (!text && !reasoningText && toolRequests.length === 0) {
return undefined;
}
const content: AssistantMessage["content"] = [];
if (reasoningText) {
content.push({ thinking: reasoningText, type: "thinking" });
}
if (text) {
content.push({ text, type: "text" });
}
for (const request of toolRequests) {
content.push({
arguments: request.arguments ?? {},
id: request.toolCallId,
name: request.name,
type: "toolCall",
});
}
return {
api: params.modelRef.api ?? "openai-responses",
content,
model: event?.data.model ?? params.modelRef.id,
provider: params.modelRef.provider,
role: "assistant",
stopReason: toolRequests.length > 0 ? "toolUse" : "stop",
timestamp: params.now(),
usage: buildCopilotAssistantUsage({
fallbackOutputTokens: event?.data.outputTokens,
usage: params.usage,
}),
};
}
export function resolveAssistantUsage(
event: Extract<SessionEvent, { type: "assistant.message" }> | undefined,
latest: AssistantUsageSnapshot | undefined,
byApiCallId: Map<string, AssistantUsageSnapshot>,
): AssistantUsageSnapshot | undefined {
const apiCallId = readString(event?.data.apiCallId);
return apiCallId ? (byApiCallId.get(apiCallId) ?? latest) : latest;
}
export function resolveEventTimestamp(timestamp: string, now: () => number): number {
const parsed = Date.parse(timestamp);
return Number.isFinite(parsed) ? parsed : now();
}
export function hasOwnKeys(value: unknown): value is Record<string, unknown> {
return Boolean(value && typeof value === "object" && Object.keys(value).length > 0);
}
export function projectSdkUserMetadata(
attachments: Attachment[] | undefined,
source: string | undefined,
): Record<string, unknown> | undefined {
const summaries = (attachments ?? []).map((attachment) => {
const {
data: _data,
payload: _payload,
text: _text,
...summary
} = attachment as Attachment & {
data?: unknown;
payload?: unknown;
text?: unknown;
};
return summary;
});
const media = (attachments ?? []).flatMap<{
contentType?: string;
kind?: string;
path: string;
}>((attachment) => {
if (attachment.type === "file") {
return [{ path: attachment.path, contentType: attachment.mimeType }];
}
return attachment.type === "selection" ? [{ path: attachment.filePath, kind: "document" }] : [];
});
if (!source && summaries.length === 0) {
return undefined;
}
return {
...(source ? { copilotSource: source } : {}),
...(summaries.length > 0 ? { copilotAttachments: summaries } : {}),
...(media.length > 0 ? { media } : {}),
};
}
export function projectToolResultDetails(
data: Extract<SessionEvent, { type: "tool.execution_complete" }>["data"],
): unknown {
const result = data.result;
const sanitizedContents = result?.contents
? (sanitizeToolResult({ content: result.contents }) as { content?: unknown }).content
: undefined;
const binaryResultsForLlm = result?.binaryResultsForLlm?.map((entry) => {
const { data: _data, ...descriptor } = entry as typeof entry & { data?: unknown };
return descriptor;
});
const citableSources = result?.citableSources?.map((source) =>
Object.assign({}, source, { content: sanitizeToolDetailText(source.content) }),
);
return sanitizeToolResult({
...(result?.detailedContent
? { content: [{ type: "text", text: result.detailedContent }] }
: {}),
...(result?.structuredContent ? { structuredContent: result.structuredContent } : {}),
...(sanitizedContents ? { contents: sanitizedContents } : {}),
...(binaryResultsForLlm?.length ? { binaryResultsForLlm } : {}),
...(citableSources?.length ? { citableSources } : {}),
...(data.mcpMeta || result?.mcpMeta ? { mcpMeta: data.mcpMeta ?? result?.mcpMeta } : {}),
});
}
export function sanitizeToolDetailText(text: string): string {
const sanitized = sanitizeToolResult({ content: [{ type: "text", text }] }) as {
content?: Array<{ text?: unknown }>;
};
const value = sanitized.content?.[0]?.text;
return typeof value === "string" ? value : "";
}
function readString(value: unknown): string | undefined {
return typeof value === "string" && value.length > 0 ? value : undefined;
}

View File

@@ -10,10 +10,17 @@ const MODEL_REF = {
provider: "github-copilot",
} as const;
const REGISTERED_EVENT_TYPES = [
"user.message",
"system.message",
"skill.invoked",
"system.notification",
"assistant.message_delta",
"assistant.reasoning_delta",
"assistant.reasoning",
"assistant.turn_start",
"assistant.message",
"assistant.usage",
"tool.user_requested",
"tool.execution_start",
"tool.execution_complete",
"session.plan_changed",
@@ -188,6 +195,12 @@ describe("attachEventBridge", () => {
expect(bridge.snapshot().assistantTexts).toEqual(["root"]);
expect(bridge.snapshot().startedCount).toBe(0);
expect(bridge.snapshot().toolMetas).toEqual([{ meta: "child write", toolName: "write" }]);
expect(
bridge.recordSendResult({
...makeAssistantMessageEvent("child final"),
agentId: "child-1",
} as SessionEvent),
).toBe(false);
await bridge.awaitDeltaChain();
expect(onAssistantDelta).toHaveBeenCalledTimes(1);
});
@@ -215,6 +228,42 @@ describe("attachEventBridge", () => {
expect(bridge.snapshot().assistantTexts).toEqual(["ab", "x"]);
});
it("ignored child and ephemeral users do not split a root assistant API call", () => {
const session = createFakeSession();
const bridge = attachEventBridge(session, {
getSdkSessionId: () => "sdk-session-id",
isAborted: () => false,
});
session.emit("assistant.message", {
...makeAssistantMessageEvent("first", {
apiCallId: "shared-call",
messageId: "chunk-a",
}),
id: "assistant-chunk-a",
} as SessionEvent);
session.emit("user.message", {
...makeEvent("user.message", { content: "child" }),
agentId: "child-1",
} as SessionEvent);
session.emit("user.message", {
...makeEvent("user.message", { content: "ephemeral" }),
ephemeral: true,
} as SessionEvent);
session.emit("assistant.message", {
...makeAssistantMessageEvent("second", {
apiCallId: "shared-call",
messageId: "chunk-b",
}),
id: "assistant-chunk-b",
} as SessionEvent);
bridge.flushTranscriptProjection();
expect(bridge.buildAssistantMessage({ modelRef: MODEL_REF, now: () => 9 })?.content).toEqual([
{ type: "text", text: "firstsecond" },
]);
});
it("onAssistantDelta receives appended text, live sessionId, and current usage", async () => {
const session = createFakeSession();
let sdkSessionId = "sdk-session-1";
@@ -411,6 +460,27 @@ describe("attachEventBridge", () => {
expect(longerBridge.finalizeAssistantTexts()).toEqual(["longer text"]);
});
it("does not let an ephemeral assistant replace the final root response", () => {
const session = createFakeSession();
const bridge = attachEventBridge(session, {
getSdkSessionId: () => "sdk-session-id",
isAborted: () => false,
});
const persisted = makeAssistantMessageEvent("persisted final");
session.emit("assistant.message", persisted);
expect(
bridge.recordSendResult({
...makeAssistantMessageEvent("ephemeral final"),
ephemeral: true,
id: "ephemeral-final",
} as SessionEvent),
).toBe(false);
expect(bridge.buildAssistantMessage({ modelRef: MODEL_REF, now: () => 9 })?.content).toEqual([
{ text: "persisted final", type: "text" },
]);
});
it("assistant.message with toolRequests produces toolCall content and toolUse stopReason", () => {
const session = createFakeSession();
const bridge = attachEventBridge(session, {
@@ -605,7 +675,9 @@ describe("attachEventBridge", () => {
isAborted: () => false,
});
bridge.recordSendResult(makeAssistantMessageEvent("done", { outputTokens: 7 }));
bridge.recordSendResult(
makeAssistantMessageEvent("done", { apiCallId: "usage-without-id", outputTokens: 7 }),
);
session.emit(
"assistant.usage",
makeEvent("assistant.usage", {
@@ -1042,6 +1114,41 @@ describe("attachEventBridge", () => {
expect(bridge.buildAssistantMessage({ modelRef: MODEL_REF, now: () => 13 })).toBeUndefined();
});
it("keeps ephemeral deltas live without folding their text into the terminal message", async () => {
const session = createFakeSession();
const onAssistantDelta = vi.fn();
const bridge = attachEventBridge(session, {
getSdkSessionId: () => "sdk-session-id",
isAborted: () => false,
onAssistantDelta,
});
session.emit("assistant.message_delta", {
...makeEvent("assistant.message_delta", {
deltaContent: "hidden text",
messageId: "ephemeral-message",
}),
ephemeral: true,
} as SessionEvent);
session.emit("assistant.reasoning_delta", {
...makeEvent("assistant.reasoning_delta", {
deltaContent: "hidden reasoning",
reasoningId: "ephemeral-reasoning",
}),
ephemeral: true,
} as SessionEvent);
bridge.recordSendResult(makeAssistantMessageEvent("visible"));
await bridge.awaitDeltaChain();
expect(onAssistantDelta).toHaveBeenCalledWith(
expect.objectContaining({ delta: "hidden text", text: "hidden text" }),
);
expect(bridge.buildAssistantMessage({ modelRef: MODEL_REF, now: () => 13 })?.content).toEqual([
{ type: "thinking", thinking: "hidden reasoning" },
{ type: "text", text: "visible" },
]);
});
it("detach is idempotent after the first unsubscribe pass", () => {
const order: string[] = [];
const session = createFakeSession({

View File

@@ -5,14 +5,20 @@ import type {
AgentMessage,
} from "openclaw/plugin-sdk/agent-harness-runtime";
import {
buildCopilotAssistantUsage,
normalizeCopilotUsage,
type CopilotUsageSnapshot,
} from "./usage-bridge.js";
buildAssistantMessage,
hasOwnKeys,
projectSdkUserMetadata,
projectToolResultDetails,
resolveAssistantUsage,
resolveEventTimestamp,
sanitizeToolDetailText,
type AssistantMessage,
type AssistantUsageSnapshot,
type AttemptTranscriptJournalProjection,
} from "./event-bridge-transcript.js";
import { normalizeCopilotUsage } from "./usage-bridge.js";
export type AssistantMessage = Extract<AgentMessage, { role: "assistant" }>;
export type AssistantUsageSnapshot = CopilotUsageSnapshot;
export type { AssistantMessage, AssistantUsageSnapshot } from "./event-bridge-transcript.js";
export interface OnAssistantDeltaPayload {
delta: string;
@@ -62,6 +68,11 @@ interface EventBridgeOptions {
onContextCompacted?: () => void;
getSdkSessionId: () => string | undefined;
isAborted: () => boolean;
transcriptProjection?: {
journal: AttemptTranscriptJournalProjection;
modelRef: { api?: string; id: string; provider: string };
now: () => number;
};
}
interface EventBridgeSnapshot {
@@ -87,6 +98,7 @@ interface EventBridgeController {
settleCompactionWait(): void;
awaitDeltaChain(): Promise<void>;
awaitAgentEventChain(): Promise<void>;
flushTranscriptProjection(): void;
hasObservedCompaction(): boolean;
hasObservedSessionIdle(): boolean;
isCompacting(): boolean;
@@ -97,6 +109,17 @@ interface EventBridgeController {
}
type MessageAccumulator = { messageId: string; text: string };
type AssistantProjectionChunk = {
assistantTexts: string[];
event: Extract<SessionEvent, { type: "assistant.message" }>;
reasoningText?: string;
transcriptAssistantTexts: string[];
transcriptReasoningText?: string;
};
type AssistantProjectionGroup = {
apiCallId?: string;
chunks: AssistantProjectionChunk[];
};
type PromptErrorWithCode = Error & { code?: string; cause?: unknown };
export function attachEventBridge(
@@ -107,11 +130,21 @@ export function attachEventBridge(
const messagesById = new Map<string, MessageAccumulator>();
const reasoningOrder: string[] = [];
const reasoningById = new Map<string, string>();
const durableReasoningOrder: string[] = [];
const durableReasoningById = new Map<string, string>();
let lastAssistantEvent: Extract<SessionEvent, { type: "assistant.message" }> | undefined;
let lastAssistantReasoningText: string | undefined;
let usage: AssistantUsageSnapshot | undefined;
const usageByApiCallId = new Map<string, AssistantUsageSnapshot>();
const handledAssistantEventIds = new Set<string>();
const projectedAssistantMessageIdsWithoutApiCall = new Set<string>();
let pendingAssistantProjection: AssistantProjectionGroup | undefined;
let lastAssistantProjection: AssistantProjectionGroup | undefined;
let streamError: Error | undefined;
const toolMetas: AgentHarnessAttemptResult["toolMetas"] = [];
const toolMetaIndexByCallId = new Map<string, number>();
const projectedToolNamesByCallId = new Map<string, string>();
const userRequestedToolCallIds = new Set<string>();
let startedCount = 0;
let completedCount = 0;
let activeCompactionCount = 0;
@@ -129,8 +162,66 @@ export function attachEventBridge(
});
let firstDeltaError: unknown;
let detached = false;
let unconsumedDurableReasoning = false;
const unsubscribeFns: Array<() => void> = [];
registerListener(session, unsubscribeFns, "user.message", (event) => {
if (!isRootSessionEvent(event) || event.ephemeral === true) {
return;
}
flushPendingAssistantProjection();
const projection = options.transcriptProjection;
if (!projection) {
return;
}
const source = readString(event.data.source);
const transformedContent =
typeof event.data.transformedContent === "string" ? event.data.transformedContent : undefined;
const openClawMeta = projectSdkUserMetadata(event.data.attachments, source);
const idempotencyKey = `copilot-sdk:${options.getSdkSessionId() ?? "unknown"}:${event.id}`;
// `source` is open-ended provenance, not a visibility enum. Hide the one
// documented injected source; unknown sources stay visible without guessing.
const hidden = event.data.isAutopilotContinuation === true || source === "skill-pdf";
projection.journal.recordSdkUser({
eventId: event.id,
autopilotContinuation: event.data.isAutopilotContinuation === true,
replayIncomplete: Boolean(
event.data.attachments?.length ||
(event.data.agentMode !== undefined && event.data.agentMode !== "interactive") ||
(transformedContent !== undefined && transformedContent !== event.data.content),
),
message: {
role: "user",
content: event.data.content,
timestamp: resolveEventTimestamp(event.timestamp, projection.now),
idempotencyKey,
...(hidden ? { display: false } : {}),
...(openClawMeta ? { __openclaw: openClawMeta } : {}),
} as Extract<AgentMessage, { role: "user" }>,
});
});
registerListener(session, unsubscribeFns, "system.message", (event) => {
if (!isRootSessionEvent(event) || event.ephemeral === true) {
return;
}
// System/developer prompts affect native history but AgentMessage has no
// lossless canonical role for them, so keep native replay fail-closed.
options.transcriptProjection?.journal.markReplayIncomplete();
});
registerListener(session, unsubscribeFns, "skill.invoked", (event) => {
if (isRootSessionEvent(event) && event.ephemeral !== true) {
options.transcriptProjection?.journal.markReplayIncomplete();
}
});
registerListener(session, unsubscribeFns, "system.notification", (event) => {
if (isRootSessionEvent(event) && event.ephemeral !== true) {
options.transcriptProjection?.journal.markReplayIncomplete();
}
});
registerListener(session, unsubscribeFns, "assistant.message_delta", (event) => {
if (!isRootSessionEvent(event)) {
return;
@@ -184,15 +275,32 @@ export function attachEventBridge(
reasoningById.set(reasoningId, `${reasoningById.get(reasoningId) ?? ""}${delta}`);
});
registerListener(session, unsubscribeFns, "assistant.message", (event) => {
if (!isRootSessionEvent(event)) {
registerListener(session, unsubscribeFns, "assistant.reasoning", (event) => {
if (!isRootSessionEvent(event) || event.ephemeral === true) {
return;
}
lastAssistantEvent = event;
const entry = ensureMessageAccumulator(messagesById, messageOrder, event.data.messageId);
if (typeof event.data.content === "string" && event.data.content.length >= entry.text.length) {
entry.text = event.data.content;
if (!reasoningById.has(event.data.reasoningId)) {
reasoningOrder.push(event.data.reasoningId);
}
reasoningById.set(event.data.reasoningId, event.data.content);
if (!durableReasoningById.has(event.data.reasoningId)) {
durableReasoningOrder.push(event.data.reasoningId);
}
durableReasoningById.set(event.data.reasoningId, event.data.content);
unconsumedDurableReasoning = true;
});
registerListener(session, unsubscribeFns, "assistant.turn_start", (event) => {
if (isRootSessionEvent(event)) {
markUnconsumedReasoningIncomplete();
}
});
registerListener(session, unsubscribeFns, "assistant.message", (event) => {
if (!isRootSessionEvent(event) || event.ephemeral === true) {
return;
}
handleAssistantMessage(event);
});
registerListener(session, unsubscribeFns, "assistant.usage", (event) => {
@@ -200,9 +308,24 @@ export function attachEventBridge(
return;
}
usage = normalizeCopilotUsage(event.data);
const apiCallId = readString(event.data.apiCallId);
if (apiCallId && usage) {
usageByApiCallId.set(apiCallId, usage);
}
if (apiCallId) {
flushPendingAssistantProjection(apiCallId);
}
});
registerListener(session, unsubscribeFns, "tool.user_requested", (event) => {
if (isRootSessionEvent(event) && event.ephemeral !== true) {
userRequestedToolCallIds.add(event.data.toolCallId);
options.transcriptProjection?.journal.markReplayIncomplete();
}
});
registerListener(session, unsubscribeFns, "tool.execution_start", (event) => {
flushPendingAssistantProjectionForToolCall(event.data.toolCallId);
if (isRootSessionEvent(event)) {
startedCount += 1;
}
@@ -211,6 +334,7 @@ export function attachEventBridge(
});
registerListener(session, unsubscribeFns, "tool.execution_complete", (event) => {
flushPendingAssistantProjectionForToolCall(event.data.toolCallId);
if (isRootSessionEvent(event)) {
completedCount += 1;
}
@@ -226,6 +350,40 @@ export function attachEventBridge(
...(event.data.success ? {} : { isError: true }),
};
}
const projection = options.transcriptProjection;
const isDurableRootCompletion = isRootSessionEvent(event) && event.ephemeral !== true;
const wasTrackedUserRequest = userRequestedToolCallIds.has(event.data.toolCallId);
const isUserRequested = event.data.isUserRequested === true || wasTrackedUserRequest;
const projectedToolName = projectedToolNamesByCallId.get(event.data.toolCallId);
if (isDurableRootCompletion) {
userRequestedToolCallIds.delete(event.data.toolCallId);
projectedToolNamesByCallId.delete(event.data.toolCallId);
}
if (projection && isDurableRootCompletion && isUserRequested) {
projection.journal.markReplayIncomplete();
}
if (projection && isDurableRootCompletion && !isUserRequested) {
const resultText = event.data.success
? (event.data.result?.content ?? "")
: (event.data.error?.message ?? "Tool execution failed");
const details = projectToolResultDetails(event.data);
const replayIncomplete = Boolean(
event.data.result?.binaryResultsForLlm?.length || event.data.result?.citableSources?.length,
);
projection.journal.recordToolResult({
eventId: event.id,
replayIncomplete,
message: {
role: "toolResult",
toolCallId: event.data.toolCallId,
toolName: toolName ?? event.data.toolDescription?.name ?? projectedToolName ?? "unknown",
content: [{ type: "text", text: sanitizeToolDetailText(resultText) }],
...(hasOwnKeys(details) ? { details } : {}),
isError: !event.data.success,
timestamp: resolveEventTimestamp(event.timestamp, projection.now),
},
});
}
});
registerListener(session, unsubscribeFns, "session.plan_changed", (event) => {
@@ -341,12 +499,15 @@ export function attachEventBridge(
if (!isRootCompactionEvent(event)) {
return;
}
markUnconsumedReasoningIncomplete();
flushPendingAssistantProjection();
observedSessionIdle = true;
resolveSessionIdle?.();
resolveSessionIdle = undefined;
});
registerListener(session, unsubscribeFns, "session.error", (event) => {
markUnconsumedReasoningIncomplete();
if (!options.isAborted()) {
streamError = createPromptError(
event.data.errorCode ?? event.data.errorType,
@@ -356,6 +517,7 @@ export function attachEventBridge(
});
registerListener(session, unsubscribeFns, "abort", (event) => {
markUnconsumedReasoningIncomplete();
if (!options.isAborted()) {
streamError = createPromptError(
"session_aborted",
@@ -366,10 +528,15 @@ export function attachEventBridge(
return {
recordSendResult(result) {
if (!isAssistantMessageEvent(result)) {
if (
!isAssistantMessageEvent(result) ||
!isRootSessionEvent(result) ||
result.ephemeral === true
) {
return false;
}
lastAssistantEvent = result;
handleAssistantMessage(result);
flushPendingAssistantProjection();
return true;
},
awaitCompactionChain() {
@@ -392,6 +559,10 @@ export function attachEventBridge(
awaitAgentEventChain() {
return agentEventChain;
},
flushTranscriptProjection() {
markUnconsumedReasoningIncomplete();
flushPendingAssistantProjection();
},
hasObservedCompaction() {
return observedCompaction;
},
@@ -413,15 +584,24 @@ export function attachEventBridge(
};
},
buildAssistantMessage(args) {
return buildAssistantMessage({
event: lastAssistantEvent,
modelRef: args.modelRef,
now: args.now,
reasoningById,
reasoningOrder,
usage,
assistantTexts: finalizeAssistantTexts(messageOrder, messagesById, lastAssistantEvent),
});
const group = pendingAssistantProjection ?? lastAssistantProjection;
return group
? buildAssistantProjectionGroup(
group,
args.modelRef,
() => args.now(),
usageByApiCallId,
usage,
false,
).message
: buildAssistantMessage({
event: lastAssistantEvent,
modelRef: args.modelRef,
now: args.now,
reasoningText: lastAssistantReasoningText,
usage: resolveAssistantUsage(lastAssistantEvent, usage, usageByApiCallId),
assistantTexts: finalizeAssistantTexts(messageOrder, messagesById, lastAssistantEvent),
});
},
finalizeAssistantTexts() {
return finalizeAssistantTexts(messageOrder, messagesById, lastAssistantEvent);
@@ -442,6 +622,135 @@ export function attachEventBridge(
},
};
function handleAssistantMessage(
event: Extract<SessionEvent, { type: "assistant.message" }>,
): void {
if (!isRootSessionEvent(event) || event.ephemeral === true) {
return;
}
lastAssistantEvent = event;
if (handledAssistantEventIds.has(event.id)) {
return;
}
handledAssistantEventIds.add(event.id);
for (const request of event.data.toolRequests ?? []) {
projectedToolNamesByCallId.set(request.toolCallId, request.name);
}
const entry = ensureMessageAccumulator(messagesById, messageOrder, event.data.messageId);
if (typeof event.data.content === "string" && event.data.content.length >= entry.text.length) {
entry.text = event.data.content;
}
lastAssistantReasoningText =
event.data.reasoningText ?? (joinReasoning(reasoningOrder, reasoningById) || undefined);
const transcriptReasoningText =
event.data.reasoningText ??
(joinReasoning(durableReasoningOrder, durableReasoningById) || undefined);
reasoningOrder.length = 0;
reasoningById.clear();
durableReasoningOrder.length = 0;
durableReasoningById.clear();
unconsumedDurableReasoning = false;
const chunk: AssistantProjectionChunk = {
event,
assistantTexts: [messagesById.get(event.data.messageId)?.text ?? ""],
...(lastAssistantReasoningText ? { reasoningText: lastAssistantReasoningText } : {}),
transcriptAssistantTexts: [event.data.content ?? ""],
...(transcriptReasoningText ? { transcriptReasoningText } : {}),
};
const apiCallId = readString(event.data.apiCallId);
if (!apiCallId) {
if (projectedAssistantMessageIdsWithoutApiCall.has(event.data.messageId)) {
options.transcriptProjection?.journal.markReplayIncomplete();
lastAssistantProjection = { chunks: [chunk] };
return;
}
projectedAssistantMessageIdsWithoutApiCall.add(event.data.messageId);
flushPendingAssistantProjection();
const group = { chunks: [chunk] } satisfies AssistantProjectionGroup;
lastAssistantProjection = group;
recordAssistantProjection(group);
return;
}
if (pendingAssistantProjection?.apiCallId !== apiCallId) {
flushPendingAssistantProjection();
pendingAssistantProjection = { apiCallId, chunks: [] };
}
const priorChunkIndex = pendingAssistantProjection.chunks.findIndex(
(candidate) => candidate.event.data.messageId === event.data.messageId,
);
if (priorChunkIndex === -1) {
pendingAssistantProjection.chunks.push(chunk);
} else {
pendingAssistantProjection.chunks[priorChunkIndex] = chunk;
}
}
function flushPendingAssistantProjection(apiCallId?: string): void {
const group = pendingAssistantProjection;
if (!group || (apiCallId !== undefined && group.apiCallId !== apiCallId)) {
return;
}
pendingAssistantProjection = undefined;
lastAssistantProjection = group;
recordAssistantProjection(group);
}
function markUnconsumedReasoningIncomplete(): void {
if (unconsumedDurableReasoning) {
options.transcriptProjection?.journal.markReplayIncomplete();
}
reasoningOrder.length = 0;
reasoningById.clear();
durableReasoningOrder.length = 0;
durableReasoningById.clear();
unconsumedDurableReasoning = false;
}
function flushPendingAssistantProjectionForToolCall(toolCallId: string): void {
const group = pendingAssistantProjection;
if (
group?.chunks.some((chunk) =>
chunk.event.data.toolRequests?.some((request) => request.toolCallId === toolCallId),
)
) {
flushPendingAssistantProjection();
}
}
function recordAssistantProjection(group: AssistantProjectionGroup): void {
const projection = options.transcriptProjection;
if (!projection) {
return;
}
const { message, replayIncomplete, toolCallIds } = buildAssistantProjectionGroup(
group,
projection.modelRef,
(event) => resolveEventTimestamp(event.timestamp, projection.now),
usageByApiCallId,
// Optional assistant.usage must never delay canonical content. Later
// unkeyed usage stays attempt metadata; this row uses per-call data only.
undefined,
true,
);
const eventId = group.chunks[0]?.event.id;
if (!eventId) {
return;
}
if (!message) {
if (replayIncomplete) {
projection.journal.markReplayIncomplete();
}
projection.journal.recordAssistantProjectionGap();
return;
}
projection.journal.recordAssistant({
eventId,
message,
replayIncomplete,
toolCallIds,
});
}
function enqueueCompactionCallback(callback: (() => void | Promise<void>) | undefined): void {
if (!callback) {
return;
@@ -490,57 +799,6 @@ export function attachEventBridge(
}
}
function buildAssistantMessage(params: {
assistantTexts: string[];
event?: Extract<SessionEvent, { type: "assistant.message" }>;
modelRef: { api?: string; id: string; provider: string };
now: () => number;
reasoningById: Map<string, string>;
reasoningOrder: string[];
usage?: AssistantUsageSnapshot;
}): AssistantMessage | undefined {
const event = params.event;
const text = event
? event.data.content || params.assistantTexts[params.assistantTexts.length - 1] || ""
: "";
const reasoningText =
event?.data.reasoningText ?? joinReasoning(params.reasoningOrder, params.reasoningById);
const toolRequests = event?.data.toolRequests ?? [];
if (!text && !reasoningText && toolRequests.length === 0) {
return undefined;
}
const content: AssistantMessage["content"] = [];
if (reasoningText) {
content.push({ thinking: reasoningText, type: "thinking" });
}
if (text) {
content.push({ text, type: "text" });
}
for (const request of toolRequests) {
content.push({
arguments: request.arguments ?? {},
id: request.toolCallId,
name: request.name,
type: "toolCall",
});
}
return {
api: params.modelRef.api ?? "openai-responses",
content,
model: event?.data.model ?? params.modelRef.id,
provider: params.modelRef.provider,
role: "assistant",
stopReason: toolRequests.length > 0 ? "toolUse" : "stop",
timestamp: params.now(),
usage: buildCopilotAssistantUsage({
fallbackOutputTokens: event?.data.outputTokens,
usage: params.usage,
}),
};
}
function createPromptError(code: string, message: string, cause?: unknown): PromptErrorWithCode {
const error = new Error(message) as PromptErrorWithCode;
error.code = code;
@@ -581,6 +839,102 @@ function finalizeAssistantTexts(
return [];
}
function buildAssistantProjectionGroup(
group: AssistantProjectionGroup,
modelRef: { api?: string; id: string; provider: string },
resolveTimestamp: (event: Extract<SessionEvent, { type: "assistant.message" }>) => number,
usageByApiCallId: Map<string, AssistantUsageSnapshot>,
latestUsage: AssistantUsageSnapshot | undefined,
forTranscript: boolean,
): {
message: AssistantMessage | undefined;
replayIncomplete: boolean;
toolCallIds: string[];
} {
const messages = group.chunks.flatMap((chunk) => {
const message = buildAssistantMessage({
event: chunk.event,
modelRef,
now: () => resolveTimestamp(chunk.event),
reasoningText: forTranscript ? chunk.transcriptReasoningText : chunk.reasoningText,
// Usage is keyed to the complete API call, so every chunk resolves to the
// same snapshot and the merged message keeps the terminal copy.
usage: resolveAssistantUsage(chunk.event, latestUsage, usageByApiCallId),
assistantTexts: forTranscript ? chunk.transcriptAssistantTexts : chunk.assistantTexts,
});
return message ? [message] : [];
});
const replayIncomplete = group.chunks.some(({ event }) =>
hasUnprojectedAssistantReplayState(event),
);
const last = messages.at(-1);
if (!last) {
return { message: undefined, replayIncomplete, toolCallIds: [] };
}
const narrative: AssistantMessage["content"] = [];
let terminalThinking:
| Extract<AssistantMessage["content"][number], { type: "thinking" }>
| undefined;
const toolCallOrder: string[] = [];
const toolCallsById = new Map<
string,
Extract<AssistantMessage["content"][number], { type: "toolCall" }>
>();
for (const message of messages) {
for (const part of message.content) {
if (part.type === "toolCall") {
if (!toolCallsById.has(part.id)) {
toolCallOrder.push(part.id);
}
toolCallsById.set(part.id, part);
continue;
}
if (part.type === "thinking") {
// Reasoning is an accumulated snapshot, not a per-message delta. Keep
// only the terminal snapshot when one API call emits phased chunks.
terminalThinking = part;
continue;
}
const previous = narrative.at(-1);
if (part.type === "text" && previous?.type === "text") {
narrative[narrative.length - 1] = { ...previous, text: previous.text + part.text };
} else {
narrative.push(part);
}
}
}
const toolCalls = toolCallOrder.flatMap((id) => {
const toolCall = toolCallsById.get(id);
return toolCall ? [toolCall] : [];
});
const content = [...(terminalThinking ? [terminalThinking] : []), ...narrative, ...toolCalls];
const toolCallIds = [...toolCallOrder];
return {
message: {
...last,
content,
stopReason: toolCallIds.length > 0 ? "toolUse" : "stop",
},
replayIncomplete,
toolCallIds,
};
}
function hasUnprojectedAssistantReplayState(
event: Extract<SessionEvent, { type: "assistant.message" }>,
): boolean {
// The SDK contract marks these as provider/session-bound state or custom
// call shape. AgentMessage cannot represent them, so native replay must stay.
return (
event.data.citations !== undefined ||
event.data.serverTools !== undefined ||
event.data.reasoningWireField !== undefined ||
event.data.reasoningOpaque !== undefined ||
event.data.encryptedContent !== undefined ||
event.data.toolRequests?.some((request) => request.type === "custom") === true
);
}
function isAssistantMessageEvent(
event: SessionEvent | undefined,
): event is Extract<SessionEvent, { type: "assistant.message" }> {

View File

@@ -1779,6 +1779,52 @@ describe("CLI attempt execution", () => {
);
});
it("does not gap-fill an assistant already owned by the runtime", async () => {
const sessionKey = "agent:main:direct:runtime-owned-assistant";
const sessionEntry: SessionEntry = {
sessionId: "session-runtime-owned-assistant",
updatedAt: Date.now(),
};
await appendTranscriptMessage(
{ agentId: "main", sessionId: sessionEntry.sessionId, sessionKey, storePath },
{
message: {
role: "assistant",
content: [{ type: "text", text: "runtime answer" }],
timestamp: Date.now(),
},
cwd: tmpDir,
},
);
await persistCliTurnTranscript({
body: "ignored prompt",
result: makeCliResult("runtime answer"),
sessionId: sessionEntry.sessionId,
sessionKey,
sessionEntry,
storePath,
sessionAgentId: "main",
sessionCwd: tmpDir,
config: {},
embeddedAssistantGapFill: true,
skipAssistantTurn: true,
});
const messages = await readSessionMessages(
formatSqliteSessionFileMarker({
agentId: "main",
sessionId: sessionEntry.sessionId,
storePath,
}),
);
expect(messages).toHaveLength(1);
expect(messages[0]).toMatchObject({
role: "assistant",
content: [{ type: "text", text: "runtime answer" }],
});
});
it("persists a media-only ACP user turn when the reply is empty", async () => {
const sessionKey = "agent:main:direct:acp-media-only";
const sessionFile = path.join(tmpDir, "session-acp-media-only.jsonl");

View File

@@ -156,6 +156,7 @@ type PersistTextTurnTranscriptParams = {
sessionCwd: string;
config: OpenClawConfig;
embeddedAssistantGapFill?: boolean;
skipAssistantTurn?: boolean;
assistant: {
api: string;
provider: string;
@@ -274,7 +275,7 @@ async function persistTextTurnTranscript(
params: PersistTextTurnTranscriptParams,
): Promise<PersistTextTurnTranscriptResult> {
const promptText = params.transcriptBody ?? params.body;
const replyText = params.finalText;
const replyText = params.skipAssistantTurn === true ? "" : params.finalText;
const userMessage =
params.userMessage ??
(promptText
@@ -416,6 +417,7 @@ export async function persistCliTurnTranscript(params: {
config: OpenClawConfig;
embeddedAssistantGapFill?: boolean;
skipUserTurn?: boolean;
skipAssistantTurn?: boolean;
}): Promise<PersistTextTurnTranscriptResult> {
const replyText = resolveCliTranscriptReplyText(params.result);
const provider = params.result.meta.agentMeta?.provider?.trim() ?? "cli";
@@ -445,6 +447,7 @@ export async function persistCliTurnTranscript(params: {
model,
usage: params.result.meta.agentMeta?.usage,
},
skipAssistantTurn: params.skipAssistantTurn,
});
}

View File

@@ -1,3 +1,4 @@
import { getReplyPayloadMetadata } from "../../auto-reply/reply-payload.js";
import type { CliDeps } from "../../cli/deps.types.js";
import type { RestartRecoveryTerminalDeliveryEvidenceResult } from "../../config/sessions/restart-recovery-types.js";
import type { SessionEntry } from "../../config/sessions/types.js";
@@ -97,6 +98,12 @@ export async function finalizeEmbeddedAgentCommand(params: {
try {
await fallbackTrajectoryRecorder?.flush();
const finalVisiblePayload = result.payloads
?.toReversed()
.find((payload) => !payload.isError && !payload.isReasoning && payload.text?.trim());
const assistantTranscriptOwned =
finalVisiblePayload !== undefined &&
getReplyPayloadMetadata(finalVisiblePayload)?.assistantTranscriptOwned === true;
if (params.opts.internalDeliveryMediaUrls !== undefined) {
result = {
...result,
@@ -189,6 +196,7 @@ export async function finalizeEmbeddedAgentCommand(params: {
sessionCwd: effectiveCwd,
config: cfg,
embeddedAssistantGapFill,
skipAssistantTurn: assistantTranscriptOwned,
skipUserTurn:
suppressUserTurnPersistence ||
userTurnTranscriptRecorder.hasPersisted() ||

View File

@@ -328,6 +328,7 @@ export function appendSqliteTranscriptEventSync(
export async function appendSqliteExpectedSessionTranscriptTurn(
scope: SessionTranscriptWriteScope,
options: {
atomicGroup?: boolean;
config?: import("../types.openclaw.js").OpenClawConfig;
cwd?: string;
expectedLifecycleRevision?: string;
@@ -376,6 +377,7 @@ export async function appendSqliteExpectedSessionTranscriptTurn(
const { shouldAppend: _shouldAppend, ...appendOptions } = append;
const appended = appendSqliteTranscriptMessageInTransaction(transactionDb, resolved, {
...appendOptions,
messageAlreadyRedacted: options.atomicGroup === true,
...((append.cwd ?? options.cwd) ? { cwd: append.cwd ?? options.cwd } : {}),
...((append.config ?? options.config) ? { config: append.config ?? options.config } : {}),
});
@@ -383,6 +385,14 @@ export async function appendSqliteExpectedSessionTranscriptTurn(
appendedMessages.push(appended);
}
}
if (
options.atomicGroup &&
(appendedMessages.length !== messages.length ||
appendedMessages.some((message) => message.appended) !==
appendedMessages.every((message) => message.appended))
) {
throw new Error("SQLite transcript batch was not wholly inserted or replayed");
}
const sessionPatch = buildExpectedTranscriptTurnSessionPatch({
appendedMessages,
@@ -590,7 +600,7 @@ function assertSqliteTranscriptSnapshotUnchanged(
function appendSqliteTranscriptMessageInTransaction<TMessage>(
database: OpenClawAgentDatabase,
resolved: ResolvedTranscriptScope,
options: TranscriptMessageAppendOptions<TMessage>,
options: TranscriptMessageAppendOptions<TMessage> & { messageAlreadyRedacted?: boolean },
): TranscriptMessageAppendResult<TMessage> | undefined {
const idempotencyKey = readMessageIdempotencyKey(options.message);
if (idempotencyKey && options.idempotencyLookup !== "caller-checked") {
@@ -618,7 +628,9 @@ function appendSqliteTranscriptMessageInTransaction<TMessage>(
const messageId = options.eventId ?? randomUUID();
const now = options.now ?? Date.now();
const finalMessage = redactTranscriptMessageForStorage(prepared, options);
const finalMessage = options.messageAlreadyRedacted
? prepared
: redactTranscriptMessageForStorage(prepared, options);
ensureTranscriptHeader(database, resolved, options.cwd, now);
const parentId =
options.parentId === undefined

View File

@@ -1,3 +1,4 @@
import { randomUUID } from "node:crypto";
import { resolveDefaultAgentId } from "../../agents/agent-scope-config.js";
import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
import { getRuntimeConfig } from "../io.js";
@@ -8,6 +9,7 @@ import {
resolveSessionEntryFromStore,
resolveSessionEntrySelection,
} from "./session-accessor.entry.js";
import { redactTranscriptMessageForStorage } from "./session-accessor.sqlite-transcript-store.js";
import { appendSqliteExpectedSessionTranscriptTurn } from "./session-accessor.sqlite.js";
import { shouldUseExplicitTranscriptFile } from "./session-accessor.transcript-target.js";
import { appendTranscriptMessage, emitTranscriptUpdate } from "./session-accessor.transcript.js";
@@ -24,6 +26,42 @@ import { formatSqliteSessionFileMarker, parseSqliteSessionFileMarker } from "./s
import { runWithOwnedSessionTranscriptWriteLock } from "./transcript-write-context.js";
import type { SessionEntry } from "./types.js";
/** Appends one prepared ordered group in the existing transcript turn transaction. */
export async function appendTranscriptMessages<TMessage>(
scope: SessionTranscriptWriteScope,
options: Pick<SessionTranscriptTurnPersistOptions, "config" | "cwd"> & {
messages: readonly Omit<
SessionTranscriptTurnMessageAppend,
"config" | "cwd" | "parentId" | "prepareMessageAfterIdempotencyCheck" | "shouldAppend"
>[];
},
): Promise<TranscriptMessageAppendResult<TMessage>[]> {
if (options.messages.length === 0) {
return [];
}
const expectedSessionId = scope.sessionId?.trim();
if (!expectedSessionId) {
throw new Error("Cannot append a transcript batch without an exact session id");
}
const turn = await persistExpectedSessionTranscriptTurn(scope, {
atomicGroup: true,
config: options.config,
cwd: options.cwd,
expectedSessionId,
messages: options.messages.map((append) => ({
...append,
eventId: append.eventId ?? randomUUID(),
message: redactTranscriptMessageForStorage(append.message, options),
now: append.now ?? Date.now(),
})),
updateMode: "none",
});
if (turn.rejectedReason) {
throw new Error("Transcript session changed before batch append");
}
return turn.messages as TranscriptMessageAppendResult<TMessage>[];
}
/**
* Persists one logical transcript turn through the SQLite-backed session target.
* Transcript row append(s), the synthetic sessionFile marker, and the requested
@@ -134,7 +172,10 @@ async function persistExpectedSessionTranscriptTurn(
sessionEntry?: SessionEntry;
sessionStore?: Record<string, SessionEntry>;
},
options: SessionTranscriptTurnPersistOptions & { expectedSessionId: string },
options: SessionTranscriptTurnPersistOptions & {
atomicGroup?: boolean;
expectedSessionId: string;
},
): Promise<SessionTranscriptTurnPersistResult> {
const sessionKey = scope.sessionKey?.trim();
if (!scope.storePath || !sessionKey) {
@@ -190,6 +231,7 @@ async function persistExpectedSessionTranscriptTurn(
expectedLifecycleRevision: options.expectedLifecycleRevision,
expectedSessionState: options.expectedSessionState,
expectedSessionId,
atomicGroup: options.atomicGroup,
messages: options.messages,
sessionLifecyclePatch: options.sessionLifecyclePatch,
sessionFile: target.sessionFile,

View File

@@ -204,7 +204,10 @@ export {
withTranscriptWriteLock,
withTranscriptWriteTransaction,
} from "./session-accessor.transcript.js";
export { persistSessionTranscriptTurn } from "./session-accessor.transcript-turn.js";
export {
appendTranscriptMessages,
persistSessionTranscriptTurn,
} from "./session-accessor.transcript-turn.js";
export {
isSessionTranscriptProjectionUnavailableError,
readRecentSessionTranscriptMessageEvents,

View File

@@ -13,6 +13,8 @@ import * as transcriptEvents from "../sessions/transcript-events.js";
import {
appendAssistantMirrorMessageByIdentity,
appendSessionTranscriptMessageByIdentity,
appendSessionTranscriptMessageByIdentityStrict,
appendSessionTranscriptMessagesByIdentity,
formatSessionTranscriptMemoryHitKey,
parseSessionTranscriptMemoryHitKey,
publishSessionTranscriptUpdateByIdentity,
@@ -81,6 +83,74 @@ describe("session transcript runtime SDK", () => {
expect(loadSessionEntry(scope)?.sessionFile).toBeUndefined();
});
it("atomically appends and idempotently replays an ordered message group", async () => {
const scope = {
agentId: "main",
sessionId: "batch-session",
sessionKey: "agent:main:batch",
storePath,
};
await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
const messages = [
{
eventId: "batch-assistant",
idempotencyLookup: "scan" as const,
message: { role: "assistant", content: "checking", idempotencyKey: "batch:assistant" },
now: 1_000,
},
{
eventId: "batch-result",
idempotencyLookup: "scan" as const,
message: { role: "toolResult", content: "done", idempotencyKey: "batch:result" },
now: 2_000,
},
];
const appended = await appendSessionTranscriptMessagesByIdentity({ ...scope, messages });
const replayed = await appendSessionTranscriptMessagesByIdentity({ ...scope, messages });
expect(appended.map((result) => result.appended)).toEqual([true, true]);
expect(replayed.map((result) => result.appended)).toEqual([false, false]);
const events = await readSessionTranscriptEvents(scope);
expect(events).toHaveLength(3);
expect(events.slice(1)).toMatchObject([
{ id: "batch-assistant", parentId: null },
{ id: "batch-result", parentId: "batch-assistant" },
]);
});
it("distinguishes strict singleton results, suppression, and session rebound", async () => {
const scope = {
agentId: "main",
sessionId: "strict-session",
sessionKey: "agent:main:strict",
storePath,
};
await upsertSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 10 });
await expect(
appendSessionTranscriptMessageByIdentityStrict({
...scope,
message: { role: "user", content: "blocked" },
prepareMessageAfterIdempotencyCheck: () => undefined,
}),
).resolves.toEqual({ kind: "suppressed" });
await expect(
appendSessionTranscriptMessageByIdentityStrict({
...scope,
message: { role: "user", content: "persisted", idempotencyKey: "strict:user" },
}),
).resolves.toMatchObject({ kind: "result", result: { appended: true } });
await upsertSessionEntry(scope, { sessionId: "replacement-session", updatedAt: 20 });
await expect(
appendSessionTranscriptMessageByIdentityStrict({
...scope,
message: { role: "assistant", content: "stale" },
}),
).resolves.toEqual({ kind: "rejected", reason: "session-rebound" });
});
it("pages raw events across appends and resets after replacement", async () => {
const scope = {
agentId: "main",

View File

@@ -2,10 +2,12 @@ import { isRecord } from "@openclaw/normalization-core/record-coerce";
import { redactTranscriptMessage } from "../agents/transcript-redact.js";
import {
appendTranscriptMessage,
appendTranscriptMessages,
isSessionTranscriptProjectionUnavailableError,
loadSessionEntry,
loadTranscriptEvents,
publishTranscriptUpdate,
persistSessionTranscriptTurn,
readTranscriptRawDelta,
readSessionTranscriptVisibleMessageDelta as readVisibleMessageDelta,
readLatestTranscriptAssistantText,
@@ -128,6 +130,20 @@ export type SessionTranscriptTarget = SessionTranscriptIdentity & {
export type SessionTranscriptAppendMessageParams<TMessage> = SessionTranscriptTargetParams &
TranscriptMessageAppendOptions<TMessage>;
export type SessionTranscriptAppendMessagesParams<TMessage> = SessionTranscriptTargetParams & {
config?: TranscriptMessageAppendOptions<TMessage>["config"];
cwd?: string;
messages: readonly Omit<
TranscriptMessageAppendOptions<TMessage>,
"config" | "cwd" | "parentId" | "prepareMessageAfterIdempotencyCheck" | "useRawWhenLinear"
>[];
};
export type SessionTranscriptStrictMessageAppendResult<TMessage> =
| { kind: "result"; result: TranscriptMessageAppendResult<TMessage> }
| { kind: "suppressed" }
| { kind: "rejected"; reason: "session-rebound" };
export type SessionTranscriptAssistantMirrorAppendParams = SessionTranscriptReadParams & {
config?: OpenClawConfig;
deliveryMirror?: SessionTranscriptDeliveryMirror;
@@ -345,6 +361,57 @@ export async function appendSessionTranscriptMessageByIdentity<TMessage>(
return await appendTranscriptMessage(params, params);
}
/** Appends one message while preserving distinct suppression and session-rebind outcomes. */
export async function appendSessionTranscriptMessageByIdentityStrict<TMessage>(
params: SessionTranscriptAppendMessageParams<TMessage>,
): Promise<SessionTranscriptStrictMessageAppendResult<TMessage>> {
const expectedSessionId = params.sessionId?.trim();
if (!expectedSessionId) {
throw new Error("Cannot strictly append a transcript message without an exact session id");
}
const turn = await persistSessionTranscriptTurn(params, {
...(params.config ? { config: params.config } : {}),
...(params.cwd ? { cwd: params.cwd } : {}),
expectedSessionId,
messages: [
{
...(params.eventId !== undefined ? { eventId: params.eventId } : {}),
...(params.idempotencyLookup !== undefined
? { idempotencyLookup: params.idempotencyLookup }
: {}),
message: params.message,
...(params.now !== undefined ? { now: params.now } : {}),
...(params.parentId !== undefined ? { parentId: params.parentId } : {}),
...(params.prepareMessageAfterIdempotencyCheck
? {
prepareMessageAfterIdempotencyCheck: (message: unknown) =>
params.prepareMessageAfterIdempotencyCheck?.(message as TMessage),
}
: {}),
...(params.useRawWhenLinear !== undefined
? { useRawWhenLinear: params.useRawWhenLinear }
: {}),
},
],
updateMode: "none",
});
if (turn.rejectedReason) {
return { kind: "rejected", reason: turn.rejectedReason };
}
const result = turn.messages[0] as TranscriptMessageAppendResult<TMessage> | undefined;
return result ? { kind: "result", result } : { kind: "suppressed" };
}
/**
* Atomically appends one ordered, already-hooked message group. Preparation and
* redaction finish before SQLite begins; this is the canonical future harness seam.
*/
export async function appendSessionTranscriptMessagesByIdentity<TMessage>(
params: SessionTranscriptAppendMessagesParams<TMessage>,
): Promise<TranscriptMessageAppendResult<TMessage>[]> {
return await appendTranscriptMessages(params, params);
}
/**
* Publishes a transcript update by scoped transcript target.
*/

View File

@@ -221,20 +221,36 @@ describe("oxlint config", () => {
const maxLinesOverrides = (config.overrides ?? []).filter(
(override) => override.rules?.["max-lines"],
);
const scopedBudgets = maxLinesOverrides.filter((override) => override.excludeFiles);
const exactExceptions = maxLinesOverrides.filter((override) => !override.excludeFiles);
expect(maxLinesOverrides).toHaveLength(4);
expect(maxLinesOverrides.map((override) => override.rules?.["max-lines"])).toEqual([
expect(scopedBudgets).toHaveLength(4);
expect(scopedBudgets.map((override) => override.rules?.["max-lines"])).toEqual([
["error", { max: 700, skipBlankLines: true, skipComments: true }],
["error", { max: 700, skipBlankLines: true, skipComments: true }],
["error", { max: 800, skipBlankLines: true, skipComments: true }],
["error", { max: 1000, skipBlankLines: true, skipComments: true }],
]);
for (const override of maxLinesOverrides) {
for (const override of scopedBudgets) {
expect(override.excludeFiles).toContain("**/protocol-gen/**");
expect(override.excludeFiles).toContain("**/*.generated.*");
expect(override.excludeFiles).toContain("ui/src/i18n/locales/**");
expect(override.excludeFiles).toContain("src/wizard/i18n/locales/**");
}
expect(exactExceptions).toEqual([
{
files: ["extensions/copilot/src/event-bridge.ts"],
rules: {
"max-lines": ["error", { max: 950, skipBlankLines: true, skipComments: true }],
},
},
{
files: ["extensions/copilot/src/attempt-transcript-journal.test.ts"],
rules: {
"max-lines": ["error", { max: 1200, skipBlankLines: true, skipComments: true }],
},
},
]);
});
it("enables strict empty object type lint with named single-extends interfaces allowed", () => {

View File

@@ -6,7 +6,6 @@ function createExtensionCodexAppServerAttemptExtraVitestConfig(
) {
return createScopedVitestConfig(
[
"extensions/codex/src/app-server/run-attempt-client-prewarm.test.ts",
"extensions/codex/src/app-server/run-attempt-lifecycle-controller.test.ts",
"extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts",
"extensions/codex/src/app-server/run-attempt.context-engine.test.ts",