fix(coding-agents): write back only the new turns (append + idempotent retain) (#3336)

* feat(coding-agents): write back only the new turns (append + idempotent retain)

The live write-back re-uploaded the WHOLE conversation on every Stop, and every
N turns under the persistent-plugin runtime. A long session therefore re-sent its
entire transcript each time, which is what turns a large session into an
unretainable one rather than merely a slow one.

Retains now carry a per-session cursor. A session that has already been written
appends only the turns added since the last successful write, using the server's
`update_mode: "append"` (supported since #932); the server concatenates them onto
the stored document with "\n", which is exactly why the transcript is JSONL.

Append is only correct while our view of the document matches the server's, so
every uncertain case falls back to the full REPLACE this always did - no cursor,
a transcript that was rewritten rather than extended (compaction, a truncated
rollout), or a previous write whose outcome is unknown. Replace is idempotent by
construction and so is always the safe recovery.

Two supporting changes:

- Conversation retains carry a deterministic v5 `operation_id`, so a resubmitted
  write is collapsed into the original operation instead of being applied twice.
  That is what makes append safe: the client aborts at 15s, and a server that
  committed the write anyway would otherwise get the same turns again. The field
  landed in v0.8.6 (#2937/#2947) and is silently IGNORED by anything older, so
  the append path is gated on a cached GET /version probe and older servers keep
  replacing.
- `RetainOpts.async` is gone. Nothing ever passed `false`; retains are always
  async, and nothing in this plugin can afford to block a hook on extraction.

Backfill, git, knowledge and survey retains are untouched: they keep replacing,
and deliberately do not take a deterministic operation id, so re-retaining
identical content after a document is deleted still restores it.

* fix(coding-agents): serialise a session's write-backs so appends cannot overlap

Found reviewing the append cursor: the runtime fires retains without awaiting
them, and reading the cursor was not atomic with claiming it — the capability
probe awaits in between. Two overlapping write-backs therefore both planned an
append from the SAME position and submitted overlapping slices, duplicating turns
inside the document:

  replace(REF-ID + turns 0-4), append(turns 5-7), append(turns 5-8)

Serialising only the claim would not have fixed it either: that leaves an append
racing a replace on the wire, where the order they land in decides the outcome.
The whole read-plan-send-confirm cycle is now chained per session, so each
write-back plans against the previous one's CONFIRMED cursor.

The runtime's idle test now waits a tick before asserting on the fire-and-forget
retain, as its sibling assertions already did — one extra microtask hop.

* fix(coding-agents): key the write-back cursor to the bank it wrote to

The cursor is keyed by (harness, session id), but the bank is re-derived from
each hook event's cwd — so a session that moves between repos (#3133) keeps its
id and changes bank. The new bank holds no document for that session, and the
cursor still claimed a position in it:

  bank repo-a: replace(REF-ID + turns 0-4)
  bank repo-b: append(turns 5-7)      <- turns 0-4 never existed here

The cursor now records the bank it wrote to, and a mismatch replaces. Same
reasoning as the fingerprint and dirty checks: anything that makes our view of
the document unreliable falls back to the full write.

Also covers a config change (mapPathToBank, an explicit bankId) that re-points a
live session at a different bank.

* fix(coding-agents): re-sync the whole document every 20 appends

Review follow-up. Retains are async: the server acknowledges the submission and
extracts later, so a write can be confirmed to us and still fail afterwards — and
_resolve_retain_replay returns a prior operation whatever its status, so
resubmitting the same payload will not redo it. Replacing everything used to be
self-healing precisely because each write re-sent the whole document; appending
gives that up, and a single lost write would otherwise cost the rest of the
session.

A full write every MAX_APPENDS_BEFORE_RESYNC appends bounds that to the turns
since the last re-sync. A replace of any kind resets the count.

Also from the review:
- drop a session's chain entry once it settles, so a host that outlives many
  sessions (opencode runs for days) does not keep one resolved promise per
  session id forever
- pin the version test to MIN_IDEMPOTENT_RETAIN_VERSION rather than repeating
  the literal, which also gives the exported constant a consumer

Author: nicoloboschi

hindsight-integrations/coding-agents/src/core/chat.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it, vi } from "vitest";
 import type { HindsightClient } from "./hindsight";
 import { renderSessionJsonl, retainLiveSession, type TransportTurn } from "./chat";
+import { memoryCursorStore, type RetainCursorStore } from "./retain-cursor";
 
 describe("renderSessionJsonl", () => {
   const turns: TransportTurn[] = [
@@ -68,11 +69,219 @@ describe("retainLiveSession", () => {
     expect(documentId).toBe("conversation:s2");
     expect(tags).toEqual(["source:chat"]);
     expect(strategy).toBe("conversation");
-    expect(opts).toMatchObject({ async: true, timestamp: "2026-01-01T00:00:00Z" });
+    expect(opts).toMatchObject({ timestamp: "2026-01-01T00:00:00Z" });
     expect(opts.metadata).toMatchObject({
       source: "chat",
       session_id: "s2",
       ref_id: "conversation:s2",
     });
   });
 });
+
+describe("retainLiveSession — incremental write-back", () => {
+  const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
+  const turn = (i: number): TransportTurn => ({ role: "user", content: `turn ${i}` });
+  const turns = (n: number) => Array.from({ length: n }, (_, i) => turn(i));
+
+  /** Client double: `supported` is what GET /version would have told us about operation_id. */
+  const stubClient = (supported = true) => {
+    const retain = vi.fn().mockResolvedValue(undefined);
+    return {
+      retain,
+      client: {
+        retain,
+        bank: "coding-agent::repo",
+        supportsIdempotentRetain: async () => supported,
+      } as unknown as HindsightClient,
+    };
+  };
+
+  const write = (client: HindsightClient, turnList: TransportTurn[], cursors: RetainCursorStore) =>
+    retainLiveSession(client, "s1", turnList, "2026-01-01T00:00:00Z", "codex", { cursors });
+
+  it("replaces on the first write, then appends only the new turns", async () => {
+    const { retain, client } = stubClient();
+    const cursors = memoryCursorStore();
+
+    await write(client, turns(2), cursors);
+    const first = retain.mock.calls[0];
+    expect(first[5].updateMode).toBeUndefined();
+    expect(first[0]).toBe(renderSessionJsonl("conversation:s1", turns(2), "2026-01-01T00:00:00Z"));
+
+    await write(client, turns(5), cursors);
+    const second = retain.mock.calls[1];
+    expect(second[5].updateMode).toBe("append");
+    // Only the three new turns, and no REF-ID header: the document already carries one.
+    expect((second[0] as string).split("\n").map((l) => JSON.parse(l) as TransportTurn)).toEqual([
+      turn(2),
+      turn(3),
+      turn(4),
+    ]);
+    expect(second[2]).toBe("conversation:s1"); // same document id — append targets it
+  });
+
+  it("sends a stable v5 operation_id so a resubmitted write is not applied twice", async () => {
+    const a = stubClient();
+    const b = stubClient();
+    await write(a.client, turns(3), memoryCursorStore());
+    await write(b.client, turns(3), memoryCursorStore());
+    const opId = a.retain.mock.calls[0][5].operationId;
+    expect(opId).toMatch(UUID_RE);
+    expect(b.retain.mock.calls[0][5].operationId).toBe(opId);
+  });
+
+  it("gives a different operation_id to a different payload", async () => {
+    const { retain, client } = stubClient();
+    const cursors = memoryCursorStore();
+    await write(client, turns(2), cursors);
+    await write(client, turns(5), cursors);
+    expect(retain.mock.calls[0][5].operationId).not.toBe(retain.mock.calls[1][5].operationId);
+  });
+
+  it("skips the write entirely when no turn was added", async () => {
+    const { retain, client } = stubClient();
+    const cursors = memoryCursorStore();
+    await write(client, turns(3), cursors);
+    await write(client, turns(3), cursors);
+    expect(retain).toHaveBeenCalledTimes(1);
+  });
+
+  it("replaces the whole document after a failed write, instead of appending onto an unknown state", async () => {
+    const { retain, client } = stubClient();
+    const cursors = memoryCursorStore();
+    await write(client, turns(2), cursors);
+
+    retain.mockRejectedValueOnce(new Error("timeout"));
+    await expect(write(client, turns(4), cursors)).rejects.toThrow("timeout");
+    expect(cursors.read("s1")?.dirty).toBe(true);
+
+    await write(client, turns(6), cursors);
+    const recovery = retain.mock.calls[2];
+    expect(recovery[5].updateMode).toBeUndefined();
+    expect(recovery[0]).toBe(
+      renderSessionJsonl("conversation:s1", turns(6), "2026-01-01T00:00:00Z")
+    );
+    expect(cursors.read("s1")).toEqual({
+      turns: 6,
+      fingerprint: expect.any(String),
+      bank: "coding-agent::repo",
+      appends: 0, // the recovery replace re-established the document, restarting the re-sync count
+    });
+  });
+
+  it("never appends against a server that ignores operation_id", async () => {
+    const { retain, client } = stubClient(false);
+    const cursors = memoryCursorStore();
+    await write(client, turns(2), cursors);
+    await write(client, turns(5), cursors);
+    expect(retain.mock.calls.map((c) => c[5].updateMode)).toEqual([undefined, undefined]);
+    expect(retain.mock.calls[1][0]).toBe(
+      renderSessionJsonl("conversation:s1", turns(5), "2026-01-01T00:00:00Z")
+    );
+  });
+
+  it("replaces (never appends) when no cursor store is supplied", async () => {
+    const { retain, client } = stubClient();
+    await retainLiveSession(client, "s1", turns(2), "2026-01-01T00:00:00Z", "codex");
+    await retainLiveSession(client, "s1", turns(5), "2026-01-01T00:00:00Z", "codex");
+    expect(retain.mock.calls.map((c) => c[5].updateMode)).toEqual([undefined, undefined]);
+  });
+
+  it("serialises overlapping write-backs so neither appends a slice the other already sent", async () => {
+    // The runtime fires retains without awaiting them: a turn-driven one and an idle-driven one can
+    // overlap. Unserialised, both planned an append from the same cursor position and submitted
+    // overlapping slices, duplicating turns inside the document.
+    const submitted: { mode: string; turns: number }[] = [];
+    const gates: (() => void)[] = [];
+    const retain = vi.fn((content: string, ...rest: unknown[]) => {
+      const o = rest[4] as { updateMode?: string };
+      submitted.push({ mode: o.updateMode ?? "replace", turns: content.split("\n").length });
+      return new Promise<void>((resolve) => gates.push(resolve));
+    });
+    const client = {
+      retain,
+      bank: "b",
+      supportsIdempotentRetain: async () => true,
+    } as unknown as HindsightClient;
+    const cursors = memoryCursorStore();
+
+    const first = write(client, turns(5), cursors);
+    await vi.waitFor(() => expect(gates).toHaveLength(1));
+    gates[0]();
+    await first;
+
+    const a = write(client, turns(8), cursors);
+    const b = write(client, turns(9), cursors);
+    // Only ONE request is in flight: the second write-back waits for the first to be confirmed.
+    await vi.waitFor(() => expect(gates).toHaveLength(2));
+    expect(submitted).toHaveLength(2);
+    gates[1]();
+    await a;
+    await vi.waitFor(() => expect(gates).toHaveLength(3));
+    gates[2]();
+    await b;
+
+    // 6 = REF-ID + 5 turns, then turns 5-7, then turn 8 alone — every turn sent exactly once.
+    expect(submitted).toEqual([
+      { mode: "replace", turns: 6 },
+      { mode: "append", turns: 3 },
+      { mode: "append", turns: 1 },
+    ]);
+  });
+
+  it("keeps two sessions in the same directory independent", async () => {
+    // Same repo => same bank, but each session owns its own document and its own cursor, so two
+    // agents running side by side in one checkout never append into each other's conversation.
+    const { retain, client } = stubClient();
+    const cursors = memoryCursorStore();
+    const writeAs = (id: string, list: TransportTurn[]) =>
+      retainLiveSession(client, id, list, "2026-01-01T00:00:00Z", "codex", { cursors });
+
+    await writeAs("sess-a", turns(2));
+    await writeAs("sess-b", turns(4));
+    await writeAs("sess-a", turns(3));
+
+    const docs = retain.mock.calls.map((c) => c[2]);
+    expect(docs).toEqual(["conversation:sess-a", "conversation:sess-b", "conversation:sess-a"]);
+    // sess-b's longer transcript did not advance sess-a's cursor: its append is turn 2 alone.
+    expect(retain.mock.calls[1][5].updateMode).toBeUndefined(); // first write for sess-b
+    expect(retain.mock.calls[2][5].updateMode).toBe("append");
+    expect((retain.mock.calls[2][0] as string).split("\n")).toHaveLength(1);
+    expect(cursors.read("sess-a")?.turns).toBe(3);
+    expect(cursors.read("sess-b")?.turns).toBe(4);
+  });
+
+  it("replaces rather than appends when the session moved to another bank", async () => {
+    const sent: { bank: string; mode: string }[] = [];
+    const mk = (bank: string) =>
+      ({
+        bank,
+        supportsIdempotentRetain: async () => true,
+        retain: vi.fn(async (_c: string, ...rest: unknown[]) => {
+          sent.push({ bank, mode: (rest[4] as { updateMode?: string }).updateMode ?? "replace" });
+        }),
+      }) as unknown as HindsightClient;
+    const cursors = memoryCursorStore();
+
+    await write(mk("repo-a"), turns(5), cursors);
+    await write(mk("repo-b"), turns(8), cursors); // user cd'd into another repo mid-session
+    expect(sent).toEqual([
+      { bank: "repo-a", mode: "replace" },
+      { bank: "repo-b", mode: "replace" },
+    ]);
+  });
+
+  it("keeps the write-back when the capability probe itself fails", async () => {
+    const retain = vi.fn().mockResolvedValue(undefined);
+    const client = {
+      retain,
+      bank: "b",
+      supportsIdempotentRetain: async () => {
+        throw new Error("unreachable");
+      },
+    } as unknown as HindsightClient;
+    await write(client, turns(2), memoryCursorStore());
+    expect(retain).toHaveBeenCalledTimes(1);
+    expect(retain.mock.calls[0][5].updateMode).toBeUndefined();
+  });
+});

hindsight-integrations/coding-agents/src/core/chat.ts

@@ -4,7 +4,9 @@
  * the REF-ID tracer; every turn gets an ABSOLUTE timestamp.
  */
 import type { HindsightClient } from "./hindsight";
+import { fingerprintTurns, planRetain, type RetainCursorStore } from "./retain-cursor";
 import type { ChatSession } from "./types";
+import { uuidV5 } from "./uuid";
 import { pool } from "./util";
 
 export interface TransportTurn {
@@ -86,32 +88,125 @@ export async function ingestChats(
   return failures;
 }
 
+/** Capability probe for the append path. Any failure answers "no": a write-back must never be lost
+ *  because we couldn't work out whether the cheaper form of it was available. */
+async function supportsAppend(client: HindsightClient): Promise<boolean> {
+  try {
+    return await client.supportsIdempotentRetain();
+  } catch {
+    return false;
+  }
+}
+
+/**
+ * One write-back at a time per session, keyed by the store the cursors live in.
+ *
+ * The persistent-plugin runtime fires retains without awaiting them (a turn-driven one and an
+ * idle-driven one can overlap), and reading the cursor is not atomic with claiming it — there is an
+ * await in between. Two overlapping calls therefore both planned an append from the SAME position
+ * and submitted overlapping slices, duplicating turns inside the document; and even serialising the
+ * claim alone would leave an append racing a replace on the wire, where the order they land in
+ * decides whether the result is correct. Chaining the whole read-plan-send-confirm cycle is what
+ * makes the cursor mean what it says: the second call sees the first call's CONFIRMED position.
+ */
+const writeBacks = new WeakMap<RetainCursorStore, Map<string, Promise<void>>>();
+
+function serialize(
+  cursors: RetainCursorStore,
+  sessionId: string,
+  write: () => Promise<void>
+): Promise<void> {
+  const perSession = writeBacks.get(cursors) ?? new Map<string, Promise<void>>();
+  writeBacks.set(cursors, perSession);
+  // Chain off the previous write-back whether it succeeded or failed — a failure leaves the cursor
+  // dirty, which the next call needs to see so it can replace rather than append.
+  const next = (perSession.get(sessionId) ?? Promise.resolve()).then(write, write);
+  const tail = next.catch(() => {});
+  perSession.set(sessionId, tail);
+  // Drop the entry once it is settled AND still the tail, so a host that outlives many sessions
+  // (opencode runs for days) doesn't accumulate one resolved promise per session id forever.
+  void tail.then(() => {
+    if (perSession.get(sessionId) === tail) perSession.delete(sessionId);
+  });
+  return next;
+}
+
 /**
- * Live write-back: upsert a running session under a stable document_id. Same id => Hindsight
- * reprocesses the FULL conversation, so the settled decision is extracted from the whole thing.
+ * Live write-back: upsert a running session under a stable document_id, sending only what is new.
+ *
+ * Given a cursor store, a session that has already been written APPENDS the turns added since the
+ * last successful write; the server concatenates them onto the stored document (with "\n", which is
+ * why the transcript is JSONL) and re-chunks. Without one — first write, a rewritten transcript, an
+ * unconfirmed previous write, or a server too old to be idempotent — it falls back to REPLACING the
+ * whole document, which is what this always used to do. See core/retain-cursor.ts for why every
+ * uncertain case resolves that way.
  *
  * Uses the same `conversation` strategy as backfilled chats — one strategy for all developer
  * conversations; the mission scales extraction to the substance. The content is a JSON transcript
- * (renderSessionJson) whose tool activity is compacted into `role:"action"` turns
+ * (renderSessionJsonl) whose tool activity is compacted into `role:"action"` turns
  * (see core/transcript*.ts).
  */
 export async function retainLiveSession(
   client: HindsightClient,
   sessionId: string,
   turns: TransportTurn[],
   startTs: string,
-  harness?: string
+  harness?: string,
+  opts: { cursors?: RetainCursorStore } = {}
+): Promise<void> {
+  const cursors = opts.cursors;
+  if (!cursors) return writeSession(client, sessionId, turns, startTs, harness);
+  // Serialised so the plan is made against the previous write-back's CONFIRMED cursor (see above).
+  return serialize(cursors, sessionId, () =>
+    writeSession(client, sessionId, turns, startTs, harness, cursors)
+  );
+}
+
+async function writeSession(
+  client: HindsightClient,
+  sessionId: string,
+  turns: TransportTurn[],
+  startTs: string,
+  harness?: string,
+  cursors?: RetainCursorStore
 ): Promise<void> {
   const refId = `conversation:${sessionId}`;
+  const appendSupported = Boolean(cursors) && (await supportsAppend(client));
+  const prior = cursors?.read(sessionId);
+  const plan = planRetain(turns, prior, { appendSupported, bank: client.bank });
+  if (plan.mode === "skip") return;
+
+  const content =
+    plan.mode === "append"
+      ? turns
+          .slice(plan.fromTurn)
+          .map((t) => JSON.stringify(t))
+          .join("\n")
+      : renderSessionJsonl(refId, turns, startTs);
+
+  // Claim the new position BEFORE the write and mark it unconfirmed, so a client that times out on
+  // a request the server did commit replaces next time instead of appending the same turns twice.
+  const next = {
+    turns: turns.length,
+    fingerprint: fingerprintTurns(turns, turns.length),
+    bank: client.bank,
+    // A full write re-establishes the document, so the re-sync countdown starts over.
+    appends: plan.mode === "append" ? (prior?.appends ?? 0) + 1 : 0,
+  };
+  cursors?.write(sessionId, { ...next, dirty: true });
+
   await client.retain(
-    renderSessionJsonl(refId, turns, startTs),
+    content,
     "coding agent session",
     refId,
     ["source:chat", ...(harness ? [`harness:${harness}`] : [])],
     "conversation",
     {
       timestamp: startTs,
-      async: true,
+      updateMode: plan.mode === "append" ? "append" : undefined,
+      // Identity of THIS payload: a resubmission of the same bytes is collapsed server-side into
+      // the original operation instead of extracting (or appending) twice.
+      operationId: uuidV5(`${client.bank}\n${refId}\n${plan.mode}\n${content}`),
       metadata: {
         source: "chat",
         session_id: sessionId,
@@ -120,4 +215,5 @@ export async function retainLiveSession(
       },
     }
   );
+  cursors?.write(sessionId, next);
 }

hindsight-integrations/coding-agents/src/core/git.ts

@@ -193,8 +193,7 @@ export async function ingestGitLog(
       `git commit-message history (last ${n}) for ${repoName}`,
       `gitlog:${repoName}`,
       ["source:git", "source:git-log", ...(head ? [`gitlog-head:${head}`] : [])],
-      "gitlog",
-      { async: true }
+      "gitlog"
     );
     log(`[gitlog] done: ${n} commit messages ingested as 1 document under strategy 'gitlog'`);
     return 0;