diff --git a/docs/code/story-points-estimation.md b/docs/code/story-points-estimation.md index 8934aff..fb03f1f 100644 --- a/docs/code/story-points-estimation.md +++ b/docs/code/story-points-estimation.md @@ -122,6 +122,8 @@ query = "sprint in openSprints() AND \"Story Points\" is EMPTY" Each entry needs a unique `id`, boolean `enabled`, a non-empty `query`, and exactly one of `cron` or `interval`. There is no `prompt` and no `repo`: a due entry runs one-shot `devintern --estimate --query ""`, which estimates — never implements, branches, or opens PRs. Estimating does not depend on `[defaults].task_query`; an omitted or empty `[[estimations]]` table simply means estimation is off. +You can add, remove, enable, disable, reschedule, or change the query of an estimation entry while the worker is running. The worker validates and reconciles the edit automatically without interrupting an in-progress sweep. + Notes: - Works with Jira, Linear, Azure DevOps, Asana, and GitHub (comment-only). Trello/markdown workspaces fail at startup. diff --git a/docs/code/worker.md b/docs/code/worker.md index a8e5e83..c2c26cf 100644 --- a/docs/code/worker.md +++ b/docs/code/worker.md @@ -60,7 +60,7 @@ Do not change production code.""" Every entry needs a stable unique `id`, boolean `enabled`, non-empty `prompt`, and exactly one schedule. Intervals use positive minutes, hours, or days (`15m`, `6h`, `1d`). Cron expressions have five fields and use the worker host's timezone in v1; persisted occurrence times are UTC. -Configuration is validated as a group at worker startup and changes require a restart. Automations are a valid event source, so `devintern worker` stays running without a task query when at least one automation entry is configured (disabled entries are validated but not scheduled). +Configuration is validated on load; while the worker runs it revalidates edits to `workspace.toml` automatically (SIGHUP forces a reload) — see [Workspaces → Editing workspace.toml while running](./workspaces.md#editing-workspace.toml-while-running). Automations are a valid event source, so `devintern worker` stays running without a task query when at least one automation entry is configured (disabled entries are validated but not scheduled). ### What an automation is @@ -101,7 +101,7 @@ On shutdown the scheduler stops its timer, terminates active automation subproce | Symptom | Likely cause | | ---------------------------- | ------------------------------------------------------------ | -| No occurrences fire after editing the TOML | Config is loaded at startup — restart the worker. Startup validation errors name the offending entry. | +| No occurrences fire after editing the TOML | Check the worker log: the reload logs validation errors naming the offending entry, and changing a schedule resets its cursor (the next run is the next scheduled time, not immediately). | | `occurrence skipped: previous run is active` | The previous occurrence still runs (or its lease is stale). Long prompts may simply need a longer schedule. | | `occurrence skipped: repository is busy` | Another task holds the repo run lock; the next occurrence will retry. | | Scheduled runs missing from the dashboard | Filter the run list by origin `scheduled`; check the worker has an automation license (startup log). | @@ -129,6 +129,8 @@ query = "sprint in openSprints() AND \"Story Points\" is EMPTY" Each entry needs a unique `id`, boolean `enabled`, non-empty `query`, and exactly one of `cron` or `interval`. Omitting the table (or leaving every entry disabled) changes nothing: estimation is simply off, and `[defaults].task_query` is never estimated as a side effect. The workspace tracker must support estimation (Jira, Linear, Azure DevOps, Asana, GitHub comment-only); Trello/markdown workspaces fail at startup with a clear error. `worker init` does not ask about estimations — add tables by hand. +Estimation entries live-reload with `workspace.toml`: added and re-enabled entries schedule their next future occurrence, changed schedules reset their cursor, query edits apply to the next sweep, and removed entries stop scheduling without interrupting a sweep already in progress. + When an entry comes due the worker runs one-shot `devintern --estimate --query ""` from the workspace home. That path keeps all of the interactive behavior: tickets younger than 24 hours are skipped, already-estimated tickets are skipped unless the ticket changed since the estimate, changed tickets are re-estimated with the estimate comment updated in place, points are written to the tracker field, and usage-limit aborts exit cleanly so the next occurrence retries. Each sweep is recorded with its own `estimate` origin (plus the schedule id) in the [dashboard](./dashboard.md) — it never shows up as a scheduled implement run. ### Serialization @@ -250,7 +252,7 @@ conflict_resolution_cron = "0 3 * * *" # or conflict_resolution_interval = "1d In scheduled mode the poller still detects every conflict on the first tick it appears and queues it durably (a pending base-sync event), but the agent is not invoked. When the scheduled window arrives — cron uses the worker host timezone, intervals are relative — the worker resolves all queued conflicts in one pass and logs the active mode and next window at startup. The window stays open for a grace period (`WORKER_RESOLVE_WINDOW_GRACE_MINUTES`, default 60) so quiet-period waits and retry backoffs inside the pass can still complete; anything unresolved when it closes waits for the next window. A window that arrives while the worker is down (missed nightly run) catches up on the first tick after restart. Stale resolutions cannot happen: before invoking the agent the worker re-fetches the PR, and closed/merged PRs or a conflict that resolved itself are dropped from the queue without spending tokens. The manual `devintern resolve-conflicts ` command always works on demand, and once GitHub reports a PR conflict-free its queued event never triggers an agent run. -The setting applies to the whole workspace and takes effect on worker restart, like the rest of `workspace.toml`. Between windows a conflicted PR cannot be merged, so teams that rely on instant rebases should keep `auto`. See [Workspaces → Automatic conflict resolution](./workspaces.md#automatic-conflict-resolution-auto-vs-scheduled-vs-disabled) for the config reference and tradeoffs. +The setting applies to the whole workspace and live-reloads with `workspace.toml`. Between windows a conflicted PR cannot be merged, so teams that rely on instant rebases should keep `auto`. See [Workspaces → Automatic conflict resolution](./workspaces.md#automatic-conflict-resolution-auto-vs-scheduled-vs-disabled) for the config reference and tradeoffs. To turn automatic conflict resolution off entirely — no detection, no queuing, no agent runs — set `conflict_resolution = "disabled"`: conflicted PRs stay conflicted until resolved by hand or via `devintern resolve-conflicts `. diff --git a/docs/code/workspaces.md b/docs/code/workspaces.md index ec6f5ca..fb190e9 100644 --- a/docs/code/workspaces.md +++ b/docs/code/workspaces.md @@ -103,7 +103,7 @@ The schedule uses the same format as `[[automations]]`: a five-field cron expres Two things are never delayed by scheduled mode: review feedback on the agent's PRs is addressed immediately as usual, and you can always run `devintern resolve-conflicts ` by hand to fix one PR without waiting for the window — once GitHub reports the PR conflict-free, the queued event never triggers an agent run. -The setting is workspace-wide (per-repo overrides are not supported in v1) and, like the rest of `workspace.toml`, requires a worker restart to take effect. The tradeoff to keep in mind: between windows a conflicted PR cannot be merged, so on fast-moving branches where an instant rebase unblocks a waiting reviewer, `auto` stays the better choice. See [Worker Daemon → Merge conflicts on the agent's PRs](./worker.md#merge-conflicts-on-the-agents-prs) for how resolution itself works. +The setting is workspace-wide (per-repo overrides are not supported in v1) and live-reloads with the rest of the runtime configuration. The tradeoff to keep in mind: between windows a conflicted PR cannot be merged, so on fast-moving branches where an instant rebase unblocks a waiting reviewer, `auto` stays the better choice. See [Worker Daemon → Merge conflicts on the agent's PRs](./worker.md#merge-conflicts-on-the-agents-prs) for how resolution itself works. Set `conflict_resolution = "disabled"` to turn automatic conflict resolution off entirely: the worker stops watching for conflicts on the agent's PRs altogether — no detection, no queuing, no agent runs. A PR that conflicts with its base simply stays conflicted until someone resolves it (by hand, or on demand via `devintern resolve-conflicts `). Review feedback and @mention handling are unaffected. This is a valid choice when the team prefers to rebase manually, or when the agent is not trusted to resolve conflicts in a sensitive repository. @@ -150,7 +150,16 @@ devintern worker # auto-detects ~/.devintern/workspace.toml devintern worker --workspace /path/to/workspace.toml ``` -The fleet query comes from `[defaults].task_query`. A workspace with automations or estimations can omit the query and run as a schedules-only worker. Poll interval, per-task flags, and the embedded dashboard are also set in `workspace.toml` (`poll_interval`, `worker_task_args`, `[workspace].dashboard` / `dashboard_port`). Direct webhooks are an advanced repo-local service: run `devintern webhook serve` from that repository as a separate process. Workspace, automation, and estimation configuration is loaded at startup; restart the worker after editing it. Schedule state and leases for automations and estimations live in the central workspace database. +The fleet query comes from `[defaults].task_query`. A workspace with automations or estimations can omit the query and run as a schedules-only worker. Poll interval, per-task flags, and the embedded dashboard are also set in `workspace.toml` (`poll_interval`, `worker_task_args`, `[workspace].dashboard` / `dashboard_port`). Direct webhooks are an advanced repo-local service: run `devintern webhook serve` from that repository as a separate process. Schedule state and leases for automations and estimations live in the central workspace database. + +### Editing workspace.toml while running + +The worker watches `workspace.toml` and reloads it automatically a moment after you save — no restart, and no missed tracker events or relay messages during the bounce: + +- **Routing rules, repos, `task_query`, `[[automations]]`, `[[estimations]]`, `worker_task_args`, `poll_interval`, `worktrees_ttl_days`, and conflict-resolution mode/schedules apply to subsequent work.** Runs already in progress finish under the configuration they started with; everything picked up afterwards uses the new one. Changing a repo's `remote` updates its managed bare clone the next time that repo is prepared. +- **A broken edit never takes the daemon down.** The reload validates the file first; parse or schema errors are logged (naming the offending entries) and the last valid configuration keeps serving until you fix it. Rewriting identical content is ignored. +- **Manual fallback:** send SIGHUP (`kill -HUP `) to force an immediate reload if file watching is unavailable on your system. +- **Startup-only settings** still require a restart: tracker credentials in the workspace `.env` and `[defaults].tracker` (the tracker client and its detector are built once), plus `[workspace].dashboard` / `dashboard_port`. A reload that changes one of these settings is rejected in full, so the active config remains internally consistent. `devintern worker init` can generate a user-level systemd unit on Linux or launchd agent on macOS. One service runs the whole workspace. For a hand-written Linux unit: diff --git a/packages/code/src/lib/automation-acquirer.ts b/packages/code/src/lib/automation-acquirer.ts index fb63797..d100004 100644 --- a/packages/code/src/lib/automation-acquirer.ts +++ b/packages/code/src/lib/automation-acquirer.ts @@ -42,8 +42,11 @@ export interface SpawnedAutomationRun { export interface AutomationAcquirerOptions { automations: AutomationConfig[]; dbPath: string; - /** Per-task CLI flags; defaults to `--create-pr`. */ - extraArgs?: string[]; + /** + * Per-task CLI flags; defaults to `--create-pr`. May be a factory so live + * config reloads (e.g. `[defaults].worker_task_args`) apply to later runs. + */ + extraArgs?: string[] | (() => string[]); resolveContext: (automation: AutomationConfig) => Promise; now?: () => number; spawnRun?: (automation: AutomationConfig, context: AutomationRunContext) => SpawnedAutomationRun; @@ -127,6 +130,59 @@ export class AutomationAcquirer implements Acquirer { } } + /** + * Replace the automation set at runtime (live workspace.toml reload). + * + * Reconciles durable schedule state without interrupting active work: + * + * - Newly added or rescheduled entries register (a changed schedule resets + * the cursor; an unchanged one retains the prior interval anchor). + * - Entries removed from the config retire: removed-but-active runs finish + * their current occurrence naturally, then their schedule state is + * dropped. Fully removed inactive entries are unregistered immediately. + * - Entries merely disabled keep their schedule rows so re-enabling later + * keeps the anchor, matching restart semantics. + */ + applyAutomations(next: AutomationConfig[]): void { + const previous = this.options.automations; + this.options.automations = next; + + const now = this.now(); + for (const automation of next.filter((item) => item.enabled)) { + try { + this.store.register( + automation, + nextAutomationDue(automation, now), + this.stateId(automation), + ); + } catch (error) { + console.error( + `❌ [${this.jobKind()}:${automation.id}] invalid schedule after reload: ` + + `${(error as Error).message}`, + ); + } + } + + for (const retired of previous.filter((item) => !next.some((n) => n.id === item.id))) { + const active = this.active.get(retired.id); + if (active) { + // Let the in-flight occurrence run to completion; drop its state only + // after it ends and only if the id was not re-added in the meantime. + void active.lifecycle + .catch(() => undefined) + .then(() => { + if (!this.options.automations.some((item) => item.id === retired.id)) { + this.store.unregister(this.stateId(retired)); + } + }); + } else { + this.store.unregister(this.stateId(retired)); + } + } + + this.scheduleNext(); + } + private async runTick(): Promise { const leaseMs = this.options.leaseMs ?? LEASE_MS; const kind = this.jobKind(); @@ -199,14 +255,12 @@ export class AutomationAcquirer implements Acquirer { continue; } console.log(`\n⏰ [${kind}:${automation.id}] starting scheduled run`); + const extraArgs = this.options.extraArgs; + const args = + typeof extraArgs === "function" ? extraArgs() : (extraArgs ?? workerTaskArgs()); const run = this.options.spawnRun ? this.options.spawnRun(automation, context) - : defaultSpawnRun( - automation, - context, - this.options.extraArgs ?? workerTaskArgs(), - this.options.terminationGraceMs, - ); + : defaultSpawnRun(automation, context, args, this.options.terminationGraceMs); const active: ActiveAutomationRun = { run, lifecycle: Promise.resolve(), diff --git a/packages/code/src/lib/automation-state.ts b/packages/code/src/lib/automation-state.ts index e6a66bb..6d6a16c 100644 --- a/packages/code/src/lib/automation-state.ts +++ b/packages/code/src/lib/automation-state.ts @@ -55,6 +55,16 @@ export class AutomationStateStore { } } + /** + * Remove schedule state for an automation retired by a live config reload + * (removed from the config entirely). Rows for entries still present in + * the config — even disabled — are kept so re-enabling retains the + * interval anchor. + */ + unregister(automationId: string): void { + this.db.run("DELETE FROM automation_schedules WHERE automation_id = ?", [automationId]); + } + get(automationId: string): AutomationScheduleState | null { const row = this.db .query("SELECT * FROM automation_schedules WHERE automation_id = ?") diff --git a/packages/code/src/lib/estimation-acquirer.ts b/packages/code/src/lib/estimation-acquirer.ts index 97c2f61..da64c53 100644 --- a/packages/code/src/lib/estimation-acquirer.ts +++ b/packages/code/src/lib/estimation-acquirer.ts @@ -84,23 +84,16 @@ export function spawnEstimationSweep( */ export class EstimationAcquirer extends AutomationAcquirer { constructor(options: EstimationAcquirerOptions) { - const byId = new Map(options.estimations.map((item) => [item.id, item])); - // The scheduler's config shape requires a prompt; scheduled estimates - // always use the sweep runner below, so it stays empty and unread. - const schedules: AutomationConfig[] = options.estimations.map((item) => ({ - ...item, - prompt: "", - })); + const schedules = estimationSchedules(options.estimations); super({ automations: schedules, dbPath: options.dbPath, name: "scheduled-estimations", jobKind: "estimation", stateId: (schedule) => `${STATE_PREFIX}${schedule.id}`, - resolveContext: (schedule) => - options.resolveContext(byId.get(schedule.id) as EstimationConfig), + resolveContext: (schedule) => options.resolveContext(schedule as ScheduledEstimation), spawnRun: (schedule, context) => { - const estimation = byId.get(schedule.id) as EstimationConfig; + const estimation = schedule as ScheduledEstimation; return options.spawnRun ? options.spawnRun(estimation, context) : spawnEstimationSweep(estimation, context); @@ -115,4 +108,16 @@ export class EstimationAcquirer extends AutomationAcquirer { clearInterval: options.clearInterval, }); } + + /** Reconcile scheduled estimation entries after a live workspace reload. */ + applyEstimations(estimations: EstimationConfig[]): void { + this.applyAutomations(estimationSchedules(estimations)); + } +} + +type ScheduledEstimation = AutomationConfig & EstimationConfig; + +/** Adapt estimation bodies to the shared durable scheduler's config shape. */ +function estimationSchedules(estimations: EstimationConfig[]): ScheduledEstimation[] { + return estimations.map((item) => ({ ...item, prompt: "" })); } diff --git a/packages/code/src/lib/mention-sweep-acquirer.ts b/packages/code/src/lib/mention-sweep-acquirer.ts index 348bb5b..ecb109b 100644 --- a/packages/code/src/lib/mention-sweep-acquirer.ts +++ b/packages/code/src/lib/mention-sweep-acquirer.ts @@ -141,6 +141,7 @@ export class MentionSweepAcquirer implements Acquirer { /** Start sweeping: immediate first tick, then on the configured interval. */ async start(): Promise { + if (this.timer) return; console.log( `🔎 Sweeping ${this.options.repo} for @mentions every ${this.options.intervalSeconds}s`, ); @@ -156,6 +157,17 @@ export class MentionSweepAcquirer implements Acquirer { } } + /** + * Apply a new sweep cadence without restarting (live workspace config + * reload). Re-arms the repeating timer with the new interval. + */ + updateInterval(intervalSeconds: number): void { + this.options.intervalSeconds = intervalSeconds; + if (!this.timer) return; + clearInterval(this.timer); + this.timer = setInterval(() => void this.tick(), intervalSeconds * 1000); + } + /** One repo-wide sweep over both comment feeds. Skipped while busy. */ async tick(): Promise { if (this.busy) { diff --git a/packages/code/src/lib/review-polling-acquirer.ts b/packages/code/src/lib/review-polling-acquirer.ts index 5fe51bc..07e67a1 100644 --- a/packages/code/src/lib/review-polling-acquirer.ts +++ b/packages/code/src/lib/review-polling-acquirer.ts @@ -104,9 +104,10 @@ export interface ReviewPollingAcquirerOptions { * Repo slugs (`owner/repo`) this worker manages. Open registry rows for * any other repo (e.g. left behind after a rename/transfer) are * auto-unwatched at startup and skipped on every tick. Omit to watch the - * whole registry (previous behavior). + * whole registry (previous behavior). May be a factory so live config + * reloads (repos added to `workspace.toml`) apply on later ticks. */ - allowedRepos?: string[]; + allowedRepos?: string[] | (() => string[]); verbose?: boolean; /** * Schedule gating automatic conflict resolution (workspace scheduled @@ -432,11 +433,12 @@ export class ReviewPollingAcquirer implements Acquirer { /** Start polling: immediate first tick, then on the configured interval. */ async start(): Promise { + if (this.timer) return; // Drop stale registry rows for repos this worker no longer manages // (e.g. after a rename/transfer) so they never hit the API again. - const { allowedRepos, workerState } = this.options; + const allowedRepos = this.resolveAllowedRepos(); if (allowedRepos && allowedRepos.length > 0) { - for (const pr of workerState.closeForeignAgentPrs(allowedRepos)) { + for (const pr of this.options.workerState.closeForeignAgentPrs(allowedRepos)) { console.log( `🧹 [${this.name}] ${pr.repo}#${pr.prNumber} is not part of this project; unwatching`, ); @@ -460,6 +462,35 @@ export class ReviewPollingAcquirer implements Acquirer { } } + /** + * Apply a new poll cadence without restarting (live workspace config + * reload). Re-arms the repeating timer with the new interval. + */ + updateInterval(intervalSeconds: number): void { + this.options.intervalSeconds = intervalSeconds; + if (!this.timer) return; + clearInterval(this.timer); + this.timer = setInterval(() => void this.tick(), intervalSeconds * 1000); + } + + /** Apply conflict-resolution mode and schedule changes from workspace reloads. */ + updateConflictResolution( + conflictResolution: ConflictResolutionMode, + conflictSchedule?: CronOrIntervalSchedule, + ): void { + this.options.conflictResolution = conflictResolution; + this.options.conflictSchedule = conflictSchedule; + this.conflictWindowState = null; + this.deferCounts.clear(); + this.syncConflictWindow(); + this.logConflictMode(); + } + + private resolveAllowedRepos(): string[] | undefined { + const raw = this.options.allowedRepos; + return typeof raw === "function" ? raw() : raw; + } + /** One polling cycle over all watched PRs. Skipped while busy. */ async tick(): Promise { if (this.busy) { @@ -469,7 +500,8 @@ export class ReviewPollingAcquirer implements Acquirer { try { this.syncConflictWindow(); - const { workerState, allowedRepos } = this.options; + const { workerState } = this.options; + const allowedRepos = this.resolveAllowedRepos(); // Reconcile the registry with GitHub first: one conditional GET per // watched PR whose result is shared with the poll loop below, so PRs @@ -626,7 +658,7 @@ export class ReviewPollingAcquirer implements Acquirer { private async maybeSyncBase(repo: string, prNumber: number, pr: PolledPr): Promise { const { github, queue, resolveConflicts, runStore } = this.options; - if (!resolveConflicts) return; + if (!resolveConflicts || this.options.conflictResolution === "disabled") return; if (!pr.head?.sha || !pr.base?.sha || !pr.head.ref) return; // The event key includes the head SHA so that new commits on the branch @@ -792,6 +824,7 @@ export class ReviewPollingAcquirer implements Acquirer { * window opens so pending conflicts resolve on the following ticks. */ private syncConflictWindow(): void { + if (this.options.conflictResolution === "disabled") return; const schedule = this.options.conflictSchedule; if (!schedule) return; const now = this.now(); diff --git a/packages/code/src/lib/run-coordinator.ts b/packages/code/src/lib/run-coordinator.ts index 97babda..b51bd53 100644 --- a/packages/code/src/lib/run-coordinator.ts +++ b/packages/code/src/lib/run-coordinator.ts @@ -11,9 +11,27 @@ */ export class RunCoordinator { private tail: Promise = Promise.resolve(); + private transition = 0; + + constructor(private enabled = true) {} + + /** Enable serialization for newly started runs. */ + enable(): void { + this.transition += 1; + this.enabled = true; + } + + /** Disable after already-held and queued slots drain. */ + disableWhenIdle(): void { + const transition = ++this.transition; + void this.tail.then(() => { + if (this.transition === transition) this.enabled = false; + }); + } /** Wait for a free slot; returns the release function for that slot. */ async acquire(): Promise<() => void> { + if (!this.enabled) return () => undefined; let release!: () => void; const turn = new Promise((resolve) => { release = resolve; diff --git a/packages/code/src/lib/task-polling-acquirer.ts b/packages/code/src/lib/task-polling-acquirer.ts index b11db32..24060be 100644 --- a/packages/code/src/lib/task-polling-acquirer.ts +++ b/packages/code/src/lib/task-polling-acquirer.ts @@ -43,7 +43,7 @@ function hasUpdateStamp(task: ReadyTask): boolean { export interface TaskPollingAcquirerOptions { trackerType: string; /** The user's task-selection query (same language as `--query`). */ - query: string; + query: string | (() => string | undefined); intervalSeconds: number; detector: ChangeDetector; workerState: WorkerState; @@ -107,9 +107,11 @@ export class TaskPollingAcquirer implements Acquirer { /** Start polling: immediate first tick, then on the configured interval. */ async start(): Promise { + if (this.timer) return; + const query = this.resolveQuery(); console.log( `🔎 Polling ${this.options.trackerType} every ${this.options.intervalSeconds}s ` + - `(query: ${this.options.query})`, + `(query: ${query ?? "disabled until task_query is configured"})`, ); await this.tick(); this.timer = setInterval(() => void this.tick(), this.options.intervalSeconds * 1000); @@ -123,14 +125,27 @@ export class TaskPollingAcquirer implements Acquirer { } } + /** + * Apply a new poll cadence without restarting (live workspace config + * reload). Re-arms the repeating timer with the new interval. + */ + updateInterval(intervalSeconds: number): void { + this.options.intervalSeconds = intervalSeconds; + if (!this.timer) return; + clearInterval(this.timer); + this.timer = setInterval(() => void this.tick(), intervalSeconds * 1000); + } + /** One detect → evaluate → dedupe → execute cycle. Skipped while busy. */ async tick(): Promise { if (this.busy) { return; } + const query = this.resolveQuery(); + if (!query) return; this.busy = true; - const { detector, workerState, queue, query, searchTasks, executeTask, verbose } = this.options; + const { detector, workerState, queue, searchTasks, executeTask, verbose } = this.options; try { const cursor = workerState.getCursor(detector.source)?.cursorValue ?? null; const detection = await detector.changesSince(cursor); @@ -190,6 +205,11 @@ export class TaskPollingAcquirer implements Acquirer { } } + private resolveQuery(): string | undefined { + const raw = this.options.query; + return typeof raw === "function" ? raw() : raw; + } + /** * Always-on skip/stamp diagnosis. Silent skips made "ticket matches query * but was never picked up" undebuggable from the worker log. diff --git a/packages/code/src/lib/workspace/config-reload.ts b/packages/code/src/lib/workspace/config-reload.ts new file mode 100644 index 0000000..1d96792 --- /dev/null +++ b/packages/code/src/lib/workspace/config-reload.ts @@ -0,0 +1,217 @@ +/** + * Live configuration reload for the fleet worker. + * + * The worker is an unattended daemon: tuning `workspace.toml` (routing + * rules, repos, automations, cadence) must not require restarting it, or + * tracker events and relay messages get missed during the bounce. + * + * Design: every consumer holds a reference to one shared `WorkspaceConfig` + * instance for the lifetime of the process. {@link applyWorkspaceConfig} + * swaps that instance's top-level fields in place on each successful reload, + * so routing decisions, repo lookups, per-repo environments, and relay + * handling read the current config without any rewiring. Consumers that + * snapshot values at construction (the automation scheduler, polling + * intervals) reconcile through the {@link WorkspaceConfigReloaderOptions.onApplied} + * callback. + * + * Failure policy: a malformed or semantically invalid file is logged and + * ignored — the worker keeps serving with the last valid configuration and + * never crashes on a bad edit. + */ + +import { watch } from "fs"; +import { basename, dirname } from "path"; + +import type { WorkspaceConfig } from "./config"; +import { loadWorkspaceConfig } from "./config"; + +/** Coalescing window for rapid successive edits before reloads run. */ +export const DEFAULT_RELOAD_DEBOUNCE_MS = 300; + +/** + * Swap {@linkcode target}'s top-level sections in place so every holder of + * the original object observes the updated configuration. Fields are + * assigned synchronously (no awaits), so no reader sees a half-applied set. + */ +export function applyWorkspaceConfig(target: WorkspaceConfig, next: WorkspaceConfig): void { + target.workspace = next.workspace; + target.defaults = next.defaults; + target.repos = next.repos; + target.routing = next.routing; + target.automations = next.automations; + target.estimations = next.estimations; +} + +/** Deterministic serialization used to skip no-op saves/touches. */ +export function serializeWorkspaceConfig(config: WorkspaceConfig): string { + return JSON.stringify(config); +} + +export interface ReloadOutcome { + /** True when a new configuration was parsed, validated, and applied. */ + applied: boolean; + /** True when disk content parsed but matched the already-active config. */ + unchanged?: boolean; +} + +export interface WorkspaceConfigReloaderOptions { + /** Absolute path of the watched `workspace.toml`. */ + configPath: string; + /** + * The live config instance the worker consumers share; mutated in place + * by each applied reload (see {@link applyWorkspaceConfig}). + */ + current: WorkspaceConfig; + /** Parse/validate step (injected for tests). Defaults to loadWorkspaceConfig. */ + load?: (path: string) => WorkspaceConfig; + /** Coalescing window in milliseconds for rapid successive edits. */ + debounceMs?: number; + /** Called after a changed config was applied (reconcile acquirers here). */ + onApplied?: (config: WorkspaceConfig) => void; + /** Final runtime validation before the shared config is mutated. */ + validate?: (next: WorkspaceConfig, current: WorkspaceConfig) => void; + /** Error sink (injected for tests). Defaults to console.error. */ + onError?: (message: string) => void; +} + +/** + * Watches `workspace.toml` and applies validated edits to the running + * worker. + * + * - File watching follows the config's directory (editors replace files via + * rename, which does not fire on the inode being watched). + * - Rapid successive edits are coalesced (trailing-edge debounce). + * - Invalid edits are surfaced through the error sink and dropped; the last + * valid configuration keeps serving. + * - `SIGHUP` triggers a manual reload as a fallback when file watching is + * unavailable (e.g. inotify limits). + */ +export class WorkspaceConfigReloader { + private readonly options: Required> & + WorkspaceConfigReloaderOptions; + private watcher: ReturnType | null = null; + private debounceTimer: ReturnType | null = null; + private signalHandler: (() => void) | null = null; + private stopped = false; + + constructor(options: WorkspaceConfigReloaderOptions) { + this.options = { ...options, debounceMs: options.debounceMs ?? DEFAULT_RELOAD_DEBOUNCE_MS }; + } + + /** Start watching the config directory and install the SIGHUP fallback. */ + start(): void { + if (this.stopped) return; + this.installSignalFallback(); + try { + this.watcher = watch(dirname(this.options.configPath), (_event, filename) => { + if (!filename || filename === basename(this.options.configPath)) { + this.scheduleReload("file change"); + } + }); + this.watcher.on("error", (error) => { + this.reportError( + `workspace.toml watcher failed (${(error as Error).message}); ` + + "send SIGHUP to force a config reload.", + ); + this.closeWatcher(); + }); + // Catch up on an edit that lands while the watcher is attaching (an + // edit racing the initial registration would otherwise be missed); + // identical content is filtered out by the no-op check. + this.scheduleReload("watcher attached"); + } catch (error) { + this.reportError( + `Could not watch ${this.options.configPath} for changes ` + + `(${(error as Error).message}); send SIGHUP to reload after editing.`, + ); + } + } + + /** Stop watching and clear pending reloads. */ + stop(): void { + this.stopped = true; + if (this.debounceTimer) clearTimeout(this.debounceTimer); + this.debounceTimer = null; + this.closeWatcher(); + } + + /** + * Read, validate, and apply the on-disk configuration. + * + * @param reason - Human-readable trigger for log messages ("SIGHUP", + * "file change", …). + * @returns Whether a new config was applied, and whether it was a no-op. + */ + reload(reason: string): ReloadOutcome { + let next: WorkspaceConfig; + try { + next = (this.options.load ?? loadWorkspaceConfig)(this.options.configPath); + } catch (error) { + this.reportError( + `Failed to reload workspace config (${reason}): ` + + `${(error as Error).message}\n` + + " Continuing with the previously loaded configuration.", + ); + return { applied: false }; + } + + if (serializeWorkspaceConfig(next) === serializeWorkspaceConfig(this.options.current)) { + return { applied: false, unchanged: true }; + } + + try { + this.options.validate?.(next, this.options.current); + } catch (error) { + this.reportError( + `Failed to reload workspace config (${reason}): ${(error as Error).message}\n` + + " Continuing with the previously loaded configuration.", + ); + return { applied: false }; + } + + applyWorkspaceConfig(this.options.current, next); + console.log( + `🔄 [config] Reloaded ${this.options.configPath} ` + + `(${next.repos.length} repo(s), ${next.routing.length} routing rule(s), ` + + `${next.automations.length} automation(s), ${next.estimations.length} estimation(s))`, + ); + this.options.onApplied?.(this.options.current); + return { applied: true }; + } + + /** Queue a debounced reload (coalesces rapid successive edits). */ + scheduleReload(reason: string): void { + if (this.stopped) return; + if (this.debounceTimer) clearTimeout(this.debounceTimer); + this.debounceTimer = setTimeout(() => { + this.debounceTimer = null; + this.reload(reason); + }, this.options.debounceMs); + } + + /** + * Manual reload fallback for environments where file watching is + * unavailable. Installed once per process (the daemon creates a single + * reloader; the handler outlives stop(), which only tears down the watcher). + */ + private installSignalFallback(): void { + if (process.listenerCount("SIGHUP") > 0 || this.signalHandler) return; + this.signalHandler = () => this.reload("SIGHUP"); + process.on("SIGHUP" as const, this.signalHandler as (signal: NodeJS.Signals) => void); + } + + private closeWatcher(): void { + if (!this.watcher) return; + try { + this.watcher.close(); + } catch { + // Already closed. + } + this.watcher = null; + } + + private reportError(message: string): void { + const onError = this.options.onError ?? ((text: string) => console.error(text)); + onError(message); + } +} diff --git a/packages/code/src/lib/workspace/fleet-events.ts b/packages/code/src/lib/workspace/fleet-events.ts index 57be37f..4418d98 100644 --- a/packages/code/src/lib/workspace/fleet-events.ts +++ b/packages/code/src/lib/workspace/fleet-events.ts @@ -164,13 +164,21 @@ export function createFleetMentionHandler( * executor when it is ready. */ export function createFleetTaskEvaluator(options: { - query: string; + query: string | (() => string | undefined); searchTasks: (query: string) => Promise<{ tasks: FleetTask[] }>; execute: ReturnType; verbose?: boolean; }): (taskKey: string) => Promise { return async (taskKey) => { - const { tasks } = await options.searchTasks(options.query); + const rawQuery = options.query; + const query = typeof rawQuery === "function" ? rawQuery() : rawQuery; + if (!query) { + if (options.verbose) { + console.log(` [fleet] task ${taskKey} changed but task_query is not configured.`); + } + return; + } + const { tasks } = await options.searchTasks(query); const task = tasks.find((candidate) => candidate.key === taskKey); if (!task) { if (options.verbose) { diff --git a/packages/code/src/lib/workspace/init.ts b/packages/code/src/lib/workspace/init.ts index d120d25..250adfa 100644 --- a/packages/code/src/lib/workspace/init.ts +++ b/packages/code/src/lib/workspace/init.ts @@ -33,7 +33,7 @@ dashboard = true # "auto" (default) resolves as soon as a conflict is detected; "scheduled" # queues conflicts during polling and resolves them in one off-peak window # to cut AI token spend. Requires exactly one schedule below, and a worker -# restart to take effect. "disabled" turns it off entirely — conflicts +# changes apply to the running worker. "disabled" turns it off entirely — conflicts # stay for manual resolution (devintern resolve-conflicts ). # conflict_resolution = "scheduled" # conflict_resolution_cron = "0 3 * * *" # worker host timezone @@ -65,8 +65,9 @@ default_branch = "main" # project = "BACK" # task key prefix (BACK-123) # labels = ["backend"] # any-of; AND-ed with the other criteria -# Recurring work is loaded once when the worker starts. Each occurrence runs -# the prompt through the normal task pipeline as a local markdown task. +# Recurring work is hot-reloaded: edits apply to the running worker without a +# restart. Each occurrence runs the prompt through the normal task pipeline as +# a local markdown task. # Cron uses the worker host timezone; interval values support m, h, and d. # # [[automations]] diff --git a/packages/code/src/lib/workspace/repo-manager.ts b/packages/code/src/lib/workspace/repo-manager.ts index d9a29e4..b682a34 100644 --- a/packages/code/src/lib/workspace/repo-manager.ts +++ b/packages/code/src/lib/workspace/repo-manager.ts @@ -59,6 +59,26 @@ export class RepoManager { async ensureBareClone(repo: RepoConfig): Promise { const clonePath = this.bareClonePath(repo.name); if (existsSync(clonePath)) { + const currentRemote = await Utils.executeGitCommand(["remote", "get-url", "origin"], { + cwd: clonePath, + }); + if (!currentRemote.success) { + throw new Error( + `Failed to read origin for repo "${repo.name}": ${currentRemote.error || currentRemote.output}`, + ); + } + if (currentRemote.output.trim() !== repo.remote) { + const updateRemote = await Utils.executeGitCommand( + ["remote", "set-url", "origin", repo.remote], + { cwd: clonePath }, + ); + if (!updateRemote.success) { + throw new Error( + `Failed to update origin for repo "${repo.name}": ${updateRemote.error || updateRemote.output}`, + ); + } + console.log(`🔄 [config] updated origin for repo ${repo.name}`); + } return clonePath; } diff --git a/packages/code/src/lib/workspace/workspace-worker.ts b/packages/code/src/lib/workspace/workspace-worker.ts index cb1164c..ce69c5b 100644 --- a/packages/code/src/lib/workspace/workspace-worker.ts +++ b/packages/code/src/lib/workspace/workspace-worker.ts @@ -39,6 +39,7 @@ import { } from "./paths"; import { routeTask, toRoutableTask } from "./router"; import type { RoutableTask } from "./router"; +import { WorkspaceConfigReloader } from "./config-reload"; import { createRepoRunLock, createWorkspaceLock, openWorkspaceState } from "./state"; import type { RoutingSkipStore } from "./state"; import { BASE_WORKTREE_NAME, RepoManager } from "./repo-manager"; @@ -167,7 +168,7 @@ export interface WorkspaceTaskAcquirerDeps { repoManager: RepoManagerLike; detector: ChangeDetector; searchTasks: (query: string) => Promise<{ tasks: FleetTask[] }>; - query: string; + query: string | (() => string | undefined); intervalSeconds: number; verbose?: boolean; /** Task runner (injected for tests; defaults to the CLI subprocess). */ @@ -308,6 +309,43 @@ export async function warnOnPushAuthIssues( } } +/** + * Enabled-and-shape-valid automations for the current fleet, plus any + * semantic problems. Shared by worker startup (problems are fatal) and the + * live-reload path (problems surface as errors; offending entries do not + * schedule, so a repo-less automation can never run outside every repo). + */ +export function resolveFleetAutomations(config: WorkspaceConfig): { + automations: AutomationConfig[]; + problems: string[]; +} { + const problems: string[] = []; + const fleetAutomations: AutomationConfig[] = []; + for (const automation of config.automations) { + if (!automation.repo && config.repos.length !== 1) { + problems.push( + `Automation "${automation.id}" must set repo when the workspace has multiple repositories.`, + ); + continue; + } + fleetAutomations.push(automation); + } + return { automations: fleetAutomations, problems }; +} + +/** + * Reconciliation hooks exposed to the live config reload path by the fleet + * event wiring (see {@linkcode buildFleetEventAcquirers}). + */ +export interface FleetEventReloadHooks { + /** Re-run mention-sweep reconciliation against the live config's repos. */ + reconcileMentionSweeps(): void; + /** Apply live conflict-resolution mode and schedule settings. */ + reconcileConflictResolution(): void; + /** Slugs currently served by a mention sweep (sorted). */ + mentionSweepRepos(): string[]; +} + /** * Build the fleet task acquirer: detect-then-evaluate (reusing * {@link TaskPollingAcquirer}) with routing between evaluate and execute. @@ -380,14 +418,21 @@ export type FleetExecutorDeps = Pick< */ export function createFleetTaskExecutor( deps: FleetExecutorDeps, - options: { extraArgs?: string[] } = {}, + options: { extraArgs?: string[] | (() => string[]) } = {}, ): (taskKey: string, routable: RoutableTask) => Promise { const { config, workspaceDir, skips, repoManager } = deps; const runTask = deps.runTask ?? runTaskViaCli; const repoLock = deps.repoLock ?? ((name: string) => createRepoRunLock(name, workspaceDir)); - const extraArgs = options.extraArgs ?? fleetTaskArgs(config); return async (taskKey, routable) => { + // Read per run: live config reloads must apply to subsequent work. + // Explicit overrides can also be factories (dashboard retries prepend + // `--force` while still following live worker_task_args). + const configuredArgs = options.extraArgs; + const extraArgs = + typeof configuredArgs === "function" + ? configuredArgs() + : (configuredArgs ?? fleetTaskArgs(config)); const decision = routeTask(routable, config); if (decision.kind !== "routed") { @@ -491,13 +536,15 @@ export async function sweepAllWorktrees( * never keeps the process alive on its own. */ export function startWorktreeSweeper( - repos: RepoConfig[], + repos: RepoConfig[] | (() => RepoConfig[]), repoManager: RepoManagerLike, - ttlDays: number, + ttlDays: number | (() => number), intervalMs: number = WORKTREE_SWEEP_INTERVAL_MS, ): ReturnType { const timer = setInterval(() => { - sweepAllWorktrees(repos, repoManager, ttlDays).catch((error) => + const activeRepos = typeof repos === "function" ? repos() : repos; + const activeTtlDays = typeof ttlDays === "function" ? ttlDays() : ttlDays; + sweepAllWorktrees(activeRepos, repoManager, activeTtlDays).catch((error) => console.warn(`⚠️ [fleet] periodic worktree sweep failed: ${(error as Error).message}`), ); }, intervalMs); @@ -546,14 +593,18 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr // In-process consumers (dashboard, run records) follow the fleet DB. process.env.WEBHOOK_QUEUE_DB = workspaceDbPath(workspaceDir); - const query = config.defaults.taskQuery; + const initialQuery = config.defaults.taskQuery; const intervalSeconds = config.defaults.pollIntervalSeconds; + const initialFleetAutomations = resolveFleetAutomations(config); + if (initialFleetAutomations.problems.length > 0) { + throw new Error(`Invalid ${configPath}:\n- ${initialFleetAutomations.problems.join("\n- ")}`); + } // Dashboard retries ride the shared workspace DB: the dashboard inserts a // pending row, this worker drains it through the fleet executor below. const retryQueue = new ScheduledRetryStore(workspaceDbPath(workspaceDir)); - if (!query && config.automations.length === 0 && config.estimations.length === 0) { + if (!initialQuery && config.automations.length === 0 && config.estimations.length === 0) { if (retryQueue.hasPending()) { console.warn( "⚠️ No task query or automations configured; the worker will only drain scheduled dashboard retries.", @@ -572,9 +623,8 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr // absent or fully disabled. The account-global gate is needed only once an // enabled schedule joins the process and must serialize with every other // agent run. - const coordinator = config.estimations.some((item) => item.enabled) - ? new RunCoordinator() - : undefined; + const coordinator = new RunCoordinator(false); + if (config.estimations.some((item) => item.enabled)) coordinator.enable(); // Recover what the previous worker left behind before acquiring new work. await recoverOrphanedWorkspaceRuns({ @@ -587,7 +637,11 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr // Keep sweeping while the worker runs, not only at startup: a long-lived // worker would otherwise accumulate worktrees that age past the TTL until // the next restart. - startWorktreeSweeper(config.repos, repoManager, config.workspace.worktreesTtlDays); + startWorktreeSweeper( + () => config.repos, + repoManager, + () => config.workspace.worktreesTtlDays, + ); await warnOnPushAuthIssues(config, repoManager); @@ -602,103 +656,151 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr { config, workspaceDir, skips: state.skips, repoManager }, // `--force` bypasses the incomplete-attempt retry gate, exactly like // the manual `devintern --force` the dashboard action mirrors. - { extraArgs: ["--force", ...fleetTaskArgs(config)] }, + { extraArgs: () => ["--force", ...fleetTaskArgs(config)] }, ), intervalSeconds: parseEnvInteger("WORKER_RETRY_INTERVAL_SECONDS", 5, { min: 1 }), verbose: options.verbose, }), ); - if (config.automations.length > 0) { - const semanticErrors: string[] = []; - for (const automation of config.automations) { - if (!automation.repo && config.repos.length !== 1) { - semanticErrors.push( - `Automation "${automation.id}" must set repo when the workspace has multiple repositories.`, - ); - } - } - if (semanticErrors.length > 0) { - throw new Error(`Invalid ${configPath}:\n- ${semanticErrors.join("\n- ")}`); - } - - acquirers.push( - new AutomationAcquirer({ - automations: config.automations, - dbPath: state.dbPath, - extraArgs: fleetTaskArgs(config), - resolveContext: async (automation) => - withCoordinatorSlot( - await resolveWorkspaceAutomationContext(automation, config, workspaceDir, repoManager), - coordinator, - ), - }), - ); - } - - if (config.estimations.length > 0) { - // Scheduled story-point sweeps: the automation scheduler (durable cursors - // and leases), a one-shot search of each entry's query, and the regular - // `--estimate` engine. No repo, no worktree, no branch, no PR. - console.log( - `📊 Scheduling ${config.estimations.filter((item) => item.enabled).length} enabled estimation schedule(s)`, - ); - acquirers.push( - new EstimationAcquirer({ - estimations: config.estimations, - dbPath: state.dbPath, - resolveContext: () => - withCoordinatorSlot( - Promise.resolve({ cwd: workspaceDir, env: { ...process.env }, release() {} }), - coordinator, - ), - }), - ); - } - - if (query) { - const { TaskTrackerManager } = await import("../task-tracker-manager"); - const { createChangeDetector } = await import("../change-detector"); - const tracker = new TaskTrackerManager().getClient(); - const detector = createChangeDetector(config.defaults.tracker, (q) => tracker.searchTasks(q)); - if (!detector) { - console.error( - `❌ Could not initialize the ${config.defaults.tracker} change detector. ` + - "Check the tracker's required variables in the workspace .env.", - ); - process.exit(1); - } - acquirers.push( - createWorkspaceTaskAcquirer({ - config, - workspaceDir, - workerState: state.workerState, - queue: state.queue, - skips: state.skips, - repoManager, - detector, - searchTasks: (q) => tracker.searchTasks(q), - query, - intervalSeconds, - verbose: options.verbose, + // Always assembled (even with no automations yet): a live reload can add + // [[automations]] without restarting, and applyAutomations schedules them. + // Semantic problems were already rejected at startup above and are + // re-checked on the reload path via resolveFleetAutomations. + const fleetAutomationAcquirer = new AutomationAcquirer({ + automations: initialFleetAutomations.automations, + dbPath: state.dbPath, + extraArgs: () => fleetTaskArgs(config), + resolveContext: async (automation) => + withCoordinatorSlot( + await resolveWorkspaceAutomationContext(automation, config, workspaceDir, repoManager), coordinator, - }), + ), + }); + acquirers.push(fleetAutomationAcquirer); + + // Cadence reconciliation: applied on successful reloads of poll_interval. + const intervalUpdaters: Array<(seconds: number) => void> = []; + // Reload hooks published by buildFleetEventAcquirers (mention sweeps). + const eventReloadHooks: { hooks?: FleetEventReloadHooks } = {}; + let pollIntervalSeconds = config.defaults.pollIntervalSeconds; + + // Always assemble the estimation scheduler so entries can be added to an + // already-running schedules-only worker. + const estimationAcquirer = new EstimationAcquirer({ + estimations: config.estimations, + dbPath: state.dbPath, + resolveContext: () => + withCoordinatorSlot( + Promise.resolve({ cwd: workspaceDir, env: { ...process.env }, release() {} }), + coordinator, + ), + }); + acquirers.push(estimationAcquirer); + + // The tracker type is startup-only, but task_query itself is live. Keep a + // dormant poller when the detector prerequisites are available so adding + // a query needs no restart. The tracker client itself remains lazy, which + // preserves automation-only workspaces without tracker credentials. + const { TaskTrackerManager } = await import("../task-tracker-manager"); + const { createChangeDetector } = await import("../change-detector"); + const trackerManager = new TaskTrackerManager(); + const searchTasks = (q: string) => trackerManager.getClient().searchTasks(q); + const detector = createChangeDetector(config.defaults.tracker, searchTasks); + if (initialQuery && !detector) { + console.error( + `❌ Could not initialize the ${config.defaults.tracker} change detector. ` + + "Check the tracker's required variables in the workspace .env.", ); + process.exit(1); + } + if (detector) { + const taskAcquirer = createWorkspaceTaskAcquirer({ + config, + workspaceDir, + workerState: state.workerState, + queue: state.queue, + skips: state.skips, + repoManager, + detector, + searchTasks, + query: () => config.defaults.taskQuery, + intervalSeconds, + verbose: options.verbose, + coordinator, + }); + intervalUpdaters.push((seconds) => taskAcquirer.updateInterval(seconds)); + acquirers.push(taskAcquirer); acquirers.push( ...(await buildFleetEventAcquirers({ config, workspaceDir, state, repoManager, - searchTasks: (q) => tracker.searchTasks(q), - query, + searchTasks, + query: () => config.defaults.taskQuery, intervalSeconds, verbose: options.verbose, + intervalUpdaters, + reloadHooksOut: eventReloadHooks, coordinator, })), ); } + /** + * Apply a freshly validated config to consumers that snapshot values: + * reconciles the automation set, surfaces semantic problems, and refreshes + * cadence-driven acquirers when `[defaults].poll_interval` changed. + */ + const applyReloadedConfig = (updated: WorkspaceConfig): void => { + const fleet = resolveFleetAutomations(updated); + fleetAutomationAcquirer.applyAutomations(fleet.automations); + if (updated.estimations.some((item) => item.enabled)) coordinator.enable(); + else coordinator.disableWhenIdle(); + estimationAcquirer.applyEstimations(updated.estimations); + + if (updated.defaults.pollIntervalSeconds !== pollIntervalSeconds) { + pollIntervalSeconds = updated.defaults.pollIntervalSeconds; + console.log(`⏱️ [config] Poll interval is now ${pollIntervalSeconds}s`); + for (const update of intervalUpdaters) update(pollIntervalSeconds); + } + eventReloadHooks.hooks?.reconcileMentionSweeps(); + eventReloadHooks.hooks?.reconcileConflictResolution(); + }; + + // Live reload: watch workspace.toml and apply validated edits in place. + // Routing rules, repos, automations, worker_task_args, and poll_interval + // take effect without a restart; malformed edits keep the last-good + // config. SIGHUP forces a manual reload as a fallback. + const reloader = new WorkspaceConfigReloader({ + configPath, + current: config, + validate: (next, current) => { + const fleet = resolveFleetAutomations(next); + if (fleet.problems.length > 0) throw new Error(fleet.problems.join("\n- ")); + if (next.defaults.tracker !== current.defaults.tracker) { + throw new Error("[defaults].tracker is startup-only; restart the worker to change it."); + } + if (next.defaults.taskQuery && !detector) { + throw new Error( + `task_query cannot be enabled live because the ${current.defaults.tracker} change detector ` + + "could not be initialized; fix its required workspace .env settings and restart the worker.", + ); + } + if ( + next.workspace.dashboard !== current.workspace.dashboard || + next.workspace.dashboardPort !== current.workspace.dashboardPort + ) { + throw new Error( + "[workspace].dashboard and dashboard_port are startup-only; restart the worker to change them.", + ); + } + }, + onApplied: applyReloadedConfig, + }); + reloader.start(); + if (config.workspace.dashboard) { try { const { startDashboardServer } = await import("../../dashboard-server"); @@ -713,6 +815,9 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr } console.log(`🗂️ Workspace: ${configPath} (${config.repos.length} repo(s))`); + console.log( + "🔄 Live config reload armed: edits to workspace.toml apply automatically (SIGHUP forces one)", + ); const { startWorker } = await import("../../worker"); await startWorker( { @@ -751,13 +856,19 @@ export async function buildFleetEventAcquirers(options: { state: ReturnType; repoManager: RepoManagerLike; searchTasks: (query: string) => Promise<{ tasks: FleetTask[] }>; - query: string; + query: string | (() => string | undefined); intervalSeconds: number; verbose?: boolean; + /** Collectors of cadence changes, applied on live config reloads. */ + intervalUpdaters?: Array<(seconds: number) => void>; + /** Published once event acquirers are wired (mention-sweep reconcile). */ + reloadHooksOut?: { hooks?: FleetEventReloadHooks }; + /** Process-level agent-run gate; only set when scheduled estimation exists. */ coordinator?: RunCoordinator; }): Promise { const { config, workspaceDir, state, repoManager, searchTasks, query, intervalSeconds, verbose } = options; + const intervalUpdaters = options.intervalUpdaters ?? []; const acquirers: import("../../worker").Acquirer[] = []; const { @@ -776,7 +887,9 @@ export async function buildFleetEventAcquirers(options: { | ((repo: string, comment: { user: { login: string } }, prNumber: number) => Promise) | undefined; - if (hasGitHubCreds && slugs.length > 0) { + // Built whenever credentials exist — even with zero GitHub repos today — + // so a repo added to the config at runtime gets full event coverage. + if (hasGitHubCreds) { const { GitHubReviewsClient } = await import("../github-reviews"); github = new GitHubReviewsClient({ preferAppAuth: true }); const gh = github; @@ -803,129 +916,169 @@ export async function buildFleetEventAcquirers(options: { const { ReviewPollingAcquirer } = await import("../review-polling-acquirer"); const { isGitHubNotFound } = await import("../github-reviews"); const runStore = new RunStore(state.dbPath); - acquirers.push( - new ReviewPollingAcquirer({ + const reviewAcquirer = new ReviewPollingAcquirer({ + intervalSeconds, + workerState: state.workerState, + queue: state.queue, + github: { + fetchPr: async (repo, n, etag) => { + try { + return await gh.conditionalGet( + `/repos/${repo}/pulls/${n}`, + ownerOf(repo), + nameOf(repo), + etag, + ); + } catch (error) { + if (isGitHubNotFound(error)) { + // Renamed/transferred/deleted repo or PR (or lost App + // access): report gone so the reconciler unregisters the + // row instead of erroring on every tick. + return { data: null, notModified: false, gone: true }; + } + throw error; + } + }, + fetchReviews: (repo, n, etag) => + gh.conditionalGet( + `/repos/${repo}/pulls/${n}/reviews?per_page=100`, + ownerOf(repo), + nameOf(repo), + etag, + ), + fetchReviewCommentsSince: async (repo, n, sinceIso) => { + const result = await gh.conditionalGet< + Array<{ id: number; user: { login: string; type: string }; created_at: string }> + >( + `/repos/${repo}/pulls/${n}/comments?since=${encodeURIComponent(sinceIso)}&per_page=100`, + ownerOf(repo), + nameOf(repo), + ); + return result.data ?? []; + }, + }, + addressPr: fleetAddressPr, + resolveConflicts, + conflictSchedule: config.workspace.conflictSchedule, + conflictResolution: config.workspace.conflictResolution, + quietPeriodSeconds: parseEnvInteger("WORKER_BASE_SYNC_QUIET_SECONDS", 30, { min: 0 }), + runStore, + // Factory form: repos added at runtime become watchable without a + // restart (a static list would pin the startup slug set). + allowedRepos: () => fleetGitHubSlugs(config), + verbose, + }); + acquirers.push(reviewAcquirer); + + // Tier 2: one mention sweep per GitHub repo (cursor sources are already + // namespaced by slug). The permission gate runs in the fleet handler. + // Sweeps are map-managed so live config reloads can attach sweeps for + // newly added repos and stop them for removed ones. + const { MentionSweepAcquirer } = await import("../mention-sweep-acquirer"); + type MentionSweep = import("../mention-sweep-acquirer").MentionSweepAcquirer; + const mentionSweeps = new Map(); + const createMentionSweep = (slug: string): MentionSweep => { + const [repoOwner, repoName] = slug.split("/") as [string, string]; + return new MentionSweepAcquirer({ + repo: slug, intervalSeconds, workerState: state.workerState, queue: state.queue, github: { - fetchPr: async (repo, n, etag) => { - try { - return await gh.conditionalGet( - `/repos/${repo}/pulls/${n}`, - ownerOf(repo), - nameOf(repo), - etag, - ); - } catch (error) { - if (isGitHubNotFound(error)) { - // Renamed/transferred/deleted repo or PR (or lost App - // access): report gone so the reconciler unregisters the - // row instead of erroring on every tick. - return { data: null, notModified: false, gone: true }; - } - throw error; - } + fetchIssueCommentsSince: async (sinceIso) => { + const result = await gh.conditionalGet< + Array<{ + id: number; + body: string | null; + user: { login: string; type: string }; + created_at: string; + html_url: string; + issue_url?: string; + }> + >( + `/repos/${slug}/issues/comments?since=${encodeURIComponent(sinceIso)}&per_page=100&sort=created&direction=asc`, + repoOwner, + repoName, + ); + return result.data ?? []; }, - fetchReviews: (repo, n, etag) => - gh.conditionalGet( - `/repos/${repo}/pulls/${n}/reviews?per_page=100`, - ownerOf(repo), - nameOf(repo), - etag, - ), - fetchReviewCommentsSince: async (repo, n, sinceIso) => { + fetchReviewCommentsSince: async (sinceIso) => { const result = await gh.conditionalGet< - Array<{ id: number; user: { login: string; type: string }; created_at: string }> + Array<{ + id: number; + body: string | null; + user: { login: string; type: string }; + created_at: string; + html_url: string; + pull_request_url?: string; + }> >( - `/repos/${repo}/pulls/${n}/comments?since=${encodeURIComponent(sinceIso)}&per_page=100`, - ownerOf(repo), - nameOf(repo), + `/repos/${slug}/pulls/comments?since=${encodeURIComponent(sinceIso)}&per_page=100&sort=created&direction=asc`, + repoOwner, + repoName, ); return result.data ?? []; }, + getBotUsername: () => gh.getBotUsername(repoOwner, repoName), + getPr: async (prNumber) => { + const pr = await gh.getPullRequest(repoOwner, repoName, prNumber); + return { + number: pr.number, + state: pr.state, + headRepoFullName: pr.head.repo?.full_name, + maintainerCanModify: pr.maintainer_can_modify, + }; + }, + postComment: async (prNumber, body) => { + await gh.postPullRequestComment(repoOwner, repoName, prNumber, body); + }, }, - addressPr: fleetAddressPr, - resolveConflicts: - config.workspace.conflictResolution === "disabled" ? undefined : resolveConflicts, - conflictSchedule: config.workspace.conflictSchedule, - conflictResolution: config.workspace.conflictResolution, - quietPeriodSeconds: parseEnvInteger("WORKER_BASE_SYNC_QUIET_SECONDS", 30, { min: 0 }), - runStore, - allowedRepos: slugs, + handleMention: (comment, prNumber) => fleetHandleMention(slug, comment, prNumber), verbose, - }), - ); - - // Tier 2: one mention sweep per GitHub repo (cursor sources are already - // namespaced by slug). The permission gate runs in the fleet handler. - const { MentionSweepAcquirer } = await import("../mention-sweep-acquirer"); + }); + }; for (const slug of slugs) { - const [repoOwner, repoName] = slug.split("/") as [string, string]; - acquirers.push( - new MentionSweepAcquirer({ - repo: slug, - intervalSeconds, - workerState: state.workerState, - queue: state.queue, - github: { - fetchIssueCommentsSince: async (sinceIso) => { - const result = await gh.conditionalGet< - Array<{ - id: number; - body: string | null; - user: { login: string; type: string }; - created_at: string; - html_url: string; - issue_url?: string; - }> - >( - `/repos/${slug}/issues/comments?since=${encodeURIComponent(sinceIso)}&per_page=100&sort=created&direction=asc`, - repoOwner, - repoName, - ); - return result.data ?? []; - }, - fetchReviewCommentsSince: async (sinceIso) => { - const result = await gh.conditionalGet< - Array<{ - id: number; - body: string | null; - user: { login: string; type: string }; - created_at: string; - html_url: string; - pull_request_url?: string; - }> - >( - `/repos/${slug}/pulls/comments?since=${encodeURIComponent(sinceIso)}&per_page=100&sort=created&direction=asc`, - repoOwner, - repoName, - ); - return result.data ?? []; - }, - getBotUsername: () => gh.getBotUsername(repoOwner, repoName), - getPr: async (prNumber) => { - const pr = await gh.getPullRequest(repoOwner, repoName, prNumber); - return { - number: pr.number, - state: pr.state, - headRepoFullName: pr.head.repo?.full_name, - maintainerCanModify: pr.maintainer_can_modify, - }; - }, - postComment: async (prNumber, body) => { - await gh.postPullRequestComment(repoOwner, repoName, prNumber, body); - }, - }, - handleMention: (comment, prNumber) => fleetHandleMention(slug, comment, prNumber), - verbose, - }), - ); + const sweep = createMentionSweep(slug); + mentionSweeps.set(slug, sweep); + acquirers.push(sweep); + intervalUpdaters.push((seconds) => sweep.updateInterval(seconds)); + } + + if (options.reloadHooksOut && !options.reloadHooksOut.hooks) { + options.reloadHooksOut.hooks = { + reconcileMentionSweeps: () => { + const wanted = new Set(fleetGitHubSlugs(config)); + for (const [slug, sweep] of [...mentionSweeps]) { + if (!wanted.has(slug)) { + // A stale updater calling updateInterval on a stopped sweep + // only mutates options (no timer) and is harmless. + sweep.stop(); + mentionSweeps.delete(slug); + console.log(`🧹 [config] stopped @mention sweep for removed repo ${slug}`); + } + } + for (const slug of wanted) { + if (!mentionSweeps.has(slug)) { + const sweep = createMentionSweep(slug); + mentionSweeps.set(slug, sweep); + intervalUpdaters.push((seconds) => sweep.updateInterval(seconds)); + void sweep.start(); + console.log(`➕ [config] watching @mentions on newly added repo ${slug}`); + } + } + }, + reconcileConflictResolution: () => + reviewAcquirer.updateConflictResolution( + config.workspace.conflictResolution, + config.workspace.conflictSchedule, + ), + mentionSweepRepos: () => [...mentionSweeps.keys()].sort(), + }; } } else if (verbose) { console.log( hasGitHubCreds - ? " [fleet] no GitHub repos in the workspace; review/mention acquirers disabled." + ? " [fleet] no GitHub repos configured yet; mention sweeps attach on config changes." : " [fleet] GITHUB_TOKEN/GITHUB_APP_ID not set; review/mention acquirers disabled.", ); } diff --git a/packages/code/tests/automation-acquirer.test.ts b/packages/code/tests/automation-acquirer.test.ts index 43fdfc5..3d92fb1 100644 --- a/packages/code/tests/automation-acquirer.test.ts +++ b/packages/code/tests/automation-acquirer.test.ts @@ -501,4 +501,148 @@ describe("AutomationAcquirer", () => { expect(AUTOMATION_ORIGIN_ENV).toBe("DEVINTERN_RUN_ORIGIN"); expect(AUTOMATION_ID_ENV).toBe("DEVINTERN_AUTOMATION_ID"); }); + + describe("applyAutomations (live config reload)", () => { + const newDb = () => { + const dbPath = join(tmpdir(), `acquirer-${Date.now()}-${Math.random()}.db`); + dbPaths.push(dbPath); + return dbPath; + }; + + test("schedules an automation added at runtime", async () => { + const dbPath = newDb(); + let now = 0; + const acquirer = new AutomationAcquirer({ + automations: [], + dbPath, + now: () => now, + setTimer: () => 1 as unknown as ReturnType, + clearTimer: () => {}, + resolveContext: async () => null, + }); + await acquirer.start(); + const added: AutomationConfig = { + id: "added", + enabled: true, + prompt: "p", + interval: "15m", + intervalMs: 900_000, + }; + acquirer.applyAutomations([added]); + + const store = new AutomationStateStore(dbPath); + expect(store.get("added")?.nextDueAt).toBe(900_000); + store.close(); + await acquirer.stop(); + }); + + test("drops schedule state for a fully removed automation and forgets its timer", async () => { + const dbPath = newDb(); + let now = 0; + let timerCleared = false; + const automation: AutomationConfig = { + id: "retired", + enabled: true, + prompt: "p", + interval: "1h", + intervalMs: 3_600_000, + }; + const acquirer = new AutomationAcquirer({ + automations: [automation], + dbPath, + now: () => now, + setTimer: () => 1 as unknown as ReturnType, + clearTimer: () => { + timerCleared = true; + }, + resolveContext: async () => null, + }); + await acquirer.start(); + + acquirer.applyAutomations([]); + const store = new AutomationStateStore(dbPath); + expect(store.get("retired")).toBeNull(); + store.close(); + expect(timerCleared).toBe(true); + await acquirer.stop(); + }); + + test("lets an active run of a removed automation finish before dropping state", async () => { + const dbPath = newDb(); + let now = 0; + let resolveRun!: (ok: boolean) => void; + const automation: AutomationConfig = { + id: "in-flight", + enabled: true, + prompt: "p", + interval: "10ms", + intervalMs: 10, + }; + const acquirer = new AutomationAcquirer({ + automations: [automation], + dbPath, + now: () => now, + setTimer: () => 1 as unknown as ReturnType, + clearTimer: () => {}, + resolveContext: async () => ({ cwd: "/tmp", env: {}, release() {} }), + spawnRun: () => ({ + completion: new Promise((resolve) => (resolveRun = resolve)), + terminate() {}, + }), + }); + await acquirer.start(); + now = 100; + await acquirer.tick(); + + // Removed from the config mid-run: the occurrence finishes naturally. + acquirer.applyAutomations([]); + const storeWhileRunning = new AutomationStateStore(dbPath); + expect(storeWhileRunning.get("in-flight")).not.toBeNull(); + storeWhileRunning.close(); + + resolveRun(true); + await new Promise((resolve) => setTimeout(resolve, 0)); + await new Promise((resolve) => setTimeout(resolve, 0)); + const storeAfterFinish = new AutomationStateStore(dbPath); + expect(storeAfterFinish.get("in-flight")).toBeNull(); + storeAfterFinish.close(); + await acquirer.stop(); + }); + + test("resets the cursor when the schedule spec changes; keeps it when unchanged", async () => { + const dbPath = newDb(); + const fixedNow = 50_000; + const hourly: AutomationConfig = { + id: "resched", + enabled: true, + prompt: "p", + interval: "1h", + intervalMs: 3_600_000, + }; + const halfHourly: AutomationConfig = { ...hourly, interval: "30m", intervalMs: 1_800_000 }; + const acquirer = new AutomationAcquirer({ + automations: [hourly], + dbPath, + now: () => fixedNow, + setTimer: () => 1 as unknown as ReturnType, + clearTimer: () => {}, + resolveContext: async () => null, + }); + await acquirer.start(); + + const store = new AutomationStateStore(dbPath); + const anchoredAt = store.get("resched")?.nextDueAt; + expect(anchoredAt).toBe(fixedNow + 3_600_000); + + // Unchanged spec (same definition object shape) keeps the anchor. + acquirer.applyAutomations([hourly]); + expect(store.get("resched")?.nextDueAt).toBe(anchoredAt); + + // Changed spec resets the interval anchor. + acquirer.applyAutomations([halfHourly]); + expect(store.get("resched")?.nextDueAt).toBe(fixedNow + 1_800_000); + store.close(); + await acquirer.stop(); + }); + }); }); diff --git a/packages/code/tests/estimation-acquirer.test.ts b/packages/code/tests/estimation-acquirer.test.ts index 140bec1..b9e6442 100644 --- a/packages/code/tests/estimation-acquirer.test.ts +++ b/packages/code/tests/estimation-acquirer.test.ts @@ -154,6 +154,38 @@ describe("EstimationAcquirer", () => { store.close(); }); + test("reconciles entries and updated queries after a live reload", async () => { + const dbPath = newDb(); + let now = 0; + const queries: string[] = []; + const acquirer = new EstimationAcquirer({ + estimations: [], + dbPath, + now: () => now, + ...noopTimers(), + resolveContext: async () => ({ cwd: "/tmp", env: {}, release() {} }), + spawnRun: (estimation) => { + queries.push(estimation.query); + return { completion: Promise.resolve(true), terminate() {} }; + }, + }); + + await acquirer.start(); + acquirer.applyEstimations([ + intervalEntry("groom", { query: "labels = Fresh", interval: "1m", intervalMs: 60_000 }), + ]); + now = 60_000; + await acquirer.tick(); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(queries).toEqual(["labels = Fresh"]); + + acquirer.applyEstimations([]); + const store = new AutomationStateStore(dbPath); + expect(store.get("estimation:groom")).toBeNull(); + store.close(); + await acquirer.stop(); + }); + test("caps long timer delays at the runtime maximum", async () => { const dbPath = newDb(); const delays: number[] = []; diff --git a/packages/code/tests/repo-manager.test.ts b/packages/code/tests/repo-manager.test.ts index 438cb1b..3563756 100644 --- a/packages/code/tests/repo-manager.test.ts +++ b/packages/code/tests/repo-manager.test.ts @@ -71,6 +71,17 @@ describe("RepoManager", () => { expect(git(clonePath, "rev-parse origin/main")).toBe(git(originDir, "rev-parse main")); }); + test("ensureBareClone applies a changed remote URL to an existing clone", async () => { + const clonePath = await manager.ensureBareClone(repo); + const nextOrigin = join(rootDir, "next-origin"); + mkdirSync(nextOrigin); + git(nextOrigin, "init -b main"); + + await manager.ensureBareClone({ ...repo, remote: `file://${nextOrigin}` }); + + expect(git(clonePath, "remote get-url origin")).toBe(`file://${nextOrigin}`); + }); + test("task worktrees are detached at origin/ and see the origin remote", async () => { await manager.ensureBareClone(repo); const worktree = await manager.createTaskWorktree(repo, "BACK-42"); diff --git a/packages/code/tests/review-polling-acquirer.test.ts b/packages/code/tests/review-polling-acquirer.test.ts index f98aeda..159efed 100644 --- a/packages/code/tests/review-polling-acquirer.test.ts +++ b/packages/code/tests/review-polling-acquirer.test.ts @@ -874,6 +874,20 @@ describe("ReviewPollingAcquirer", () => { expect(queue.getBaseSyncEvent("base-sync:acme/widgets#42:base1:head1")).toBeNull(); }); + test("live reload can disable and re-enable conflict resolution", async () => { + workerState.recordAgentPr({ repo: "acme/widgets", prNumber: 42 }); + const made = makeAcquirer(conflictGh()); + + made.acquirer.updateConflictResolution("disabled"); + await made.acquirer.tick(); + expect(made.resolved).toEqual([]); + expect(queue.getBaseSyncEvent("base-sync:acme/widgets#42:base1:head1")).toBeNull(); + + made.acquirer.updateConflictResolution("auto"); + await made.acquirer.tick(); + expect(made.resolved).toEqual(["acme/widgets#42"]); + }); + test("start() surfaces the active conflict-resolution mode", async () => { workerState.recordAgentPr({ repo: "acme/widgets", prNumber: 42 }); const logs: string[] = []; diff --git a/packages/code/tests/run-coordinator.test.ts b/packages/code/tests/run-coordinator.test.ts index 40b53c2..cbb063f 100644 --- a/packages/code/tests/run-coordinator.test.ts +++ b/packages/code/tests/run-coordinator.test.ts @@ -3,6 +3,37 @@ import { describe, expect, test } from "bun:test"; import { RunCoordinator } from "../src/lib/run-coordinator"; describe("RunCoordinator", () => { + test("stays transparent until live estimations enable it", async () => { + const coordinator = new RunCoordinator(false); + const release = await coordinator.acquire(); + let ran = false; + await coordinator.run(async () => { + ran = true; + }); + expect(ran).toBe(true); + release(); + + coordinator.enable(); + const held = await coordinator.acquire(); + const waiting = coordinator.run(async () => "ran"); + let settled = false; + void waiting.then(() => (settled = true)); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(settled).toBe(false); + held(); + expect(await waiting).toBe("ran"); + + coordinator.disableWhenIdle(); + await new Promise((resolve) => setTimeout(resolve, 0)); + const transparentRelease = await coordinator.acquire(); + let transparentRan = false; + await coordinator.run(async () => { + transparentRan = true; + }); + expect(transparentRan).toBe(true); + transparentRelease(); + }); + test("runs held operations one at a time in FIFO order", async () => { const coordinator = new RunCoordinator(); const events: string[] = []; diff --git a/packages/code/tests/task-polling-acquirer.test.ts b/packages/code/tests/task-polling-acquirer.test.ts index 4319b9e..97aa0e6 100644 --- a/packages/code/tests/task-polling-acquirer.test.ts +++ b/packages/code/tests/task-polling-acquirer.test.ts @@ -137,6 +137,35 @@ describe("TaskPollingAcquirer", () => { return { acquirer, seenCursors }; } + test("reads task_query dynamically and stays dormant while it is absent", async () => { + let query: string | undefined; + const seenQueries: string[] = []; + const acquirer = new TaskPollingAcquirer({ + trackerType: "markdown", + query: () => query, + intervalSeconds: 60, + detector: { + source: "markdown", + async changesSince() { + return { changed: true, nextCursor: "1" }; + }, + }, + workerState, + queue, + searchTasks: async (activeQuery) => { + seenQueries.push(activeQuery); + return { tasks: [] }; + }, + executeTask: async () => true, + }); + + await acquirer.tick(); + expect(seenQueries).toEqual([]); + query = "status=ready"; + await acquirer.tick(); + expect(seenQueries).toEqual(["status=ready"]); + }); + test("executes query matches when a change is detected", async () => { const executed: string[] = []; const { acquirer } = makeAcquirer({ diff --git a/packages/code/tests/workspace-config-reload.test.ts b/packages/code/tests/workspace-config-reload.test.ts new file mode 100644 index 0000000..a205ca5 --- /dev/null +++ b/packages/code/tests/workspace-config-reload.test.ts @@ -0,0 +1,406 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdirSync, readFileSync, rmSync, writeFileSync } from "fs"; +import { tmpdir } from "os"; +import { join } from "path"; + +import { + WorkspaceConfigReloader, + applyWorkspaceConfig, + serializeWorkspaceConfig, +} from "../src/lib/workspace/config-reload"; +import { parseWorkspaceConfig } from "../src/lib/workspace/config"; +import type { RepoConfig, WorkspaceConfig } from "../src/lib/workspace/config"; +import type { RepoManagerLike } from "../src/lib/workspace/workspace-worker"; +import { + createFleetTaskExecutor, + resolveFleetAutomations, +} from "../src/lib/workspace/workspace-worker"; +import { toRoutableTask } from "../src/lib/workspace/router"; + +const V1 = ` +[defaults] +tracker = "markdown" +task_query = "status=todo" + +[[repos]] +name = "backend" +remote = "git@github.com:acme/backend.git" + +[[repos]] +name = "frontend" +remote = "git@github.com:acme/frontend.git" + +[[routing.rules]] +repo = "backend" +labels = ["backend"] +`; + +const V2 = ` +[defaults] +tracker = "markdown" +task_query = "status=todo" +worker_task_args = "--create-pr --fast" +poll_interval = 15 + +[[repos]] +name = "backend" +remote = "git@github.com:acme/backend.git" + +[[repos]] +name = "frontend" +remote = "git@github.com:acme/frontend.git" + +[[routing.rules]] +repo = "frontend" +labels = ["backend"] + +[[automations]] +id = "sweep" +enabled = true +interval = "1h" +prompt = "Tidy up." +`; + +const dirs: string[] = []; +afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); +}); + +function freshDir(): string { + const dir = join(tmpdir(), `reload-${Date.now()}-${Math.random()}`); + mkdirSync(dir, { recursive: true }); + dirs.push(dir); + return dir; +} + +function writeToml(path: string, text: string): void { + writeFileSync(path, text); +} + +const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); + +async function waitFor(check: () => boolean, timeoutMs = 5000): Promise { + const deadline = Date.now() + timeoutMs; + while (!check()) { + if (Date.now() > deadline) throw new Error("condition not met in time"); + await sleep(20); + } +} + +describe("WorkspaceConfigReloader", () => { + test("applies edited workspace.toml to the shared config instance", () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V1); + const current: WorkspaceConfig = parseWorkspaceConfig(readFileSync(path, "utf8")); + const reloader = new WorkspaceConfigReloader({ configPath: path, current }); + + expect(reloader.reload("test").applied).toBe(false); + + writeToml(path, V2); + const outcome = reloader.reload("test"); + expect(outcome.applied).toBe(true); + // The same instance every consumer holds was mutated in place. + expect(current.repos).toHaveLength(2); + expect(current.defaults.pollIntervalSeconds).toBe(15); + expect(current.routing).toHaveLength(1); + expect(current.routing[0]?.repo).toBe("frontend"); + expect(current.automations.map((automation) => automation.id)).toEqual(["sweep"]); + }); + + test("copies scheduled estimations into the shared config", () => { + const current = parseWorkspaceConfig(V1); + const next = parseWorkspaceConfig(V2); + next.estimations = [ + { + id: "groom", + enabled: true, + query: "status=todo", + interval: "1h", + intervalMs: 3_600_000, + }, + ]; + + applyWorkspaceConfig(current, next); + + expect(current.estimations.map((item) => item.id)).toEqual(["groom"]); + }); + + test("rejects runtime-incompatible changes before mutating active config", () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V2); + const current = parseWorkspaceConfig(V1); + const errors: string[] = []; + const reloader = new WorkspaceConfigReloader({ + configPath: path, + current, + validate: () => { + throw new Error("tracker is startup-only"); + }, + onError: (message) => errors.push(message), + }); + + expect(reloader.reload("test").applied).toBe(false); + expect(current.defaults.pollIntervalSeconds).toBe(60); + expect(errors[0]).toContain("tracker is startup-only"); + }); + + test("keeps the last good config when a reload produces invalid TOML", () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V2); + const current: WorkspaceConfig = parseWorkspaceConfig(readFileSync(path, "utf8")); + const errors: string[] = []; + const reloader = new WorkspaceConfigReloader({ + configPath: path, + current, + onError: (message) => errors.push(message), + }); + + writeToml(path, "[[repos]\nname = broken"); + const outcome = reloader.reload("mid-edit save"); + + expect(outcome.applied).toBe(false); + expect(errors).toHaveLength(1); + expect(errors[0]).toContain("Failed to reload workspace config (mid-edit save)"); + expect(errors[0]).toContain(path); + // Still serving the previous configuration. + expect(current.repos).toHaveLength(2); + expect(current.defaults.pollIntervalSeconds).toBe(15); + }); + + test("keeps the last good config on semantic errors and surfaces them", () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V2); + const current: WorkspaceConfig = parseWorkspaceConfig(readFileSync(path, "utf8")); + const errors: string[] = []; + const reloader = new WorkspaceConfigReloader({ + configPath: path, + current, + onError: (message) => errors.push(message), + }); + + writeToml( + path, + `${V2}\n[[repos]]\nname = "frontend"\nremote = "git@github.com:acme/other.git"\n`, + ); + const outcome = reloader.reload("bad edit"); + + expect(outcome.applied).toBe(false); + expect(errors.join("\n")).toContain("Duplicate repo name"); + expect(current.repos).toHaveLength(2); + }); + + test("treats rewriting identical content as a no-op", () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V2); + const current: WorkspaceConfig = parseWorkspaceConfig(readFileSync(path, "utf8")); + const reloader = new WorkspaceConfigReloader({ configPath: path, current }); + + writeToml(path, V2); + const outcome = reloader.reload("identical rewrite"); + expect(outcome.applied).toBe(false); + expect(outcome.unchanged).toBe(true); + }); + + test("serializations of equal configs match regardless of parse instance", () => { + const first = parseWorkspaceConfig(V2); + const second = parseWorkspaceConfig(V2); + expect(serializeWorkspaceConfig(first)).toBe(serializeWorkspaceConfig(second)); + expect(serializeWorkspaceConfig(first)).not.toBe( + serializeWorkspaceConfig(parseWorkspaceConfig(V1)), + ); + }); + + test("coalesces rapid successive edits into one reload", async () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V1); + const current: WorkspaceConfig = parseWorkspaceConfig(readFileSync(path, "utf8")); + let loads = 0; + const reloader = new WorkspaceConfigReloader({ + configPath: path, + current, + debounceMs: 25, + load: (p) => { + loads += 1; + return parseWorkspaceConfig(readFileSync(p, "utf8")); + }, + }); + + reloader.scheduleReload("edit 1"); + await sleep(5); + reloader.scheduleReload("edit 2"); + reloader.scheduleReload("edit 3"); + expect(loads).toBe(0); + + await waitFor(() => loads === 1); + await sleep(50); + expect(loads).toBe(1); + }); + + test("watches the file and applies edits while running; SIGHUP still works after stopping", async () => { + const dir = freshDir(); + const path = join(dir, "workspace.toml"); + writeToml(path, V1); + const current: WorkspaceConfig = parseWorkspaceConfig(readFileSync(path, "utf8")); + const reloader = new WorkspaceConfigReloader({ + configPath: path, + current, + debounceMs: 20, + }); + + reloader.start(); + + // Automatic file watching: an edit applies within the debounce window. + writeToml(path, V2); + await waitFor(() => current.defaults.pollIntervalSeconds === 15); + expect(current.repos).toHaveLength(2); + expect(current.automations).toHaveLength(1); + + // Stop() tears down the watcher but the daemon-lifetime SIGHUP fallback + // keeps working for manual reloads. + reloader.stop(); + writeToml(path, V1); + process.kill(process.pid, "SIGHUP"); + await waitFor(() => current.defaults.pollIntervalSeconds === 60); + expect(current.repos).toHaveLength(2); + expect(current.automations).toHaveLength(0); + }); +}); + +describe("resolveFleetAutomations", () => { + test("requires repo targeting once the fleet has multiple repos", () => { + const withProblem = parseWorkspaceConfig(` +[defaults] +tracker = "markdown" + +[[repos]] +name = "one" +remote = "git@github.com:acme/one.git" + +[[repos]] +name = "two" +remote = "git@github.com:acme/two.git" + +[[automations]] +id = "loose" +enabled = true +interval = "1h" +prompt = "p" +`); + const result = resolveFleetAutomations(withProblem); + expect(result.problems).toHaveLength(1); + expect(result.problems[0]).toContain('"loose"'); + expect(result.problems[0]).toContain("must set repo"); + expect(result.automations).toHaveLength(0); + + // A single-repo workspace implies its only checkout. + const single = parseWorkspaceConfig(` +[defaults] +tracker = "markdown" + +[[repos]] +name = "only" +remote = "git@github.com:acme/only.git" + +[[automations]] +id = "implicit" +enabled = true +interval = "1h" +prompt = "p" +`); + const okResult = resolveFleetAutomations(single); + expect(okResult.problems).toHaveLength(0); + expect(okResult.automations).toHaveLength(1); + }); +}); + +describe("createFleetTaskExecutor with live-reloaded config", () => { + class StubRepoManager implements RepoManagerLike { + worktrees: string[] = []; + constructor(private root: string) {} + async ensureBareClone(): Promise { + return this.root; + } + async fetch(): Promise {} + async ensureBaseWorktree(repo: RepoConfig): Promise { + return join(this.root, "worktrees", repo.name, "base"); + } + async createTaskWorktree(repo: RepoConfig, taskKey: string): Promise { + const path = join(this.root, "worktrees", repo.name, taskKey.toLowerCase()); + this.worktrees.push(path); + return path; + } + async removeTaskWorktree(_repoName: string, worktreePath: string): Promise { + rmSync(worktreePath, { recursive: true, force: true }); + } + async sweepStaleWorktrees(): Promise { + return []; + } + } + + test("routes subsequent tasks through updated rules, args, and env", async () => { + const root = freshDir(); + const config: WorkspaceConfig = parseWorkspaceConfig( + V2.replace(/\n\[\[automations\]\][\s\S]*$/, "\n"), + ); + const repoManager = new StubRepoManager(root); + const runs: Array<{ args: string[]; cwd: string; env: Record }> = + []; + const executor = createFleetTaskExecutor({ + config, + workspaceDir: root, + skips: { record() {} } as unknown as import("../src/lib/workspace/state").RoutingSkipStore, + repoManager, + repoLock: () => + ({ + acquire: () => ({ success: true, message: "" }), + release() {}, + }) as never, + runTask: async (_taskKey, args, opts) => { + runs.push({ args, cwd: opts.cwd, env: opts.env }); + return true; + }, + }); + + const routable = toRoutableTask({ key: "PROJ-12", components: [], labels: ["backend"] }); + + // V2's rule sends the "backend" label to frontend. + let ok = await executor("PROJ-12", routable); + expect(ok).toBe(true); + expect(runs.at(-1)?.cwd).toContain(join("worktrees", "frontend")); + expect(runs.at(-1)?.args).toEqual(["--create-pr", "--fast"]); + + // Simulate a live reload flipping routing back to backend. + const next: WorkspaceConfig = parseWorkspaceConfig(` +[defaults] +tracker = "markdown" +task_query = "status=todo" +worker_task_args = "--create-pr --auto-review" + +[[repos]] +name = "backend" +remote = "git@github.com:acme/backend.git" + +[[repos]] +name = "frontend" +remote = "git@github.com:acme/frontend.git" + +[[routing.rules]] +repo = "backend" +labels = ["backend"] +`); + applyWorkspaceConfig(config, next); + + ok = await executor("PROJ-12", routable); + expect(ok).toBe(true); + expect(runs.at(-1)?.cwd).toContain(join("worktrees", "backend")); + expect(runs.at(-1)?.args).toEqual(["--create-pr", "--auto-review"]); + // Per-repo env follows the reloaded repo definitions too. + expect(Object.keys(runs.at(-1)?.env ?? {})).toContain("GITHUB_REPO"); + }); +}); diff --git a/packages/code/tests/workspace-worker.test.ts b/packages/code/tests/workspace-worker.test.ts index 95d90a2..6753bb9 100644 --- a/packages/code/tests/workspace-worker.test.ts +++ b/packages/code/tests/workspace-worker.test.ts @@ -5,6 +5,7 @@ import { tmpdir } from "os"; import { parseWorkspaceConfig } from "../src/lib/workspace/config"; import type { RepoConfig } from "../src/lib/workspace/config"; +import { applyWorkspaceConfig } from "../src/lib/workspace/config-reload"; import { buildFleetEventAcquirers, createFleetTaskExecutor, @@ -45,6 +46,20 @@ repo = "frontend" labels = ["frontend"] `); +const FRONTEND_ONLY_CONFIG = parseWorkspaceConfig(` +[defaults] +tracker = "markdown" +task_query = "status=todo" + +[[repos]] +name = "frontend" +remote = "git@github.com:acme/frontend.git" + +[[routing.rules]] +repo = "frontend" +labels = ["frontend"] +`); + class FakeRepoManager implements RepoManagerLike { calls: string[] = []; worktrees: string[] = []; @@ -376,6 +391,78 @@ describe("buildFleetEventAcquirers", () => { rmSync(workspaceDir, { recursive: true, force: true }); } }); + + test("reconciles mention sweeps when repos are added or removed while running", async () => { + const workspaceDir = join( + tmpdir(), + `ws-sweep-${Date.now()}-${Math.random().toString(36).slice(2)}`, + ); + mkdirSync(workspaceDir, { recursive: true }); + const state = openWorkspaceState(workspaceDir); + const repoManager = new FakeRepoManager(workspaceDir); + const savedToken = process.env.GITHUB_TOKEN; + process.env.GITHUB_TOKEN = "test-token"; + + // No GitHub remotes yet: review poller idles, zero mention sweeps. + const nonGithub = parseWorkspaceConfig(` +[defaults] +tracker = "markdown" + +[[repos]] +name = "gitlab" +remote = "https://gitlab.com/acme/gitlab.git" + +[[repos]] +name = "forgejo" +remote = "https://forgejo.example/acme/forgejo.git" +`); + + // Block all network access: reconciliation starts new sweeps eagerly, + // and their initial poll must not reach api.github.com in tests. + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => { + throw new Error("network disabled in tests"); + }) as unknown as typeof fetch; + + try { + const intervalUpdaters: Array<(seconds: number) => void> = []; + const hooksOut: { + hooks?: import("../src/lib/workspace/workspace-worker").FleetEventReloadHooks; + } = {}; + const acquirers = await buildFleetEventAcquirers({ + config: nonGithub, + workspaceDir, + state, + repoManager, + searchTasks: async () => ({ tasks: [] }), + query: "status=todo", + intervalSeconds: 60, + intervalUpdaters, + reloadHooksOut: hooksOut, + }); + + expect(acquirers.map((acquirer) => acquirer.name)).toEqual(["poll:reviews"]); + expect(hooksOut.hooks?.mentionSweepRepos()).toEqual([]); + + // Live reload adds two GitHub repos; reconciling attaches their sweeps + // without a restart. + applyWorkspaceConfig(nonGithub, CONFIG); + hooksOut.hooks?.reconcileMentionSweeps(); + expect(hooksOut.hooks?.mentionSweepRepos()).toEqual(["acme/backend", "acme/frontend"]); + expect(intervalUpdaters.length).toBeGreaterThanOrEqual(2); + + // Removing a repo stops its sweep on the next reload. + applyWorkspaceConfig(nonGithub, FRONTEND_ONLY_CONFIG); + hooksOut.hooks?.reconcileMentionSweeps(); + expect(hooksOut.hooks?.mentionSweepRepos()).toEqual(["acme/frontend"]); + } finally { + globalThis.fetch = originalFetch; + if (savedToken === undefined) delete process.env.GITHUB_TOKEN; + else process.env.GITHUB_TOKEN = savedToken; + state.close(); + rmSync(workspaceDir, { recursive: true, force: true }); + } + }); }); describe("resolveWorkspaceAutomationContext", () => { @@ -481,4 +568,26 @@ describe("worktree sweeping", () => { await new Promise((resolve) => setTimeout(resolve, 40)); expect(swept.length).toBe(countAfterClear); }); + + test("startWorktreeSweeper reads live repos and TTL on every pass", async () => { + const { swept, repoManager } = fakeSweeper({}); + let repos = [REPOS[0]!]; + let ttlDays = 7; + const timer = startWorktreeSweeper( + () => repos, + repoManager, + () => ttlDays, + 10, + ); + try { + await new Promise((resolve) => setTimeout(resolve, 25)); + repos = [REPOS[1]!]; + ttlDays = 3; + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(swept).toContainEqual({ repo: "backend", ttlDays: 7 }); + expect(swept).toContainEqual({ repo: "frontend", ttlDays: 3 }); + } finally { + clearInterval(timer); + } + }); });