refactor(sync): split transcript cursor decisions

Author: kushagrchitkar

docs/intentional-architecture-rewrite-2026-06-27/worklog.md

@@ -592,3 +592,9 @@ Eighty-sixth production slice:
 - Split Codex app-server startup into `src/harness/providers/codex/app-server-session.ts`.
 - Moved `initialize`, `thread/start`, `turn/start`, root thread/turn state assignment, provider-session event emission, and prompt/sandbox request construction out of `app-server.ts`.
 - Kept `app-server.ts` focused on process lifetime, JSON-RPC transport, request dispatch, notification handling, timeout failure, and cleanup.
+
+Eighty-seventh production slice:
+
+- Split sync transcript snapshot and cursor decision logic into `src/sync/transcript-cursor.ts`.
+- Moved transcript file reading, line counting, unchanged/pending/prefix-mismatch decisions, pending ledger entry construction, and failed-entry construction out of `src/sync/sweep.ts`.
+- Kept `src/sync/sweep.ts` focused on candidate iteration, eligibility gates, repo locks, ledger loading/reconciliation, Absorb enqueue orchestration, summary recording, and lock cleanup.

src/sync/sweep.ts

@@ -1,15 +1,19 @@
-import { readFile } from "node:fs/promises";
-
 import type { SessionCandidate, SweepApp } from "./discovery/index.js";
 import {
+  type LedgerEntry,
   type SyncLedger,
-  countLines,
-  syncCursor,
   freshLedgerEntry,
   ledgerKey,
   reconcileLedger,
-  sha256,
 } from "./ledger.js";
+import {
+  type SyncCursorDecision,
+  type TranscriptSnapshot,
+  evaluateSyncCursor,
+  failedLedgerEntry,
+  pendingLedgerEntry,
+  readTranscriptSnapshot,
+} from "./transcript-cursor.js";
 import {
   loadLedgerForRepo,
   writeLedger,
@@ -68,21 +72,6 @@ export type StartSyncAbsorbFn = (
   args: StartSyncAbsorbArgs,
 ) => Promise<StartSyncAbsorbResult>;
 
-interface TranscriptSnapshot {
-  content: Buffer;
-  currentSize: number;
-  currentLine: number;
-}
-
-type TranscriptReadResult =
-  | { ok: true; snapshot: TranscriptSnapshot }
-  | { ok: false; reason: string };
-
-type SyncCursorDecision =
-  | { kind: "skip"; reason: string }
-  | { kind: "needs_attention"; reason: string; entry: ReturnType<typeof markPrefixMismatch> }
-  | { kind: "ready"; fromLine: number; toLine: number };
-
 export async function executeSyncSweep(args: {
   candidates: SessionCandidate[];
   syncSince: Date | null;
@@ -225,73 +214,16 @@ function candidateEligibility(
   return null;
 }
 
-async function readTranscriptSnapshot(
-  candidate: SessionCandidate,
-): Promise<TranscriptReadResult> {
-  try {
-    const content = await readFile(candidate.transcriptPath);
-    return {
-      ok: true,
-      snapshot: {
-        content,
-        currentSize: content.length,
-        currentLine: countLines(content.toString("utf8")),
-      },
-    };
-  } catch (err: unknown) {
-    const reason = `read-failed: ${err instanceof Error ? err.message : String(err)}`;
-    return { ok: false, reason };
-  }
-}
-
-function evaluateSyncCursor(
-  entry: SyncLedger["sessions"][string],
-  snapshot: TranscriptSnapshot,
-): SyncCursorDecision {
-  if (entry.status === "pending") {
-    return { kind: "skip", reason: "sync-already-pending" };
-  }
-
-  if (snapshot.currentSize <= entry.lastAbsorbedSize) {
-    return { kind: "skip", reason: "unchanged" };
-  }
-
-  const prefixHash = sha256(snapshot.content.subarray(0, entry.lastAbsorbedSize));
-  if (prefixHash !== entry.lastAbsorbedPrefixHash) {
-    return {
-      kind: "needs_attention",
-      reason: "prefix-mismatch",
-      entry: markPrefixMismatch(entry),
-    };
-  }
-
-  return {
-    kind: "ready",
-    fromLine: entry.lastAbsorbedLine + 1,
-    toLine: snapshot.currentLine,
-  };
-}
-
-function markPrefixMismatch(
-  entry: SyncLedger["sessions"][string],
-): SyncLedger["sessions"][string] {
-  return {
-    ...entry,
-    status: "needs_attention",
-    lastError: "transcript prefix no longer matches ledger cursor",
-  };
-}
-
 async function enqueueAbsorb(args: {
   candidate: SessionCandidate;
-  entry: SyncLedger["sessions"][string];
+  entry: LedgerEntry;
   decision: Extract<SyncCursorDecision, { kind: "ready" }>;
   snapshot: TranscriptSnapshot;
   now: Date;
   startAbsorb: StartSyncAbsorbFn;
 }): Promise<
-  | { ok: true; jobId: string; entry: SyncLedger["sessions"][string] }
-  | { ok: false; reason: string; entry: SyncLedger["sessions"][string] }
+  | { ok: true; jobId: string; entry: LedgerEntry }
+  | { ok: false; reason: string; entry: LedgerEntry }
 > {
   const result = await args.startAbsorb({
     candidate: args.candidate,
@@ -306,28 +238,18 @@ async function enqueueAbsorb(args: {
     return {
       ok: false,
       reason: "absorb-start-failed",
-      entry: {
-        ...args.entry,
-        status: "failed",
-        lastError: result.error,
-      },
+      entry: failedLedgerEntry(args.entry, result.error),
     };
   }
-  const pendingCursor = syncCursor(args.snapshot.content, args.snapshot.currentLine);
   return {
     ok: true,
     jobId: result.jobId,
-    entry: {
-      ...args.entry,
-      status: "pending",
-      pendingToSize: pendingCursor.size,
-      pendingToLine: pendingCursor.line,
-      pendingPrefixHash: pendingCursor.prefixHash,
-      pendingJobId: result.jobId,
-      pendingStartedAt: args.now.toISOString(),
-      lastJobId: result.jobId,
-      lastError: undefined,
-    },
+    entry: pendingLedgerEntry({
+      entry: args.entry,
+      snapshot: args.snapshot,
+      jobId: result.jobId,
+      now: args.now,
+    }),
   };
 }
 

src/sync/transcript-cursor.ts

@@ -0,0 +1,107 @@
+import { readFile } from "node:fs/promises";
+
+import type { SessionCandidate } from "./discovery/index.js";
+import {
+  type LedgerEntry,
+  countLines,
+  sha256,
+  syncCursor,
+} from "./ledger.js";
+
+export interface TranscriptSnapshot {
+  content: Buffer;
+  currentSize: number;
+  currentLine: number;
+}
+
+export type TranscriptReadResult =
+  | { ok: true; snapshot: TranscriptSnapshot }
+  | { ok: false; reason: string };
+
+export type SyncCursorDecision =
+  | { kind: "skip"; reason: string }
+  | { kind: "needs_attention"; reason: string; entry: LedgerEntry }
+  | { kind: "ready"; fromLine: number; toLine: number };
+
+export async function readTranscriptSnapshot(
+  candidate: SessionCandidate,
+): Promise<TranscriptReadResult> {
+  try {
+    const content = await readFile(candidate.transcriptPath);
+    return {
+      ok: true,
+      snapshot: {
+        content,
+        currentSize: content.length,
+        currentLine: countLines(content.toString("utf8")),
+      },
+    };
+  } catch (err: unknown) {
+    const reason = `read-failed: ${err instanceof Error ? err.message : String(err)}`;
+    return { ok: false, reason };
+  }
+}
+
+export function evaluateSyncCursor(
+  entry: LedgerEntry,
+  snapshot: TranscriptSnapshot,
+): SyncCursorDecision {
+  if (entry.status === "pending") {
+    return { kind: "skip", reason: "sync-already-pending" };
+  }
+
+  if (snapshot.currentSize <= entry.lastAbsorbedSize) {
+    return { kind: "skip", reason: "unchanged" };
+  }
+
+  const prefixHash = sha256(snapshot.content.subarray(0, entry.lastAbsorbedSize));
+  if (prefixHash !== entry.lastAbsorbedPrefixHash) {
+    return {
+      kind: "needs_attention",
+      reason: "prefix-mismatch",
+      entry: markPrefixMismatch(entry),
+    };
+  }
+
+  return {
+    kind: "ready",
+    fromLine: entry.lastAbsorbedLine + 1,
+    toLine: snapshot.currentLine,
+  };
+}
+
+export function pendingLedgerEntry(args: {
+  entry: LedgerEntry;
+  snapshot: TranscriptSnapshot;
+  jobId: string;
+  now: Date;
+}): LedgerEntry {
+  const pendingCursor = syncCursor(args.snapshot.content, args.snapshot.currentLine);
+  return {
+    ...args.entry,
+    status: "pending",
+    pendingToSize: pendingCursor.size,
+    pendingToLine: pendingCursor.line,
+    pendingPrefixHash: pendingCursor.prefixHash,
+    pendingJobId: args.jobId,
+    pendingStartedAt: args.now.toISOString(),
+    lastJobId: args.jobId,
+    lastError: undefined,
+  };
+}
+
+export function failedLedgerEntry(entry: LedgerEntry, error: string): LedgerEntry {
+  return {
+    ...entry,
+    status: "failed",
+    lastError: error,
+  };
+}
+
+function markPrefixMismatch(entry: LedgerEntry): LedgerEntry {
+  return {
+    ...entry,
+    status: "needs_attention",
+    lastError: "transcript prefix no longer matches ledger cursor",
+  };
+}

test/architecture-boundaries.test.ts

@@ -370,6 +370,22 @@ describe("architecture boundaries", () => {
     expect(syncSweep).not.toContain("sync.lock");
   });
 
+  it("keeps sync transcript cursor decisions out of the sweep coordinator", async () => {
+    const syncSweep = await readSource("src/sync/sweep.ts");
+    const transcriptCursor = await readSource("src/sync/transcript-cursor.ts");
+
+    expect(existsSync(join(ROOT, "src/sync/transcript-cursor.ts"))).toBe(true);
+    expect(syncSweep).toContain("transcript-cursor.js");
+    expect(syncSweep).not.toContain("from \"node:fs/promises\"");
+    expect(syncSweep).not.toContain("function readTranscriptSnapshot");
+    expect(syncSweep).not.toContain("function evaluateSyncCursor");
+    expect(syncSweep).not.toContain("lastAbsorbedPrefixHash");
+    expect(syncSweep).not.toContain("pendingPrefixHash");
+    expect(transcriptCursor).toContain("export async function readTranscriptSnapshot");
+    expect(transcriptCursor).toContain("export function evaluateSyncCursor");
+    expect(transcriptCursor).toContain("export function pendingLedgerEntry");
+  });
+
   it("keeps local process signaling in the platform layer", async () => {
     const jobsService = await readSource("src/services/jobs/jobs.ts");
     const viewerJobs = await readSource("src/viewer/jobs.ts");