From 7aaaf95cf266080111615843bed0a2525375de9b Mon Sep 17 00:00:00 2001 From: Daniil Pokrovsky Date: Thu, 27 Aug 2026 17:21:54 +0700 Subject: [PATCH] feat: implement DEV-95 - Worker scheduled estimation via [[estimations]] in workspace.toml --- docs/code/dashboard.md | 4 +- docs/code/story-points-estimation.md | 147 +++---------- docs/code/worker.md | 33 +++ docs/code/workspaces.md | 3 +- packages/code/src/index.ts | 19 +- packages/code/src/lib/analytics.ts | 10 +- packages/code/src/lib/automation-acquirer.ts | 74 ++++--- packages/code/src/lib/automation-config.ts | 5 + packages/code/src/lib/automation-state.ts | 8 +- packages/code/src/lib/dashboard-api.ts | 8 +- packages/code/src/lib/estimation-acquirer.ts | 118 ++++++++++ packages/code/src/lib/estimation-config.ts | 111 ++++++++++ packages/code/src/lib/run-coordinator.ts | 41 ++++ packages/code/src/lib/run-recorder.ts | 3 +- packages/code/src/lib/workspace/config.ts | 30 ++- .../code/src/lib/workspace/fleet-events.ts | 27 ++- .../src/lib/workspace/workspace-worker.ts | 92 +++++++- packages/code/src/worker.ts | 4 +- packages/code/tests/automation-config.test.ts | 18 ++ packages/code/tests/dashboard-api.test.ts | 3 + .../code/tests/estimation-acquirer.test.ts | 207 ++++++++++++++++++ packages/code/tests/estimation-config.test.ts | 125 +++++++++++ packages/code/tests/run-coordinator.test.ts | 62 ++++++ packages/code/tests/run-recorder.test.ts | 14 ++ packages/code/tests/workspace-config.test.ts | 130 +++++++++++ packages/code/tests/workspace-worker.test.ts | 70 +++++- packages/dashboard-ui/src/lib/api.ts | 2 +- .../dashboard-ui/src/lib/run-origin.test.ts | 1 + packages/dashboard-ui/src/lib/run-origin.ts | 1 + packages/dashboard-ui/src/views/RunsView.tsx | 1 + packages/dashboard-ui/src/views/StatsView.tsx | 8 +- 31 files changed, 1203 insertions(+), 176 deletions(-) create mode 100644 packages/code/src/lib/estimation-acquirer.ts create mode 100644 packages/code/src/lib/estimation-config.ts create mode 100644 packages/code/src/lib/run-coordinator.ts create mode 100644 packages/code/tests/estimation-acquirer.test.ts create mode 100644 packages/code/tests/estimation-config.test.ts create mode 100644 packages/code/tests/run-coordinator.test.ts diff --git a/docs/code/dashboard.md b/docs/code/dashboard.md index 4e884f4..cd68a51 100644 --- a/docs/code/dashboard.md +++ b/docs/code/dashboard.md @@ -28,7 +28,7 @@ The standalone command reads the database in read-only mode, so it is safe to ru ## What it shows -- **Run list**: every run with its status, task key or automation id, origin (tracker task, PR mention, or scheduled), agent harness, PR link, and duration. Filter by status or origin (`origin=scheduled` isolates automation runs). +- **Run list**: every run with its status, task key or automation id, origin (tracker task, PR mention, scheduled, or estimate), agent harness, PR link, and duration. Filter by status or origin (`origin=scheduled` isolates automation runs; `origin=estimate` isolates story-point sweeps). - **Run detail**: a stage-by-stage timeline for one run: the feasibility verdict, the implementation summary, each self-review iteration, each human change request and how it was handled, and the final outcome. - **Stats**: runs per week, success and escalation rates, median run duration, and a per-harness breakdown over a selectable window (7, 30, or 90 days, or all time). - **Worker status**: whether the daemon is running, queued and failed events, open agent PRs, and per-source poll cursors. @@ -52,7 +52,7 @@ The dashboard is backed by a small read-only JSON API you can use directly, for | Endpoint | Returns | | --------------------------- | --------------------------------------------------------------------- | -| `GET /api/runs` | Paginated run list (`limit`, `offset`, `status`, `origin`, `taskKey`); `origin=scheduled` is supported | +| `GET /api/runs` | Paginated run list (`limit`, `offset`, `status`, `origin`, `taskKey`); `origin=scheduled` and `origin=estimate` are supported | | `GET /api/runs/:id` | One run with its stage timeline | | `GET /api/stats?window=30d` | Aggregate stats (`7d`, `30d`, `90d`, or `all`) | | `GET /api/worker` | Worker liveness, queue counts, agent PRs, poll cursors | diff --git a/docs/code/story-points-estimation.md b/docs/code/story-points-estimation.md index 8c993d3..8934aff 100644 --- a/docs/code/story-points-estimation.md +++ b/docs/code/story-points-estimation.md @@ -1,14 +1,22 @@ --- title: "Story Points Estimation" -description: "Automatically estimate JIRA story points with AI using the --estimate flag" +description: "Schedule unattended AI story-point estimation via [[estimations]] in workspace.toml, or run one-shot --estimate" section: "Server Automation" order: 6 -dateModified: 2026-08-26 +dateModified: 2026-08-27 --- # Story Points Estimation -Use the `--estimate` flag to have your AI agent analyze JIRA tasks and automatically assign story point estimates. This is useful for backlog grooming, sprint planning, or keeping estimates up to date as tasks evolve. +Let the [workspace worker](./worker.md) estimate stories on a schedule with `[[estimations]]` in `workspace.toml`, or use the `--estimate` flag as a CLI one-shot. In both modes your AI agent analyzes tasks and automatically assigns story point estimates — useful for backlog grooming, sprint planning, or keeping estimates up to date as tasks evolve. + +```toml +[[estimations]] +id = "weekday-groom" +enabled = true +cron = "0 9 * * 1-5" +query = "status = 'To Do' AND labels IN (NeedsEstimate)" +``` ## What It Does @@ -18,7 +26,9 @@ Use the `--estimate` flag to have your AI agent analyze JIRA tasks and automatic - Sets the story points field directly in JIRA - Posts a rich estimation comment with full context -## Usage +## Usage (CLI one-shot) + +`--estimate` remains the interactive/one-shot path; scheduled sweeps run the same engine from the worker. ### Single Task @@ -90,123 +100,36 @@ If your JIRA instance uses a custom field name, you can override it in `.devinte } ``` -## Automated Estimation - -`--estimate` is still a CLI one-shot. The [worker](./worker.md) does not schedule estimation yet. Until it does, the unattended path is a systemd timer or crontab of `devintern --estimate --query`. Prefer that over inventing a second daemon; drain ready tickets with the worker, and keep this timer only for story points. - -### systemd timers - -A systemd job is a one-shot `.service` plus a `.timer` that triggers it. Below is the daily morning run; the weekly grooming and sprint-planning helpers follow the same shape with different `OnCalendar` expressions and JQL. - -#### Daily estimation (weekday mornings) - -Run every weekday at 9 AM to estimate new tasks. - -`/etc/systemd/system/devintern-estimate-daily.service`: - -```ini -[Unit] -Description=Daily estimation of NeedsEstimate tasks -After=network-online.target -Wants=network-online.target - -[Service] -Type=oneshot -User=devintern -WorkingDirectory=/path/to/your/project -ExecStart=/usr/local/bin/devintern --estimate \ - --query 'project = PROJ AND status = "To Do" AND labels IN (NeedsEstimate)' -StandardOutput=journal -StandardError=journal -``` - -`/etc/systemd/system/devintern-estimate-daily.timer`: - -```ini -[Unit] -Description=Run daily estimation at 9 AM on weekdays - -[Timer] -OnCalendar=Mon..Fri *-*-* 09:00:00 -Persistent=true - -[Install] -WantedBy=timers.target -``` - -Enable and inspect: - -```bash -sudo systemctl daemon-reload -sudo systemctl enable --now devintern-estimate-daily.timer -systemctl list-timers devintern-estimate-daily.timer -journalctl -u devintern-estimate-daily.service -f -``` - -#### Weekly backlog grooming - -`OnCalendar` for Monday at 8 AM, JQL covers the last 7 days: - -```ini -# devintern-estimate-weekly.timer -[Timer] -OnCalendar=Mon *-*-* 08:00:00 -Persistent=true - -[Install] -WantedBy=timers.target -``` - -```ini -# devintern-estimate-weekly.service, ExecStart line -ExecStart=/usr/local/bin/devintern --estimate \ - --query 'project = PROJ AND status = "Backlog" AND updated >= -7d' -``` - -#### Sprint planning helper +--- -Wednesday at 10 AM, fills in any remaining gaps before planning: +## Scheduled Estimation with the Worker -```ini -# devintern-estimate-sprint.timer -[Timer] -OnCalendar=Wed *-*-* 10:00:00 -Persistent=true +The [workspace worker](./worker.md) runs unattended estimation for you via `[[estimations]]` in `workspace.toml`. No more cron of `--estimate`: the worker owns the timer, the durable schedule state, and serialization with all other agent work. -[Install] -WantedBy=timers.target -``` +```toml +[[estimations]] +id = "weekday-groom" +enabled = true +cron = "0 9 * * 1-5" +query = "status = 'To Do' AND labels IN (NeedsEstimate)" -```ini -# devintern-estimate-sprint.service, ExecStart line -ExecStart=/usr/local/bin/devintern --estimate \ - --query 'project = PROJ AND sprint in openSprints() AND "Story Points" is EMPTY' +[[estimations]] +id = "sprint-gaps" +enabled = true +cron = "0 10 * * 3" +query = "sprint in openSprints() AND \"Story Points\" is EMPTY" ``` -### Cron +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. -If you're not on systemd, the same three schedules work as crontab entries: - -```bash -# In crontab (run `crontab -e`) - -# Daily at 9 AM, weekdays: estimate new tasks -0 9 * * 1-5 cd /path/to/your/project && devintern --estimate --query 'project = PROJ AND status = "To Do" AND labels IN (NeedsEstimate)' >> /tmp/devintern-estimate.log 2>&1 - -# Weekly on Monday at 8 AM: re-estimate recently updated backlog -0 8 * * 1 cd /path/to/your/project && devintern --estimate --query 'project = PROJ AND status = "Backlog" AND updated >= -7d' >> /tmp/devintern-estimate.log 2>&1 - -# Weekly on Wednesday at 10 AM: sprint planning helper -0 10 * * 3 cd /path/to/your/project && devintern --estimate --query 'project = PROJ AND sprint in openSprints() AND "Story Points" is EMPTY' >> /tmp/devintern-estimate.log 2>&1 -``` +Notes: -### Important Notes +- Works with Jira, Linear, Azure DevOps, Asana, and GitHub (comment-only). Trello/markdown workspaces fail at startup. +- Runs use the full skip/re-estimate behavior above (24h gate, update-in-place). +- Usage-limit aborts end the sweep cleanly; the next occurrence retries. Tickets already written stay done. +- Each sweep is visible in the [dashboard](./dashboard.md) under its own `estimate` origin. -- Set `WorkingDirectory` (systemd) or `cd` (cron) to your project directory so the correct `.devintern-code/.env` is loaded -- Use absolute paths to `devintern` and your agent binary, or pin `PATH` explicitly in the unit file -- For systemd, `journalctl -u ` gives you logs; for cron, redirect stdout/stderr to a log file -- Test your JQL query manually before scheduling -- Tasks created less than 24 hours ago are automatically skipped, so frequent runs are safe +See [Worker Daemon → Scheduled story-point estimation](./worker.md#scheduled-story-point-estimation) for semantics and troubleshooting. ## Example Output diff --git a/docs/code/worker.md b/docs/code/worker.md index bd94723..dac35f6 100644 --- a/docs/code/worker.md +++ b/docs/code/worker.md @@ -107,6 +107,39 @@ On shutdown the scheduler stops its timer, terminates active automation subproce | Scheduled runs missing from the dashboard | Filter the run list by origin `scheduled`; check the worker has an automation license (startup log). | | Task files pile up under `~/.devintern/automations/` | They are small and safe to delete — they are only run inputs; the durable record is the run history in `queue.db`. | +## Scheduled story-point estimation + +`[[estimations]]` in `workspace.toml` runs unattended `--estimate` sweeps on a schedule. Entries use the same schedule grammar as `[[automations]]`, but the body is a **query**, not a prompt — there is no implementation, no branch, no worktree, no PR, and no `repo` key even in multi-repo workspaces: + +```toml +[[estimations]] +id = "weekday-groom" +enabled = true +cron = "0 9 * * 1-5" +query = "status = 'To Do' AND labels IN (NeedsEstimate)" + +[[estimations]] +id = "sprint-gaps" +enabled = true +cron = "0 10 * * 3" +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. + +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 + +Agent usage limits are account-global. While scheduled estimation is configured, every agent run in the worker process — implement tasks, automation occurrences, PR reviews, conflict resolutions, and estimation sweeps — takes turns through one process-level gate, so at most one agent subprocess burns quota at a time. + +Estimation shares the [automation scheduler](#schedule-semantics): durable cursors, leases, missed-occurrence coalescing, and overlap skipping live under `estimation:` keys in `queue.db`. + +```bash +# The CLI one-shot stays available for ad-hoc estimates: +devintern --estimate --query "project = PROJ AND status = 'To Do'" +``` + ## Polling mode With `[defaults].task_query` in `workspace.toml`, the worker polls your tracker on an interval (`[defaults].poll_interval`, default 60 seconds) and runs every task that matches the query. The query uses the same language as batch `--query` runs for your tracker, so "ready" means whatever your query says, for example a status or label. diff --git a/docs/code/workspaces.md b/docs/code/workspaces.md index ced8e84..281331d 100644 --- a/docs/code/workspaces.md +++ b/docs/code/workspaces.md @@ -77,6 +77,7 @@ prompt = "Review the frontend and clean up one source of recurring noise." - Repo names must be unique and filesystem-safe; they become directory names under `repos/` and `worktrees/`. - Rule criteria combine with AND; list values (`components`, `labels`) match when the task carries any of them. Comparisons are case-insensitive. `project` matches the task key prefix for `PROJ-123` style keys (Jira, Linear); trackers with numeric or opaque ids route via labels or components. - `[[automations]]` uses the same schema as single-repo `.devintern-code/automations.toml`. An entry must name `repo` when the workspace has more than one repository. See [Worker Daemon → Recurring automations](./worker.md#recurring-automations) for prompt-writing guidance and schedule semantics. +- `[[estimations]]` schedules unattended story-point sweeps (tracker query + cron/interval, no `prompt`, no `repo`). The workspace tracker must support estimation. See [Worker Daemon → Scheduled story-point estimation](./worker.md#scheduled-story-point-estimation). ### How workspace automations differ from single-repo ones @@ -121,7 +122,7 @@ 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 can omit the query and run as an automation-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 and automation configuration is loaded at startup; restart the worker after editing it. Schedule state and leases for automations 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. 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. `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/index.ts b/packages/code/src/index.ts index 3b73b74..41e7c6e 100755 --- a/packages/code/src/index.ts +++ b/packages/code/src/index.ts @@ -150,7 +150,7 @@ function buildCliRunProps(tracker: string): Record { estimationResults.total++; + // Structured run record for this attempt (skips above are not + // attempts). Estimation is its own dashboard origin — never an + // implement run — and scheduled sweeps carry the schedule id. + const estimationScheduleId = process.env.DEVINTERN_AUTOMATION_ID; + beginRun({ + origin: "estimate", + taskKey, + tracker: activeTrackerType, + ...(estimationScheduleId ? { automationId: estimationScheduleId } : {}), + }); + // Fetch comments and linked resources const comments = await tracker.getComments(taskKey); const linkedResources = tracker.extractLinkedResources(task); @@ -2182,10 +2193,15 @@ async function main(): Promise { error: "Failed to parse estimation response", }); } + await finishTaskRun( + result ? "succeeded" : "failed", + result ? undefined : "Failed to parse estimation response", + ); } catch (error) { // Usage limit is account-global — abort the rest of the estimation // batch and exit 0 so the scheduler retries next window. if (error instanceof UsageLimitError) { + await finishTaskRun("deferred", error.message); console.warn(`\n⏳ ${error.message}. Aborting estimation batch; will retry next run.`); if (lockManager) { lockManager.release(); @@ -2199,6 +2215,7 @@ async function main(): Promise { error: (error as Error).message, }); console.error(`❌ Failed to estimate ${taskKey}: ${(error as Error).message}`); + await finishTaskRun("failed", (error as Error).message); } } diff --git a/packages/code/src/lib/analytics.ts b/packages/code/src/lib/analytics.ts index ab0b263..b72ba6f 100644 --- a/packages/code/src/lib/analytics.ts +++ b/packages/code/src/lib/analytics.ts @@ -27,7 +27,7 @@ export type AnalyticsEvent = "cli_run" | "worker_started" | "worker_task_run" | export type WorkerTaskOutcome = "succeeded" | "failed" | "deferred" | "escalated" | "abandoned"; -export type WorkerTaskTrigger = "task" | "scheduled"; +export type WorkerTaskTrigger = "task" | "scheduled" | "estimate"; export type WorkerMode = "polling" | "relay" | "hybrid" | "scheduled"; /** Internal marker inherited only by task subprocesses launched by the worker. */ @@ -243,7 +243,13 @@ export function trackWorkerTaskRun( ): boolean { const runOrigin = env[RUN_ORIGIN_ENV]; const workerTrigger: WorkerTaskTrigger | undefined = - runOrigin === "worker" ? "task" : runOrigin === "scheduled" ? "scheduled" : undefined; + runOrigin === "worker" + ? "task" + : runOrigin === "scheduled" + ? "scheduled" + : runOrigin === "estimate" + ? "estimate" + : undefined; if (!workerTrigger) return false; void track("worker_task_run", { diff --git a/packages/code/src/lib/automation-acquirer.ts b/packages/code/src/lib/automation-acquirer.ts index b475eef..8b5d72b 100644 --- a/packages/code/src/lib/automation-acquirer.ts +++ b/packages/code/src/lib/automation-acquirer.ts @@ -54,6 +54,12 @@ export interface AutomationAcquirerOptions { clearTimer?: (timer: ReturnType) => void; setInterval?: (callback: () => void, ms: number) => ReturnType; clearInterval?: (timer: ReturnType) => void; + /** Namespace durable schedule state when another job type reuses this scheduler. */ + stateId?: (automation: AutomationConfig) => string; + /** User-facing job kind used in log lines; defaults to "automation". */ + jobKind?: string; + /** Acquirer name override (worker banner / analytics worker mode). */ + name?: string; } interface ActiveAutomationRun { @@ -72,7 +78,7 @@ export function nextAutomationDue(automation: AutomationConfig, afterMs: number) /** One-timer scheduler with durable UTC cursors and per-automation leases. */ export class AutomationAcquirer implements Acquirer { - readonly name = "scheduled-automations"; + readonly name: string; private options: AutomationAcquirerOptions; private store: AutomationStateStore; private owner = `${process.pid}:${randomUUID()}`; @@ -83,6 +89,7 @@ export class AutomationAcquirer implements Acquirer { constructor(options: AutomationAcquirerOptions) { this.options = options; + this.name = options.name ?? "scheduled-automations"; this.store = new AutomationStateStore(options.dbPath); } @@ -90,10 +97,10 @@ export class AutomationAcquirer implements Acquirer { this.stopped = false; const now = this.now(); for (const automation of this.options.automations.filter((item) => item.enabled)) { - this.store.register(automation, nextAutomationDue(automation, now)); + this.store.register(automation, nextAutomationDue(automation, now), this.stateId(automation)); } console.log( - `⏰ Scheduling ${this.options.automations.filter((item) => item.enabled).length} enabled automation(s)`, + `⏰ Scheduling ${this.options.automations.filter((item) => item.enabled).length} enabled ${this.jobKind()}(s)`, ); await this.tick(); } @@ -123,18 +130,20 @@ export class AutomationAcquirer implements Acquirer { private async runTick(): Promise { const leaseMs = this.options.leaseMs ?? LEASE_MS; + const kind = this.jobKind(); for (const automation of this.options.automations.filter((item) => item.enabled)) { - let state = this.store.get(automation.id); + const stateId = this.stateId(automation); + let state = this.store.get(stateId); if (!state) continue; if (this.active.has(automation.id)) { const active = this.active.get(automation.id) as ActiveAutomationRun; - if (!this.store.heartbeat(automation.id, this.owner, this.now(), leaseMs)) { - console.warn(`⏭️ [automation:${automation.id}] terminating: lease was lost`); + if (!this.store.heartbeat(stateId, this.owner, this.now(), leaseMs)) { + console.warn(`⏭️ [${kind}:${automation.id}] terminating: lease was lost`); active.run.terminate(); await active.lifecycle; } - state = this.store.get(automation.id); + state = this.store.get(stateId); if (!state) continue; } if (state.nextDueAt > this.now()) continue; @@ -143,16 +152,14 @@ export class AutomationAcquirer implements Acquirer { if (state.leaseOwner && (state.leaseExpiresAt ?? 0) > overlapNow) { const skipNow = this.now(); const nextDue = nextAutomationDue(automation, skipNow); - if (this.store.skipOverlap(automation.id, skipNow, nextDue)) { - console.warn( - `⏭️ [automation:${automation.id}] occurrence skipped: previous run is active`, - ); + if (this.store.skipOverlap(stateId, skipNow, nextDue)) { + console.warn(`⏭️ [${kind}:${automation.id}] occurrence skipped: previous run is active`); } continue; } const claimNow = this.now(); const nextDue = nextAutomationDue(automation, claimNow); - if (!this.store.claim(automation.id, this.owner, claimNow, nextDue, leaseMs)) continue; + if (!this.store.claim(stateId, this.owner, claimNow, nextDue, leaseMs)) continue; let context: AutomationRunContext | null = null; try { @@ -162,7 +169,7 @@ export class AutomationAcquirer implements Acquirer { const clearHeartbeatInterval = this.options.clearInterval ?? clearInterval; const preparationHeartbeat = setHeartbeatInterval( () => { - if (!this.store.heartbeat(automation.id, this.owner, this.now(), leaseMs)) { + if (!this.store.heartbeat(stateId, this.owner, this.now(), leaseMs)) { ownsClaim = false; } }, @@ -175,22 +182,24 @@ export class AutomationAcquirer implements Acquirer { clearHeartbeatInterval(preparationHeartbeat); } if (!context) { - console.warn(`⏭️ [automation:${automation.id}] occurrence skipped: repository is busy`); - this.store.release(automation.id, this.owner); + console.warn( + `⏭️ [${kind}:${automation.id}] occurrence skipped: ${this.busySkipReason()}`, + ); + this.store.release(stateId, this.owner); continue; } if (this.stopped) { - this.store.release(automation.id, this.owner); + this.store.release(stateId, this.owner); await context.release(); continue; } - ownsClaim &&= this.store.heartbeat(automation.id, this.owner, this.now(), leaseMs); + ownsClaim &&= this.store.heartbeat(stateId, this.owner, this.now(), leaseMs); if (!ownsClaim) { - console.warn(`⏭️ [automation:${automation.id}] occurrence skipped: lease was lost`); + console.warn(`⏭️ [${kind}:${automation.id}] occurrence skipped: lease was lost`); await context.release(); continue; } - console.log(`\n⏰ [automation:${automation.id}] starting scheduled run`); + console.log(`\n⏰ [${kind}:${automation.id}] starting scheduled run`); const run = this.options.spawnRun ? this.options.spawnRun(automation, context) : defaultSpawnRun( @@ -207,16 +216,16 @@ export class AutomationAcquirer implements Acquirer { .then((ok) => console.log( ok - ? `✅ [automation:${automation.id}] completed` - : `⚠️ [automation:${automation.id}] did not complete cleanly`, + ? `✅ [${kind}:${automation.id}] completed` + : `⚠️ [${kind}:${automation.id}] did not complete cleanly`, ), ) .catch((error) => - console.error(`❌ [automation:${automation.id}] ${(error as Error).message}`), + console.error(`❌ [${kind}:${automation.id}] ${(error as Error).message}`), ) .finally(async () => { try { - this.store.release(automation.id, this.owner); + this.store.release(stateId, this.owner); await context?.release(); } finally { if (this.active.get(automation.id) === active) this.active.delete(automation.id); @@ -225,9 +234,9 @@ export class AutomationAcquirer implements Acquirer { }); this.active.set(automation.id, active); } catch (error) { - this.store.release(automation.id, this.owner); + this.store.release(stateId, this.owner); await context?.release(); - console.error(`❌ [automation:${automation.id}] ${(error as Error).message}`); + console.error(`❌ [${kind}:${automation.id}] ${(error as Error).message}`); } } this.scheduleNext(); @@ -237,13 +246,26 @@ export class AutomationAcquirer implements Acquirer { return (this.options.now ?? Date.now)(); } + private jobKind(): string { + return this.options.jobKind ?? "automation"; + } + + /** Reason logged when `resolveContext` declines a due occurrence. */ + private busySkipReason(): string { + return this.options.jobKind === undefined ? "repository is busy" : "worker is busy"; + } + + private stateId(automation: AutomationConfig): string { + return this.options.stateId?.(automation) ?? automation.id; + } + private scheduleNext(): void { if (this.stopped) return; if (this.timer) (this.options.clearTimer ?? clearTimeout)(this.timer); const now = this.now(); const dueTimes = this.options.automations .filter((item) => item.enabled) - .map((item) => this.store.get(item.id)?.nextDueAt) + .map((item) => this.store.get(this.stateId(item))?.nextDueAt) .filter((value): value is number => value !== undefined); const heartbeatAt = this.active.size > 0 ? now + (this.options.heartbeatMs ?? HEARTBEAT_MS) : Infinity; diff --git a/packages/code/src/lib/automation-config.ts b/packages/code/src/lib/automation-config.ts index 26ff287..fb95105 100644 --- a/packages/code/src/lib/automation-config.ts +++ b/packages/code/src/lib/automation-config.ts @@ -74,6 +74,11 @@ export function parseAutomationEntries( const enabledValue = table.enabled; if (typeof enabledValue !== "boolean") errors.push(`${label}.enabled must be a boolean.`); + if (table.kind !== undefined) { + errors.push( + `${label}.kind is not supported; scheduled estimation belongs in [[estimations]].`, + ); + } const prompt = stringValue("prompt"); if (!prompt) errors.push(`${label}.prompt is required.`); diff --git a/packages/code/src/lib/automation-state.ts b/packages/code/src/lib/automation-state.ts index c2385ed..e6a66bb 100644 --- a/packages/code/src/lib/automation-state.ts +++ b/packages/code/src/lib/automation-state.ts @@ -35,22 +35,22 @@ export class AutomationStateStore { } /** Create schedule state once; restarts retain the prior interval anchor. */ - register(automation: AutomationConfig, nextDueAt: number): void { + register(automation: AutomationConfig, nextDueAt: number, stateId = automation.id): void { const spec = automation.cron ? `cron:${automation.cron}` : `interval:${automation.interval}`; const existing = this.db .query("SELECT schedule_spec FROM automation_schedules WHERE automation_id = ?") - .get(automation.id) as { schedule_spec: string } | null; + .get(stateId) as { schedule_spec: string } | null; if (!existing) { this.db.run( `INSERT INTO automation_schedules (automation_id, schedule_spec, next_due_at) VALUES (?, ?, ?)`, - [automation.id, spec, nextDueAt], + [stateId, spec, nextDueAt], ); } else if (existing.schedule_spec !== spec) { this.db.run( `UPDATE automation_schedules SET schedule_spec = ?, next_due_at = ?, last_scheduled_at = NULL WHERE automation_id = ?`, - [spec, nextDueAt, automation.id], + [spec, nextDueAt, stateId], ); } } diff --git a/packages/code/src/lib/dashboard-api.ts b/packages/code/src/lib/dashboard-api.ts index ef0b45c..0fce3cd 100644 --- a/packages/code/src/lib/dashboard-api.ts +++ b/packages/code/src/lib/dashboard-api.ts @@ -26,7 +26,13 @@ const RUN_STATUSES: RunStatus[] = [ "escalated", "abandoned", ]; -const RUN_ORIGINS: RunOrigin[] = ["task", "pr_mention", "conflict_resolution", "scheduled"]; +const RUN_ORIGINS: RunOrigin[] = [ + "task", + "pr_mention", + "conflict_resolution", + "scheduled", + "estimate", +]; const STATS_WINDOWS: Record = { "7d": 7 * 24 * 60 * 60 * 1000, diff --git a/packages/code/src/lib/estimation-acquirer.ts b/packages/code/src/lib/estimation-acquirer.ts new file mode 100644 index 0000000..97c2f61 --- /dev/null +++ b/packages/code/src/lib/estimation-acquirer.ts @@ -0,0 +1,118 @@ +import { + AUTOMATION_ID_ENV, + AUTOMATION_ORIGIN_ENV, + AutomationAcquirer, + spawnAutomationProcess, +} from "./automation-acquirer"; +import type { AutomationRunContext, SpawnedAutomationRun } from "./automation-acquirer"; +import type { AutomationConfig } from "./automation-config"; +import type { EstimationConfig } from "./estimation-config"; + +/** Value of the run-origin marker that attributes CLI runs to scheduled estimation. */ +export const ESTIMATION_ORIGIN_ENV_VALUE = "estimate"; +/** Durable-schedule namespace so estimations never collide with automation ids. */ +const STATE_PREFIX = "estimation:"; + +export interface EstimationAcquirerOptions { + estimations: EstimationConfig[]; + dbPath: string; + /** + * Resolve where/how one due sweep runs. Unlike automations this never + * clones a repository or takes a repo lock; returning null still means + * "skip this occurrence". + */ + resolveContext: (estimation: EstimationConfig) => Promise; + /** Test-only runner override; defaults to spawning the CLI estimate sweep. */ + spawnRun?: (estimation: EstimationConfig, context: AutomationRunContext) => SpawnedAutomationRun; + /** Remaining fields are test-only scheduler/process overrides. */ + now?: () => number; + leaseMs?: number; + heartbeatMs?: number; + terminationGraceMs?: number; + setTimer?: (callback: () => void, delay: number) => ReturnType; + clearTimer?: (timer: ReturnType) => void; + setInterval?: (callback: () => void, ms: number) => ReturnType; + clearInterval?: (timer: ReturnType) => void; +} + +/** CLI arguments for one due estimation sweep (`--estimate` stays the engine). */ +export function estimationCliArgs(estimation: EstimationConfig): string[] { + return ["--estimate", "--query", estimation.query, "--no-git"]; +} + +/** + * Attribution env for the sweep subprocess: a distinct run origin (never an + * implement "scheduled" run) plus the owning schedule id. + */ +export function estimationRunEnv( + estimation: EstimationConfig, + base?: Record, +): Record { + return { + ...base, + [AUTOMATION_ORIGIN_ENV]: ESTIMATION_ORIGIN_ENV_VALUE, + [AUTOMATION_ID_ENV]: estimation.id, + }; +} + +/** + * Run one due sweep as the regular one-shot `devintern --estimate --query` + * pipeline in the workspace home. That path already implements the skip + * gates (<24h old / unchanged since last estimate), writes points + updates + * the estimate comment in place, treats usage-limit aborts as exit 0, and + * never creates branches, worktrees, or PRs. + */ +export function spawnEstimationSweep( + estimation: EstimationConfig, + context: AutomationRunContext, + terminationGraceMs?: number, +): SpawnedAutomationRun { + return spawnAutomationProcess( + process.execPath, + [process.argv[1] as string, ...estimationCliArgs(estimation)], + { + cwd: context.cwd, + env: estimationRunEnv(estimation, context.env), + terminationGraceMs, + }, + ); +} + +/** + * Scheduled story-point sweeps driven by the automation **scheduler** + * (durable cursors, leases, overlap coalescing) — never its prompt pipeline. + */ +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: "", + })); + 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), + spawnRun: (schedule, context) => { + const estimation = byId.get(schedule.id) as EstimationConfig; + return options.spawnRun + ? options.spawnRun(estimation, context) + : spawnEstimationSweep(estimation, context); + }, + now: options.now, + leaseMs: options.leaseMs, + heartbeatMs: options.heartbeatMs, + terminationGraceMs: options.terminationGraceMs, + setTimer: options.setTimer, + clearTimer: options.clearTimer, + setInterval: options.setInterval, + clearInterval: options.clearInterval, + }); + } +} diff --git a/packages/code/src/lib/estimation-config.ts b/packages/code/src/lib/estimation-config.ts new file mode 100644 index 0000000..2b05873 --- /dev/null +++ b/packages/code/src/lib/estimation-config.ts @@ -0,0 +1,111 @@ +import { CronExpressionParser } from "cron-parser"; + +import { parseAutomationInterval } from "./automation-config"; + +/** + * One workspace `[[estimations]]` entry: a scheduled story-point sweep. + * + * Same schedule grammar as `[[automations]]`, but the job body is a single + * `query` — the automation prompt pipeline is never used, and there is no + * `repo` because estimation never touches a repository. + */ +export interface EstimationConfig { + id: string; + enabled: boolean; + /** Tracker query selecting the tickets to estimate (same language as `--query`). */ + query: string; + cron?: string; + interval?: string; + intervalMs?: number; +} + +const ESTIMATION_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]*$/; + +/** Validate and normalize workspace `[[estimations]]` tables, collecting every error. */ +export function parseEstimationEntries(value: unknown): { + estimations: EstimationConfig[]; + errors: string[]; +} { + const errors: string[] = []; + const estimations: EstimationConfig[] = []; + const ids = new Set(); + if (value === undefined || value === null) return { estimations, errors }; + if (!Array.isArray(value)) { + return { estimations, errors: ["[[estimations]] must be an array of tables."] }; + } + + for (const [index, raw] of value.entries()) { + const label = `[[estimations]][${index}]`; + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + errors.push(`${label} must be a table.`); + continue; + } + const table = raw as Record; + const stringValue = (key: string): string | undefined => { + const item = table[key]; + if (item === undefined || item === null) return undefined; + if (typeof item !== "string" || !item.trim()) { + errors.push(`${label}.${key} must be a non-empty string.`); + return undefined; + } + return item.trim(); + }; + + // Estimation is not an implement job: no prompt pipeline, no repository, + // and no `kind` selector. + for (const forbidden of ["prompt", "repo", "kind"] as const) { + if (table[forbidden] !== undefined) errors.push(`${label}.${forbidden} is not supported.`); + } + + const id = stringValue("id"); + if (!id) errors.push(`${label}.id is required.`); + else if (!ESTIMATION_ID_PATTERN.test(id)) + errors.push(`${label}.id must contain only letters, digits, ".", "_", or "-".`); + else if (ids.has(id)) errors.push(`Duplicate estimation id "${id}".`); + else ids.add(id); + + const enabledValue = table.enabled; + if (typeof enabledValue !== "boolean") errors.push(`${label}.enabled must be a boolean.`); + const query = stringValue("query"); + if (!query) errors.push(`${label}.query is required.`); + + const cron = stringValue("cron"); + const interval = stringValue("interval"); + if (Boolean(cron) === Boolean(interval)) { + errors.push(`${label} must set exactly one of cron or interval.`); + } + if (cron) { + if (cron.split(/\s+/).length !== 5) { + errors.push(`${label}.cron must be a five-field cron expression.`); + } else { + try { + CronExpressionParser.parse(cron); + } catch (error) { + errors.push(`${label}.cron is invalid: ${(error as Error).message}`); + } + } + } + const intervalMs = interval ? parseAutomationInterval(interval) : undefined; + if (interval && intervalMs === null) { + errors.push(`${label}.interval must use a positive duration such as 15m, 6h, or 1d.`); + } + + if ( + id && + typeof enabledValue === "boolean" && + query && + Boolean(cron) !== Boolean(interval) && + (!interval || intervalMs !== null) + ) { + estimations.push({ + id, + enabled: enabledValue, + query, + cron, + interval, + intervalMs: intervalMs ?? undefined, + }); + } + } + return { estimations, errors }; +} diff --git a/packages/code/src/lib/run-coordinator.ts b/packages/code/src/lib/run-coordinator.ts new file mode 100644 index 0000000..97babda --- /dev/null +++ b/packages/code/src/lib/run-coordinator.ts @@ -0,0 +1,41 @@ +/** + * Process-level serialization gate for agent runs. + * + * Usage limits are account-global: two concurrent agent subprocesses (an + * implement run and a scheduled estimation sweep, say) burn the same quota + * and risk tripping it twice. The coordinator hands out one slot at a time + * in FIFO order, so whoever asked first runs next. + * + * Only wired when scheduled estimation joins the worker process; without + * `[[estimations]]` every acquirer keeps today's behavior. + */ +export class RunCoordinator { + private tail: Promise = Promise.resolve(); + + /** Wait for a free slot; returns the release function for that slot. */ + async acquire(): Promise<() => void> { + let release!: () => void; + const turn = new Promise((resolve) => { + release = resolve; + }); + const previous = this.tail; + this.tail = previous.then(() => turn); + await previous; + let released = false; + return () => { + if (released) return; + released = true; + release(); + }; + } + + /** Run `operation` while holding exactly one slot. */ + async run(operation: () => Promise): Promise { + const release = await this.acquire(); + try { + return await operation(); + } finally { + release(); + } + } +} diff --git a/packages/code/src/lib/run-recorder.ts b/packages/code/src/lib/run-recorder.ts index 45c6a17..eebd719 100644 --- a/packages/code/src/lib/run-recorder.ts +++ b/packages/code/src/lib/run-recorder.ts @@ -15,7 +15,7 @@ import { Database } from "bun:sqlite"; import { prepareQueueDbDirectory, resolveQueueDbPath } from "./webhook-queue"; -export type RunOrigin = "task" | "pr_mention" | "conflict_resolution" | "scheduled"; +export type RunOrigin = "task" | "pr_mention" | "conflict_resolution" | "scheduled" | "estimate"; export type RunStatus = | "in_progress" @@ -454,6 +454,7 @@ export class RunStore { pr_mention: 0, conflict_resolution: 0, scheduled: 0, + estimate: 0, }; const weekCounts = new Map(); const harnesses = new Map< diff --git a/packages/code/src/lib/workspace/config.ts b/packages/code/src/lib/workspace/config.ts index 2f9118d..0125c74 100644 --- a/packages/code/src/lib/workspace/config.ts +++ b/packages/code/src/lib/workspace/config.ts @@ -1,8 +1,15 @@ import { readFileSync } from "fs"; -import { supportsPolling, trackersSupportingPolling } from "../tracker-capabilities"; +import { + supportsEstimate, + supportsPolling, + trackersSupportingEstimate, + trackersSupportingPolling, +} from "../tracker-capabilities"; import { parseAutomationEntries } from "../automation-config"; import type { AutomationConfig } from "../automation-config"; +import { parseEstimationEntries } from "../estimation-config"; +import type { EstimationConfig } from "../estimation-config"; import { parseToml } from "./toml"; /** Workspace-wide settings from the `[workspace]` table. */ @@ -62,6 +69,7 @@ export interface WorkspaceConfig { repos: RepoConfig[]; routing: RoutingRule[]; automations: AutomationConfig[]; + estimations: EstimationConfig[]; } export const DEFAULT_WORKTREES_TTL_DAYS = 7; @@ -244,6 +252,11 @@ export function parseWorkspaceConfig( `Pollable trackers: ${trackersSupportingPolling().join(", ")}.`, ); } + if (defaultsTable.estimate_query !== undefined) { + errors.push( + "[defaults].estimate_query is not supported; scheduled estimation queries belong in [[estimations]].", + ); + } const defaults: WorkspaceDefaults = { tracker: tracker ?? "", taskQuery: readString(defaultsTable, "task_query", "[defaults]", errors), @@ -325,6 +338,20 @@ export function parseWorkspaceConfig( }); errors.push(...automationResult.errors); + const estimationResult = parseEstimationEntries(document.estimations); + errors.push(...estimationResult.errors); + // Estimation sweeps run the one-shot `--estimate` engine, so only trackers + // with estimate support can serve them; fail the job at startup otherwise. + if (tracker && !supportsEstimate(tracker) && estimationResult.estimations.length > 0) { + for (const estimation of estimationResult.estimations) { + errors.push( + `Estimation "${estimation.id}" requires a tracker that supports --estimate ` + + `(supported: ${trackersSupportingEstimate().join(", ")}); ` + + `[defaults].tracker "${tracker}" does not.`, + ); + } + } + if (errors.length > 0) { throw new Error(`Invalid ${sourceLabel}:\n- ${errors.join("\n- ")}`); } @@ -335,6 +362,7 @@ export function parseWorkspaceConfig( repos, routing, automations: automationResult.automations, + estimations: estimationResult.estimations, }; } diff --git a/packages/code/src/lib/workspace/fleet-events.ts b/packages/code/src/lib/workspace/fleet-events.ts index f5a29fb..4ab2fb7 100644 --- a/packages/code/src/lib/workspace/fleet-events.ts +++ b/packages/code/src/lib/workspace/fleet-events.ts @@ -16,6 +16,7 @@ import type { RepoConfig, WorkspaceConfig } from "./config"; import { buildRepoEnv, gitHubSlugFromRemote } from "./env"; import { toRoutableTask } from "./router"; import type { createFleetTaskExecutor, FleetTask, RepoManagerLike } from "./workspace-worker"; +import type { RunCoordinator } from "../run-coordinator"; export interface FleetEventDeps { config: WorkspaceConfig; @@ -32,6 +33,8 @@ export interface FleetEventDeps { /** Base-sync runner (injected for tests; defaults to the CLI subprocess). */ runResolve?: typeof runResolveConflictsViaCli; verbose?: boolean; + /** Process-level agent-run gate; only set when scheduled estimation exists. */ + coordinator?: RunCoordinator; } /** @@ -80,10 +83,12 @@ export function createFleetAddressPr( await repoManager.ensureBareClone(repo); await repoManager.fetch(repo.name); const base = await repoManager.ensureBaseWorktree(repo); - return runReview(slug, prNumber, { - cwd: base, - env: buildRepoEnv(repo, workspaceDir), - }); + const invoke = () => + runReview(slug, prNumber, { + cwd: base, + env: buildRepoEnv(repo, workspaceDir), + }); + return deps.coordinator ? deps.coordinator.run(invoke) : invoke(); }; } @@ -105,12 +110,14 @@ export function createFleetResolveConflicts( await repoManager.ensureBareClone(repo); await repoManager.fetch(repo.name); const base = await repoManager.ensureBaseWorktree(repo); - return runResolve(slug, prNumber, { - cwd: base, - env: buildRepoEnv(repo, workspaceDir), - expectedHeadSha: expected.headSha, - expectedBaseSha: expected.baseSha, - }); + const invoke = () => + runResolve(slug, prNumber, { + cwd: base, + env: buildRepoEnv(repo, workspaceDir), + expectedHeadSha: expected.headSha, + expectedBaseSha: expected.baseSha, + }); + return deps.coordinator ? deps.coordinator.run(invoke) : invoke(); }; } diff --git a/packages/code/src/lib/workspace/workspace-worker.ts b/packages/code/src/lib/workspace/workspace-worker.ts index 7899b96..eedb377 100644 --- a/packages/code/src/lib/workspace/workspace-worker.ts +++ b/packages/code/src/lib/workspace/workspace-worker.ts @@ -34,6 +34,9 @@ import { RepoManager } from "./repo-manager"; import { probePushAccess } from "../github-push-probe"; import { AutomationAcquirer } from "../automation-acquirer"; import type { AutomationConfig } from "../automation-config"; +import { EstimationAcquirer } from "../estimation-acquirer"; +import { RunCoordinator } from "../run-coordinator"; +import type { AutomationRunContext } from "../automation-acquirer"; import { flushAnalytics, RUN_ORIGIN_ENV, trackWorkerStarted } from "../analytics"; /** Task shape the fleet acquirer needs (structural subset of `Task`). */ @@ -74,6 +77,8 @@ export interface WorkspaceTaskAcquirerDeps { ) => Promise; /** Repo run lock factory (injected for tests). */ repoLock?: (repoName: string) => LockManager; + /** Process-level agent-run gate; only set when scheduled estimation exists. */ + coordinator?: RunCoordinator; } interface RepoRunLockLike { @@ -81,6 +86,34 @@ interface RepoRunLockLike { release(): void; } +/** + * Hold a process-level agent slot for a scheduled run's lifetime. + * + * Without a coordinator (no [[estimations]] configured) the context passes + * through untouched, so long-lived gates are never introduced silently. + * With one, acquisition happens after any repo lock is held, and release + * always runs — even when the caller's own release throws. + */ +async function withCoordinatorSlot( + context: AutomationRunContext | null | Promise, + coordinator?: RunCoordinator, +): Promise { + const resolved = await context; + if (!coordinator || !resolved) return resolved; + const releaseRun = await coordinator.acquire(); + const releaseContext = resolved.release; + return { + ...resolved, + release: async () => { + try { + await releaseContext(); + } finally { + releaseRun(); + } + }, + }; +} + /** Resolve a scheduled run context while holding the repo lock during preparation. */ export async function resolveWorkspaceAutomationContext( automation: AutomationConfig, @@ -228,7 +261,7 @@ export function createWorkspaceTaskAcquirer(deps: WorkspaceTaskAcquirerDeps): Ta export type FleetExecutorDeps = Pick< WorkspaceTaskAcquirerDeps, "config" | "workspaceDir" | "skips" | "repoManager" | "runTask" | "repoLock" ->; +> & { coordinator?: RunCoordinator }; /** * Build the fleet execute step: route a task to its repo and run it in a @@ -289,13 +322,15 @@ export function createFleetTaskExecutor( const worktree = await repoManager.createTaskWorktree(repo, taskKey); console.log(`🏗️ [fleet] ${taskKey} → ${repo.name} (${worktree})`); - const ok = await runTask(taskKey, extraArgs, { - cwd: worktree, - env: { - ...buildRepoEnv(repo, workspaceDir), - [RUN_ORIGIN_ENV]: "worker", - }, - }); + const invoke = () => + runTask(taskKey, extraArgs, { + cwd: worktree, + env: { + ...buildRepoEnv(repo, workspaceDir), + [RUN_ORIGIN_ENV]: "worker", + }, + }); + const ok = deps.coordinator ? await deps.coordinator.run(invoke) : await invoke(); if (ok) { await repoManager.removeTaskWorktree(repo.name, worktree); @@ -357,7 +392,7 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr const query = config.defaults.taskQuery; const intervalSeconds = config.defaults.pollIntervalSeconds; - if (!query && config.automations.length === 0) { + if (!query && config.automations.length === 0 && config.estimations.length === 0) { console.error( "❌ Workspace mode needs a task query: set [defaults].task_query in workspace.toml.", ); @@ -366,6 +401,13 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr const state = openWorkspaceState(workspaceDir); const repoManager = new RepoManager(workspaceDir); + // Preserve the worker's existing concurrency when scheduled estimation is + // 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; for (const repo of config.repos) { const removed = await repoManager.sweepStaleWorktrees( @@ -399,8 +441,31 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr automations: config.automations, dbPath: state.dbPath, extraArgs: fleetTaskArgs(config), - resolveContext: (automation) => - resolveWorkspaceAutomationContext(automation, config, workspaceDir, repoManager), + 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, + ), }), ); } @@ -430,6 +495,7 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr query, intervalSeconds, verbose: options.verbose, + coordinator, }), ); acquirers.push( @@ -442,6 +508,7 @@ export async function runWorkspaceWorker(options: RunWorkspaceWorkerOptions): Pr query, intervalSeconds, verbose: options.verbose, + coordinator, })), ); } @@ -496,6 +563,7 @@ export async function buildFleetEventAcquirers(options: { query: string; intervalSeconds: number; verbose?: boolean; + coordinator?: RunCoordinator; }): Promise { const { config, workspaceDir, state, repoManager, searchTasks, query, intervalSeconds, verbose } = options; @@ -531,6 +599,7 @@ export async function buildFleetEventAcquirers(options: { userHasPushAccess: (owner: string, repo: string, user: string) => gh.userHasPushAccess(owner, repo, user), verbose, + coordinator: options.coordinator, }; const fleetAddressPr = createFleetAddressPr(eventDeps); addressPr = fleetAddressPr; @@ -678,6 +747,7 @@ export async function buildFleetEventAcquirers(options: { workspaceDir, skips: state.skips, repoManager, + coordinator: options.coordinator, }); const evaluateTask = createFleetTaskEvaluator({ query, searchTasks, execute, verbose }); diff --git a/packages/code/src/worker.ts b/packages/code/src/worker.ts index e6d43f8..19c31fa 100644 --- a/packages/code/src/worker.ts +++ b/packages/code/src/worker.ts @@ -63,7 +63,9 @@ export async function startWorker( if (acquirers.length === 0) { console.error("❌ No event sources enabled."); - console.error(" Set [defaults].task_query in workspace.toml, or add [[automations]]."); + console.error( + " Set [defaults].task_query in workspace.toml, or add [[automations]] / [[estimations]].", + ); console.error(" Direct webhooks are a separate command: devintern webhook serve"); lock.release(); process.exit(1); diff --git a/packages/code/tests/automation-config.test.ts b/packages/code/tests/automation-config.test.ts index dc3de79..33c16db 100644 --- a/packages/code/tests/automation-config.test.ts +++ b/packages/code/tests/automation-config.test.ts @@ -85,4 +85,22 @@ repo = "web" `), ).toThrow(/does not match any \[\[repos\]\] name/); }); + + test("rejects the kind selector — scheduled estimation lives in [[estimations]]", () => { + let message = ""; + try { + parseAutomationConfig(` +[[automations]] +id = "groom" +enabled = true +interval = "1d" +kind = "estimate" +prompt = "estimate stories" +`); + } catch (error) { + message = (error as Error).message; + } + expect(message).toContain("kind is not supported"); + expect(message).toContain("[[estimations]]"); + }); }); diff --git a/packages/code/tests/dashboard-api.test.ts b/packages/code/tests/dashboard-api.test.ts index a17825d..3c8e547 100644 --- a/packages/code/tests/dashboard-api.test.ts +++ b/packages/code/tests/dashboard-api.test.ts @@ -88,6 +88,9 @@ describe("dashboard API", () => { const scheduled = handleRuns(data, new URLSearchParams("origin=scheduled")); expect((scheduled.body as { total: number }).total).toBe(1); + const estimate = handleRuns(data, new URLSearchParams("origin=estimate")); + expect((estimate.body as { total: number }).total).toBe(0); + const byKey = handleRuns(data, new URLSearchParams("taskKey=PROJ-1")); expect((byKey.body as { total: number }).total).toBe(1); }); diff --git a/packages/code/tests/estimation-acquirer.test.ts b/packages/code/tests/estimation-acquirer.test.ts new file mode 100644 index 0000000..140bec1 --- /dev/null +++ b/packages/code/tests/estimation-acquirer.test.ts @@ -0,0 +1,207 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { rmSync } from "fs"; +import { join } from "path"; +import { tmpdir } from "os"; + +import { + AUTOMATION_ID_ENV, + AUTOMATION_ORIGIN_ENV, + MAX_TIMER_DELAY_MS, +} from "../src/lib/automation-acquirer"; +import { AutomationStateStore } from "../src/lib/automation-state"; +import { + estimationCliArgs, + estimationRunEnv, + EstimationAcquirer, +} from "../src/lib/estimation-acquirer"; +import type { EstimationConfig } from "../src/lib/estimation-config"; + +function intervalEntry(id: string, overrides: Partial = {}): EstimationConfig { + return { + id, + enabled: true, + query: "labels IN (NeedsEstimate)", + interval: "15m", + intervalMs: 900_000, + ...overrides, + }; +} + +describe("estimationCliArgs", () => { + test("runs the one-shot estimate engine with git disabled", () => { + expect(estimationCliArgs(intervalEntry("groom"))).toEqual([ + "--estimate", + "--query", + "labels IN (NeedsEstimate)", + "--no-git", + ]); + }); + + test("stamps the distinct estimate origin plus the owning schedule id", () => { + const env = estimationRunEnv(intervalEntry("groom"), { TASK_TRACKER: "jira" }); + expect(env[AUTOMATION_ORIGIN_ENV]).toBe("estimate"); + expect(env[AUTOMATION_ID_ENV]).toBe("groom"); + expect(env.TASK_TRACKER).toBe("jira"); + }); +}); + +describe("EstimationAcquirer", () => { + const dbPaths: string[] = []; + afterEach(() => { + for (const path of dbPaths.splice(0)) { + for (const suffix of ["", "-wal", "-shm"]) rmSync(`${path}${suffix}`, { force: true }); + } + }); + + function newDb(): string { + const dbPath = join(tmpdir(), `estimator-${Date.now()}-${Math.random()}.db`); + dbPaths.push(dbPath); + return dbPath; + } + + function noopTimers() { + return { + setTimer: () => 1 as unknown as ReturnType, + clearTimer: () => {}, + }; + } + + test("is a distinct scheduled source beside the automations scheduler", () => { + const acquirer = new EstimationAcquirer({ + estimations: [intervalEntry("groom")], + dbPath: newDb(), + resolveContext: async () => null, + }); + expect(acquirer.name).toBe("scheduled-estimations"); + }); + + test("namespaces durable schedule state away from automation ids", async () => { + const dbPath = newDb(); + let now = 0; + let runs = 0; + const acquirer = new EstimationAcquirer({ + estimations: [intervalEntry("weekday-groom")], + dbPath, + now: () => now, + ...noopTimers(), + resolveContext: async () => ({ cwd: "/tmp", env: {}, release() {} }), + spawnRun: () => { + runs += 1; + return { completion: Promise.resolve(true), terminate() {} }; + }, + }); + + await acquirer.start(); + now = 900_000; + await acquirer.tick(); + + // State lives under "estimation:weekday-groom"; an unrelated automation + // row sharing the plain id is untouched. + const store = new AutomationStateStore(dbPath); + expect(store.get("weekday-groom")).toBeNull(); + expect(store.get("estimation:weekday-groom")?.nextDueAt).toBe(1_800_000); + store.close(); + + expect(runs).toBe(1); + await acquirer.stop(); + }); + + test("hands the sweep runner the estimation entry and run context", async () => { + const dbPath = newDb(); + let now = 0; + const calls: Array<{ id: string; cwd: string; origin?: string }> = []; + const finishers: Array<(ok: boolean) => void> = []; + const acquirer = new EstimationAcquirer({ + estimations: [intervalEntry("sprint-gaps"), intervalEntry("off", { enabled: false })], + dbPath, + now: () => now, + ...noopTimers(), + resolveContext: async () => ({ cwd: "/workspace-home", env: {}, release() {} }), + spawnRun: (estimation, context) => { + calls.push({ + id: estimation.id, + cwd: context.cwd, + origin: context.env[AUTOMATION_ORIGIN_ENV], + }); + let finish!: (ok: boolean) => void; + const completion = new Promise((resolve) => (finish = resolve)); + finishers.push(finish); + return { completion, terminate() {} }; + }, + }); + + await acquirer.start(); + now = 900_000; + await acquirer.tick(); + + expect(calls).toEqual([{ id: "sprint-gaps", cwd: "/workspace-home", origin: undefined }]); + for (const finish of finishers) finish(true); + await new Promise((resolve) => setTimeout(resolve, 0)); + await acquirer.stop(); + }); + + test("disabled entries are never scheduled", async () => { + const dbPath = newDb(); + const acquirer = new EstimationAcquirer({ + estimations: [intervalEntry("off", { enabled: false })], + dbPath, + resolveContext: async () => null, + }); + await acquirer.start(); + await acquirer.stop(); + const store = new AutomationStateStore(dbPath); + expect(store.get("estimation:off")).toBeNull(); + store.close(); + }); + + test("caps long timer delays at the runtime maximum", async () => { + const dbPath = newDb(); + const delays: number[] = []; + const acquirer = new EstimationAcquirer({ + estimations: [intervalEntry("long", { interval: "30d", intervalMs: 30 * 86_400_000 })], + dbPath, + now: () => 0, + setTimer: (_callback, delay) => { + delays.push(delay); + return 1 as unknown as ReturnType; + }, + clearTimer: () => {}, + resolveContext: async () => null, + }); + await acquirer.start(); + expect(delays.at(-1)).toBe(MAX_TIMER_DELAY_MS); + await acquirer.stop(); + }); + + test("context decline releases the claim so later occurrences still run", async () => { + const dbPath = newDb(); + let now = 0; + let runs = 0; + let declinedOnce = false; + const acquirer = new EstimationAcquirer({ + estimations: [intervalEntry("busy")], + dbPath, + now: () => now, + ...noopTimers(), + resolveContext: async () => { + if (!declinedOnce) { + declinedOnce = true; + return null; + } + return { cwd: "/tmp", env: {}, release() {} }; + }, + spawnRun: () => { + runs += 1; + return { completion: Promise.resolve(true), terminate() {} }; + }, + }); + + await acquirer.start(); + now = 900_000; + await acquirer.tick(); // occurrence skipped; cursor advances to 1_800_000 + now = 1_800_000; + await acquirer.tick(); + expect(runs).toBe(1); + await acquirer.stop(); + }); +}); diff --git a/packages/code/tests/estimation-config.test.ts b/packages/code/tests/estimation-config.test.ts new file mode 100644 index 0000000..5fbc472 --- /dev/null +++ b/packages/code/tests/estimation-config.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, test } from "bun:test"; + +import { parseEstimationEntries } from "../src/lib/estimation-config"; + +function firstError(value: unknown): string { + const { errors } = parseEstimationEntries(value); + return errors.join("\n"); +} + +describe("parseEstimationEntries", () => { + test("omitted or null means estimation is off", () => { + expect(parseEstimationEntries(undefined)).toEqual({ estimations: [], errors: [] }); + expect(parseEstimationEntries(null)).toEqual({ estimations: [], errors: [] }); + }); + + test("parses cron and interval entries", () => { + const { estimations, errors } = parseEstimationEntries([ + { + id: "weekday-groom", + enabled: true, + cron: "0 9 * * 1-5", + query: "status = 'To Do' AND labels IN (NeedsEstimate)", + }, + { + id: "sprint-gaps", + enabled: false, + interval: "6h", + query: "sprint in openSprints()", + }, + ]); + expect(errors).toEqual([]); + expect(estimations).toHaveLength(2); + expect(estimations[0]).toMatchObject({ + id: "weekday-groom", + enabled: true, + cron: "0 9 * * 1-5", + }); + expect(estimations[1]).toMatchObject({ + id: "sprint-gaps", + enabled: false, + intervalMs: 21_600_000, + }); + }); + + test("requires a non-empty query", () => { + const error = firstError([{ id: "a", enabled: true, cron: "0 9 * * 1-5" }]); + expect(error).toContain("[[estimations]][0].query is required."); + expect(firstError([{ id: "a", enabled: true, cron: "0 9 * * 1-5", query: " " }])).toContain( + "[[estimations]][0].query must be a non-empty string.", + ); + }); + + test("rejects prompt and repo — estimation is not an implement job", () => { + const error = firstError([ + { + id: "a", + enabled: true, + cron: "0 9 * * 1-5", + query: "q", + prompt: "implement this", + repo: "backend", + }, + ]); + expect(error).toContain("[[estimations]][0].prompt is not supported."); + expect(error).toContain("[[estimations]][0].repo is not supported."); + }); + + test("rejects kind — there is no estimate flavor of automations", () => { + const error = firstError([ + { id: "a", enabled: true, cron: "0 9 * * 1-5", query: "q", kind: "estimate" }, + ]); + expect(error).toContain("[[estimations]][0].kind is not supported."); + }); + + test("rejects duplicate ids and invalid ids", () => { + const error = firstError([ + { id: "dup", enabled: true, cron: "0 9 * * 1-5", query: "q" }, + { id: "dup", enabled: true, cron: "0 10 * * 1-5", query: "q" }, + ]); + expect(error).toContain('Duplicate estimation id "dup"'); + expect(firstError([{ id: "", enabled: true, cron: "0 9 * * 1-5", query: "q" }])).toContain( + "[[estimations]][0].id is required.", + ); + expect( + firstError([{ id: "has space", enabled: true, cron: "0 9 * * 1-5", query: "q" }]), + ).toContain("[[estimations]][0].id must contain only letters"); + }); + + test.each([ + [ + "enabled missing", + { id: "a", cron: "0 9 * * 1-5", query: "q" }, + ".enabled must be a boolean.", + ], + [ + "neither cron nor interval", + { id: "a", enabled: true, query: "q" }, + "must set exactly one of cron or interval.", + ], + [ + "both cron and interval", + { id: "a", enabled: true, cron: "0 9 * * 1-5", interval: "6h", query: "q" }, + "must set exactly one of cron or interval.", + ], + ])("%s", (_label, entry, expected) => { + expect(firstError([entry])).toContain(expected); + }); + + test("validates the schedule grammar like [[automations]]", () => { + expect(firstError([{ id: "a", enabled: true, cron: "daily noon", query: "q" }])).toContain( + "[[estimations]][0].cron must be a five-field cron expression.", + ); + expect(firstError([{ id: "a", enabled: true, cron: "99 99 * * *", query: "q" }])).toContain( + "[[estimations]][0].cron is invalid:", + ); + expect(firstError([{ id: "a", enabled: true, interval: "45x", query: "q" }])).toContain( + "[[estimations]][0].interval must use a positive duration such as 15m, 6h, or 1d.", + ); + }); + + test("reports malformed table shapes", () => { + expect(firstError("nope")).toContain("[[estimations]] must be an array of tables."); + expect(firstError(["nope"])).toContain("[[estimations]][0] must be a table."); + }); +}); diff --git a/packages/code/tests/run-coordinator.test.ts b/packages/code/tests/run-coordinator.test.ts new file mode 100644 index 0000000..40b53c2 --- /dev/null +++ b/packages/code/tests/run-coordinator.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, test } from "bun:test"; + +import { RunCoordinator } from "../src/lib/run-coordinator"; + +describe("RunCoordinator", () => { + test("runs held operations one at a time in FIFO order", async () => { + const coordinator = new RunCoordinator(); + const events: string[] = []; + + const first = coordinator.run(async () => { + events.push("first:start"); + await new Promise((resolve) => setTimeout(resolve, 10)); + events.push("first:end"); + return "a"; + }); + const second = coordinator.run(async () => { + events.push("second:start"); + events.push("second:end"); + return "b"; + }); + + // Both are queued before either finishes; FIFO order must hold. + await Promise.all([first, second]); + expect(events).toEqual(["first:start", "first:end", "second:start", "second:end"]); + expect(await first).toBe("a"); + }); + + test("releases on failure so the next operation still runs", async () => { + const coordinator = new RunCoordinator(); + await expect( + coordinator.run(async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + let ran = false; + await coordinator.run(async () => { + ran = true; + }); + expect(ran).toBe(true); + }); + + test("waits for an explicitly acquired slot to be released", async () => { + const coordinator = new RunCoordinator(); + let secondRan = false; + let firstRelease!: () => void; + + const first = coordinator.acquire().then((release) => { + firstRelease = release; + }); + await first; + + const second = coordinator.run(async () => { + secondRan = true; + }); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(secondRan).toBe(false); + + firstRelease(); + await second; + expect(secondRan).toBe(true); + }); +}); diff --git a/packages/code/tests/run-recorder.test.ts b/packages/code/tests/run-recorder.test.ts index aec3932..cc78a21 100644 --- a/packages/code/tests/run-recorder.test.ts +++ b/packages/code/tests/run-recorder.test.ts @@ -91,6 +91,20 @@ describe("RunStore", () => { expect(store.getStats(null).byOrigin.scheduled).toBe(1); }); + test("estimate runs are a distinct origin carrying the schedule id", () => { + const id = store.createRun({ + origin: "estimate", + taskKey: "PROJ-9", + automationId: "weekday-groom", + }); + store.finishRun(id, "succeeded"); + + expect(store.getRun(id)).toMatchObject({ origin: "estimate", automationId: "weekday-groom" }); + const stats = store.getStats(null).byOrigin; + expect(stats.estimate).toBe(1); + expect(stats.scheduled).toBe(0); + }); + test("stages accumulate in order with structured detail", () => { const id = store.createRun({ origin: "task", taskKey: "PROJ-2" }); store.addStage(id, "feasibility", "succeeded", "clear enough", '{"clarityScore":8}'); diff --git a/packages/code/tests/workspace-config.test.ts b/packages/code/tests/workspace-config.test.ts index b628d66..0692e61 100644 --- a/packages/code/tests/workspace-config.test.ts +++ b/packages/code/tests/workspace-config.test.ts @@ -160,6 +160,136 @@ tracker = "fossil" ).toThrow(/does not support polling/); }); + describe("[[estimations]]", () => { + function configWithEstimations(table: string): string { + return ` +[defaults] +tracker = "jira" +task_query = "labels = devintern" + +[[repos]] +name = "backend" +remote = "git@github.com:acme/backend.git" + +${table} +`; + } + + test("parses scheduled estimation sweeps", () => { + const config = parseWorkspaceConfig( + configWithEstimations(` +[[estimations]] +id = "weekday-groom" +enabled = true +cron = "0 9 * * 1-5" +query = "status = 'To Do' AND labels IN (NeedsEstimate)" + +[[estimations]] +id = "sprint-gaps" +enabled = true +cron = "0 10 * * 3" +query = "sprint in openSprints() AND \\"Story Points\\" is EMPTY" +`), + ); + + expect(config.estimations).toHaveLength(2); + expect(config.estimations[0]).toMatchObject({ + id: "weekday-groom", + enabled: true, + cron: "0 9 * * 1-5", + query: "status = 'To Do' AND labels IN (NeedsEstimate)", + }); + expect(config.estimations[1]?.query).toContain("is EMPTY"); + }); + + test("an omitted table leaves estimation off", () => { + expect(parseWorkspaceConfig(VALID_CONFIG).estimations).toEqual([]); + }); + + test("requires no repo or routing even with several repos", () => { + const config = parseWorkspaceConfig( + configWithEstimations(` +[[estimations]] +id = "groom" +enabled = true +interval = "1d" +query = "labels IN (NeedsEstimate)" +`), + ); + // Estimation entries carry only schedule + query fields. + expect(config.estimations[0]).toEqual({ + id: "groom", + enabled: true, + interval: "1d", + intervalMs: 86_400_000, + query: "labels IN (NeedsEstimate)", + }); + }); + + test("rejects prompt and repo keys", () => { + expect(() => + parseWorkspaceConfig( + configWithEstimations(` +[[estimations]] +id = "groom" +enabled = true +interval = "1d" +query = "q" +prompt = "implement it" +repo = "backend" +`), + ), + ).toThrow(/prompt is not supported[\s\S]*repo is not supported/s); + }); + + test("rejects the kind selector", () => { + expect(() => + parseWorkspaceConfig( + configWithEstimations(` +[[estimations]] +id = "groom" +enabled = true +interval = "1d" +query = "q" +kind = "estimate" +`), + ), + ).toThrow(/kind is not supported/); + }); + + test("rejects [defaults].estimate_query", () => { + expect(() => + parseWorkspaceConfig(` +[defaults] +tracker = "jira" +estimate_query = "labels = NeedsEstimate" +`), + ).toThrow(/\[defaults\]\.estimate_query is not supported/); + }); + + test.each(["trello", "markdown"])( + "fails startup on trackers that cannot estimate (%s)", + (tracker) => { + expect(() => + parseWorkspaceConfig(` +[defaults] +tracker = "${tracker}" + +[[repos]] +name = "board" +remote = "git@github.com:acme/board.git" + +[[estimations]] +id = "groom" +enabled = true +interval = "1d" +query = "list:'To Do'" +`), + ).toThrow(/requires a tracker that supports --estimate/); + }, + ); + }); + test("rejects duplicate repo names", () => { expect(() => parseWorkspaceConfig(` diff --git a/packages/code/tests/workspace-worker.test.ts b/packages/code/tests/workspace-worker.test.ts index a48082e..368b08a 100644 --- a/packages/code/tests/workspace-worker.test.ts +++ b/packages/code/tests/workspace-worker.test.ts @@ -7,6 +7,7 @@ import { parseWorkspaceConfig } from "../src/lib/workspace/config"; import type { RepoConfig } from "../src/lib/workspace/config"; import { buildFleetEventAcquirers, + createFleetTaskExecutor, createWorkspaceTaskAcquirer, fleetTaskArgs, resolveWorkspaceAutomationContext, @@ -15,6 +16,8 @@ import type { FleetTask, RepoManagerLike } from "../src/lib/workspace/workspace- import { createRepoRunLock, openWorkspaceState } from "../src/lib/workspace/state"; import type { WorkspaceState } from "../src/lib/workspace/state"; import type { ChangeDetector } from "../src/lib/change-detector"; +import { RunCoordinator } from "../src/lib/run-coordinator"; +import { toRoutableTask } from "../src/lib/workspace/router"; import { saveRelayState } from "../src/lib/relay-connect"; const CONFIG = parseWorkspaceConfig(` @@ -244,11 +247,76 @@ describe("createWorkspaceTaskAcquirer", () => { }); }); +describe("createFleetTaskExecutor serialization", () => { + let workspaceDir: string; + let state: WorkspaceState; + let repoManager: FakeRepoManager; + + beforeEach(() => { + workspaceDir = join(tmpdir(), `ws-coord-${Date.now()}-${Math.random().toString(36).slice(2)}`); + mkdirSync(workspaceDir, { recursive: true }); + state = openWorkspaceState(workspaceDir); + repoManager = new FakeRepoManager(workspaceDir); + }); + + afterEach(() => { + state.close(); + rmSync(workspaceDir, { recursive: true, force: true }); + }); + + test("with a coordinator, task runs wait for held agent slots (account-global limits)", async () => { + const coordinator = new RunCoordinator(); + let agentStarted = false; + const executor = createFleetTaskExecutor({ + config: CONFIG, + workspaceDir, + skips: state.skips, + repoManager, + runTask: async () => { + agentStarted = true; + await new Promise((resolve) => setTimeout(resolve, 10)); + return true; + }, + repoLock: (name) => createRepoRunLock(name, workspaceDir), + coordinator, + }); + + // A scheduled estimation sweep holds the process-level slot first. + const releaseEstimation = await coordinator.acquire(); + const routable = toRoutableTask({ key: "T-C1", labels: ["backend"], components: [] }); + const running = executor("T-C1", routable); + + await new Promise((resolve) => setTimeout(resolve, 30)); + // Repo preparation may proceed, but no agent process starts while the + // estimation sweep holds the account-global slot. + expect(agentStarted).toBe(false); + expect(repoManager.calls).toContain("worktree:backend:T-C1"); + + releaseEstimation(); + await running; + expect(agentStarted).toBe(true); + }); + + test("without a coordinator, runs start immediately (unchanged legacy behavior)", async () => { + const executor = createFleetTaskExecutor({ + config: CONFIG, + workspaceDir, + skips: state.skips, + repoManager, + runTask: async () => true, + repoLock: (name) => createRepoRunLock(name, workspaceDir), + }); + const routable = toRoutableTask({ key: "T-C2", labels: ["backend"], components: [] }); + const result = await executor("T-C2", routable); + expect(result).toBe(true); + expect(repoManager.calls).toContain("worktree:backend:T-C2"); + }); +}); + describe("fleetTaskArgs", () => { test("uses worker_task_args from the workspace config", () => { expect(fleetTaskArgs(CONFIG)).toEqual(["--create-pr", "--auto-review"]); }); - test("defaults to --create-pr when worker_task_args is omitted", () => { const config = parseWorkspaceConfig(` [defaults] diff --git a/packages/dashboard-ui/src/lib/api.ts b/packages/dashboard-ui/src/lib/api.ts index 9600f11..4c18fce 100644 --- a/packages/dashboard-ui/src/lib/api.ts +++ b/packages/dashboard-ui/src/lib/api.ts @@ -5,7 +5,7 @@ import { useCallback, useEffect, useRef, useState } from "react"; -export type RunOrigin = "task" | "pr_mention" | "conflict_resolution" | "scheduled"; +export type RunOrigin = "task" | "pr_mention" | "conflict_resolution" | "scheduled" | "estimate"; export type RunStatus = | "in_progress" diff --git a/packages/dashboard-ui/src/lib/run-origin.test.ts b/packages/dashboard-ui/src/lib/run-origin.test.ts index 1abcee3..ffb446f 100644 --- a/packages/dashboard-ui/src/lib/run-origin.test.ts +++ b/packages/dashboard-ui/src/lib/run-origin.test.ts @@ -8,6 +8,7 @@ test.each([ ["pr_mention", "PR mention"], ["conflict_resolution", "Conflict resolution"], ["scheduled", "Scheduled automation"], + ["estimate", "Estimate run"], ] satisfies [RunOrigin, string][])("labels the %s run origin", (origin, label) => { expect(formatRunOrigin(origin)).toBe(label); }); diff --git a/packages/dashboard-ui/src/lib/run-origin.ts b/packages/dashboard-ui/src/lib/run-origin.ts index 43b1a31..cec8a66 100644 --- a/packages/dashboard-ui/src/lib/run-origin.ts +++ b/packages/dashboard-ui/src/lib/run-origin.ts @@ -5,6 +5,7 @@ export const RUN_ORIGIN_LABELS = { pr_mention: "PR mention", conflict_resolution: "Conflict resolution", scheduled: "Scheduled automation", + estimate: "Estimate run", } satisfies Record; export function formatRunOrigin(origin: RunOrigin): string { diff --git a/packages/dashboard-ui/src/views/RunsView.tsx b/packages/dashboard-ui/src/views/RunsView.tsx index 74a8bbf..3f04cb8 100644 --- a/packages/dashboard-ui/src/views/RunsView.tsx +++ b/packages/dashboard-ui/src/views/RunsView.tsx @@ -35,6 +35,7 @@ const ORIGIN_FILTERS: (RunOrigin | "all")[] = [ "pr_mention", "conflict_resolution", "scheduled", + "estimate", ]; /** Paginated, filterable list of worker runs. */ diff --git a/packages/dashboard-ui/src/views/StatsView.tsx b/packages/dashboard-ui/src/views/StatsView.tsx index 8748185..64868f0 100644 --- a/packages/dashboard-ui/src/views/StatsView.tsx +++ b/packages/dashboard-ui/src/views/StatsView.tsx @@ -16,7 +16,13 @@ import { formatRunOrigin } from "@/lib/run-origin"; import { formatDuration, formatRate } from "@/lib/utils"; const WINDOWS = ["7d", "30d", "90d", "all"] as const; -const RUN_ORIGINS: RunOrigin[] = ["task", "pr_mention", "conflict_resolution", "scheduled"]; +const RUN_ORIGINS: RunOrigin[] = [ + "task", + "pr_mention", + "conflict_resolution", + "scheduled", + "estimate", +]; /** Hand-rolled SVG bar chart of runs per week (keeps dependencies minimal). */ function WeeklyBars({ weeks }: { weeks: { weekStart: string; count: number }[] }) {