refactor(sync): split sweep candidate ownership

Author: kushagrchitkar

docs/intentional-architecture-rewrite-2026-06-27/decision-log.md

@@ -131,6 +131,8 @@ Platform transcript discovery returns `DiscoveredTranscript` facts: app, session
 
 The top-level sync workflow file coordinates the sync verb; it should not be a bucket for all sync helper behavior. Source/quiet parsing, transcript-to-repo candidate shaping, sweep-summary projection, and sync Absorb prompt context each have named files under `src/services/sync/`. `src/services/sync/sync.ts` keeps config activation lookup and lifecycle Absorb handoff because those are workflow decisions of the sync product verb.
 
+The sync sweep file coordinates candidate iteration. It may show the order of eligibility checks, repo locks, ledger reconciliation, transcript snapshot reads, cursor decisions, Absorb enqueue attempts, and summary accounting. It should not own the implementation details for candidate eligibility, internal-Almanac-session lookup, or ready-cursor-to-Absorb enqueue transitions. Those live in `candidate-eligibility.ts`, `internal-sessions.ts`, and `absorb-enqueue.ts`, respectively. Sweep-result files own summary shapes only, not prompt context text.
+
 ### Automation scheduler contracts are service-owned; launchd is platform
 
 `src/services/automation/` owns automation product workflows: task selection, interval validation, install/status/uninstall results, legacy migration semantics, and setup cleanup verbs. It talks to an injected `AutomationScheduler` contract from `src/services/automation/scheduler.ts`. `src/platform/automation/scheduler.ts` implements that contract with launchd/plist mechanics: default plist paths, log paths, launch PATH construction, plist writes, launchctl activation/removal/status, legacy capture detection, and XML-to-command-array normalization. CLI/setup/uninstall edges create the launchd scheduler because they are the concrete runtime composition points.

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

@@ -5,7 +5,7 @@ Branch: `codex/intentional-architecture-rewrite`
 
 ## Current State
 
-The branch has more than 280 committed rewrite commits past `dev`. The worklog records 232 production slices so far.
+The branch has more than 280 committed rewrite commits past `dev`. The worklog records 233 production slices so far.
 
 The diff is broad: more than 490 files changed, with tens of thousands of lines reshaped.
 
@@ -56,6 +56,7 @@ This is no longer a small cleanup branch. It is a real ownership rewrite.
 - Moved concrete transcript discovery/snapshot composition behind an injected sync transcript runtime contract, with `src/platform/transcripts/runtime.ts` wired by the CLI sync edge.
 - Moved sync's internal-Almanac-session lookup behind a jobs-service provider-session helper, so sync no longer reads job record storage shape directly.
 - Split sync workflow helpers into owned files for input parsing, transcript candidate shaping, summary projection, and Absorb context rendering.
+- Split sync sweep helpers into owned files for candidate eligibility, internal-session detection, and Absorb enqueue/ledger transitions.
 - Moved lifecycle operation construction and Absorb input/source handling into `src/services/lifecycle/` and removed the old top-level `src/operations/` and `src/absorb/` source buckets.
 - Normalized lifecycle operation failures into lifecycle-owned result contracts before command rendering sees them.
 - Moved init prompt context construction out of the operation command adapter and into lifecycle workflows, so command code only shapes flags into service requests and renders service results.
@@ -108,7 +109,7 @@ This is no longer a small cleanup branch. It is a real ownership rewrite.
 
 ## Latest Checkpoint
 
-The latest slice split the sync workflow helper logic into owned files so `src/services/sync/sync.ts` reads as orchestration instead of a mixed helper bucket.
+The latest slice split sync sweep helper logic into owned files so `src/services/sync/sweep.ts` reads as candidate orchestration instead of a mixed eligibility/session/enqueue bucket.
 
 Verification passed:
 
@@ -117,8 +118,8 @@ Verification passed:
 - `npx vitest run test/architecture-boundaries.test.ts test/sync.test.ts`
 - `npm test`
 - `npm run build`
-- `node dist/launcher.js sync status --json --quiet 1m`
 - `node dist/launcher.js sync status --help`
+- `node dist/launcher.js sync status --from codex --quiet 999999999s --json`
 - `node dist/launcher.js search transcript --limit 1`
 
 ## Immediate Next Work

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

@@ -1630,3 +1630,13 @@ Two-hundred-thirty-second production slice:
 - Kept config `sync_since` lookup and lifecycle Absorb handoff in `sync.ts` because those are the workflow decisions the top-level sync verb coordinates.
 - Reduced `src/services/sync/sync.ts` from a large mixed file to a small orchestrator over named sync helpers.
 - Strengthened architecture-boundary tests so sync parsing, summary projection, repo candidate shaping, and Absorb context rendering stay in their owned files.
+
+Two-hundred-thirty-third production slice:
+
+- Split `src/services/sync/sweep.ts` so it stays focused on per-candidate sweep orchestration instead of owning every helper decision inline.
+- Added `src/services/sync/candidate-eligibility.ts` for activation and quiet-window eligibility checks.
+- Added `src/services/sync/internal-sessions.ts` for the internal-Almanac-session skip check over job provider session ids.
+- Added `src/services/sync/absorb-enqueue.ts` for ready-cursor-to-Absorb-job handoff, sync cursor context text, and the resulting ledger transition.
+- Kept ledger loading/reconciliation, repo sync locking, transcript snapshot reads, and summary accounting visible in `sweep.ts` because those are the sweep workflow steps.
+- Moved cursor context text out of `sweep-results.ts`, leaving that file as summary projection only.
+- Strengthened architecture-boundary tests so `sweep.ts` does not regain provider-session lookup or cursor context text.

src/services/sync/absorb-enqueue.ts

@@ -0,0 +1,82 @@
+import type {
+  TranscriptCandidate,
+  TranscriptSnapshot,
+} from "../../shared/transcripts.js";
+import type { LedgerEntry } from "./ledger.js";
+import {
+  failedLedgerEntry,
+  pendingLedgerEntry,
+  type SyncCursorDecision,
+} from "./transcript-cursor.js";
+
+export interface StartSyncAbsorbArgs {
+  candidate: TranscriptCandidate;
+  contextNote: string;
+}
+
+export type StartSyncAbsorbResult =
+  | { ok: true; jobId: string }
+  | { ok: false; error: string };
+
+export type StartSyncAbsorbFn = (
+  args: StartSyncAbsorbArgs,
+) => Promise<StartSyncAbsorbResult>;
+
+export async function enqueueSyncAbsorb(args: {
+  candidate: TranscriptCandidate;
+  entry: LedgerEntry;
+  decision: Extract<SyncCursorDecision, { kind: "ready" }>;
+  snapshot: TranscriptSnapshot;
+  now: Date;
+  startAbsorb: StartSyncAbsorbFn;
+}): Promise<
+  | { ok: true; jobId: string; entry: LedgerEntry }
+  | { ok: false; reason: string; entry: LedgerEntry }
+> {
+  const result = await args.startAbsorb({
+    candidate: args.candidate,
+    contextNote: syncCursorContext({
+      candidate: args.candidate,
+      fromLine: args.decision.fromLine,
+      lastAbsorbedLine: args.entry.lastAbsorbedLine,
+      lastAbsorbedSize: args.entry.lastAbsorbedSize,
+    }),
+  });
+  if (!result.ok) {
+    return {
+      ok: false,
+      reason: "absorb-start-failed",
+      entry: failedLedgerEntry(args.entry, result.error),
+    };
+  }
+  return {
+    ok: true,
+    jobId: result.jobId,
+    entry: pendingLedgerEntry({
+      entry: args.entry,
+      snapshot: args.snapshot,
+      jobId: result.jobId,
+      now: args.now,
+    }),
+  };
+}
+
+function syncCursorContext(args: {
+  candidate: TranscriptCandidate;
+  fromLine: number;
+  lastAbsorbedLine: number;
+  lastAbsorbedSize: number;
+}): string {
+  return [
+    "Scheduled sync cursor:",
+    `- App: ${args.candidate.app}`,
+    `- Session id: ${args.candidate.sessionId}`,
+    `- Transcript: ${args.candidate.transcriptPath}`,
+    `- Previously absorbed through line: ${args.lastAbsorbedLine}`,
+    `- Previously absorbed through byte: ${args.lastAbsorbedSize}`,
+    `- Focus on line ${args.fromLine} onward.`,
+    "- You may inspect earlier lines only for context.",
+    "- Do not re-document decisions already absorbed unless newer lines amend, " +
+      "invalidate, or add important nuance to them.",
+  ].join("\n");
+}

src/services/sync/candidate-eligibility.ts

@@ -0,0 +1,21 @@
+import type { TranscriptCandidate } from "../../shared/transcripts.js";
+import { syncSkippedSummary, type SyncSkipped } from "./sweep-results.js";
+
+export function syncCandidateEligibility(
+  candidate: TranscriptCandidate,
+  args: {
+    syncSince: Date | null;
+    quietMs: number;
+    now: Date;
+  },
+): SyncSkipped | null {
+  if (args.syncSince !== null && candidate.mtimeMs < args.syncSince.getTime()) {
+    return syncSkippedSummary(candidate, "before-automation-activation");
+  }
+
+  const quietForMs = args.now.getTime() - candidate.mtimeMs;
+  if (quietForMs < args.quietMs) {
+    return syncSkippedSummary(candidate, "quiet-window");
+  }
+  return null;
+}

src/services/sync/internal-sessions.ts

@@ -0,0 +1,14 @@
+import type { TranscriptCandidate } from "../../shared/transcripts.js";
+import { listJobProviderSessionIds } from "../jobs/index.js";
+
+export async function isInternalAlmanacSession(
+  candidate: TranscriptCandidate,
+  cache: Map<string, Set<string>>,
+): Promise<boolean> {
+  let ids = cache.get(candidate.repoRoot);
+  if (ids === undefined) {
+    ids = await listJobProviderSessionIds(candidate.repoRoot);
+    cache.set(candidate.repoRoot, ids);
+  }
+  return ids.has(candidate.sessionId);
+}

src/services/sync/sweep-results.ts

@@ -1,4 +1,7 @@
-import type { TranscriptCandidate, TranscriptSourceApp } from "../../shared/transcripts.js";
+import type {
+  TranscriptCandidate,
+  TranscriptSourceApp,
+} from "../../shared/transcripts.js";
 import type { SyncCursorDecision } from "./transcript-cursor.js";
 
 export interface SyncStarted {
@@ -98,23 +101,3 @@ export function syncSkippedSummary(
     reason,
   };
 }
-
-export function syncCursorContext(args: {
-  candidate: TranscriptCandidate;
-  fromLine: number;
-  lastAbsorbedLine: number;
-  lastAbsorbedSize: number;
-}): string {
-  return [
-    "Scheduled sync cursor:",
-    `- App: ${args.candidate.app}`,
-    `- Session id: ${args.candidate.sessionId}`,
-    `- Transcript: ${args.candidate.transcriptPath}`,
-    `- Previously absorbed through line: ${args.lastAbsorbedLine}`,
-    `- Previously absorbed through byte: ${args.lastAbsorbedSize}`,
-    `- Focus on line ${args.fromLine} onward.`,
-    "- You may inspect earlier lines only for context.",
-    "- Do not re-document decisions already absorbed unless newer lines amend, " +
-      "invalidate, or add important nuance to them.",
-  ].join("\n");
-}

src/services/sync/sweep.ts

@@ -1,50 +1,34 @@
 import {
   type TranscriptCandidate,
   type TranscriptReadResult,
-  type TranscriptSnapshot,
 } from "../../shared/transcripts.js";
 import {
-  type LedgerEntry,
   type SyncLedger,
   freshLedgerEntry,
   ledgerKey,
   reconcileLedger,
 } from "./ledger.js";
+import { evaluateSyncCursor } from "./transcript-cursor.js";
 import {
-  type SyncCursorDecision,
-  evaluateSyncCursor,
-  failedLedgerEntry,
-  pendingLedgerEntry,
-} from "./transcript-cursor.js";
-import {
-  type SyncSkipped,
   type SyncSummary,
   emptySyncSummary,
-  syncCursorContext,
   syncReadySummary,
   syncSkippedSummary,
   syncStartedSummary,
 } from "./sweep-results.js";
 import type { IsPidAlive } from "../../shared/pid-liveness.js";
+import {
+  enqueueSyncAbsorb,
+  type StartSyncAbsorbFn,
+} from "./absorb-enqueue.js";
+import { syncCandidateEligibility } from "./candidate-eligibility.js";
+import { isInternalAlmanacSession } from "./internal-sessions.js";
 import {
   loadLedgerForRepo,
   writeLedger,
 } from "../../stores/sync/ledger.js";
 import { acquireRepoSyncLock, releaseRepoSyncLock } from "../../stores/sync/lock.js";
-import { listJobProviderSessionIds } from "../jobs/index.js";
-
-export interface StartSyncAbsorbArgs {
-  candidate: TranscriptCandidate;
-  contextNote: string;
-}
-
-export type StartSyncAbsorbResult =
-  | { ok: true; jobId: string }
-  | { ok: false; error: string };
 
-export type StartSyncAbsorbFn = (
-  args: StartSyncAbsorbArgs,
-) => Promise<StartSyncAbsorbResult>;
 export type ReadSyncTranscriptSnapshotFn = (
   transcriptPath: string,
 ) => Promise<TranscriptReadResult>;
@@ -71,7 +55,7 @@ export async function executeSyncSweep(args: {
   const heldLocks = new Set<string>();
   try {
     for (const candidate of args.candidates) {
-      const eligibilitySkip = candidateEligibility(candidate, args);
+      const eligibilitySkip = syncCandidateEligibility(candidate, args);
       if (eligibilitySkip !== null) {
         summary.skipped.push(eligibilitySkip);
         continue;
@@ -102,9 +86,13 @@ export async function executeSyncSweep(args: {
       await reconcileLedger(candidate.repoRoot, ledger, args.now);
       const key = ledgerKey(candidate);
 
-      const transcript = await args.readTranscriptSnapshot(candidate.transcriptPath);
+      const transcript = await args.readTranscriptSnapshot(
+        candidate.transcriptPath,
+      );
       if (!transcript.ok) {
-        summary.needsAttention.push(syncSkippedSummary(candidate, transcript.reason));
+        summary.needsAttention.push(
+          syncSkippedSummary(candidate, transcript.reason),
+        );
         continue;
       }
       const entry = ledger.sessions[key] ??
@@ -130,7 +118,7 @@ export async function executeSyncSweep(args: {
         continue;
       }
 
-      const enqueue = await enqueueAbsorb({
+      const enqueue = await enqueueSyncAbsorb({
         candidate,
         entry,
         decision,
@@ -140,7 +128,9 @@ export async function executeSyncSweep(args: {
       });
       if (!enqueue.ok) {
         ledger.sessions[key] = enqueue.entry;
-        summary.needsAttention.push(syncSkippedSummary(candidate, enqueue.reason));
+        summary.needsAttention.push(
+          syncSkippedSummary(candidate, enqueue.reason),
+        );
         await writeLedger(candidate.repoRoot, ledger, args.now);
         continue;
       }
@@ -162,73 +152,3 @@ export async function executeSyncSweep(args: {
 
   return summary;
 }
-
-async function isInternalAlmanacSession(
-  candidate: TranscriptCandidate,
-  cache: Map<string, Set<string>>,
-): Promise<boolean> {
-  let ids = cache.get(candidate.repoRoot);
-  if (ids === undefined) {
-    ids = await listJobProviderSessionIds(candidate.repoRoot);
-    cache.set(candidate.repoRoot, ids);
-  }
-  return ids.has(candidate.sessionId);
-}
-
-function candidateEligibility(
-  candidate: TranscriptCandidate,
-  args: {
-    syncSince: Date | null;
-    quietMs: number;
-    now: Date;
-  },
-): SyncSkipped | null {
-  if (args.syncSince !== null && candidate.mtimeMs < args.syncSince.getTime()) {
-    return syncSkippedSummary(candidate, "before-automation-activation");
-  }
-
-  const quietForMs = args.now.getTime() - candidate.mtimeMs;
-  if (quietForMs < args.quietMs) {
-    return syncSkippedSummary(candidate, "quiet-window");
-  }
-  return null;
-}
-
-async function enqueueAbsorb(args: {
-  candidate: TranscriptCandidate;
-  entry: LedgerEntry;
-  decision: Extract<SyncCursorDecision, { kind: "ready" }>;
-  snapshot: TranscriptSnapshot;
-  now: Date;
-  startAbsorb: StartSyncAbsorbFn;
-}): Promise<
-  | { ok: true; jobId: string; entry: LedgerEntry }
-  | { ok: false; reason: string; entry: LedgerEntry }
-> {
-  const result = await args.startAbsorb({
-    candidate: args.candidate,
-    contextNote: syncCursorContext({
-      candidate: args.candidate,
-      fromLine: args.decision.fromLine,
-      lastAbsorbedLine: args.entry.lastAbsorbedLine,
-      lastAbsorbedSize: args.entry.lastAbsorbedSize,
-    }),
-  });
-  if (!result.ok) {
-    return {
-      ok: false,
-      reason: "absorb-start-failed",
-      entry: failedLedgerEntry(args.entry, result.error),
-    };
-  }
-  return {
-    ok: true,
-    jobId: result.jobId,
-    entry: pendingLedgerEntry({
-      entry: args.entry,
-      snapshot: args.snapshot,
-      jobId: result.jobId,
-      now: args.now,
-    }),
-  };
-}