refactor(gateway): route chat transcript injection through the session accessor (#101688)

jalehman · web-flow · commit 453f5968bbca · 2026-07-07T09:10:30.000-07:00
* refactor(gateway): route chat transcript injection through the session accessor

* fix(gateway): avoid map-spread in source reply mirror rewrite results

* perf(sessions): find injected transcript duplicates with a reverse early-exit scan
diff --git a/scripts/check-session-accessor-boundary.mjs b/scripts/check-session-accessor-boundary.mjs
@@ -193,6 +193,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
   "src/config/sessions/cleanup-service.ts",
   "src/config/sessions/goals.ts",
   "src/gateway/boot.ts",
+  "src/gateway/server-methods/chat.ts",
   "src/gateway/server-methods/sessions.ts",
   "src/gateway/server-node-events.ts",
   "src/gateway/session-compaction-checkpoints.ts",
diff --git a/scripts/lib/session-accessor-debt-baseline.json b/scripts/lib/session-accessor-debt-baseline.json
@@ -58,7 +58,6 @@
     "src/config/sessions/session-file.ts": 2,
     "src/config/sessions/session-registry-maintenance.ts": 2,
     "src/gateway/server-methods/agent.ts": 2,
-    "src/gateway/server-methods/chat.ts": 2,
     "src/gateway/session-lifecycle-state.ts": 2,
     "src/gateway/test-helpers.mocks.ts": 1,
     "src/infra/heartbeat-runner.ts": 5,
diff --git a/src/config/sessions/session-accessor.test.ts b/src/config/sessions/session-accessor.test.ts
@@ -18,9 +18,11 @@ import {
   cleanupSessionLifecycleArtifacts,
   commitReplySessionInitialization,
   createSessionEntryWithTranscript,
+  findTranscriptEvent,
   listSessionEntries,
   loadReplySessionInitializationSnapshot,
   loadSessionEntry,
+  loadTranscriptEvents,
   markSessionAbortTarget,
   openSessionEntryReadView,
   patchSessionEntry,
@@ -101,6 +103,68 @@ describe("session accessor file-backed seam", () => {
     });
   });
 
+  it("loads parsed transcript events from explicit and store-derived targets", async () => {
+    const header = { type: "session", id: "session-events", timestamp: 1 };
+    const message = { type: "message", id: "m1", message: { role: "assistant" } };
+    fs.writeFileSync(
+      transcriptPath,
+      `${JSON.stringify(header)}\n${JSON.stringify(message)}\nnot-json\n`,
+      "utf-8",
+    );
+    await upsertSessionEntry(
+      { sessionKey: "agent:main:main", storePath },
+      { sessionId: "session-events", sessionFile: transcriptPath, updatedAt: 10 },
+    );
+
+    const explicit = await loadTranscriptEvents({
+      sessionFile: transcriptPath,
+      sessionId: "session-events",
+    });
+    expect(explicit).toEqual([header, message]);
+
+    const derived = await loadTranscriptEvents({
+      sessionId: "session-events",
+      sessionKey: "agent:main:main",
+      storePath,
+    });
+    expect(derived).toEqual([header, message]);
+
+    const missing = await loadTranscriptEvents({
+      sessionFile: path.join(tempDir, "missing.jsonl"),
+      sessionId: "session-events",
+    });
+    expect(missing).toEqual([]);
+  });
+
+  it("finds the newest matching transcript event without loading the whole file", async () => {
+    const header = { type: "session", id: "session-find", timestamp: 1 };
+    const older = { type: "message", id: "m1", message: { role: "assistant", tag: "old" } };
+    const newer = { type: "message", id: "m2", message: { role: "assistant", tag: "new" } };
+    fs.writeFileSync(
+      transcriptPath,
+      `${JSON.stringify(header)}\n${JSON.stringify(older)}\n${JSON.stringify(newer)}\n`,
+      "utf-8",
+    );
+
+    const seen: unknown[] = [];
+    const found = await findTranscriptEvent(
+      { sessionFile: transcriptPath, sessionId: "session-find" },
+      (event) => {
+        seen.push(event);
+        return (event as { type?: string }).type === "message";
+      },
+    );
+    // Newest-first with early exit: the older message is never visited.
+    expect(found).toEqual({ event: newer });
+    expect(seen).toEqual([newer]);
+
+    const missing = await findTranscriptEvent(
+      { sessionFile: path.join(tempDir, "missing.jsonl"), sessionId: "session-find" },
+      () => true,
+    );
+    expect(missing).toBeUndefined();
+  });
+
   it("opens a borrowed read view with raw exact-key probes and deferred enumeration", async () => {
     const mixedKey = "agent:main:matrix:channel:!RoomAbC:example.org";
     await upsertSessionEntry(
diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts
@@ -96,7 +96,10 @@ import { resolveSessionTranscriptFile } from "./transcript-file-resolve.js";
 import { createSessionTranscriptHeader } from "./transcript-header.js";
 import { serializeJsonlLine, writeJsonlLines } from "./transcript-jsonl.js";
 import { replayRecentUserAssistantMessages } from "./transcript-replay.js";
-import { streamSessionTranscriptLines } from "./transcript-stream.js";
+import {
+  streamSessionTranscriptLines,
+  streamSessionTranscriptLinesReverse,
+} from "./transcript-stream.js";
 import {
   scanSessionTranscriptTree,
   selectSessionTranscriptTreePathNodes,
@@ -2089,6 +2092,48 @@ export async function appendTranscriptMessage<TMessage>(
   });
 }
 
+/**
+ * Finds the newest transcript record accepted by the matcher. Reads newest-first
+ * with early exit so hot append-path lookups never materialize the whole
+ * transcript; missing transcripts match nothing. The match is wrapped so parsed
+ * falsy records stay distinguishable from "no match".
+ */
+export async function findTranscriptEvent(
+  scope: SessionTranscriptReadScope,
+  match: (event: TranscriptEvent) => boolean,
+): Promise<{ event: TranscriptEvent } | undefined> {
+  const target = resolveSessionTranscriptReadTarget(scope);
+  for await (const line of streamSessionTranscriptLinesReverse(target.sessionFile)) {
+    try {
+      const event = JSON.parse(line) as TranscriptEvent;
+      if (match(event)) {
+        return { event };
+      }
+    } catch {
+      // Malformed lines are skipped, matching transcript index tolerance.
+    }
+  }
+  return undefined;
+}
+
+/** Reads parsed transcript records from an explicit or derived transcript target. */
+export async function loadTranscriptEvents(
+  scope: SessionTranscriptReadScope,
+): Promise<TranscriptEvent[]> {
+  const target = resolveSessionTranscriptReadTarget(scope);
+  const events: TranscriptEvent[] = [];
+  // Missing transcripts stream zero lines, so readers get an empty event list
+  // instead of a filesystem error; that keeps the read contract storage-neutral.
+  for await (const line of streamSessionTranscriptLines(target.sessionFile)) {
+    try {
+      events.push(JSON.parse(line) as TranscriptEvent);
+    } catch {
+      // Malformed lines are skipped, matching transcript index tolerance.
+    }
+  }
+  return events;
+}
+
 /** Emits a transcript update after resolving the current transcript target. */
 export async function publishTranscriptUpdate(
   scope: SessionTranscriptWriteScope,
diff --git a/src/gateway/server-methods/chat-transcript-inject.ts b/src/gateway/server-methods/chat-transcript-inject.ts
@@ -1,7 +1,11 @@
 // Chat transcript injection appends gateway-authored assistant rows while
 // preserving agent-session parent links and transcript update notifications.
 import type { SessionManager } from "../../agents/sessions/session-manager.js";
-import { persistSessionTranscriptTurn } from "../../config/sessions/session-accessor.js";
+import {
+  findTranscriptEvent,
+  persistSessionTranscriptTurn,
+  type TranscriptEvent,
+} from "../../config/sessions/session-accessor.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { formatErrorMessage } from "../../infra/errors.js";
 
@@ -53,9 +57,71 @@ function resolveInjectedAssistantContent(params: {
   return [{ type: "text", text: `${labelPrefix}${params.message}` }];
 }
 
+function transcriptEventRecord(event: TranscriptEvent): Record<string, unknown> | undefined {
+  return event && typeof event === "object" && !Array.isArray(event)
+    ? (event as Record<string, unknown>)
+    : undefined;
+}
+
+function transcriptEventMessage(event: TranscriptEvent): Record<string, unknown> | undefined {
+  const message = transcriptEventRecord(event)?.message;
+  return message && typeof message === "object" && !Array.isArray(message)
+    ? (message as Record<string, unknown>)
+    : undefined;
+}
+
+function transcriptEventId(event: TranscriptEvent): string | undefined {
+  const id = transcriptEventRecord(event)?.id;
+  return typeof id === "string" && id.trim().length > 0 ? id : undefined;
+}
+
+type InjectedAssistantIdempotencyTarget = {
+  agentId?: string;
+  sessionFile?: string;
+  sessionId?: string;
+  sessionKey?: string;
+  storePath?: string;
+};
+
+async function findInjectedAssistantMessageByIdempotencyKey(params: {
+  idempotencyKey: string;
+  target: InjectedAssistantIdempotencyTarget;
+}): Promise<{ messageId: string; message: Record<string, unknown> } | undefined> {
+  if (!params.target.sessionId || !params.target.sessionKey) {
+    return undefined;
+  }
+  // The in-lock duplicate check resolves through the already-resolved
+  // sessionFile when present so the lookup reads the file being appended to.
+  // findTranscriptEvent scans newest-first with early exit, keeping the hot
+  // idempotent append path from materializing the whole transcript.
+  const found = await findTranscriptEvent(
+    {
+      ...(params.target.agentId ? { agentId: params.target.agentId } : {}),
+      ...(params.target.sessionFile ? { sessionFile: params.target.sessionFile } : {}),
+      sessionId: params.target.sessionId,
+      sessionKey: params.target.sessionKey,
+      ...(params.target.storePath ? { storePath: params.target.storePath } : {}),
+    },
+    (candidate) => {
+      const message = transcriptEventMessage(candidate);
+      return message?.role === "assistant" && message.idempotencyKey === params.idempotencyKey;
+    },
+  );
+  const message = found ? transcriptEventMessage(found.event) : undefined;
+  if (!message) {
+    return undefined;
+  }
+  // Legacy shipped transcripts can carry assistant rows without top-level ids;
+  // fall back to the idempotency key so re-issued aborts still dedupe there.
+  const messageId = (found ? transcriptEventId(found.event) : undefined) ?? params.idempotencyKey;
+  return { messageId, message };
+}
+
 /** Append a gateway-authored assistant message while preserving transcript parent links. */
 export async function appendInjectedAssistantMessageToTranscript(params: {
-  transcriptPath: string;
+  transcriptPath?: string;
+  storePath?: string;
+  sessionId?: string;
   sessionKey?: string;
   agentId?: string;
   message: string;
@@ -118,26 +184,60 @@ export async function appendInjectedAssistantMessageToTranscript(params: {
   };
 
   try {
+    if (!params.transcriptPath && (!params.storePath || !params.sessionId || !params.sessionKey)) {
+      return { ok: false, error: "transcript identity not resolved" };
+    }
+    const assistantScopedIdempotency =
+      params.idempotencyKey && params.storePath && params.sessionId && params.sessionKey
+        ? {
+            idempotencyKey: params.idempotencyKey,
+            target: {
+              ...(params.agentId ? { agentId: params.agentId } : {}),
+              sessionId: params.sessionId,
+              sessionKey: params.sessionKey,
+              storePath: params.storePath,
+            },
+          }
+        : undefined;
     const turn = await persistSessionTranscriptTurn(
       {
-        sessionFile: params.transcriptPath,
         sessionKey: params.sessionKey ?? "",
+        ...(params.transcriptPath ? { sessionFile: params.transcriptPath } : {}),
+        ...(params.storePath ? { storePath: params.storePath } : {}),
+        ...(params.sessionId ? { sessionId: params.sessionId } : {}),
         ...(params.agentId ? { agentId: params.agentId } : {}),
       },
       {
         updateMode: "inline",
+        touchSessionEntry: Boolean(params.storePath && params.sessionId && params.sessionKey),
         ...(params.config ? { config: params.config } : {}),
         messages: [
           {
             message: messageBody,
+            idempotencyLookup: assistantScopedIdempotency ? "caller-checked" : "scan",
             now,
             useRawWhenLinear: true,
+            shouldAppend: assistantScopedIdempotency
+              ? async (target) =>
+                  !(await findInjectedAssistantMessageByIdempotencyKey({
+                    idempotencyKey: assistantScopedIdempotency.idempotencyKey,
+                    target,
+                  }))
+              : undefined,
           },
         ],
       },
     );
     const appended = turn.messages[0];
     if (!appended) {
+      if (assistantScopedIdempotency) {
+        const existing = await findInjectedAssistantMessageByIdempotencyKey(
+          assistantScopedIdempotency,
+        );
+        if (existing) {
+          return { ok: true, messageId: existing.messageId, message: existing.message };
+        }
+      }
       return { ok: false, error: "gateway-injected assistant message was not appended" };
     }
     return {
diff --git a/src/gateway/server-methods/chat.abort-persistence.test.ts b/src/gateway/server-methods/chat.abort-persistence.test.ts
@@ -6,6 +6,7 @@ import path from "node:path";
 import { CURRENT_SESSION_VERSION } from "openclaw/plugin-sdk/agent-sessions";
 import { afterAll, afterEach, describe, expect, it, vi } from "vitest";
 import { cleanupTempDirs, makeTempDir } from "../../../test/helpers/temp-dir.js";
+import { replaceSessionEntry } from "../../config/sessions/session-accessor.js";
 import { onAgentEvent, resetAgentEventsForTest } from "../../infra/agent-events.js";
 import {
   createActiveRun,
@@ -171,6 +172,12 @@ async function createTranscriptFixture(prefix: string) {
   const sessionId = "sess-main";
   const transcriptPath = path.join(dir, `${sessionId}.jsonl`);
   await writeTranscriptHeader(transcriptPath, sessionId);
+  // The accessor resolves transcript targets from the persisted store, so the
+  // fixture seeds a real entry instead of relying on the mocked gateway wrapper.
+  await replaceSessionEntry(
+    { agentId: "main", sessionKey: "main", storePath: path.join(dir, "sessions.json") },
+    { sessionId, sessionFile: transcriptPath, updatedAt: Date.now() },
+  );
   setMockSessionEntry(transcriptPath, sessionId);
   return { transcriptPath, sessionId };
 }
diff --git a/src/gateway/server-methods/chat.directive-tags.test.ts b/src/gateway/server-methods/chat.directive-tags.test.ts
diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts
diff --git a/test/scripts/check-session-accessor-boundary.test.ts b/test/scripts/check-session-accessor-boundary.test.ts
