From 8267cb4a02caa4b045aa2d07068e946b7cb0eb1f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=BE=E5=B1=B1=E8=B2=B4=E8=87=B3?= Date: Sun, 7 Jun 2026 22:10:45 +0900 Subject: [PATCH] basou: add 'basou refresh --watch' to keep the corpus current Import is a pull, so the corpus only advances when 'basou refresh' runs. After multi-root capture fixed which repos are scanned, the remaining gap is currency: with no scheduler, the dogfood corpus drifts stale the moment more AI work happens. --watch closes that gap. 'basou refresh --watch [--interval ]' runs a catch-up refresh on start, then polls the native-log stores (~/.codex/sessions, ~/.claude/projects) and re-imports + regenerates only when they have SETTLED (unchanged since the previous poll, so nothing is captured mid-write) AND changed since the last import. Handoff / decisions regenerate only when something was actually imported, so AI work in unrelated projects never rewrites this workspace's files. Polling is dependency-free and cross-platform; latency is the interval (default 30s, min 5s, max 1 day), not real-time. Ctrl-C / SIGTERM stops after the in-flight cycle, never mid-write. Design (v1): per-file {mtime,size} scan for settle + change detection; the baseline scan is taken before the initial import so a session that appears during the import window is never mistaken for "already seen" and missed; a pending-regenerate flag retries regeneration if an import succeeded but the regenerate failed; scan errors are surfaced pathlessly and a root that is missing or not a directory contributes nothing; the abortable sleep leaves no listener behind across polls. Because import is idempotent, a session still active at start-up or one that resumes after it settled is recorded only up to that point -- 'basou refresh --force' rebuilds those. Event-driven watching and vendor session hooks are deferred. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 15 ++ docs/spec/terminal-and-import.md | 24 ++ .../cli/src/commands/refresh-watch.test.ts | 246 ++++++++++++++++++ packages/cli/src/commands/refresh-watch.ts | 200 ++++++++++++++ packages/cli/src/commands/refresh.test.ts | 26 +- packages/cli/src/commands/refresh.ts | 106 +++++++- 6 files changed, 614 insertions(+), 3 deletions(-) create mode 100644 packages/cli/src/commands/refresh-watch.test.ts create mode 100644 packages/cli/src/commands/refresh-watch.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f74a7b49..01441831 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ All notable changes to **basou** are recorded here. The project follows ### Added +- `basou refresh --watch [--interval ]` โ€” keep the workspace current + automatically instead of re-running `basou refresh` by hand. It does one + catch-up refresh on start, then polls the native-log stores (`~/.codex/sessions`, + `~/.claude/projects`) and re-imports + regenerates only when they have settled + (unchanged since the previous poll, so no session is captured mid-write) AND + changed since the last import. Handoff / decisions are regenerated only when + something was actually imported, so unrelated AI work elsewhere never rewrites + this workspace's files. Polling (default 30s, min 5s) is dependency-free and + cross-platform; latency is the interval, not real-time. Ctrl-C / SIGTERM stops + after the in-flight cycle. `--watch` cannot be combined with `--dry-run`, + `--json`, or `--force`. Because import is idempotent, a session that was + already imported is not re-imported in watch mode โ€” so a session still active + when the watcher starts (caught by the start-up catch-up) or one that resumes + after it settled is captured only up to that point; run `basou refresh --force` + to rebuild those from the latest logs. - Multi-root capture โ€” one `.basou/` workspace can now aggregate the native logs of several sibling repositories. This matters when a single logical project spans multiple checkouts (e.g. an implementation repo, a planning diff --git a/docs/spec/terminal-and-import.md b/docs/spec/terminal-and-import.md index adb96483..c292dbfa 100644 --- a/docs/spec/terminal-and-import.md +++ b/docs/spec/terminal-and-import.md @@ -124,3 +124,27 @@ Discovery therefore accepts a set of **source roots** instead of one: Each session is sanitized against its own `working_directory`, so aggregating sibling repos never relativizes a path across repositories or leaks an absolute host path beyond the existing pathless contract. + +## ยง14.4 Keeping the corpus current (`--watch`) + +Import is a pull, so the corpus only advances when `basou refresh` runs. +`basou refresh --watch [--interval ]` keeps it current without a +manual step: + +- It runs one catch-up refresh on start, then polls the native-log stores + (`~/.codex/sessions`, `~/.claude/projects`) every `interval` seconds + (default 30, minimum 5). +- A cycle re-imports and regenerates only when the logs have **settled** + (unchanged since the previous poll, so an in-progress session is not + captured mid-write) **and** changed since the last import. Handoff / + decisions regenerate only when something was actually imported, so AI + work in unrelated projects never rewrites this workspace's files. +- Polling is dependency-free and cross-platform; the trade-off is that + capture latency is the poll interval, not real-time. Ctrl-C / SIGTERM + stops the watcher after the current cycle (never mid-write). +- `--watch` cannot be combined with `--dry-run`, `--json`, or `--force`. + Because import is idempotent (already-imported sessions are skipped), a + session that is still **active when the watcher starts** (captured by the + start-up catch-up) or that **resumes** after it settled is recorded only up + to that point and is not re-imported; run `basou refresh --force` to rebuild + those from the latest logs. diff --git a/packages/cli/src/commands/refresh-watch.test.ts b/packages/cli/src/commands/refresh-watch.test.ts new file mode 100644 index 00000000..6c0b0718 --- /dev/null +++ b/packages/cli/src/commands/refresh-watch.test.ts @@ -0,0 +1,246 @@ +import { execFile } from "node:child_process"; +import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; +import { devNull, tmpdir } from "node:os"; +import { join } from "node:path"; +import { promisify } from "node:util"; +import { + basouPaths, + createManifest, + ensureBasouDirectory, + readSessionYaml, + writeManifest, +} from "@basou/core"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { runRefreshWatch, scanSourceLogs, scansEqual, type WatchDeps } from "./refresh-watch.js"; + +const execFileAsync = promisify(execFile); +const ENV = { ...process.env, GIT_CONFIG_GLOBAL: devNull, GIT_CONFIG_SYSTEM: devNull }; +const FIXED_WS_ID = "ws_01HXABCDEF1234567890ABCDEF" as const; +const FIXED_DATE = new Date("2026-05-09T03:00:00.000Z"); + +let tmpRepo: string | undefined; +let claudeRoot: string | undefined; +let codexRoot: string | undefined; + +beforeEach(async () => { + tmpRepo = await mkdtemp(join(tmpdir(), "basou-watch-test-")); + claudeRoot = await mkdtemp(join(tmpdir(), "basou-watch-claude-")); + codexRoot = await mkdtemp(join(tmpdir(), "basou-watch-codex-")); + await execFileAsync("git", ["-c", "init.defaultBranch=main", "init"], { cwd: tmpRepo, env: ENV }); +}); + +afterEach(async () => { + for (const dir of [tmpRepo, claudeRoot, codexRoot]) { + if (dir !== undefined) await rm(dir, { recursive: true, force: true }); + } + tmpRepo = undefined; + claudeRoot = undefined; + codexRoot = undefined; + vi.restoreAllMocks(); +}); + +function getClaudeRoot(): string { + if (claudeRoot === undefined) throw new Error("claudeRoot not initialized"); + return claudeRoot; +} +function getCodexRoot(): string { + if (codexRoot === undefined) throw new Error("codexRoot not initialized"); + return codexRoot; +} + +async function setupRepo(sourceRoots?: string[]): Promise { + const { realpath } = await import("node:fs/promises"); + const repo = await realpath(tmpRepo as string); + const paths = await ensureBasouDirectory(repo); + await writeManifest( + paths, + createManifest({ + workspaceName: "fixture-ws", + now: FIXED_DATE, + workspaceId: FIXED_WS_ID, + ...(sourceRoots ? { sourceRoots } : {}), + }), + ); + return repo; +} + +async function writeCodexRolloutAt(cwd: string, id: string): Promise { + const dir = join(getCodexRoot(), "2026", "05", "10"); + await mkdir(dir, { recursive: true }); + const records = [ + { + type: "session_meta", + timestamp: "2026-05-10T00:00:00.000Z", + payload: { id, cwd, timestamp: "2026-05-10T00:00:00.000Z" }, + }, + { + type: "response_item", + timestamp: "2026-05-10T00:00:01.000Z", + payload: { + type: "function_call", + name: "exec_command", + arguments: JSON.stringify({ cmd: "ls", workdir: cwd }), + call_id: "c1", + }, + }, + { + type: "response_item", + timestamp: "2026-05-10T00:00:02.000Z", + payload: { + type: "function_call_output", + call_id: "c1", + output: "Wall time: 0.1000 seconds\nProcess exited with code 0\n", + }, + }, + ]; + await writeFile( + join(dir, `rollout-${id}.jsonl`), + records.map((r) => JSON.stringify(r)).join("\n"), + ); +} + +async function codexSessionIds(repo: string): Promise { + const paths = basouPaths(repo); + let dirs: string[]; + try { + dirs = await readdir(paths.sessions); + } catch { + return []; + } + const ids: string[] = []; + for (const dir of dirs) { + const session = await readSessionYaml(paths, dir); + if (session.session.source.kind === "codex-import") { + const ext = session.session.source.external_id; + if (typeof ext === "string") ids.push(ext); + } + } + return ids.sort(); +} + +/** + * A fake `sleep` that runs one scripted step per cycle, then aborts when the + * steps run out (so the watcher loop terminates deterministically). + */ +function scriptedSleep( + controller: AbortController, + steps: Array<() => Promise | void>, +): WatchDeps["sleep"] { + let i = 0; + return async () => { + const step = steps[i]; + i += 1; + if (step) { + await step(); + } else { + controller.abort(); + } + }; +} + +function watchDeps( + repo: string, + controller: AbortController, + sleep: WatchDeps["sleep"], +): WatchDeps { + const logs: string[] = []; + const deps: WatchDeps = { + ctx: { cwd: repo, claudeProjectsDir: getClaudeRoot(), codexSessionsDir: getCodexRoot() }, + paths: basouPaths(repo), + intervalMs: 1000, + importOptions: {}, + now: () => FIXED_DATE, + signal: controller.signal, + sleep, + log: (line) => logs.push(line), + }; + (deps as WatchDeps & { logs: string[] }).logs = logs; + return deps; +} + +describe("scanSourceLogs / scansEqual", () => { + it("collects *.jsonl signatures and tolerates a missing root", async () => { + await writeCodexRolloutAt("/some/proj", "c1"); + const scan = await scanSourceLogs([getCodexRoot(), join(getCodexRoot(), "does-not-exist")]); + expect(scan.size).toBe(1); + expect([...scan.keys()][0]?.endsWith("rollout-c1.jsonl")).toBe(true); + }); + + it("treats a root that is a file (ENOTDIR) as empty, not an error", async () => { + const file = join(getCodexRoot(), "not-a-dir"); + await writeFile(file, "x"); + const scan = await scanSourceLogs([file]); + expect(scan.size).toBe(0); + }); + + it("scansEqual is true for identical scans and false after growth", async () => { + await writeCodexRolloutAt("/some/proj", "c1"); + const a = await scanSourceLogs([getCodexRoot()]); + const b = await scanSourceLogs([getCodexRoot()]); + expect(scansEqual(a, b)).toBe(true); + // Append to the file -> size changes -> not equal. + const file = join(getCodexRoot(), "2026", "05", "10", "rollout-c1.jsonl"); + await writeFile(file, "x".repeat(10), { flag: "a" }); + const c = await scanSourceLogs([getCodexRoot()]); + expect(scansEqual(a, c)).toBe(false); + }); +}); + +describe("runRefreshWatch", () => { + it("imports a session only once its log has settled (stable for one poll)", async () => { + const repo = await setupRepo(); + const controller = new AbortController(); + // Cycle 1: a new rollout appears (not yet stable -> skipped). + // Cycle 2: unchanged (settled) -> imported. Cycle 3: no step -> abort. + const sleep = scriptedSleep(controller, [ + async () => writeCodexRolloutAt(repo, "c-settle"), + // Before the settle cycle: the just-appeared rollout must NOT be imported + // yet (it was not stable for a full poll). This guards against a watcher + // that imports on first sighting. + async () => expect(await codexSessionIds(repo)).toEqual([]), + ]); + const deps = watchDeps(repo, controller, sleep); + + await runRefreshWatch(deps); + + expect(await codexSessionIds(repo)).toEqual(["c-settle"]); + const logs = (deps as WatchDeps & { logs: string[] }).logs; + expect(logs.some((l) => l.includes("refreshed:") && l.includes("codex +1"))).toBe(true); + expect(logs[logs.length - 1]).toBe("watch stopped"); + }); + + it("does not regenerate when a change imports nothing relevant (no churn)", async () => { + // Manifest scopes to the repo only; an unrelated-cwd rollout must not import. + const repo = await setupRepo(["."]); + const controller = new AbortController(); + const sleep = scriptedSleep(controller, [ + async () => writeCodexRolloutAt("/unrelated/project", "c-other"), + () => {}, + ]); + const deps = watchDeps(repo, controller, sleep); + + await runRefreshWatch(deps); + + expect(await codexSessionIds(repo)).toEqual([]); + const logs = (deps as WatchDeps & { logs: string[] }).logs; + // Only the initial catch-up regenerates (and logs one "refreshed:" line). + // The steady-state cycle triggered by the unrelated change imported 0, so it + // regenerated nothing and added NO further "refreshed:" line (no churn). + const refreshed = logs.filter((l) => l.includes("refreshed:")); + expect(refreshed).toHaveLength(1); + expect(logs[logs.length - 1]).toBe("watch stopped"); + }); + + it("exits cleanly on the first abort with a 'watch stopped' line", async () => { + const repo = await setupRepo(); + const controller = new AbortController(); + const sleep = scriptedSleep(controller, []); // abort on the first sleep + const deps = watchDeps(repo, controller, sleep); + + await runRefreshWatch(deps); + + const logs = (deps as WatchDeps & { logs: string[] }).logs; + expect(logs[0]).toContain("watching"); + expect(logs[logs.length - 1]).toBe("watch stopped"); + }); +}); diff --git a/packages/cli/src/commands/refresh-watch.ts b/packages/cli/src/commands/refresh-watch.ts new file mode 100644 index 00000000..beee4a95 --- /dev/null +++ b/packages/cli/src/commands/refresh-watch.ts @@ -0,0 +1,200 @@ +import type { Dirent } from "node:fs"; +import { readdir, stat } from "node:fs/promises"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { type BasouPaths, findErrorCode } from "@basou/core"; +import { + type ImportOutcome, + importClaudeCode, + importCodex, + type RefreshActionOptions, + regenerateDecisions, + regenerateHandoff, +} from "../lib/provenance-actions.js"; +import type { ImportContext } from "./import.js"; + +/** Default poll interval for `--watch`, in seconds. */ +export const DEFAULT_WATCH_INTERVAL_SEC = 30; +/** Smallest accepted `--interval`, in seconds. */ +export const MIN_WATCH_INTERVAL_SEC = 5; +/** Largest accepted `--interval`, in seconds (1 day; keeps the timer well within the 32-bit ms range). */ +export const MAX_WATCH_INTERVAL_SEC = 86_400; + +/** A file's change signature: a refresh is triggered when this moves. */ +type FileSig = { mtimeMs: number; size: number }; +/** Absolute `*.jsonl` path -> its {mtime, size} signature. */ +export type SourceLogScan = Map; + +/** + * The native-log stores the importers read (Codex rollouts + Claude + * transcripts), resolved from the context or the `~` defaults. These are the + * directories the watcher polls -- a new session is a new/grown `*.jsonl` here. + */ +export function watchedRoots(ctx: ImportContext): string[] { + return [ + ctx.codexSessionsDir ?? join(homedir(), ".codex", "sessions"), + ctx.claudeProjectsDir ?? join(homedir(), ".claude", "projects"), + ]; +} + +/** + * Recursively collect a `{mtime, size}` signature of every `*.jsonl` under the + * given roots, keyed by absolute path. A missing root contributes nothing (it + * may appear later); a file that vanishes mid-walk is skipped. No file content + * is read, so this is cheap to run every poll. + */ +export async function scanSourceLogs(roots: string[]): Promise { + const out: SourceLogScan = new Map(); + const walk = async (dir: string): Promise => { + let entries: Dirent[]; + try { + entries = await readdir(dir, { withFileTypes: true }); + } catch (error: unknown) { + // Absent (ENOENT) or not a directory (ENOTDIR): nothing to scan here. + if (findErrorCode(error, "ENOENT") || findErrorCode(error, "ENOTDIR")) return; + // Surface other errors pathlessly (the native message carries the path). + throw new Error("Failed to read a source log directory", { cause: error }); + } + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + await walk(full); + } else if (entry.isFile() && entry.name.endsWith(".jsonl")) { + try { + const info = await stat(full); + out.set(full, { mtimeMs: info.mtimeMs, size: info.size }); + } catch (error: unknown) { + if (findErrorCode(error, "ENOENT")) continue; // vanished mid-walk; skip + throw new Error("Failed to stat a source log file", { cause: error }); + } + } + } + }; + for (const root of roots) await walk(root); + return out; +} + +/** Whether two scans describe the same set of files at the same size/mtime. */ +export function scansEqual(a: SourceLogScan, b: SourceLogScan): boolean { + if (a.size !== b.size) return false; + for (const [path, sig] of a) { + const other = b.get(path); + if (other === undefined || other.mtimeMs !== sig.mtimeMs || other.size !== sig.size) { + return false; + } + } + return true; +} + +function importedCount(outcome: ImportOutcome): number { + return outcome.status === "ran" ? outcome.importedCount : 0; +} + +function describeOutcome(outcome: ImportOutcome): string { + return outcome.status === "ran" + ? `${outcome.adapter} +${outcome.importedCount}` + : `${outcome.adapter} skipped`; +} + +function hms(date: Date): string { + return date.toISOString().slice(11, 19); +} + +/** Dependencies for {@link runRefreshWatch}; timers / clock / signal are injectable for tests. */ +export type WatchDeps = { + ctx: ImportContext; + paths: BasouPaths; + intervalMs: number; + /** Import options forwarded to each cycle (project source roots only; no force / dry-run in watch). */ + importOptions: RefreshActionOptions; + now: () => Date; + signal: AbortSignal; + /** Resolves after `ms`, or early when `signal` aborts. */ + sleep: (ms: number, signal: AbortSignal) => Promise; + log: (line: string) => void; +}; + +/** Import both adapters for the workspace's source roots; returns the outcomes + total imported. */ +async function runImports( + deps: WatchDeps, +): Promise<{ claude: ImportOutcome; codex: ImportOutcome; imported: number }> { + const claude = await importClaudeCode(deps.importOptions, deps.ctx); + const codex = await importCodex(deps.importOptions, deps.ctx); + return { claude, codex, imported: importedCount(claude) + importedCount(codex) }; +} + +/** Regenerate handoff + decisions; returns the handoff session count. */ +async function regenerate(deps: WatchDeps): Promise { + const nowIso = deps.now().toISOString(); + const handoff = await regenerateHandoff(deps.paths, nowIso); + await regenerateDecisions(deps.paths, nowIso); + return handoff.sessionCount; +} + +/** + * Poll the native-log stores and keep the workspace current. Does an initial + * catch-up refresh, then on each interval re-imports ONLY when the logs are + * quiescent (unchanged since the previous poll, so no session is mid-write) AND + * have changed since the last import. Handoff / decisions regenerate only when + * something was imported, so unrelated AI activity elsewhere never rewrites this + * workspace's files. A failure inside a steady-state cycle is logged and the + * loop continues; the initial refresh failing is fatal (it propagates). Returns + * when `signal` aborts (after the in-flight cycle, never mid-write). + */ +export async function runRefreshWatch(deps: WatchDeps): Promise { + const { intervalMs, ctx, signal, sleep, log } = deps; + const roots = watchedRoots(ctx); + log( + `watching ${roots.join(", ")} every ${Math.round(intervalMs / 1000)}s ` + + "(imports on change; Ctrl-C to stop)", + ); + + // Baseline BEFORE the initial import, so a session that appears during the + // import window is not mistaken for "already seen" and missed forever. + let lastScan = await scanSourceLogs(roots); + let importedScan = lastScan; + + // Initial catch-up: failure here is fatal (propagates to the caller). + const initial = await runImports(deps); + const initialSessions = await regenerate(deps); + log( + `[${hms(deps.now())}] refreshed: ${describeOutcome(initial.codex)}, ` + + `${describeOutcome(initial.claude)} (sessions: ${initialSessions})`, + ); + if (signal.aborted) { + log("watch stopped"); + return; + } + + // Set when an import succeeded but the matching regenerate has not yet (so a + // regenerate failure cannot leave handoff / decisions stale forever). + let pendingRegen = false; + while (!signal.aborted) { + await sleep(intervalMs, signal); + if (signal.aborted) break; + try { + const current = await scanSourceLogs(roots); + // Quiescent since the previous poll AND changed since the last import. + if (scansEqual(current, lastScan) && !scansEqual(current, importedScan)) { + const { claude, codex, imported } = await runImports(deps); + if (imported > 0) pendingRegen = true; + if (pendingRegen) { + const sessions = await regenerate(deps); + pendingRegen = false; + log( + `[${hms(deps.now())}] refreshed: ${describeOutcome(codex)}, ` + + `${describeOutcome(claude)} (sessions: ${sessions})`, + ); + } + importedScan = current; + } + lastScan = current; + } catch (error: unknown) { + // A transient fs error must not kill a long-running watcher. Messages from + // the scan / import / render layers are pathless by contract. + const message = error instanceof Error ? error.message : String(error); + log(`[${hms(deps.now())}] refresh cycle skipped: ${message}`); + } + } + log("watch stopped"); +} diff --git a/packages/cli/src/commands/refresh.test.ts b/packages/cli/src/commands/refresh.test.ts index 18fa402f..d21800f7 100644 --- a/packages/cli/src/commands/refresh.test.ts +++ b/packages/cli/src/commands/refresh.test.ts @@ -5,7 +5,7 @@ import { dirname, join } from "node:path"; import { promisify } from "node:util"; import { basouPaths, createManifest, ensureBasouDirectory, writeManifest } from "@basou/core"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { doRunRefresh, runRefresh } from "./refresh.js"; +import { doRunRefresh, doRunRefreshWatch, parseInterval, runRefresh } from "./refresh.js"; const execFileAsync = promisify(execFile); @@ -249,3 +249,27 @@ describe("basou refresh", () => { expect(errSpy.mock.calls.flat().join(" ")).toContain("Workspace not initialized"); }); }); + +describe("basou refresh --watch (validation)", () => { + it("rejects --watch combined with --dry-run / --json / --force", async () => { + const repo = await setupInitedRepo(); + await expect(doRunRefreshWatch({ watch: true, dryRun: true }, ctxFor(repo))).rejects.toThrow( + /--watch cannot be combined with --dry-run/, + ); + await expect(doRunRefreshWatch({ watch: true, json: true }, ctxFor(repo))).rejects.toThrow( + /--watch cannot be combined with --json/, + ); + await expect(doRunRefreshWatch({ watch: true, force: true }, ctxFor(repo))).rejects.toThrow( + /--watch cannot be combined with --force/, + ); + }); + + it("parseInterval accepts in-range integers and rejects out-of-range / non-integers", () => { + expect(parseInterval("30")).toBe(30); + expect(parseInterval("5")).toBe(5); + expect(parseInterval("86400")).toBe(86400); + for (const bad of ["4", "0", "-1", "2.5", "abc", "", "999999999"]) { + expect(() => parseInterval(bad)).toThrow(); + } + }); +}); diff --git a/packages/cli/src/commands/refresh.ts b/packages/cli/src/commands/refresh.ts index 173bb006..04ca14b2 100644 --- a/packages/cli/src/commands/refresh.ts +++ b/packages/cli/src/commands/refresh.ts @@ -1,14 +1,24 @@ import { assertBasouRootSafe, basouPaths, findErrorCode, resolveRepositoryRoot } from "@basou/core"; -import type { Command } from "commander"; +import { type Command, InvalidArgumentError } from "commander"; import { isVerbose, renderCliError } from "../lib/error-render.js"; import { type ImportOutcome, type RefreshResult, refreshAll } from "../lib/provenance-actions.js"; import type { ImportContext } from "./import.js"; +import { + DEFAULT_WATCH_INTERVAL_SEC, + MAX_WATCH_INTERVAL_SEC, + MIN_WATCH_INTERVAL_SEC, + runRefreshWatch, +} from "./refresh-watch.js"; export type RefreshOptions = { project?: string[]; force?: boolean; dryRun?: boolean; json?: boolean; + /** Run as a long-lived watcher: re-import when the native logs change. */ + watch?: boolean; + /** Poll interval in seconds for `--watch` (default {@link DEFAULT_WATCH_INTERVAL_SEC}). */ + interval?: number; verbose?: boolean; }; @@ -17,6 +27,41 @@ function collectPath(value: string, previous: string[]): string[] { return [...previous, value]; } +/** Commander parser: `--interval` is an integer count of seconds within the supported range. */ +export function parseInterval(value: string): number { + const seconds = Number(value); + if ( + !Number.isInteger(seconds) || + seconds < MIN_WATCH_INTERVAL_SEC || + seconds > MAX_WATCH_INTERVAL_SEC + ) { + throw new InvalidArgumentError( + `--interval must be an integer between ${MIN_WATCH_INTERVAL_SEC} and ${MAX_WATCH_INTERVAL_SEC} (seconds).`, + ); + } + return seconds; +} + +/** Resolve after `ms`, or early when `signal` aborts (e.g. on Ctrl-C). Leaves no listener behind. */ +function abortableSleep(ms: number, signal: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal.aborted) { + resolve(); + return; + } + let timer: ReturnType; + const onAbort = (): void => { + clearTimeout(timer); + resolve(); + }; + timer = setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, ms); + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + export type RefreshContext = ImportContext & { /** Defaults to `() => new Date()`. Injectable for tests. */ nowProvider?: () => Date; @@ -42,6 +87,15 @@ export function registerRefreshCommand(program: Command): void { .option("--force", "Re-import sessions already imported instead of skipping") .option("--dry-run", "Preview imports and skip writing handoff / decisions") .option("--json", "Output the result as JSON") + .option( + "--watch", + "Keep running: re-import + regenerate when the native logs change (Ctrl-C to stop)", + ) + .option( + "--interval ", + `Poll interval for --watch, in seconds (default ${DEFAULT_WATCH_INTERVAL_SEC}, min ${MIN_WATCH_INTERVAL_SEC})`, + parseInterval, + ) .option("-v, --verbose", "Show error causes") .action(async (options: RefreshOptions) => { await runRefresh(options); @@ -54,13 +108,61 @@ export function registerRefreshCommand(program: Command): void { */ export async function runRefresh(options: RefreshOptions, ctx: RefreshContext = {}): Promise { try { - await doRunRefresh(options, ctx); + if (options.watch === true) { + await doRunRefreshWatch(options, ctx); + } else { + await doRunRefresh(options, ctx); + } } catch (error: unknown) { renderCliError(error, { verbose: isVerbose(options) }); process.exitCode = 1; } } +/** + * `basou refresh --watch`: resolve + validate, then run the polling watcher + * until SIGINT / SIGTERM. Startup failures (bad combo, no workspace, failed + * initial refresh) propagate and exit non-zero; a steady-state cycle failure is + * logged inside the loop and the watcher keeps running. + */ +export async function doRunRefreshWatch( + options: RefreshOptions, + ctx: RefreshContext, +): Promise { + if (options.dryRun === true) throw new Error("--watch cannot be combined with --dry-run."); + if (options.json === true) throw new Error("--watch cannot be combined with --json."); + if (options.force === true) throw new Error("--watch cannot be combined with --force."); + + const cwd = ctx.cwd ?? process.cwd(); + const repositoryRoot = await resolveRepositoryRootForRefresh(cwd); + const paths = basouPaths(repositoryRoot); + await assertWorkspaceInitialized(paths.root); + + const intervalMs = (options.interval ?? DEFAULT_WATCH_INTERVAL_SEC) * 1000; + const controller = new AbortController(); + const onSignal = (): void => controller.abort(); + process.on("SIGINT", onSignal); + process.on("SIGTERM", onSignal); + try { + await runRefreshWatch({ + ctx, + paths, + intervalMs, + importOptions: + options.project !== undefined && options.project.length > 0 + ? { project: options.project } + : {}, + now: () => ctx.nowProvider?.() ?? new Date(), + signal: controller.signal, + sleep: abortableSleep, + log: (line) => console.log(line), + }); + } finally { + process.off("SIGINT", onSignal); + process.off("SIGTERM", onSignal); + } +} + /** * Pure runner: resolves the workspace, runs the shared refresh pipeline, and * prints a summary (or JSON). Returns the {@link RefreshResult} so the same