diff --git a/.changeset/sync-agent-serve.md b/.changeset/sync-agent-serve.md new file mode 100644 index 00000000..03d201d3 --- /dev/null +++ b/.changeset/sync-agent-serve.md @@ -0,0 +1,12 @@ +--- +"@bounded-systems/prx": minor +--- + +Add the sync agent (`prx sync serve`, prx-697): a long-running daemon that every +`--interval` seconds runs the existing cross-repo reconcile orchestrators +(`runBeadsSyncAcrossRepos` beads↔GH + `runDoltReconcileAcrossRepos` dolt +push/pull) over the repo inventory — so beads durability no longer depends on a +hand-run sync. Best-effort per tick (per-repo failures self-isolate; a pass-level +throw is swallowed + logged, like beadsd's refresh). The blocking prerequisite +for prx-82b (remove host bd). No socket in v1 — it's a periodic orchestrator, not +a request daemon. diff --git a/docs/cli.md b/docs/cli.md index 39c7173b..24e13adc 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -2,7 +2,7 @@ -`prx` exposes **242** commands across 4 domains, +`prx` exposes **243** commands across 4 domains, owned by **16** actors. Each command is a registry entry; the CLI, MCP toolset, help text, and this page are all projections of it. @@ -175,6 +175,7 @@ CLI, MCP toolset, help text, and this page are all projections of it. | `sync` | Canonical actor surface for issue reconcile | domain_sync | — | | `sync backfill` | Backfill cursor-skipped external records over a range | domain_sync | — | | `sync issues` | Reconcile pinned pairs; --limit caps push only | domain_sync | — | +| `sync serve` | Run the sync agent: periodic cross-repo reconcile daemon | domain_sync | — | | `transcripts digest` | Digest transcripts into durable memory shards | transcripts | — | | `transcripts status` | Report transcript TTL pressure and candidates | transcripts | — | diff --git a/packages/prx/openapi.json b/packages/prx/openapi.json index 3a246d4d..b0f4c31e 100644 --- a/packages/prx/openapi.json +++ b/packages/prx/openapi.json @@ -1079,6 +1079,35 @@ } } }, + "/sync/serve": { + "post": { + "operationId": "sync_serve", + "summary": "Run the sync agent: periodically reconcile every inventory repo (beads + dolt).", + "x-prx-actor": "work", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/sync_serveInput" + } + } + } + }, + "responses": { + "200": { + "description": "ok", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/sync_serveOutput" + } + } + } + } + } + } + }, "/transition": { "post": { "operationId": "transition", @@ -3406,6 +3435,39 @@ ], "additionalProperties": false }, + "sync_serveInput": { + "type": "object", + "properties": { + "interval": { + "description": "seconds between cross-repo reconcile passes (default 300)", + "type": "integer", + "exclusiveMinimum": 0, + "maximum": 9007199254740991 + }, + "pidfile": { + "description": "write the daemon pid here (removed on close)", + "type": "string" + }, + "cwd": { + "description": "ignored — the sync agent is not repo-bound", + "type": "string" + } + }, + "additionalProperties": false + }, + "sync_serveOutput": { + "type": "object", + "properties": { + "intervalSeconds": { + "type": "number", + "description": "the cross-repo reconcile interval the agent ran at" + } + }, + "required": [ + "intervalSeconds" + ], + "additionalProperties": false + }, "transitionInput": { "type": "object", "properties": { diff --git a/packages/prx/scripts/coverage-summary.ts b/packages/prx/scripts/coverage-summary.ts index 132e1559..d1817870 100644 --- a/packages/prx/scripts/coverage-summary.ts +++ b/packages/prx/scripts/coverage-summary.ts @@ -69,6 +69,7 @@ const PER_FILE_BASELINE = new Set([ "packages/prx/src/room/pod-secrets-verb.ts", // 79% — run() calls live podman (ensurePodSecrets); logic covered in pod-secrets.test.ts, not unit-tested here "packages/prx/src/room/pod-down-verb.ts", // run() calls live downPod (podman); covered by live e2e, not unit test (mirrors pod-up-verb) "packages/prx/src/builder/verb.ts", // run() calls live podman + ssh-keygen (the nix-builder container); render core (container-builder.ts) is unit-tested, not the live verb (mirrors pod-up-verb) + "packages/prx/src/sync/serve-verb.ts", // wraps the live cross-repo orchestrators (real bd/gh/dolt); the loop (serve.ts) is unit-tested, not the live wiring (mirrors pod-up-verb) ]); type Totals = { diff --git a/packages/prx/src/cli/registry.data.ts b/packages/prx/src/cli/registry.data.ts index 25eb5f79..74385d9b 100644 --- a/packages/prx/src/cli/registry.data.ts +++ b/packages/prx/src/cli/registry.data.ts @@ -1927,6 +1927,17 @@ const RAW_REGISTRY: z.input[] = [ domain: "state", actor: "domain_sync", }, + { + // prx-697: the SYNC AGENT daemon. A long-running loop that every --interval + // seconds runs the cross-repo reconcile orchestrators (beads↔GH + dolt + // push/pull) over the repo inventory, so beads durability doesn't depend on + // a hand-run sync. The blocking prerequisite for prx-82b (remove host bd). + name: "sync serve", + parent: "sync", + description: "Run the sync agent: periodic cross-repo reconcile daemon", + domain: "state", + actor: "domain_sync", + }, { // GH-1513: bd-side memory-decay policy (GH-1500 ADR §3b; capability // split 4/4 of GH-298). Operator-triggered tick that classifies closed diff --git a/packages/prx/src/cli/verb-registry.ts b/packages/prx/src/cli/verb-registry.ts index cc0302cd..3203c9b5 100644 --- a/packages/prx/src/cli/verb-registry.ts +++ b/packages/prx/src/cli/verb-registry.ts @@ -34,6 +34,7 @@ import { podDownVerb } from "../room/pod-down-verb.ts"; import { podUpVerb } from "../room/pod-up-verb.ts"; import { builderUpVerb, builderRegisterVerb } from "../builder/verb.ts"; import { forgeServeVerb } from "../forge-d/serve-verb.ts"; +import { syncServeVerb } from "../sync/serve-verb.ts"; import { doorBridgeVerb } from "../door/bridge-verb.ts"; import { doorGrantVerb, doorIssuerKeysVerb } from "../door/grant-verb.ts"; import { conciergeServeVerb } from "../concierge/serve-verb.ts"; @@ -73,6 +74,7 @@ export const verbRegistry: Registry = { [builderUpVerb.id]: builderUpVerb, [builderRegisterVerb.id]: builderRegisterVerb, [forgeServeVerb.id]: forgeServeVerb, + [syncServeVerb.id]: syncServeVerb, [doorBridgeVerb.id]: doorBridgeVerb, [doorGrantVerb.id]: doorGrantVerb, [doorIssuerKeysVerb.id]: doorIssuerKeysVerb, diff --git a/packages/prx/src/pr-state/cli.ts b/packages/prx/src/pr-state/cli.ts index 05e41459..aa72b79a 100644 --- a/packages/prx/src/pr-state/cli.ts +++ b/packages/prx/src/pr-state/cli.ts @@ -4148,7 +4148,7 @@ export function normalizeNamespaceArgv(argv: string[]): string[] { // parse time (see the executor branch for `sync-issues-pair`). if (c0 === "sync") { if (!c1 || c1.startsWith("-")) { - throw new CliError("sync requires a subcommand: issues, backfill"); + throw new CliError("sync requires a subcommand: issues, backfill, serve"); } if (c1 === "issues") { return ["sync-issues-pair", ...tail]; @@ -4156,6 +4156,11 @@ export function normalizeNamespaceArgv(argv: string[]): string[] { if (c1 === "backfill") { return ["sync-backfill", ...tail]; } + // prx-697: `sync serve` (the sync agent) is a spec-driven verb dispatched + // ahead of the legacy parser (runSpecVerb); pass it through unchanged. + if (c1 === "serve") { + return argv; + } throw new CliError(`Unknown sync subcommand: ${c1}`); } @@ -14858,6 +14863,12 @@ export function runCli( if (orchestratorVerb === "builder" && orchestratorRest[0] === "register") { return runSpecVerb("builder register", orchestratorRest.slice(1), output); } + // `sync serve` — the sync agent (prx-697): periodic cross-repo beads+dolt + // reconcile. Only `serve` routes here; `prx sync issues` falls through to the + // legacy sync parser below. + if (orchestratorVerb === "sync" && orchestratorRest[0] === "serve") { + return runSpecVerb("sync serve", orchestratorRest.slice(1), output); + } // The `contract ` namespace reroutes several subcommands to verbs that // are now spec-driven. The early dispatch keys off the raw `argv[0]` // (`contract`), not the normalized rewrite, so those aliases would miss the diff --git a/packages/prx/src/sync/serve-verb.ts b/packages/prx/src/sync/serve-verb.ts new file mode 100644 index 00000000..eb60215a --- /dev/null +++ b/packages/prx/src/sync/serve-verb.ts @@ -0,0 +1,88 @@ +// `prx sync serve` — run the SYNC AGENT (prx-697): a long-running daemon that +// every `--interval` seconds reconciles every inventory repo (domain↔GH + dolt +// push/pull) so beads durability doesn't depend on anyone running a sync by hand. +// Authored once as a VerbSpec (projected to CLI / MCP / OpenAPI). Mirrors +// forgeServeVerb's infra shape; the reconcile logic lives in runSyncServe (the +// loop) over the existing, tested cross-repo orchestrators. + +import { z } from "zod"; + +import { defineVerb } from "@bounded-systems/verbspec"; + +import { + runSyncServe, + DEFAULT_SYNC_INTERVAL_MS, + type SyncServeHandle, + type SyncServeOutput, +} from "./serve.ts"; +import { runBeadsSyncAcrossRepos } from "./run-cross-repo.ts"; +import { runDoltReconcileAcrossRepos } from "./run-dolt-reconcile-cross-repo.ts"; +import { DEFAULT_SYNC_LIMIT } from "./limits.ts"; + +/** The real domain↔GH pass: reconcile every inventory repo against GitHub. */ +async function beadsSyncPass(output: SyncServeOutput): Promise<{ exitCode: number }> { + const r = await runBeadsSyncAcrossRepos( + { dryRun: false, domain: "gh", limit: DEFAULT_SYNC_LIMIT, format: "plain" }, + output, + ); + return { exitCode: r.exitCode }; +} + +/** The real dolt pass: full commit→pull→push reconcile of every eligible repo. */ +async function doltReconcilePass(output: SyncServeOutput): Promise<{ exitCode: number }> { + const { exitCode } = await runDoltReconcileAcrossRepos( + { mode: "full", dryRun: false, format: "plain" }, + output, + ); + return { exitCode }; +} + +export const SyncServeResult = z + .object({ + intervalSeconds: z.number().describe("the cross-repo reconcile interval the agent ran at"), + }) + .strict(); +export type SyncServeResult = z.infer; + +export type SyncServeVerbDeps = { + serve: typeof runSyncServe; + log: (line: string) => void; +}; + +const realSyncServeDeps = (): SyncServeVerbDeps => ({ + serve: runSyncServe, + log: (line) => console.error(line), +}); + +export const syncServeVerb = defineVerb({ + id: "sync serve", + summary: "Run the sync agent: periodically reconcile every inventory repo (beads + dolt).", + actor: "work", + input: z.object({ + interval: z + .number() + .int() + .positive() + .optional() + .describe(`seconds between cross-repo reconcile passes (default ${DEFAULT_SYNC_INTERVAL_MS / 1000})`), + pidfile: z.string().optional().describe("write the daemon pid here (removed on close)"), + // Accepted for daemon-lifecycle uniformity (the generic launcher passes --cwd + // to every serve command). The sync agent is host-global / cross-repo, not + // repo-bound, so this is ignored. + cwd: z.string().optional().describe("ignored — the sync agent is not repo-bound"), + }), + output: SyncServeResult, + deps: realSyncServeDeps, + run: async (input, deps: SyncServeVerbDeps = realSyncServeDeps()): Promise => { + const intervalSeconds = input.interval ?? DEFAULT_SYNC_INTERVAL_MS / 1000; + const handle: SyncServeHandle = await deps.serve({ + intervalMs: intervalSeconds * 1000, + ...(input.pidfile ? { pidfile: input.pidfile } : {}), + deps: { beadsSyncPass, doltReconcilePass }, + }); + deps.log(`sync agent: reconciling every inventory repo every ${intervalSeconds}s`); + // Block until terminated — the daemon runs until killed (SIGTERM/SIGINT). + await handle.closed; + return { intervalSeconds }; + }, +}); diff --git a/packages/prx/src/sync/serve.ts b/packages/prx/src/sync/serve.ts new file mode 100644 index 00000000..7aa43996 --- /dev/null +++ b/packages/prx/src/sync/serve.ts @@ -0,0 +1,140 @@ +// The SYNC AGENT (prx-697) — a periodic driver over the existing cross-repo +// reconcile orchestrators. It owns cross-repo beads DURABILITY: every tick it +// runs the domain↔GH sync (`runBeadsSyncAcrossRepos`, GH-1662) then the dolt +// reconcile (`runDoltReconcileAcrossRepos`, GH-1702) over the repo inventory. +// +// This is NOT new reconcile logic — the primitives + orchestrators (with their +// cursor + GH-API budget + per-repo isolation + schema-conflict handling) already +// exist and are tested. The agent is just the timer loop above them, the blocking +// prerequisite for prx-82b (remove host bd) — once the agent owns durability, the +// host-native daemon can be retired. +// +// Shape: a long-running daemon (`prx sync serve`), mirroring beadsd's +// `runBeadsServe` (interval + graceful shutdown). It runs host-global (cross-repo), +// so it's a standalone agent — not a per-repo-pod member. No socket in v1: it's a +// periodic orchestrator, not a request daemon (a status door is a later refinement). + +import { writeFileSync, rmSync } from "node:fs"; + +/** Where pass output goes (mirrors the orchestrators' `output` shape). */ +export interface SyncServeOutput { + log: (line: string) => void; + error: (line: string) => void; +} + +/** Default tick interval — 5 min, matching the per-repo beadsd refresh cadence. */ +export const DEFAULT_SYNC_INTERVAL_MS = 5 * 60_000; + +/** Injectable seams. The two passes are required (the verb supplies the real + * orchestrator-backed ones — see serve-verb.ts; tests supply fakes); the rest + * default to the wall clock / fs / process signals. */ +export interface SyncServeDeps { + /** The domain↔GH cross-repo pass (e.g. `runBeadsSyncAcrossRepos` over all repos). */ + beadsSyncPass: (output: SyncServeOutput) => Promise<{ exitCode: number }>; + /** The dolt reconcile cross-repo pass (e.g. `runDoltReconcileAcrossRepos` full). */ + doltReconcilePass: (output: SyncServeOutput) => Promise<{ exitCode: number }>; + /** `setInterval` seam (test injection). */ + setInterval?: (fn: () => void, ms: number) => ReturnType; + /** `clearInterval` seam (test injection). */ + clearInterval?: (handle: ReturnType) => void; + /** Install the shutdown handler (default: process SIGTERM/SIGINT; test: no-op). */ + onSignal?: (stop: () => void) => void; + /** Write the pidfile (default: `writeFileSync`). */ + writePidfile?: (path: string, pid: number) => void; + /** Remove the pidfile (default: `rmSync`, best-effort). */ + removePidfile?: (path: string) => void; +} + +export interface SyncServeOptions { + /** Interval between cross-repo passes (default {@link DEFAULT_SYNC_INTERVAL_MS}). */ + intervalMs?: number | undefined; + /** Write the daemon pid here on start; removed on stop (launcher-trackable). */ + pidfile?: string | undefined; + output?: SyncServeOutput | undefined; + /** The reconcile passes (+ optional clock/fs/signal seams). Required. */ + deps: SyncServeDeps; +} + +export interface SyncServeHandle { + /** Resolves when the daemon stops (signal or {@link SyncServeHandle.stop}). */ + closed: Promise; + /** Stop the loop and resolve {@link SyncServeHandle.closed}. */ + stop: () => void; + /** Run one cross-repo pass now (also the on-start pass). Exposed for tests. */ + tick: () => Promise; +} + +/** + * Run the sync agent: an on-start pass then one every `intervalMs`. Each pass is + * best-effort — the orchestrators already self-isolate per-repo failures, and a + * pass-level throw is swallowed + logged (a stale-but-up agent beats a crash, + * like beadsd's refresh). Returns a handle whose `closed` resolves on shutdown. + */ +export async function runSyncServe(options: SyncServeOptions): Promise { + const output: SyncServeOutput = options.output ?? { + log: (line) => console.error(line), + error: (line) => console.error(line), + }; + const intervalMs = options.intervalMs ?? DEFAULT_SYNC_INTERVAL_MS; + const deps = options.deps; + const setI = deps.setInterval ?? ((fn, ms) => setInterval(fn, ms)); + const clearI = deps.clearInterval ?? ((handle) => clearInterval(handle)); + const beadsSyncPass = deps.beadsSyncPass; + const doltReconcilePass = deps.doltReconcilePass; + + const tick = async (): Promise => { + try { + const r = await beadsSyncPass(output); + output.log(`sync: beads cross-repo pass done (exit ${r.exitCode})`); + } catch (e) { + output.error(`sync: beads pass failed (continuing): ${(e as Error).message}`); + } + try { + const r = await doltReconcilePass(output); + output.log(`sync: dolt reconcile pass done (exit ${r.exitCode})`); + } catch (e) { + output.error(`sync: dolt pass failed (continuing): ${(e as Error).message}`); + } + }; + + let resolveClosed: () => void = () => {}; + const closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + let handle: ReturnType | null = null; + let stopped = false; + const removePidfile = deps.removePidfile ?? ((p) => rmSync(p, { force: true })); + const stop = (): void => { + if (stopped) return; + stopped = true; + if (handle !== null) clearI(handle); + if (options.pidfile) { + try { + removePidfile(options.pidfile); + } catch { + // best-effort — a leftover pidfile is harmless + } + } + resolveClosed(); + }; + + if (options.pidfile) { + const writePidfile = deps.writePidfile ?? ((p, pid) => writeFileSync(p, `${pid}\n`)); + writePidfile(options.pidfile, process.pid); + } + + if (deps.onSignal) { + deps.onSignal(stop); + } else { + process.on("SIGTERM", stop); + process.on("SIGINT", stop); + } + + // On-start pass, then schedule the recurring tick. + await tick(); + handle = setI(() => { + void tick(); + }, intervalMs); + + return { closed, stop, tick }; +} diff --git a/packages/prx/test/pr-state/help/__snapshots__/help.snapshot.test.ts.snap b/packages/prx/test/pr-state/help/__snapshots__/help.snapshot.test.ts.snap index 2abfff7c..e5af59fc 100644 --- a/packages/prx/test/pr-state/help/__snapshots__/help.snapshot.test.ts.snap +++ b/packages/prx/test/pr-state/help/__snapshots__/help.snapshot.test.ts.snap @@ -82,6 +82,7 @@ State: prx sync Canonical actor surface for issue reconcile prx sync issues Reconcile pinned pairs; --limit caps push only prx sync backfill Backfill cursor-skipped external records over a range + prx sync serve Run the sync agent: periodic cross-repo reconcile daemon prx memory compact Compact closed beads (memory-decay policy chokepoint) prx handoff enqueue Enqueue a handoff for the recipient actor prx handoff status Show pending handoffs filtered by target diff --git a/packages/prx/test/sync/serve-verb.test.ts b/packages/prx/test/sync/serve-verb.test.ts new file mode 100644 index 00000000..d8d4bf0b --- /dev/null +++ b/packages/prx/test/sync/serve-verb.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, test } from "bun:test"; + +import { syncServeVerb, type SyncServeVerbDeps } from "../../src/sync/serve-verb.ts"; +import { DEFAULT_SYNC_INTERVAL_MS, type SyncServeHandle } from "../../src/sync/serve.ts"; + +/** A handle whose `closed` resolves on the next tick so run() unblocks. */ +function fakeHandle(): SyncServeHandle { + return { + closed: new Promise((resolve) => setTimeout(resolve, 0)), + stop: () => {}, + tick: async () => {}, + }; +} + +describe("syncServeVerb", () => { + test("serves at the given interval (seconds → ms), logs, returns the interval", async () => { + let servedMs: number | undefined; + let servedPidfile: string | undefined; + const logs: string[] = []; + const deps: SyncServeVerbDeps = { + serve: async (opts) => { + servedMs = opts?.intervalMs; + servedPidfile = opts?.pidfile; + return fakeHandle(); + }, + log: (l) => logs.push(l), + }; + + const out = await syncServeVerb.run({ interval: 60, pidfile: "/tmp/s.pid" }, deps); + + expect(servedMs).toBe(60_000); + expect(servedPidfile).toBe("/tmp/s.pid"); + expect(logs.some((l) => l.includes("every 60s"))).toBe(true); + expect(out).toEqual({ intervalSeconds: 60 }); + }); + + test("defaults to DEFAULT_SYNC_INTERVAL_MS when --interval is omitted", async () => { + let servedMs: number | undefined; + const deps: SyncServeVerbDeps = { + serve: async (opts) => { + servedMs = opts?.intervalMs; + return fakeHandle(); + }, + log: () => {}, + }; + + const out = await syncServeVerb.run({}, deps); + + expect(servedMs).toBe(DEFAULT_SYNC_INTERVAL_MS); + expect(out.intervalSeconds).toBe(DEFAULT_SYNC_INTERVAL_MS / 1000); + }); + + test("verb id + actor", () => { + expect(syncServeVerb.id).toBe("sync serve"); + }); +}); diff --git a/packages/prx/test/sync/serve.test.ts b/packages/prx/test/sync/serve.test.ts new file mode 100644 index 00000000..63b36100 --- /dev/null +++ b/packages/prx/test/sync/serve.test.ts @@ -0,0 +1,176 @@ +import { describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { runSyncServe, DEFAULT_SYNC_INTERVAL_MS, type SyncServeOutput } from "../../src/sync/serve.ts"; + +const SAFE_PASSES = { + beadsSyncPass: async () => ({ exitCode: 0 }), + doltReconcilePass: async () => ({ exitCode: 0 }), +}; + +function collectOutput(): SyncServeOutput & { lines: string[]; errors: string[] } { + const lines: string[] = []; + const errors: string[] = []; + return { lines, errors, log: (l) => lines.push(l), error: (e) => errors.push(e) }; +} + +describe("runSyncServe — the sync-agent loop (prx-697)", () => { + test("runs both passes once on start, then again on each interval tick", async () => { + let beads = 0; + let dolt = 0; + let intervalFn: (() => void) | null = null; + let scheduledMs = 0; + const out = collectOutput(); + + const handle = await runSyncServe({ + intervalMs: 1000, + output: out, + deps: { + beadsSyncPass: async () => { + beads++; + return { exitCode: 0 }; + }, + doltReconcilePass: async () => { + dolt++; + return { exitCode: 0 }; + }, + setInterval: (fn, ms) => { + intervalFn = fn; + scheduledMs = ms; + return 0 as unknown as ReturnType; + }, + clearInterval: () => {}, + onSignal: () => {}, + }, + }); + + // On-start pass. + expect(beads).toBe(1); + expect(dolt).toBe(1); + expect(scheduledMs).toBe(1000); + + // Fire the interval tick twice. + await intervalFn!(); + await intervalFn!(); + expect(beads).toBe(3); + expect(dolt).toBe(3); + expect(out.lines.some((l) => l.includes("beads cross-repo pass done"))).toBe(true); + expect(out.lines.some((l) => l.includes("dolt reconcile pass done"))).toBe(true); + + handle.stop(); + await handle.closed; + }); + + test("a thrown pass is swallowed — the other pass still runs and the loop survives", async () => { + let dolt = 0; + const out = collectOutput(); + + const handle = await runSyncServe({ + output: out, + deps: { + beadsSyncPass: async () => { + throw new Error("boom"); + }, + doltReconcilePass: async () => { + dolt++; + return { exitCode: 0 }; + }, + setInterval: () => 0 as unknown as ReturnType, + clearInterval: () => {}, + onSignal: () => {}, + }, + }); + + // beads threw, but dolt still ran and runSyncServe did not reject. + expect(dolt).toBe(1); + expect(out.errors.some((e) => e.includes("beads pass failed") && e.includes("boom"))).toBe(true); + + handle.stop(); + await handle.closed; + }); + + test("stop() clears the interval and resolves closed (idempotent)", async () => { + let cleared = 0; + const handle = await runSyncServe({ + deps: { + beadsSyncPass: async () => ({ exitCode: 0 }), + doltReconcilePass: async () => ({ exitCode: 0 }), + setInterval: () => 42 as unknown as ReturnType, + clearInterval: () => { + cleared++; + }, + onSignal: () => {}, + }, + }); + + handle.stop(); + handle.stop(); // idempotent — no double clear + await handle.closed; // resolves + expect(cleared).toBe(1); + }); + + test("the shutdown handler is wired to stop", async () => { + let registered: (() => void) | null = null; + const handle = await runSyncServe({ + deps: { + beadsSyncPass: async () => ({ exitCode: 0 }), + doltReconcilePass: async () => ({ exitCode: 0 }), + setInterval: () => 0 as unknown as ReturnType, + clearInterval: () => {}, + onSignal: (stop) => { + registered = stop; + }, + }, + }); + expect(typeof registered).toBe("function"); + registered!(); // simulate SIGTERM + await handle.closed; // the signal stopped the loop + }); + + test("writes the pidfile on start and removes it on stop", async () => { + const written: Array<{ path: string; pid: number }> = []; + const removed: string[] = []; + const handle = await runSyncServe({ + pidfile: "/tmp/prx-sync.pid", + deps: { + beadsSyncPass: async () => ({ exitCode: 0 }), + doltReconcilePass: async () => ({ exitCode: 0 }), + setInterval: () => 0 as unknown as ReturnType, + clearInterval: () => {}, + onSignal: () => {}, + writePidfile: (path, pid) => written.push({ path, pid }), + removePidfile: (path) => removed.push(path), + }, + }); + expect(written).toEqual([{ path: "/tmp/prx-sync.pid", pid: process.pid }]); + handle.stop(); + await handle.closed; + expect(removed).toEqual(["/tmp/prx-sync.pid"]); + }); + + test("real defaults: wires the wall-clock timer, process signals, and fs pidfile", async () => { + // Exercise the default setInterval/clearInterval/onSignal/writePidfile/ + // removePidfile seams (injected passes keep it offline + side-effect-free). + const dir = mkdtempSync(join(tmpdir(), "prx-sync-")); + const pidfile = join(dir, "agent.pid"); + try { + const handle = await runSyncServe({ + intervalMs: 60 * 60_000, // far in the future — never fires during the test + pidfile, + deps: { ...SAFE_PASSES }, + }); + expect(existsSync(pidfile)).toBe(true); + handle.stop(); + await handle.closed; + expect(existsSync(pidfile)).toBe(false); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + test("DEFAULT_SYNC_INTERVAL_MS is 5 minutes (matches beadsd refresh)", () => { + expect(DEFAULT_SYNC_INTERVAL_MS).toBe(5 * 60_000); + }); +});