diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8a5c4b2..f97d9a8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,6 +14,7 @@ concurrency: jobs: test: runs-on: ubuntu-latest + timeout-minutes: 15 strategy: fail-fast: false matrix: @@ -29,14 +30,42 @@ jobs: - name: Lint and test run: | uv run --no-sync ruff check . + uv run --no-sync ruff format --check . uv run --no-sync pytest - name: Build and verify installed wheel outside the checkout run: | - uv build --no-sources + uv build --no-sources --out-dir .agent-backlog/dist uv venv --python "${{ matrix.python }}" .wheel-smoke - uv pip install --python .wheel-smoke/bin/python dist/*.whl + uv pip install --python .wheel-smoke/bin/python .agent-backlog/dist/*.whl smoke_python="$PWD/.wheel-smoke/bin/python" smoke_dir=$(mktemp -d) cd "$smoke_dir" "$smoke_python" -m agent_backlog_runner.demo --output demo "$smoke_python" -m agent_backlog_runner.cli status --db demo/tasks.sqlite3 --format json + "$smoke_python" -m agent_backlog_runner.cli --version + "$smoke_python" -m agent_backlog_runner.cli demo --output maintenance + "$smoke_python" -m agent_backlog_runner.cli status --db maintenance/checks.sqlite3 --format json + macos-processes: + runs-on: macos-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.14' + - uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0 + - name: Install dependencies + run: uv sync --locked --extra dev --python 3.14 + - name: Check attempts, parallel workers, recovery and process groups + run: uv run --no-sync pytest tests/test_batch_execution.py tests/test_dispatch.py + - name: Verify installed repair demo outside the checkout + run: | + uv build --no-sources --out-dir .agent-backlog/dist + uv venv --python 3.14 .wheel-smoke + uv pip install --python .wheel-smoke/bin/python .agent-backlog/dist/*.whl + smoke_python="$PWD/.wheel-smoke/bin/python" + smoke_dir=$(mktemp -d) + cd "$smoke_dir" + "$smoke_python" -m agent_backlog_runner.cli --version + "$smoke_python" -m agent_backlog_runner.cli demo --output maintenance + "$smoke_python" -m agent_backlog_runner.cli status --db maintenance/checks.sqlite3 --format json diff --git a/CHANGELOG.md b/CHANGELOG.md index 31f6b03..da407d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,40 @@ # Changelog +## 0.3.0 + +- Run local commands concurrently with a real shared SQLite capacity cap. `--drain` + processes a finite queued snapshot, filling slots as workers exit. +- Add idempotent finite command manifests, explicit bounded retries with capped + exponential backoff, per-task timeouts, and durable token-owned attempt history. +- Store separate capped stdout/stderr files, drain both streams without unbounded + capture, and record observed bytes/truncation plus short event tails. +- Add `cancel`, conservative `recover`, `attempts`, `logs`, `report`, `demo`, and + `--version` CLI paths. Cancellation is handled by the owning coordinator; + recovery never signals stored PIDs and never silently retries uncertain work. +- Stop owned POSIX process groups on timeout/cancel/interruption. Keep leaders + unreaped during group signalling; document the older-macOS normal-exit fallback. +- Add an executable repository-maintenance repair/resume walkthrough and a + reusable pytest/Ruff manifest, plus targeted concurrency/crash/log/retry tests. + +### Compatibility and migration + +Schema 1 migrates automatically to schema 2, preserving task/event/scheduler data. +Old tasks receive three future attempts and a one-second base backoff; past +execution history cannot be reconstructed. Old running tasks retain their state +and require explicit inspection and unknown-identity recovery. Version 0.2 +refuses schema 2; retain an earlier copy before upgrading if rollback is needed. + +Default dispatch still selects a bounded batch and defaults to preview, but its +workers now execute concurrently. Only `--drain` waits for future retry due times. +Worker failure/cancellation now returns CLI exit `1`, blocked capacity/budget `2`, +and Ctrl-C `130`. `DispatchResult` adds completed/failed/cancelled ids; existing +`started` and `skipped_reason` fields remain. Existing string commands/placeholders +and scheduler APIs remain available. Duplicate YAML/JSON fields are now rejected. + +For managed attempts, use `cancel`, `retry`, and `recover`; the CLI rejects raw +status edits that would bypass live-attempt ownership or retry budgets. Low-level +manual `mark_task`/`claim_task` bookkeeping APIs remain for compatibility. + ## 0.2.0 - Claim a task and a capacity slot atomically before a subprocess starts. diff --git a/README.md b/README.md index 70c7411..a3769d5 100644 --- a/README.md +++ b/README.md @@ -2,164 +2,142 @@ [![CI](https://github.com/matchalatteguy/agent-backlog-runner/actions/workflows/ci.yml/badge.svg)](https://github.com/matchalatteguy/agent-backlog-runner/actions/workflows/ci.yml) -Keep a small local task queue replenished from a validated JSON or YAML template -catalog. Preview the next batch, store tasks and events in SQLite, then run short -worker commands and inspect what completed or failed. +A local command queue for repository checks and repeatable maintenance. Load a finite batch, run independent commands in parallel, inspect each attempt's exit code and logs, repair failures, and resume without repeating completed work. SQLite stores the queue and attempt history across CLI invocations. -**This project is a queue replenisher with a synchronous dispatcher.** Its sibling -[Local Agent Task Runtime](https://github.com/matchalatteguy/local-agent-task-runtime) -manages registered tasks in detached sessions and contained workspaces. Choose -this runner for recurring template work and short commands; choose the runtime -when independent workers need to outlive the dispatching process. +Use it when a shell loop is becoming hard to resume or debug. Templates can also replenish a recurring backlog below a chosen depth. Commands run while the dispatcher is open; detached sessions, workspace provisioning, dependency graphs, and distributed workers are outside this tool's scope. -```text -template catalog -> depth plan -> SQLite queue -> local command -> status + events +## Install and try the repair/resume walkthrough + +Python 3.11+ on Linux or macOS. Version 0.3 is an alpha release; unfinished attempts require operator inspection and explicit recovery. + +```bash +uv tool install 'git+https://github.com/matchalatteguy/agent-backlog-runner.git@v0.3.0' +agent-backlog --help ``` -## Run a real example +From a checkout, use `uv sync --locked --extra dev` and prefix commands with `uv run`. -Requires Python 3.11+ and [uv](https://docs.astral.sh/uv/). From a fresh checkout: +The packaged demonstration requires no test framework or external service: ```bash -uv sync --locked --extra dev -uv run python -m agent_backlog_runner.demo --output .agent-backlog/demo +agent-backlog demo --output .agent-backlog/maintenance-demo ``` -The demo creates three Markdown documents, previews a queue replenishment plan, -enqueues three audits, then executes a real Python worker for each document. -Workers count headings and TODO lines, write JSON reports, and send heartbeats. -The final report is read after reopening SQLite: - -```json -{ - "completed_tasks": 3, - "created_tasks": 3, - "follow_up_items": 3, - "reports": [ - {"document": "onboarding.md", "headings": 1, "todo_items": 1}, - {"document": "release.md", "headings": 2, "todo_items": 0}, - {"document": "troubleshooting.md", "headings": 1, "todo_items": 2} - ] -} -``` +From a checkout, run `uv run agent-backlog demo --output .agent-backlog/maintenance-demo`. -Whitespace in the printed JSON differs. Outputs are under -`.agent-backlog/demo/reports/`; task state is in -`.agent-backlog/demo/tasks.sqlite3`. Use a new `--output` directory for another -run; the demo refuses to overwrite an existing directory. +The walkthrough creates a tiny Python project, runs its unit tests, reopens the queue, then runs syntax and README-link checks. The missing usage document fails its check. It writes the document, explicitly schedules one retry, and resumes through another SQLite connection: -```bash -uv run agent-backlog status --db .agent-backlog/demo/tasks.sqlite3 -uv run agent-backlog events --db .agent-backlog/demo/tasks.sqlite3 --limit 20 +```text +checks: 3 +initial_completed: 1 +resume_failed: 1 +repaired_checks: 1 +attempts: 4 +final_done: 3 ``` -## Plan and replenish your own queue +[Implementation](src/agent_backlog_runner/maintenance.py). These are real commands on illustrative files, not a claim of production use. Inspect `before-repair.md`, `after-repair.md`, and `logs/` under the output directory. The demo refuses to overwrite an existing output directory. + +## Run checks on your own repository -A template describes a repeatable unit of work: +Save a trusted manifest as `checks.yaml`: ```yaml -templates: - - slug: docs-refresh - title: "Review documentation ${sequence}" - body: "Review one documentation page and record an improvement." - lane: docs - role: writer - tags: [docs] +schema_version: 1 +name: repo-checks +tasks: + - id: unit-tests + argv: ['{python}', '-m', pytest, '-q'] + timeout_seconds: 600 + max_attempts: 3 + retry_backoff_seconds: 2 + - id: lint-check + argv: ['{python}', '-m', ruff, check, '.'] + timeout_seconds: 120 + max_attempts: 3 ``` -Try the bundled catalog in a fresh database: +Use your project's interpreter, with its test dependencies installed: ```bash -uv run agent-backlog templates validate --templates examples/templates/basic-backlog.yaml -uv run agent-backlog scheduler plan --db .agent-backlog/queue.sqlite3 \ - --templates examples/templates/basic-backlog.yaml --min-depth 2 --target-depth 4 --max-enqueue 3 --json -uv run agent-backlog scheduler run --db .agent-backlog/queue.sqlite3 \ - --templates examples/templates/basic-backlog.yaml --min-depth 2 --target-depth 4 --max-enqueue 3 +agent-backlog batch load checks.yaml --root . \ + --python "$PWD/.venv/bin/python" --run-id first-pass \ + --db .agent-backlog/checks.sqlite3 + +# Preview; commands do not run until --backend subprocess is selected. +agent-backlog dispatch --db .agent-backlog/checks.sqlite3 --max-workers 2 --drain +agent-backlog dispatch --db .agent-backlog/checks.sqlite3 \ + --backend subprocess --max-workers 2 --drain + +agent-backlog status --db .agent-backlog/checks.sqlite3 +agent-backlog attempts --db .agent-backlog/checks.sqlite3 +agent-backlog report --db .agent-backlog/checks.sqlite3 \ + --json-out .agent-backlog/checks.json --md-out .agent-backlog/checks.md ``` -Expected: `valid templates: 3`, a preview containing three planned tasks, then -`created 3`. Active depth counts `todo` and `running`. A later replenishment run -adds nothing while depth remains at or above the minimum. Titles are checked for -duplicates, including duplicates within the same plan. +`--max-workers 2` really runs two commands concurrently. Claims and the shared capacity are reserved atomically; cancelled attempts retain their slots until stopped. Concurrent coordinators must use the same cap. Avoid putting commands that write the same files into one parallel batch. -Planning creates no tasks or events and does not advance the cursor. The CLI may -initialize the selected SQLite database on first use. Applying a plan commits -its tasks, events, and cursor together; stale plans must be regenerated. +Reloading the same manifest/root/run id is idempotent. It preserves task statuses and rejects changed command definitions. Choose a new `--run-id` for a new full check pass. The [checked-in manifest](examples/repo-checks.yaml) runs this repository's tests and lint after installing development dependencies. -## Dispatch short commands +## Repair, retry, and resume -Preview selected task ids with the default backend: +A failed command stays failed. Inspect its task id and retained stderr: ```bash -uv run agent-backlog dispatch --db .agent-backlog/queue.sqlite3 --max-workers 2 +agent-backlog attempts --db .agent-backlog/checks.sqlite3 --format json +agent-backlog logs --db .agent-backlog/checks.sqlite3 TASK_ID --stream stderr +# Repair the source or configuration, then schedule failed tasks with budget remaining. +agent-backlog retry --db .agent-backlog/checks.sqlite3 --failed +agent-backlog dispatch --db .agent-backlog/checks.sqlite3 \ + --backend subprocess --max-workers 2 --drain ``` -This records `dispatch_preview` events and leaves tasks `todo`. Execute the -bundled worker explicitly: +Retry is explicit, bounded by `max_attempts`, and delayed by exponential backoff capped at 60 seconds. Spawn failures also consume attempts. `--drain` waits for scheduled retries in its initial queued snapshot, fills slots as workers finish, and leaves completed tasks untouched. Without `--drain`, one ready batch is selected; delayed work does not block later ready tasks. + +Ctrl-C stops owned workers and records interrupted attempts as failed. Queued tasks remain ready. An abrupt dispatcher crash leaves unfinished work requiring [explicit recovery](docs/batches-and-recovery.md#interruption-and-recovery), because its side effects may already have happened. This is at-least-once execution after an operator chooses recovery and retry. + +| Dispatch exit | Meaning | +|---|---| +| `0` | Selected work passed, preview completed, or no eligible work was selected. | +| `1` | A selected worker failed/cancelled, or a CLI input/operation failed. | +| `2` | Capacity or exhausted attempts prevented dispatch. | +| `130` | Ctrl-C interrupted the dispatcher. | + +Use `status` to distinguish an empty queue from future scheduled work. JSON dispatch output lists started, completed, failed, and cancelled task ids. + +## Recurring template work + +The original queue-replenishment workflow remains available: ```bash -uv run agent-backlog dispatch --db .agent-backlog/queue.sqlite3 \ - --backend subprocess --max-workers 1 --timeout 60 \ - --command "bash examples/worker_echo.sh {task_id}" +agent-backlog scheduler plan --db .agent-backlog/recurring.sqlite3 \ + --templates examples/templates/basic-backlog.yaml \ + --min-depth 2 --target-depth 4 --max-enqueue 3 --json +agent-backlog scheduler run --db .agent-backlog/recurring.sqlite3 \ + --templates examples/templates/basic-backlog.yaml \ + --min-depth 2 --target-depth 4 --max-enqueue 3 ``` -A zero exit code marks the task `done`. Nonzero exits, timeouts, missing binaries, -and invalid workdirs mark it `failed`; later tasks in the selected batch still -run. `--max-workers` bounds the batch against current running work. Commands run -one at a time inside the CLI process, even when the limit is larger than one. -Output snippets are decoded as UTF-8 with replacement characters for invalid -bytes, so binary or differently encoded output cannot strand the batch. - -Each task can have its own `--command` and `--workdir` through `enqueue`. A -command supplied to `dispatch` overrides those commands. Supported placeholders -are `{task_id}`, `{title}`, `{lane}`, `{role}`, `{priority}`, and `{workdir}`. -Shell quoting is parsed **before** substitution, preserving each placeholder as -one argument. No shell is started automatically. Treat command templates as -trusted code, especially if you invoke a shell or interpolate into program text. - -Atomic claims prevent two local dispatchers from executing the same queued task. -Completion updates preserve an operator's intervening cancellation or status -change; changing a status does not itself stop an already running process. - -## Python API - -```python -from agent_backlog_runner import BacklogPolicy, init_store, load_template_catalog -from agent_backlog_runner.scheduler import apply_backlog_plan, plan_backlog - -with init_store(".agent-backlog/api.sqlite3") as store: - catalog = load_template_catalog("examples/templates/basic-backlog.yaml") - policy = BacklogPolicy(min_queue_depth=2, target_queue_depth=4) - plan = plan_backlog(store, catalog, policy, dry_run=False) - tasks = apply_backlog_plan(store, plan) - print(len(tasks)) -``` +Templates create work descriptions. Supply per-task commands or a dispatch command when executing them. Plans check queue depth and duplicate titles; applying a stale plan is rejected. The scheduler cursor is shared across catalogs and filters. [Template schema](docs/template-schema.md) · [Scheduler policy](docs/scheduler-policy.md). + +## Logs, processes, and trust + +Each attempt retains stdout and stderr in separate files, up to 1 MiB per stream by default. Excess bytes are drained and discarded; attempts record observed byte counts and truncation flags. `--max-log-bytes` changes the per-stream cap up to 64 MiB. Full output is available only below the cap. Historical attempt files are retained until you remove or archive them yourself. -A plan marked `dry_run=True` cannot create tasks. See [CHANGELOG.md](CHANGELOG.md) -for the 0.2 API change. Every SQLite CLI command accepts `--db`; the fallback is -`AGENT_BACKLOG_DB`, then `.agent-backlog/tasks.sqlite3`. +Timeout and cancellation stop the owning dispatcher's process group on POSIX. Workers that escape their group, change privileges, or leave uninterruptible OS operations are outside that guarantee. On interpreters with `waitid(WNOWAIT)`, normal exit also cleans up remaining group members before the leader is reaped. Older macOS interpreters lack that operation: normal-exit background children cannot be safely adopted; commands must wait for their children. Windows supports queue inspection/planning but not subprocess dispatch in this release. -## Verification and boundaries +Subprocess dispatch requires the default `SIGCHLD` handler. Embedded callers must not reap its children independently; lost ownership prevents group signals and leaves unfinished work for inspection. + +Commands and manifests are trusted local code, with normal access to your files, environment, and network. Workdir containment prevents accidental manifest path escapes; it is not a process sandbox. No shell is inserted automatically. Legacy command placeholders are substituted after argument splitting. SQLite, logs, and exported reports can contain command arguments, local paths, and sensitive output; keep them in a protected ignored directory. + +## Development and documentation ```bash -uv run pytest +uv sync --locked --extra dev uv run ruff check . -uv build +uv run ruff format --check . +uv run pytest +uv build --no-sources ``` -CI tests Python 3.11–3.14 and executes the demo from an installed wheel outside -the checkout. Regression tests cover timeout recovery, competing claims, partial -plan rollback, duplicate titles, and stale plans. - -This is an alpha library for one local machine. It has no detached worker pool, -automatic retries, process sandbox, distributed leases, or exactly-once execution. -A killed dispatcher can leave `running` tasks requiring operator recovery. -`stale` reports those tasks; it does not restart them. Timeouts terminate the -direct worker process, not an arbitrary tree of child processes. The cursor is -shared across catalogs and filters; title-based deduplication is deliberately -simple. Events retain the last 500 characters of worker output, so protect the -SQLite database and keep command output free of secrets. - -[Template schema](docs/template-schema.md) · [Scheduler policy](docs/scheduler-policy.md) · -[Dispatch](docs/dispatch-backends.md) · [Python API](docs/python-api.md) · -[Troubleshooting](docs/troubleshooting.md) +CI tests Python 3.11–3.14 on Linux, plus process ownership and the repair/resume walkthrough on macOS 3.14. It covers overlapping workers, competing capacity claims, interruption, conservative recovery, bounded logs, and retries; installed-wheel examples run outside the checkout. [0.3 migration](CHANGELOG.md) · [Batch manifests and recovery](docs/batches-and-recovery.md) · [Dispatch](docs/dispatch-backends.md) · [Python API](docs/python-api.md) · [Troubleshooting](docs/troubleshooting.md) · [MIT license](LICENSE). diff --git a/docs/batches-and-recovery.md b/docs/batches-and-recovery.md new file mode 100644 index 0000000..61f70ac --- /dev/null +++ b/docs/batches-and-recovery.md @@ -0,0 +1,96 @@ +# Batch manifests and recovery + +A batch is a finite set of independent local commands. It has no dependencies or hidden retries. Use a separate batch or a script when a later command needs another command's output. + +## Manifest schema + +```yaml +schema_version: 1 +name: repo-checks +tasks: + - id: unit-tests + title: Run unit tests + argv: ['{python}', '-m', unittest, discover, '-s', tests, '-v'] + workdir: '.' + timeout_seconds: 300 + max_attempts: 3 + retry_backoff_seconds: 1 +``` + +`name` and task `id` are unique-purpose lowercase slugs of 2–64 characters. Task ids must be unique within a manifest. Unknown fields and invalid types are rejected before enqueueing. Required fields are `schema_version: 1`, `name`, a non-empty `tasks` list, and each task's `id` and non-empty `argv` list. + +| Task field | Default | Meaning | +|---|---|---| +| `title` | task id | Human-readable status label. | +| `argv` | required | Exact command arguments; no shell splitting or placeholder formatting. | +| `workdir` | `.` | Existing directory inside the batch root. | +| `timeout_seconds` | `300` | Positive seconds, maximum one day. | +| `max_attempts` | `3` | Total future attempts, including spawn failures; 1–100. | +| `retry_backoff_seconds` | `1` | Base delay between explicitly scheduled retries; 0–60. | + +The exact argument `{python}` is replaced with the current interpreter or the `--python` override. Other braces and argument strings are literal. Workdirs resolve relative to `--root`, or to the manifest's directory if no root is supplied. Containment rejects absolute paths, `..` escapes, and escaping symlinks; it does not limit what a trusted command can do. + +```bash +agent-backlog batch validate checks.yaml --root . --python "$PWD/.venv/bin/python" +agent-backlog batch load checks.yaml --root . --python "$PWD/.venv/bin/python" \ + --run-id release-check --db .agent-backlog/checks.sqlite3 +``` + +Load validates the entire manifest and commits all new tasks atomically. IDs derive from the absolute root, manifest name, run id, and manifest task id. Loading the same run again preserves existing state; changed definitions are rejected. A new `--run-id` creates a fresh pass. Moving a checkout changes its identity; use the existing queue and recorded absolute workdirs when resuming, or create a new pass after moving it. + +## Attempts and retry + +Claims atomically reserve an attempt number, token, and global capacity slot. Each attempt records the interpreted argv, workdir, owner PID, worker PID/group when spawned, start/end times, outcome, exit code, log paths, observed byte counts, and truncation flags. Completion is conditional on the attempt token, so a stale completion cannot finish a later retry. + +Workers receive `AGENT_BACKLOG_TASK_ID` and `AGENT_BACKLOG_ATTEMPT` in their environment. Failed commands remain failed until `retry TASK_ID` or `retry --failed` schedules another attempt. Cancelled work requires a specific task retry; `--failed` does not sweep cancellations. + +The delay is `min(60, base * 2 ** (attempt_count - 1))`. Delay zero is useful for an explicitly repaired local demonstration. No retry happens automatically, and the original command definition is retained. Repair inputs/configuration/source first; choose a new run id for a changed command definition. + +## Interruption and recovery + +Ctrl-C tells the active coordinator to stop all of its owned workers. It records interrupted attempts as failed and preserves pending tasks. Inspect logs, repair any partial output, explicitly retry the failed work, then resume: + +```bash +agent-backlog retry --db .agent-backlog/checks.sqlite3 --failed +agent-backlog dispatch --db .agent-backlog/checks.sqlite3 \ + --backend subprocess --max-workers 2 --drain +``` + +SIGKILL, a machine failure, or an abrupt interpreter failure can leave attempts unfinished. `stale` reports old heartbeats; it does not prove a process is dead. The coordinator updates active heartbeats about every half second. + +```bash +agent-backlog stale --db .agent-backlog/checks.sqlite3 --after 30m +agent-backlog attempts --db .agent-backlog/checks.sqlite3 TASK_ID --format json +agent-backlog recover --db .agent-backlog/checks.sqlite3 TASK_ID --after 30m +``` + +Recovery refuses while the recorded dispatcher PID, worker PID, or process group still exists. PID reuse can cause a conservative refusal; there is no force-kill switch. Inspect and stop old work yourself before trying again. Recovery never signals saved PIDs and never queues a retry by itself. It marks the abandoned attempt and preserves its history; a later retry consumes another attempt. + +A crash between claiming/spawning/binding a worker can leave its PID unknown. Untracked running tasks from the old API/schema also have no verifiable worker identity. After inspecting possible live work and partial side effects, the explicit `--acknowledge-unknown` option records your decision to abandon that uncertain work. It does not bypass a known live PID/group refusal. + +Recovered work may already have changed files or external state. Retrying gives at-least-once execution; use idempotent commands or inspect and repair their effects. There are no leases spanning multiple machines and no exactly-once guarantee. + +## Cancellation + +```bash +agent-backlog cancel --db .agent-backlog/checks.sqlite3 TASK_ID +``` + +This requests cancellation in SQLite. The owning coordinator observes it and stops its own process/session; the CLI does not kill a stored PID. `attempts` confirms when it has finished. Cancelled but unfinished attempts still consume capacity. If the coordinator is gone, inspect the remaining processes and use conservative recovery. Untracked running work must be inspected/recovered before the managed cancel path can be used. + +## Logs and reports + +`logs TASK_ID --stream stderr --attempt 1 --tail 4096` prints a bounded tail of retained file content. The latest attempt is the default. Log files retain the first configured number of bytes, not an unlimited stream; event messages retain a separate last-500-byte snippet. Excess stream data is drained and discarded to prevent pipe deadlocks and unbounded memory/disk growth per attempt. + +```bash +agent-backlog report --db .agent-backlog/checks.sqlite3 \ + --json-out .agent-backlog/report.json --md-out .agent-backlog/report.md +``` + +JSON includes task and attempt records. Markdown lists outcomes, exit codes, budgets, and links to local logs. These files can expose command arguments, paths, and output; they are local operational reports, not automatically sanitized publications. + +## Schema migration + +Opening a schema-1 database with 0.3 migrates it to schema 2. Existing tasks, events, scheduler state, and statuses are preserved. Older tasks receive a three-attempt future budget and default one-second base backoff; historical executions cannot be reconstructed. Existing running tasks remain untracked and require explicit inspection/recovery. + +Stop dispatchers and keep a copy of an older database before first opening it if you need to retain a rollback point. Version 0.2 refuses the new schema; restoring the old copy discards activity recorded since that copy. See the [changelog](../CHANGELOG.md) for API and CLI compatibility changes. diff --git a/docs/dispatch-backends.md b/docs/dispatch-backends.md index 94858f9..de1b6d5 100644 --- a/docs/dispatch-backends.md +++ b/docs/dispatch-backends.md @@ -1,63 +1,43 @@ # Dispatch backends -Dispatch turns ready `todo` tasks into local work. It is deliberately opt-in and bounded: nothing runs unless the caller chooses a backend and a worker command. - -## Dry-run backend - -The default backend is `dry-run`: +The default `dry-run` backend records preview events and leaves tasks `todo`. Commands run only with `--backend subprocess`. ```bash -uv run agent-backlog dispatch --db .agent-backlog/tasks.sqlite3 --max-workers 2 +agent-backlog dispatch --db .agent-backlog/checks.sqlite3 --max-workers 2 --drain +agent-backlog dispatch --db .agent-backlog/checks.sqlite3 \ + --backend subprocess --max-workers 2 --drain ``` -Dry-run dispatch records preview events and returns the task ids it would start. It does not execute subprocesses or change task status to `running`. +## Concurrent bounded execution -Use this mode first when checking a new backlog. +The CLI stays open while selected commands run concurrently. A shared SQLite transaction reserves each task, attempt token, and capacity slot. Competing coordinators cannot execute the same queued task, and active coordinators must agree on `--max-workers`. -## Subprocess backend +Without `--drain`, dispatch selects at most the available number of ready tasks. Future scheduled retries do not block eligible work. With `--drain`, it takes a finite snapshot of queued tasks and refills its slots until that snapshot is consumed, waiting for its explicit retry due times. Newly enqueued work is left to a later invocation. Other coordinators or unrecovered attempts can prevent dispatch; the CLI reports the capacity reason rather than adopting their processes. -The `subprocess` backend runs a local command template once for each selected task. It is synchronous: the CLI process waits for each selected command to finish or time out before moving to the next selected task. +Tasks are selected by descending priority, then creation time and insertion order. Independent commands must not concurrently overwrite the same files. There is no DAG or lock manager for command side effects. -```bash -uv run agent-backlog dispatch \ - --db .agent-backlog/tasks.sqlite3 \ - --max-workers 1 \ - --backend subprocess \ - --command "bash examples/worker_echo.sh {task_id}" \ - --timeout 60 -``` +## Command forms + +A batch's `argv` list is passed directly to `Popen`, with no shell. The exact manifest token `{python}` is resolved when loading the manifest. Legacy per-task `command` strings remain supported, with placeholders `{task_id}`, `{title}`, `{lane}`, `{role}`, `{priority}`, and `{workdir}`. Quoting is parsed before placeholder substitution, preserving each value as one argument. A dispatch `--command` overrides stored commands/argv. + +These commands are trusted local code. Do not interpolate untrusted text into shell or interpreter program arguments. Workers inherit the coordinator's environment and access, receive a closed stdin, and run in the recorded workdir. Commands requiring interactive input are unsuitable. -`{task_id}`, `{title}`, `{lane}`, `{role}`, `{priority}`, and `{workdir}` are available as explicit placeholders. The template is split with shell-style quoting before placeholder values are substituted, so task metadata cannot create extra subprocess argv tokens. Unknown placeholders fail before any subprocess starts, which keeps command templates predictable. The command runs on the local machine, using the task's `workdir` when one is set. If it exits with code `0`, the task becomes `done`; otherwise it becomes `failed` and the final output is kept in the event log. +## Timeouts, output, and cancellation -Tasks may carry their own `command` and `workdir`, either through the Python API or `agent-backlog enqueue --command ... --workdir ...`. A dispatch-level `--command` overrides per-task commands when you want one temporary worker command for the whole run. +Each task can override the dispatch timeout. Nonzero exits, timeouts, missing executables, and invalid workdirs become failed attempts. The remaining selected tasks continue. The coordinator reads both output streams concurrently into separate capped files; excess data is drained/discarded, and truncation is recorded. -Timeouts, spawn errors, and invalid working directories also mark the task -`failed`; they do not leave it `running` or abort the remaining selected batch. -Only the direct worker process is terminated on timeout. A command that starts -its own children must manage those children's cleanup. +Cancel through `agent-backlog cancel TASK_ID --db ...`. The request is observed by the owning coordinator, which stops its own child process group. It does not signal a saved PID. Cancelled work retains its capacity slot until the attempt finishes. -## Concurrency cap +POSIX workers start in a new session. Cleanup sends TERM, allows a brief grace period, then KILL while the leader remains unreaped. This retains PID ownership during group signalling. On Darwin, a zombie-only group can return EPERM; the coordinator checks only that owned group's process states before treating that condition as already stopped. An actual permission denial with live members remains a cleanup error, and the attempt is left unfinished for inspection. -`--max-workers` caps total running work and the size of this dispatch batch. Tasks -run sequentially inside the CLI process. Each start reserves its task and a -capacity slot in one SQLite transaction, so competing local dispatchers cannot -start the same task. Use the same cap across dispatchers. +Subprocess dispatch requires the default `SIGCHLD` handler and rejects ignored or custom handlers before claiming work. Embedding callers must not reap the dispatcher's children from another thread. Where `waitid(WNOWAIT)` is available, cleanup confirms child ownership before signalling; an already-reaped child leaves its attempt unfinished for inspection. The dispatcher never replaces a caller's signal handler. -Final updates only replace `running`: an operator's cancellation or intervening -status change is preserved. Changing a status does not terminate a process. If -the CLI is killed, a task can remain `running`; inspect `stale` and its output -before requeueing it. +Where `waitid(WNOWAIT)` is available, normal completion is observed before reaping and remaining group members are cleaned up. Older macOS interpreters without it reap through `Popen.poll`; after that, the old group is never signalled because its PID could be reused. On those interpreters, commands must wait for their children before exiting. Escaped sessions, changed privileges, and uninterruptible OS operations are outside cleanup guarantees. Windows subprocess dispatch is unsupported; queue/planning commands remain available. -## Command guidance +Ctrl-C attempts cleanup of every owned worker, even if one cleanup fails. Cleanup failures are reported and remain inspectable; a still-live worker is not marked stopped. Abrupt failure can leave unfinished attempts requiring [explicit recovery](batches-and-recovery.md#interruption-and-recovery). -Keep worker commands simple: +## Results -- pass all needed inputs through the task id, files, or environment variables you control; -- write outputs under a configured demo/runtime directory; -- prefer idempotent commands; -- set a timeout that matches the expected task size; -- inspect `status` and `events` after dispatch. +Dispatch prints JSON with `started`, `completed`, `failed`, `cancelled`, and `skipped_reason`. Exit `0` means the selected work/preview passed or no eligible tasks were selected; `1` reports selected failure/cancellation or a CLI operation error, `2` blocked capacity/budget, and `130` Ctrl-C. Inspect `status` for remaining delayed tasks; success of one batch is not proof the whole queue is empty. -For detached session management and contained workspaces, see -[Local Agent Task Runtime](https://github.com/matchalatteguy/local-agent-task-runtime). -Long-running services and distributed worker pools are outside this package's scope. +See [batch manifests, retry, logs, and recovery](batches-and-recovery.md). Templates still support [recurring depth-based replenishment](scheduler-policy.md), independently of finite command batches. diff --git a/docs/first-5-minutes.md b/docs/first-5-minutes.md index 571cddb..024e43a 100644 --- a/docs/first-5-minutes.md +++ b/docs/first-5-minutes.md @@ -115,7 +115,7 @@ uv run agent-backlog status --db .agent-backlog/tasks.sqlite3 uv run agent-backlog events --db .agent-backlog/tasks.sqlite3 --limit 20 ``` -The subprocess backend is synchronous: the CLI waits for each selected local command to finish or time out. +The CLI waits for the selected batch, while commands run concurrently up to `--max-workers`. Add `--drain` for a finite queued snapshot that refills slots as workers finish. See the [repair/resume batch walkthrough](../README.md#install-and-try-the-repairresume-walkthrough). ## 7. Try an environment default diff --git a/docs/python-api.md b/docs/python-api.md index ffa7459..224feda 100644 --- a/docs/python-api.md +++ b/docs/python-api.md @@ -16,7 +16,7 @@ with init_store(".agent-backlog/tasks.sqlite3") as store: print(store.schema_version()) ``` -`init_store(path)` creates parent directories when needed, opens SQLite, initializes the schema idempotently, writes `schema_version=1`, and returns a `TaskStore`. Use it as a context manager or call `close()` when the script is done. +`init_store(path)` creates parent directories when needed, opens SQLite, initializes/migrates to schema 2, and returns a `TaskStore`. Use it as a context manager or call `close()` when done. Schema-1 tasks/events remain intact; old running tasks need explicit inspection/recovery because their process identities are unknown. ## Create one task @@ -119,7 +119,29 @@ print(result.started, result.skipped_reason) store.close() ``` -The built-in subprocess backend runs synchronously inside the calling process. It is suitable for short local workers and demos, not for a durable distributed worker fleet. +The calling process stays open, while up to `max_concurrent_workers` subprocesses actually run concurrently. Set `drain=True` to consume the initial queued snapshot and fill slots as workers finish. Every selected attempt has durable history and capped per-stream log files. Process execution requires POSIX; this is a local coordinator, not a distributed worker fleet. + +## Finite batches and explicit retry + +```python +from pathlib import Path +from agent_backlog_runner import ( + DispatchPolicy, dispatch_ready, enqueue_batch, init_store, + load_batch_manifest, write_queue_report, +) + +manifest = load_batch_manifest("checks.yaml", root=".") +with init_store(".agent-backlog/checks.sqlite3") as store: + tasks = enqueue_batch(store, manifest, run_id="release-check") + result = dispatch_ready(store, DispatchPolicy(2, "subprocess", drain=True)) + for task_id in result.failed: + print(store.attempts(task_id)) + write_queue_report(store, md_out=Path(".agent-backlog/report.md")) +``` + +After repairing inputs, `store.retry_task(task_id)` schedules another bounded attempt. `store.request_cancel(task_id)` asks the active coordinator to stop its own process; it does not signal a saved PID. `recover_task(store, task_id, stale_after=1800)` refuses known live owners/workers/groups and abandons stale uncertain work without requeueing it. Read [recovery semantics](batches-and-recovery.md) before using the optional unknown-identity acknowledgement. + +`create_task(command_argv=(...))` stores exact arguments without legacy string-template substitution. Optional `timeout_seconds`, `max_attempts`, and `retry_backoff_seconds` control bounded execution. The low-level `mark_task` and old `claim_task` APIs remain manual bookkeeping; prefer managed dispatch/retry/recovery for process work. Editing raw statuses does not bypass active-attempt claims or establish that side effects stopped. ## Status snapshots diff --git a/docs/status-and-heartbeats.md b/docs/status-and-heartbeats.md index 4039cac..780d36b 100644 --- a/docs/status-and-heartbeats.md +++ b/docs/status-and-heartbeats.md @@ -79,4 +79,4 @@ uv run agent-backlog stale --db .agent-backlog/tasks.sqlite3 --after 30m uv run agent-backlog stale --db .agent-backlog/tasks.sqlite3 --after 30m --format json ``` -The command prints stale task ids and their age in seconds. It does not change task status; callers decide whether to retry, mark failed, or investigate. +The command prints stale task ids and their age in seconds. It does not change task status or prove workers are gone. Managed work needs [explicit recovery](batches-and-recovery.md#interruption-and-recovery), followed by a separate bounded retry if appropriate. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 40e43f3..2110cbc 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -139,7 +139,7 @@ Failed subprocesses mark the task as `failed` and store the tail of stderr/stdou uv run agent-backlog events --db .agent-backlog/tasks.sqlite3 --limit 20 ``` -The built-in dispatcher is synchronous. A long command blocks the current CLI process until it exits or times out. +The CLI stays open while bounded workers run concurrently. Inspect `attempts` and `logs TASK_ID --stream stderr` for full retained output below the configured cap; events contain only a short tail. Repair inputs, use `retry TASK_ID` or `retry --failed`, then dispatch again. Commands are not retried automatically. ## Stale tasks appear @@ -149,7 +149,7 @@ Stale tasks are `running` tasks whose latest update/heartbeat is older than the uv run agent-backlog stale --db .agent-backlog/tasks.sqlite3 --after 30m ``` -This command reports stale tasks only. It does not kill workers or requeue tasks automatically. Use `agent-backlog mark TASK_ID --status todo|failed|blocked` after you decide whether to retry, fail, or investigate the task. +This command reports stale tasks only. Inspect their `attempts` and side effects, then use `recover TASK_ID` after recorded processes are gone. Recovery refuses live PIDs/groups, never kills a saved PID, and does not requeue by itself. A separate `retry` schedules more work within its budget. Unknown worker identities need explicit inspection and `--acknowledge-unknown`. See [recovery](batches-and-recovery.md#interruption-and-recovery); direct status edits are unsuitable for managed attempts. ## Start the demo over diff --git a/examples/README.md b/examples/README.md index 5af9eed..0bc8a46 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,5 +1,7 @@ # Examples +For real repository check commands, start with [`repo-checks.yaml`](repo-checks.yaml): it runs pytest and Ruff with your project interpreter, bounded parallelism, and explicit retry budgets. Install that project's dependencies before loading it. The packaged `agent-backlog demo --output .agent-backlog/maintenance-demo` demonstrates unit tests, syntax checks, a missing-document failure, repair, explicit retry, and resuming without repeating successful checks. See [batch manifests](../docs/batches-and-recovery.md). + The examples are synthetic and safe to run locally. They show queue mechanics, not domain-specific automation. Use them as starting points for your own local maintenance, documentation, or agent-demo queues. ## Files diff --git a/examples/repo-checks.yaml b/examples/repo-checks.yaml new file mode 100644 index 0000000..a9c32cf --- /dev/null +++ b/examples/repo-checks.yaml @@ -0,0 +1,17 @@ +schema_version: 1 +name: repo-checks +tasks: + - id: unit-tests + title: Run the repository test suite + argv: ['{python}', '-m', pytest, '-q'] + workdir: '.' + timeout_seconds: 600 + max_attempts: 3 + retry_backoff_seconds: 2 + - id: lint-check + title: Check Python style and imports + argv: ['{python}', '-m', ruff, check, '.'] + workdir: '.' + timeout_seconds: 120 + max_attempts: 3 + retry_backoff_seconds: 2 diff --git a/pyproject.toml b/pyproject.toml index 36fa7af..44456f3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,14 +1,14 @@ [project] name = "agent-backlog-runner" -version = "0.2.0" -description = "Local SQLite backlog runner for bounded agent task queues" +version = "0.3.0" +description = "Durable local command batches with bounded parallel workers, retries, and SQLite history" readme = "README.md" requires-python = ">=3.11" license = "MIT" license-files = ["LICENSE"] authors = [{ name = "matchalatteguy" }] dependencies = ["PyYAML>=6.0"] -keywords = ["agents", "sqlite", "backlog", "scheduler", "cli"] +keywords = ["command-queue", "sqlite", "batch", "retry", "repository-maintenance", "cli"] classifiers = [ "Development Status :: 3 - Alpha", "License :: OSI Approved :: MIT License", diff --git a/src/agent_backlog_runner/__init__.py b/src/agent_backlog_runner/__init__.py index 48bb746..a571eed 100644 --- a/src/agent_backlog_runner/__init__.py +++ b/src/agent_backlog_runner/__init__.py @@ -1,5 +1,6 @@ """Local SQLite backlog runner for bounded agent task queues.""" +from .batches import BatchManifest, BatchTask, enqueue_batch, load_batch_manifest from .config import DEFAULT_DB_PATH, ENV_DB_PATH, RunnerConfig, resolve_db_path from .dispatch import ( DispatchPolicy, @@ -9,6 +10,8 @@ render_command_template, ) from .models import BacklogPolicy, TaskEvent, TaskRecord, TaskStatus, TaskTemplate +from .recovery import recover_task +from .reports import write_queue_report from .scheduler import BacklogPlan, SchedulerState, apply_backlog_plan, plan_backlog from .status import StatusSnapshot, get_status_snapshot from .store import SCHEMA_VERSION, TaskStore, init_store @@ -17,6 +20,8 @@ __all__ = [ "BacklogPlan", "BacklogPolicy", + "BatchManifest", + "BatchTask", "DEFAULT_DB_PATH", "DispatchPolicy", "DispatchResult", @@ -33,11 +38,15 @@ "TemplateCatalog", "apply_backlog_plan", "dispatch_ready", + "enqueue_batch", "get_status_snapshot", "init_store", "load_template_catalog", + "load_batch_manifest", "plan_backlog", "render_command_argv", "render_command_template", "resolve_db_path", + "recover_task", + "write_queue_report", ] diff --git a/src/agent_backlog_runner/batches.py b/src/agent_backlog_runner/batches.py new file mode 100644 index 0000000..93655a6 --- /dev/null +++ b/src/agent_backlog_runner/batches.py @@ -0,0 +1,149 @@ +"""Idempotent manifests for finite repository-maintenance batches.""" + +from __future__ import annotations + +import hashlib +import json +import sys +from dataclasses import dataclass +from pathlib import Path + +from .models import TaskRecord +from .safety import require_safe_slug, safe_child_path +from .store import TaskStore +from .templates import _load_data + + +@dataclass(frozen=True) +class BatchTask: + slug: str + title: str + argv: tuple[str, ...] + workdir: str + timeout_seconds: float + max_attempts: int + retry_backoff_seconds: float + + +@dataclass(frozen=True) +class BatchManifest: + name: str + root: str + tasks: tuple[BatchTask, ...] + + +def load_batch_manifest( + path: str | Path, *, root: str | Path | None = None, python: str | None = None +) -> BatchManifest: + path = Path(path) + data = _load_data(path) + if not isinstance(data, dict) or set(data) - {"schema_version", "name", "tasks"}: + raise ValueError("batch must be an object with schema_version, name, and tasks only") + if type(data.get("schema_version")) is not int or data["schema_version"] != 1: + raise ValueError("unsupported batch schema_version; expected 1") + name = require_safe_slug(data.get("name", ""), field="batch name") + directory = Path(root or path.parent).resolve() + if not directory.is_dir(): + raise ValueError("batch root must be an existing directory") + raw_tasks = data.get("tasks") + if not isinstance(raw_tasks, list) or not raw_tasks: + raise ValueError("batch tasks must be a non-empty list") + tasks = [] + allowed = { + "id", + "title", + "argv", + "workdir", + "timeout_seconds", + "max_attempts", + "retry_backoff_seconds", + } + for raw in raw_tasks: + if not isinstance(raw, dict) or set(raw) - allowed: + raise ValueError("unknown or malformed batch task fields") + slug = require_safe_slug(raw.get("id", ""), field="batch task id") + title = raw.get("title", slug) + if not isinstance(title, str) or not title.strip(): + raise ValueError("batch title must be a non-empty string") + argv = raw.get("argv") + if ( + not isinstance(argv, list) + or not argv + or not all(isinstance(arg, str) and "\0" not in arg for arg in argv) + or not argv[0] + ): + raise ValueError("argv must be a non-empty list of string arguments without NUL bytes") + argv = tuple((python or sys.executable) if arg == "{python}" else arg for arg in argv) + workdir = raw.get("workdir", ".") + if not isinstance(workdir, str) or not workdir: + raise ValueError("workdir must be a non-empty relative string") + resolved = safe_child_path(directory, workdir) + if not resolved.is_dir(): + raise ValueError(f"workdir does not exist: {workdir}") + timeout, budget, backoff = ( + raw.get("timeout_seconds", 300), + raw.get("max_attempts", 3), + raw.get("retry_backoff_seconds", 1), + ) + if type(timeout) not in (int, float) or not 0 < timeout <= 86400: + raise ValueError("timeout_seconds must be positive and at most 86400") + if type(budget) is not int or not 1 <= budget <= 100: + raise ValueError("max_attempts must be an integer from 1 to 100") + if type(backoff) not in (int, float) or not 0 <= backoff <= 60: + raise ValueError("retry_backoff_seconds must be between 0 and 60") + tasks.append(BatchTask(slug, title, argv, str(resolved), timeout, budget, backoff)) + if len({task.slug for task in tasks}) != len(tasks): + raise ValueError("batch task ids must be unique") + return BatchManifest(name, str(directory), tuple(tasks)) + + +def enqueue_batch( + store: TaskStore, manifest: BatchManifest, *, run_id: str = "default" +) -> tuple[TaskRecord, ...]: + run_id = require_safe_slug(run_id, field="run id") + records = [] + with store.transaction(): + for item in manifest.tasks: + identity = json.dumps([manifest.root, manifest.name, run_id, item.slug]) + task_id = "task_" + hashlib.sha256(identity.encode()).hexdigest()[:12] + body = f"batch={manifest.name}; run={run_id}; check={item.slug}" + existing = store.get_task(task_id) + if existing: + expected = ( + item.title, + item.argv, + item.workdir, + item.timeout_seconds, + item.max_attempts, + item.retry_backoff_seconds, + ) + actual = ( + existing.title, + existing.command_argv, + existing.workdir, + existing.timeout_seconds, + existing.max_attempts, + existing.retry_backoff_seconds, + ) + if expected != actual or existing.body != body: + raise ValueError( + "batch definition changed for an existing run; use a new --run-id" + ) + records.append(existing) + continue + records.append( + store.create_task( + task_id=task_id, + title=item.title, + body=body, + lane="checks", + tags=(manifest.name,), + role="worker", + command_argv=item.argv, + workdir=item.workdir, + timeout_seconds=item.timeout_seconds, + max_attempts=item.max_attempts, + retry_backoff_seconds=item.retry_backoff_seconds, + ) + ) + return tuple(records) diff --git a/src/agent_backlog_runner/cli.py b/src/agent_backlog_runner/cli.py index 8ff2c37..7f7fda3 100644 --- a/src/agent_backlog_runner/cli.py +++ b/src/agent_backlog_runner/cli.py @@ -3,12 +3,16 @@ import argparse import json import sys +from dataclasses import asdict +from importlib.metadata import version from pathlib import Path +from .batches import enqueue_batch, load_batch_manifest from .config import ENV_DB_PATH, resolve_db_path from .dispatch import DispatchPolicy, dispatch_ready from .models import BacklogPolicy, TaskStatus -from .reports import plan_to_json, snapshot_to_json, snapshot_to_table +from .recovery import recover_task +from .reports import plan_to_json, snapshot_to_json, snapshot_to_table, write_queue_report from .scheduler import apply_backlog_plan, plan_backlog from .status import get_status_snapshot from .store import init_store @@ -30,6 +34,12 @@ def _task_to_dict(task) -> dict[str, object]: "created_at": task.created_at, "updated_at": task.updated_at, "heartbeat_at": task.heartbeat_at, + "argv": list(task.command_argv) if task.command_argv else None, + "attempt_count": task.attempt_count, + "max_attempts": task.max_attempts, + "not_before": task.not_before, + "timeout_seconds": task.timeout_seconds, + "retry_backoff_seconds": task.retry_backoff_seconds, } @@ -72,11 +82,15 @@ def _add_db_argument(parser: argparse.ArgumentParser) -> None: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="agent-backlog") + parser.add_argument("--version", action="version", version=version("agent-backlog-runner")) sub = parser.add_subparsers(dest="command", required=True) init = sub.add_parser("init") _add_db_argument(init) + demo = sub.add_parser("demo", help="run the repository-check repair and resume walkthrough") + demo.add_argument("--output", type=Path, default=Path(".agent-backlog/maintenance-demo")) + enqueue = sub.add_parser("enqueue") _add_db_argument(enqueue) enqueue.add_argument("--title", required=True) @@ -88,6 +102,21 @@ def build_parser() -> argparse.ArgumentParser: enqueue.add_argument("--priority", type=int, default=0) enqueue.add_argument("--command", dest="task_command") enqueue.add_argument("--workdir") + enqueue.add_argument("--max-attempts", type=int, default=3) + enqueue.add_argument("--retry-backoff", type=float, default=1) + enqueue.add_argument("--timeout", type=float) + + batch = sub.add_parser("batch", help="load a finite, repeatable command manifest") + batch_sub = batch.add_subparsers(dest="batch_command", required=True) + for name in ("validate", "load"): + cmd = batch_sub.add_parser(name) + _add_db_argument(cmd) + cmd.add_argument("manifest") + cmd.add_argument("--root") + cmd.add_argument( + "--python", help="interpreter substituted for the exact argv token {python}" + ) + cmd.add_argument("--run-id", default="default") list_cmd = sub.add_parser("list") _add_db_argument(list_cmd) @@ -138,7 +167,50 @@ def build_parser() -> argparse.ArgumentParser: dispatch.add_argument("--max-workers", type=int, default=1) dispatch.add_argument("--backend", choices=["dry-run", "subprocess"], default="dry-run") dispatch.add_argument("--command", dest="worker_command") - dispatch.add_argument("--timeout", type=int, default=60) + dispatch.add_argument("--timeout", type=float, default=60) + dispatch.add_argument( + "--drain", + action="store_true", + help="finish the queued snapshot, filling slots as workers exit", + ) + dispatch.add_argument("--log-root", type=Path) + dispatch.add_argument("--max-log-bytes", type=int, default=1048576) + + retry = sub.add_parser("retry", help="explicitly schedule a failed or cancelled task") + _add_db_argument(retry) + retry.add_argument("task_id", nargs="?") + retry.add_argument( + "--failed", action="store_true", help="schedule all failed tasks with remaining attempts" + ) + + recover = sub.add_parser( + "recover", help="abandon stale work only after its recorded processes are gone" + ) + _add_db_argument(recover) + recover.add_argument("task_id") + recover.add_argument("--after", default="30m") + recover.add_argument("--acknowledge-unknown", action="store_true") + + cancel = sub.add_parser("cancel", help="ask the owning dispatcher to stop a task") + _add_db_argument(cancel) + cancel.add_argument("task_id") + + attempts = sub.add_parser("attempts", help="inspect durable attempt outcomes and log locations") + _add_db_argument(attempts) + attempts.add_argument("task_id", nargs="?") + attempts.add_argument("--format", choices=["table", "json"], default="table") + + logs = sub.add_parser("logs", help="read a bounded tail of a worker's retained output") + _add_db_argument(logs) + logs.add_argument("task_id") + logs.add_argument("--attempt", type=int) + logs.add_argument("--stream", choices=["stdout", "stderr"], default="stdout") + logs.add_argument("--tail", type=int, default=4096) + + report = sub.add_parser("report", help="write an inspectable JSON or Markdown queue report") + _add_db_argument(report) + report.add_argument("--json-out", type=Path) + report.add_argument("--md-out", type=Path) status = sub.add_parser("status") _add_db_argument(status) @@ -147,6 +219,7 @@ def build_parser() -> argparse.ArgumentParser: events = sub.add_parser("events") _add_db_argument(events) events.add_argument("--limit", type=int, default=20) + events.add_argument("--task-id") events.add_argument("--format", choices=["table", "json"], default="table") stale = sub.add_parser("stale") @@ -160,6 +233,93 @@ def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) db_path = resolve_db_path(args.db) if hasattr(args, "db") else None try: + if args.command == "demo": + from .maintenance import run_demo + + print(json.dumps(run_demo(args.output), indent=2)) + return 0 + if args.command == "report": + with init_store(db_path) as store: + write_queue_report(store, json_out=args.json_out, md_out=args.md_out) + return 0 + if args.command == "batch": + manifest = load_batch_manifest(args.manifest, root=args.root, python=args.python) + if args.batch_command == "validate": + print(json.dumps(asdict(manifest), indent=2)) + else: + with init_store(db_path) as store: + records = enqueue_batch(store, manifest, run_id=args.run_id) + print(json.dumps([_task_to_dict(task) for task in records], indent=2)) + return 0 + if args.command == "retry": + if bool(args.task_id) == args.failed: + raise ValueError("choose one task_id or --failed") + with init_store(db_path) as store: + tasks = ( + store.list_tasks((TaskStatus.FAILED,)) + if args.failed + else [store.get_task(args.task_id)] + ) + scheduled, skipped = [], [] + for task in tasks: + if task is None: + raise KeyError(args.task_id) + try: + scheduled.append(_task_to_dict(store.retry_task(task.id))) + except ValueError as exc: + if not args.failed: + raise + skipped.append({"id": task.id, "reason": str(exc)}) + print(json.dumps({"scheduled": scheduled, "skipped": skipped}, indent=2)) + return 0 + if args.command == "recover": + with init_store(db_path) as store: + recover_task( + store, + args.task_id, + stale_after=_duration_seconds(args.after), + acknowledge_unknown=args.acknowledge_unknown, + ) + print(f"recovered {args.task_id}; inspect outputs, then retry explicitly") + return 0 + if args.command == "cancel": + with init_store(db_path) as store: + store.request_cancel(args.task_id) + print( + f"cancellation requested for {args.task_id}; attempts shows when the worker stops" + ) + return 0 + if args.command == "attempts": + with init_store(db_path) as store: + attempts = store.attempts(args.task_id) + if args.format == "json": + print(json.dumps(attempts, indent=2)) + else: + print("task attempt outcome exit stdout / stderr") + for attempt in attempts: + print( + f"{attempt['task_id']} {attempt['attempt_no']:7} {attempt['outcome']:12} " + f"{str(attempt['exit_code']):4} {attempt['stdout_path']} / " + f"{attempt['stderr_path']}" + ) + return 0 + if args.command == "logs": + if not 1 <= args.tail <= 67108864: + raise ValueError("--tail must be from 1 to 67108864 bytes") + with init_store(db_path) as store: + attempts = store.attempts(args.task_id) + matches = [ + item + for item in attempts + if args.attempt is None or item["attempt_no"] == args.attempt + ] + if not matches or not matches[0][args.stream + "_path"]: + raise ValueError("no retained log for the selected attempt") + with Path(matches[0][args.stream + "_path"]).open("rb") as handle: + handle.seek(0, 2) + handle.seek(max(0, handle.tell() - args.tail)) + print(handle.read(args.tail).decode("utf-8", "replace"), end="") + return 0 if args.command == "init": store = init_store(db_path) store.close() @@ -181,6 +341,9 @@ def main(argv: list[str] | None = None) -> int: priority=args.priority, command=args.task_command, workdir=args.workdir, + max_attempts=args.max_attempts, + retry_backoff_seconds=args.retry_backoff, + timeout_seconds=args.timeout, ) store.close() print(task.id) @@ -199,6 +362,7 @@ def main(argv: list[str] | None = None) -> int: store = init_store(db_path) task = store.get_task(args.task_id) events = store.events(args.task_id, limit=20) if task is not None else [] + attempts = store.attempts(args.task_id) if task is not None else [] store.close() if task is None: raise KeyError(args.task_id) @@ -208,6 +372,7 @@ def main(argv: list[str] | None = None) -> int: { "task": _task_to_dict(task), "events": [_event_to_dict(event) for event in events], + "attempts": attempts, }, indent=2, sort_keys=True, @@ -220,6 +385,13 @@ def main(argv: list[str] | None = None) -> int: return 0 if args.command == "mark": store = init_store(db_path) + task = store.get_task(args.task_id) + if task is None: + raise KeyError(args.task_id) + if any(attempt["finished_at"] is None for attempt in store.attempts(args.task_id)): + raise ValueError("managed attempt is active; use cancel to request a stop") + if args.status == "todo" and task.attempt_count: + raise ValueError("use retry to respect attempt budgets and backoff") store.mark_task(args.task_id, TaskStatus(args.status), args.message) store.close() print(f"marked {args.task_id} {args.status}") @@ -257,15 +429,19 @@ def main(argv: list[str] | None = None) -> int: store = init_store(db_path) result = dispatch_ready( store, - DispatchPolicy(args.max_workers, args.backend, args.worker_command, args.timeout), + DispatchPolicy( + args.max_workers, + args.backend, + args.worker_command, + args.timeout, + args.drain, + args.log_root, + args.max_log_bytes, + ), ) store.close() - print( - json.dumps( - {"started": list(result.started), "skipped_reason": result.skipped_reason} - ) - ) - return 0 + print(json.dumps(asdict(result))) + return 1 if result.failed or result.cancelled else (2 if result.skipped_reason else 0) if args.command == "status": store = init_store(db_path) snapshot = get_status_snapshot(store) @@ -276,7 +452,7 @@ def main(argv: list[str] | None = None) -> int: return 0 if args.command == "events": store = init_store(db_path) - events = store.events(limit=args.limit) + events = store.events(args.task_id, limit=args.limit) store.close() if args.format == "json": print( @@ -310,6 +486,12 @@ def main(argv: list[str] | None = None) -> int: print(f"{item.task.id} age={item.age_seconds}s") return 0 return 2 + except KeyboardInterrupt: + print( + "dispatcher interrupted; inspect attempts and retry unfinished tasks explicitly", + file=sys.stderr, + ) + return 130 except Exception as exc: # fail closed for CLI users print(f"error: {exc}", file=sys.stderr) return 1 diff --git a/src/agent_backlog_runner/dispatch.py b/src/agent_backlog_runner/dispatch.py index 6d7895e..16d9bb2 100644 --- a/src/agent_backlog_runner/dispatch.py +++ b/src/agent_backlog_runner/dispatch.py @@ -1,10 +1,16 @@ from __future__ import annotations +import os +import selectors import shlex +import signal import string import subprocess +import sys +import time from dataclasses import dataclass from pathlib import Path +from typing import Any, BinaryIO from .models import TaskRecord, TaskStatus from .store import TaskStore @@ -17,21 +23,29 @@ class DispatchPolicy: max_concurrent_workers: int = 1 backend: str = "dry-run" command: str | None = None - timeout_seconds: int = 60 + timeout_seconds: float = 60 + drain: bool = False + log_root: Path | None = None + max_log_bytes: int = 1048576 def __post_init__(self) -> None: - if self.max_concurrent_workers < 0: - raise ValueError("max_concurrent_workers must be non-negative") + if type(self.max_concurrent_workers) is not int or self.max_concurrent_workers < 0: + raise ValueError("max_concurrent_workers must be a non-negative integer") if self.backend not in {"dry-run", "subprocess"}: raise ValueError("backend must be dry-run or subprocess") - if self.timeout_seconds <= 0: - raise ValueError("timeout_seconds must be positive") + if not 0 < self.timeout_seconds <= 86400: + raise ValueError("timeout_seconds must be positive and at most 86400") + if type(self.max_log_bytes) is not int or not 1 <= self.max_log_bytes <= 67108864: + raise ValueError("max_log_bytes must be from 1 to 67108864 per stream") @dataclass(frozen=True) class DispatchResult: started: tuple[str, ...] skipped_reason: str = "" + completed: tuple[str, ...] = () + failed: tuple[str, ...] = () + cancelled: tuple[str, ...] = () def _command_fields(task: TaskRecord) -> dict[str, str]: @@ -46,110 +60,430 @@ def _command_fields(task: TaskRecord) -> dict[str, str]: def _render_template_part(part: str, fields: dict[str, str]) -> str: - formatter = string.Formatter() - for _, field_name, _, _ in formatter.parse(part): - if field_name is None: - continue - if "." in field_name or "[" in field_name or "]" in field_name: - raise ValueError(f"unsupported command placeholder: {field_name}") - if field_name not in _ALLOWED_COMMAND_FIELDS: + for _, field_name, _, _ in string.Formatter().parse(part): + if field_name is not None and field_name not in _ALLOWED_COMMAND_FIELDS: raise ValueError(f"unsupported command placeholder: {field_name}") return part.format_map(fields) def render_command_argv(template: str, task: TaskRecord) -> tuple[str, ...]: - """Render a worker command template to subprocess argv. - - The template is split with shell-style quoting before task placeholders are - substituted. That keeps task metadata as data: a placeholder value containing - spaces, quotes, or punctuation cannot create extra argv tokens. - """ - - fields = _command_fields(task) - argv = tuple(_render_template_part(part, fields) for part in shlex.split(template)) + """Split quoting before substitution; task values cannot create new argv tokens.""" + argv = tuple( + _render_template_part(part, _command_fields(task)) for part in shlex.split(template) + ) if not argv or not argv[0]: raise ValueError("worker command must contain an executable") return argv def render_command_template(template: str, task: TaskRecord) -> str: - """Render a worker command template for display or event logging. - - Supported placeholders are ``{task_id}``, ``{title}``, ``{lane}``, ``{role}``, - ``{priority}``, and ``{workdir}``. Unknown fields fail before a subprocess is - started so typos do not turn into surprising shell arguments. - """ - return shlex.join(render_command_argv(template, task)) def _command_argv_for_task(task: TaskRecord, policy: DispatchPolicy) -> tuple[str, ...]: - template = policy.command or task.command - if not template: + if policy.command: + return render_command_argv(policy.command, task) + if task.command_argv: + return task.command_argv + if not task.command: raise ValueError("subprocess backend requires a policy command or per-task command") - return render_command_argv(template, task) + return render_command_argv(task.command, task) + + +@dataclass +class _Log: + file: BinaryIO + limit: int + observed: int = 0 + written: int = 0 + tail: bytes = b"" + + def write(self, data: bytes) -> None: + self.observed += len(data) + self.tail = (self.tail + data)[-500:] + retained = data[: max(0, self.limit - self.written)] + if retained: + self.file.write(retained) + self.file.flush() + self.written += len(retained) + + +@dataclass +class _Worker: + task: TaskRecord + attempt: dict[str, Any] + process: subprocess.Popen + started: float + timeout: float + stdout: _Log + stderr: _Log + last_heartbeat: float = 0 + + +class _SpawnFailure(Exception): + """An ordinary launch failure after owned-process cleanup succeeded.""" + + +def _note_cleanup_failure( + store: TaskStore, task_id: str, original: BaseException, cleanup_error: BaseException +) -> None: + note = f"cleanup failed for {task_id}: {cleanup_error}" + original.add_note(note) + try: + sys.stderr.write(note + "; inspect its unfinished attempt before recovery\n") + except Exception: + pass + try: + store.add_event(task_id, "cleanup_error", note, {}) + except Exception: + pass + + +_HAS_WAITID = all(hasattr(os, name) for name in ("waitid", "WNOWAIT", "P_PID")) + + +def _require_default_child_handler() -> None: + if os.name == "posix" and signal.getsignal(signal.SIGCHLD) != signal.SIG_DFL: + raise ValueError( + "subprocess dispatch requires the default SIGCHLD handler; " + "ignored/custom child reapers cannot retain process-group ownership" + ) + + +def _exit_ready(process: subprocess.Popen) -> bool: + if os.name == "posix" and _HAS_WAITID: + # Keep the leader unreaped until group cleanup: its PID cannot be reused. + result = os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) + return result is not None + return process.poll() is not None + + +def _confirm_unreaped_child(process: subprocess.Popen) -> None: + _require_default_child_handler() + try: + os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) + except ChildProcessError as exc: + raise RuntimeError( + "worker leader was already reaped; refusing process-group signals" + ) from exc + + +def _stop_owned_process(process: subprocess.Popen) -> None: + """Signal only this coordinator's unreaped child/session, never a stored PID.""" + if process.returncode is not None: + return + if os.name == "posix": + _require_default_child_handler() + if _HAS_WAITID: + _confirm_unreaped_child(process) + elif process.poll() is not None: + # Older interpreters cannot retain an exited leader. Do not signal + # its former group after poll has reaped it (or reported ECHILD). + return + _signal_owned_group(process.pid, signal.SIGTERM) + # Do not reap between TERM and KILL; retain the leader's PID ownership. + time.sleep(0.1) + if _HAS_WAITID: + _confirm_unreaped_child(process) + _signal_owned_group(process.pid, signal.SIGKILL) + else: + process.terminate() + time.sleep(0.1) + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + +def _signal_owned_group(pgid: int, sig: int) -> None: + try: + os.killpg(pgid, sig) + except ProcessLookupError: + pass + except PermissionError: + # Darwin returns EPERM for groups containing only unreaped zombies. + # Keep the leader unreaped; inspect only this owned group, not PID names. + if sys.platform != "darwin": + raise + result = subprocess.run( + ["ps", "-o", "stat=", "-g", str(pgid)], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + states = result.stdout.split() + if ( + result.returncode != 0 + or not states + or any(state[0] not in {"Z", "X"} for state in states) + ): + raise + + +def _read_pipe(pipe: BinaryIO, log: _Log, *, limit: int = 128) -> None: + for _ in range(limit): + try: + data = os.read(pipe.fileno(), 65536) + except BlockingIOError: + break + if not data: + break + log.write(data) + + +def _finish_worker( + store: TaskStore, + selector: selectors.BaseSelector, + worker: _Worker, + outcome: str, + message: str | None = None, +) -> None: + for pipe, log in ( + (worker.process.stdout, worker.stdout), + (worker.process.stderr, worker.stderr), + ): + assert pipe is not None + _read_pipe(pipe, log) + try: + selector.unregister(pipe) + except KeyError: + pass + pipe.close() + log.file.close() + store.finish_attempt( + worker.attempt["claim_token"], + outcome, + message + if message is not None + else ( + worker.stdout.tail.decode("utf-8", "replace") + if outcome == "done" + else f"worker exited {worker.process.returncode}: " + + (worker.stderr.tail or worker.stdout.tail).decode("utf-8", "replace") + ), + exit_code=worker.process.returncode, + stdout_bytes=worker.stdout.observed, + stderr_bytes=worker.stderr.observed, + stdout_truncated=worker.stdout.observed > worker.stdout.written, + stderr_truncated=worker.stderr.observed > worker.stderr.written, + ) + + +def _spawn( + store: TaskStore, + selector: selectors.BaseSelector, + task: TaskRecord, + attempt: dict[str, Any], + policy: DispatchPolicy, +) -> _Worker | None: + token = attempt["claim_token"] + root = (policy.log_root or store.path.resolve().parent / "logs").resolve() + logs: list[BinaryIO] = [] + process = None + try: + directory = root / task.id / f"{attempt['attempt_no']:04}-{token[:12]}" + directory.mkdir(parents=True, exist_ok=False) + stdout_path, stderr_path = directory / "stdout.log", directory / "stderr.log" + for path in (stdout_path, stderr_path): + logs.append(path.open("xb")) + store.bind_attempt(token, stdout_path=str(stdout_path), stderr_path=str(stderr_path)) + environment = { + **os.environ, + "AGENT_BACKLOG_TASK_ID": task.id, + "AGENT_BACKLOG_ATTEMPT": str(attempt["attempt_no"]), + } + process = subprocess.Popen( + attempt["argv"], + cwd=task.workdir, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=os.name == "posix", + env=environment, + ) + store.bind_attempt( + token, worker_pid=process.pid, process_group=process.pid if os.name == "posix" else None + ) + worker = _Worker( + task, + attempt, + process, + time.monotonic(), + task.timeout_seconds or policy.timeout_seconds, + _Log(logs[0], policy.max_log_bytes), + _Log(logs[1], policy.max_log_bytes), + ) + for pipe, log in ((process.stdout, worker.stdout), (process.stderr, worker.stderr)): + assert pipe is not None + os.set_blocking(pipe.fileno(), False) + selector.register(pipe, selectors.EVENT_READ, log) + return worker + except BaseException as original: + cleanup_failed = False + if process is not None: + try: + _stop_owned_process(process) + except BaseException as cleanup_error: + cleanup_failed = True + _note_cleanup_failure(store, task.id, original, cleanup_error) + for pipe in (process.stdout, process.stderr): + if pipe is not None: + try: + selector.unregister(pipe) + except KeyError: + pass + try: + pipe.close() + except BaseException as cleanup_error: + cleanup_failed = True + _note_cleanup_failure(store, task.id, original, cleanup_error) + for log in logs: + try: + log.close() + except BaseException as cleanup_error: + cleanup_failed = True + _note_cleanup_failure(store, task.id, original, cleanup_error) + if not cleanup_failed: + if isinstance(original, (OSError, ValueError)): + raise _SpawnFailure(str(original)) from original + try: + store.finish_attempt(token, "interrupted", "launch interrupted; retry explicitly") + except BaseException as cleanup_error: + _note_cleanup_failure(store, task.id, original, cleanup_error) + raise def dispatch_ready(store: TaskStore, policy: DispatchPolicy) -> DispatchResult: - running = len(store.list_tasks((TaskStatus.RUNNING,))) - slots = max(policy.max_concurrent_workers - running, 0) + """Run a bounded snapshot, concurrently; --drain processes its remaining ready tasks.""" + if os.name != "posix" and policy.backend == "subprocess": + raise ValueError( + "subprocess dispatch requires POSIX; preview and queue commands are portable" + ) + if policy.backend == "subprocess": + _require_default_child_handler() + slots = max(policy.max_concurrent_workers - len(store.list_tasks((TaskStatus.RUNNING,))), 0) if slots == 0: return DispatchResult((), "concurrency cap reached") - ready = store.list_tasks((TaskStatus.TODO,))[:slots] - started: list[str] = [] - for task in ready: - if policy.backend == "dry-run": + ready = store.list_tasks((TaskStatus.TODO,)) + if not policy.drain: + ready = [task for task in ready if task.not_before <= time.time()][:slots] + if policy.backend == "dry-run": + ready = [task for task in ready if task.not_before <= time.time()] + for task in ready: store.add_event(task.id, "dispatch_preview", "dry-run dispatch preview", {}) - started.append(task.id) - continue - argv = _command_argv_for_task(task, policy) - if not store.claim_task( - task.id, - policy.max_concurrent_workers, - f"started: {shlex.join(argv)}", - ): - continue - started.append(task.id) + return DispatchResult(tuple(task.id for task in ready)) + # Validate all selected command syntax before claiming or starting any work. + commands = {task.id: _command_argv_for_task(task, policy) for task in ready} + pending = {task.id: task for task in ready} + active: dict[str, _Worker] = {} + started: list[str] = [] + completed: list[str] = [] + failed: list[str] = [] + cancelled: list[str] = [] + skipped = "" + with selectors.DefaultSelector() as selector: try: - completed = subprocess.run( - list(argv), - check=False, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=policy.timeout_seconds, - cwd=Path(task.workdir) if task.workdir else None, - ) - except subprocess.TimeoutExpired: - store.mark_task( - task.id, - TaskStatus.FAILED, - f"worker timed out after {policy.timeout_seconds}s", - expected_status=TaskStatus.RUNNING, - ) - continue - except OSError as exc: - store.mark_task( - task.id, - TaskStatus.FAILED, - f"worker could not start: {exc}", - expected_status=TaskStatus.RUNNING, - ) - continue - if completed.returncode == 0: - store.mark_task( - task.id, - TaskStatus.DONE, - completed.stdout[-500:], - expected_status=TaskStatus.RUNNING, - ) - else: - store.mark_task( - task.id, - TaskStatus.FAILED, - f"worker exited {completed.returncode}: " - + (completed.stderr or completed.stdout)[-500:], - expected_status=TaskStatus.RUNNING, - ) - return DispatchResult(tuple(started)) + while pending or active: + # Stop cancellations before attempting to fill vacated slots. + for task_id, worker in list(active.items()): + task = store.get_task(task_id) + ended = _exit_ready(worker.process) + outcome, message = None, None + if not task or task.status != TaskStatus.RUNNING: + _stop_owned_process(worker.process) + outcome, message = ( + "cancelled", + "worker stopped after operator status change", + ) + elif ended: + if os.name == "posix" and _HAS_WAITID: + _stop_owned_process(worker.process) + else: + worker.process.wait() + outcome = "done" if worker.process.returncode == 0 else "failed" + elif time.monotonic() - worker.started >= worker.timeout: + _stop_owned_process(worker.process) + outcome, message = "timeout", f"worker timed out after {worker.timeout:g}s" + if outcome: + _finish_worker(store, selector, worker, outcome, message) + ( + completed + if outcome == "done" + else cancelled + if outcome == "cancelled" + else failed + ).append(task_id) + del active[task_id] + elif time.monotonic() - worker.last_heartbeat >= 0.5: + store.heartbeat_attempt(worker.attempt["claim_token"]) + worker.last_heartbeat = time.monotonic() + for task_id in list(pending): + if len(active) >= policy.max_concurrent_workers: + break + task = store.get_task(task_id) + if not task or task.status != TaskStatus.TODO: + del pending[task_id] + continue + if task.not_before > time.time(): + continue + attempt = store.claim_attempt( + task_id, policy.max_concurrent_workers, commands[task_id] + ) + if attempt is None: + if task.attempt_count >= task.max_attempts: + del pending[task_id] + skipped = "some queued tasks exhausted their attempt budget" + continue + del pending[task_id] + started.append(task_id) + try: + worker = _spawn(store, selector, task, attempt, policy) + assert worker is not None + active[task_id] = worker + except _SpawnFailure as exc: + store.finish_attempt( + attempt["claim_token"], "spawn_error", f"worker could not start: {exc}" + ) + failed.append(task_id) + if ( + pending + and not active + and all( + store.get_task(task_id).not_before <= time.time() for task_id in pending + ) + ): + skipped = "concurrency capacity held by another or unrecovered attempt" + break + for key, _mask in selector.select(timeout=0.05): + _read_pipe(key.fileobj, key.data, limit=4) + except BaseException as original: + for worker in active.values(): + try: + _stop_owned_process(worker.process) + _finish_worker( + store, + selector, + worker, + "interrupted", + "dispatcher interrupted; retry explicitly", + ) + except BaseException as cleanup_error: + _note_cleanup_failure(store, worker.task.id, original, cleanup_error) + for pipe, log in ( + (worker.process.stdout, worker.stdout), + (worker.process.stderr, worker.stderr), + ): + assert pipe is not None + try: + selector.unregister(pipe) + except KeyError: + pass + for handle in (pipe, log.file): + try: + handle.close() + except BaseException as close_error: + _note_cleanup_failure(store, worker.task.id, original, close_error) + raise + return DispatchResult( + tuple(started), skipped, tuple(completed), tuple(failed), tuple(cancelled) + ) diff --git a/src/agent_backlog_runner/maintenance.py b/src/agent_backlog_runner/maintenance.py new file mode 100644 index 0000000..5e1e588 --- /dev/null +++ b/src/agent_backlog_runner/maintenance.py @@ -0,0 +1,174 @@ +"""Small checks and an executable failure/retry/resume walkthrough.""" + +from __future__ import annotations + +import argparse +import ast +import json +import re +import sys +from pathlib import Path + +from .batches import enqueue_batch, load_batch_manifest +from .dispatch import DispatchPolicy, dispatch_ready +from .models import TaskStatus +from .reports import write_queue_report +from .store import init_store + + +def check_syntax(directory: Path) -> dict[str, object]: + files = sorted(directory.rglob("*.py")) + if not files: + raise ValueError("no Python source files found") + for path in files: + ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + return {"check": "python-syntax", "files": len(files), "passed": True} + + +def check_links(document: Path) -> dict[str, object]: + """Check local inline Markdown file links; web URLs and heading fragments are skipped.""" + missing, checked = [], 0 + for target in re.findall(r"\]\(([^)]+)\)", document.read_text(encoding="utf-8")): + target = target.strip().strip("<>") + if not target or re.match(r"[a-zA-Z]+:", target) or target.startswith("#"): + continue + target = target.split("#", 1)[0] + checked += 1 + if not (document.parent / target).exists(): + missing.append(target) + if missing: + raise ValueError("missing linked files: " + ", ".join(missing)) + return {"check": "local-document-links", "links": checked, "passed": True} + + +def run_demo(output: Path) -> dict[str, object]: + """Perform real syntax, unittest, and local-link checks on a tiny example project.""" + output = output.resolve() + output.mkdir(parents=True, exist_ok=False) + project = output / "project" + (project / "src").mkdir(parents=True) + (project / "tests").mkdir() + (project / "docs").mkdir() + (project / "src" / "totals.py").write_text( + "def total(values):\n return sum(values)\n", encoding="utf-8" + ) + (project / "tests" / "test_totals.py").write_text( + "import sys, unittest\nfrom pathlib import Path\n" + "sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'src'))\n" + "from totals import total\n\nclass TotalsTests(unittest.TestCase):\n" + " def test_empty(self):\n self.assertEqual(total([]), 0)\n" + " def test_values(self):\n self.assertEqual(total([3, -1, 4]), 6)\n", + encoding="utf-8", + ) + (project / "README.md").write_text( + "# Totals example\n\n[Usage](docs/usage.md)\n", encoding="utf-8" + ) + manifest_path = output / "checks.json" + manifest_path.write_text( + json.dumps( + { + "schema_version": 1, + "name": "example-checks", + "tasks": [ + { + "id": "unit-tests", + "title": "Run project unit tests", + "argv": ["{python}", "-m", "unittest", "discover", "-s", "tests", "-v"], + "retry_backoff_seconds": 0, + }, + { + "id": "python-syntax", + "title": "Parse source files", + "argv": [ + "{python}", + "-m", + "agent_backlog_runner.maintenance", + "syntax", + "src", + ], + "retry_backoff_seconds": 0, + }, + { + "id": "document-links", + "title": "Check README file links", + "argv": [ + "{python}", + "-m", + "agent_backlog_runner.maintenance", + "links", + "README.md", + ], + "retry_backoff_seconds": 0, + }, + ], + }, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + db = output / "checks.sqlite3" + with init_store(db) as store: + tasks = enqueue_batch(store, load_batch_manifest(manifest_path, root=project)) + first = dispatch_ready(store, DispatchPolicy(1, "subprocess")) + assert len(first.completed) == 1 + # Resume through a fresh connection: the completed check does not execute again. + with init_store(db) as store: + resumed = dispatch_ready(store, DispatchPolicy(2, "subprocess", drain=True)) + if len(resumed.failed) != 1: + raise RuntimeError("expected the missing usage document to fail its local-link check") + failure = store.get_task(resumed.failed[0]) + assert failure is not None + write_queue_report( + store, json_out=output / "before-repair.json", md_out=output / "before-repair.md" + ) + (project / "docs" / "usage.md").write_text( + "# Usage\n\nImport `total` and pass a sequence of numbers.\n", encoding="utf-8" + ) + with init_store(db) as store: + store.retry_task(failure.id) + with init_store(db) as store: + repaired = dispatch_ready(store, DispatchPolicy(2, "subprocess", drain=True)) + if repaired.failed or any(task.status != TaskStatus.DONE for task in store.list_tasks()): + raise RuntimeError("repair failed; inspect attempts and log files") + write_queue_report( + store, json_out=output / "after-repair.json", md_out=output / "after-repair.md" + ) + attempts = store.attempts() + return { + "checks": len(tasks), + "initial_completed": len(first.completed), + "resume_failed": len(resumed.failed), + "repaired_checks": len(repaired.completed), + "attempts": len(attempts), + "final_done": len(store.list_tasks((TaskStatus.DONE,))), + "database": str(db), + "report": str(output / "after-repair.md"), + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + for name in ("syntax", "links"): + cmd = sub.add_parser(name) + cmd.add_argument("path", type=Path) + demo = sub.add_parser("demo") + demo.add_argument("--output", type=Path, default=Path(".agent-backlog/maintenance-demo")) + args = parser.parse_args(argv) + try: + if args.command == "demo": + result = run_demo(args.output) + elif args.command == "syntax": + result = check_syntax(args.path) + else: + result = check_links(args.path) + print(json.dumps(result, indent=2)) + return 0 + except (OSError, ValueError, SyntaxError, RuntimeError) as exc: + print(f"check failed: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/agent_backlog_runner/models.py b/src/agent_backlog_runner/models.py index d61ff09..963378b 100644 --- a/src/agent_backlog_runner/models.py +++ b/src/agent_backlog_runner/models.py @@ -45,6 +45,12 @@ class TaskRecord: created_at: int = 0 updated_at: int = 0 heartbeat_at: int | None = None + command_argv: tuple[str, ...] | None = None + attempt_count: int = 0 + max_attempts: int = 3 + retry_backoff_seconds: float = 1 + timeout_seconds: float | None = None + not_before: float = 0 @dataclass(frozen=True) diff --git a/src/agent_backlog_runner/recovery.py b/src/agent_backlog_runner/recovery.py new file mode 100644 index 0000000..82b3e60 --- /dev/null +++ b/src/agent_backlog_runner/recovery.py @@ -0,0 +1,73 @@ +"""Conservative local recovery; this module never signals stored process IDs.""" + +from __future__ import annotations + +import os +import time + +from .models import TaskStatus +from .store import TaskStore + + +def _pid_exists(pid: int | None, *, group: bool = False) -> bool: + if pid is None: + return False + try: + if group and os.name == "posix": + os.killpg(pid, 0) + else: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def recover_task( + store: TaskStore, task_id: str, *, stale_after: float = 1800, acknowledge_unknown: bool = False +) -> None: + if stale_after < 0: + raise ValueError("stale_after must be non-negative") + with store.transaction(): + task = store.get_task(task_id) + if task is None: + raise KeyError(task_id) + active = [item for item in store.attempts(task_id) if item["finished_at"] is None] + if task.status != TaskStatus.RUNNING and not active: + raise ValueError("task has no unfinished running work") + basis = task.heartbeat_at or task.updated_at + if time.time() - basis < stale_after: + raise ValueError("task is not stale; wait or inspect with stale --after") + for attempt in active: + if _pid_exists(attempt["owner_pid"]): + raise ValueError( + "dispatcher PID still exists; cancel through its coordinator or inspect it" + ) + if _pid_exists(attempt["worker_pid"]) or _pid_exists( + attempt["process_group"], group=True + ): + raise ValueError( + "worker PID or process group still exists; stop and inspect it before recovery" + ) + if attempt["worker_pid"] is None and not acknowledge_unknown: + raise ValueError( + "worker identity was not recorded; inspect possible side effects " + "and pass --acknowledge-unknown" + ) + if not active and not acknowledge_unknown: + raise ValueError( + "legacy running task has no attempt identity; inspect it " + "and pass --acknowledge-unknown" + ) + for attempt in active: + store.finish_attempt( + attempt["claim_token"], + "abandoned", + "stale attempt explicitly recovered; inspect outputs before retry", + ) + if not active: + store.mark_task( + task_id, TaskStatus.FAILED, "untracked running work explicitly abandoned" + ) + store.add_event(task_id, "recovered", "recovered; retry is a separate explicit action", {}) diff --git a/src/agent_backlog_runner/reports.py b/src/agent_backlog_runner/reports.py index 0c1ab5b..b99d546 100644 --- a/src/agent_backlog_runner/reports.py +++ b/src/agent_backlog_runner/reports.py @@ -1,11 +1,14 @@ from __future__ import annotations import json +import os from dataclasses import asdict, is_dataclass +from pathlib import Path from typing import Any from .scheduler import BacklogPlan -from .status import StatusSnapshot, format_snapshot +from .status import StatusSnapshot, format_snapshot, get_status_snapshot +from .store import TaskStore def _default(value: Any) -> Any: @@ -26,3 +29,62 @@ def snapshot_to_json(snapshot: StatusSnapshot) -> str: def snapshot_to_table(snapshot: StatusSnapshot) -> str: return format_snapshot(snapshot) + + +def write_queue_report( + store: TaskStore, *, json_out: Path | None = None, md_out: Path | None = None +) -> None: + if not json_out and not md_out: + raise ValueError("choose --json-out or --md-out") + tasks, attempts = store.list_tasks(), store.attempts() + snapshot = get_status_snapshot(store) + if json_out: + json_out.parent.mkdir(parents=True, exist_ok=True) + json_out.write_text( + json.dumps( + { + "summary": asdict(snapshot), + "tasks": [asdict(task) for task in tasks], + "attempts": attempts, + }, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + if md_out: + md_out.parent.mkdir(parents=True, exist_ok=True) + lines = [ + "# Local command queue report", + "", + "```text", + format_snapshot(snapshot), + "```", + "", + "| Task | Status | Attempts | Latest outcome | Exit | Logs |", + "|---|---|---:|---|---|---|", + ] + for task in tasks: + latest = next((attempt for attempt in attempts if attempt["task_id"] == task.id), None) + title = task.title.replace("|", "\\|").replace("\n", " ") + logs = [] + if latest: + for stream in ("stdout", "stderr"): + if latest[stream + "_path"]: + path = os.path.relpath(latest[stream + "_path"], md_out.resolve().parent) + logs.append(f"[{stream}](<{path}>)") + outcome, code = ( + (latest["outcome"], latest["exit_code"]) if latest else ("not started", "") + ) + lines.append( + f"| {title} (`{task.id}`) | {task.status.value} | " + f"{task.attempt_count}/{task.max_attempts} | {outcome} | {code} | " + f"{' · '.join(logs)} |" + ) + lines += [ + "", + "A zero command exit is completion, not a correctness guarantee. " + "Inspect retained logs and each command's reports.", + "", + ] + md_out.write_text("\n".join(lines), encoding="utf-8") diff --git a/src/agent_backlog_runner/status.py b/src/agent_backlog_runner/status.py index bc91bc1..fd4bf32 100644 --- a/src/agent_backlog_runner/status.py +++ b/src/agent_backlog_runner/status.py @@ -1,7 +1,8 @@ from __future__ import annotations import time -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import Any from .models import TaskEvent, TaskRecord, TaskStatus from .store import TaskStore @@ -19,6 +20,10 @@ class StatusSnapshot: active_workers: int stale_tasks: tuple[StaleTask, ...] recent_events: tuple[TaskEvent, ...] + ready_tasks: int = 0 + delayed_tasks: int = 0 + attempt_outcomes: dict[str, int] = field(default_factory=dict) + recent_attempts: tuple[dict[str, Any], ...] = () def get_status_snapshot( @@ -28,6 +33,10 @@ def get_status_snapshot( counts = {status.value: 0 for status in TaskStatus} now = int(time.time()) stale: list[StaleTask] = [] + attempts = store.attempts() + outcomes: dict[str, int] = {} + for attempt in attempts: + outcomes[attempt["outcome"]] = outcomes.get(attempt["outcome"], 0) + 1 for task in tasks: counts[task.status.value] = counts.get(task.status.value, 0) + 1 if task.status == TaskStatus.RUNNING: @@ -37,9 +46,25 @@ def get_status_snapshot( stale.append(StaleTask(task, age)) return StatusSnapshot( counts=counts, - active_workers=counts.get(TaskStatus.RUNNING.value, 0), + active_workers=sum(attempt["finished_at"] is None for attempt in attempts) + + sum( + task.status == TaskStatus.RUNNING + and not any( + attempt["task_id"] == task.id and attempt["finished_at"] is None + for attempt in attempts + ) + for task in tasks + ), stale_tasks=tuple(stale), recent_events=tuple(store.events(limit=event_limit)), + ready_tasks=sum( + task.status == TaskStatus.TODO and task.not_before <= time.time() for task in tasks + ), + delayed_tasks=sum( + task.status == TaskStatus.TODO and task.not_before > time.time() for task in tasks + ), + attempt_outcomes=outcomes, + recent_attempts=tuple(attempts[:event_limit]), ) @@ -49,4 +74,8 @@ def format_snapshot(snapshot: StatusSnapshot) -> str: lines.append(f"{key} {snapshot.counts[key]}") lines.append(f"active_workers {snapshot.active_workers}") lines.append(f"stale_tasks {len(snapshot.stale_tasks)}") + lines.append(f"ready_tasks {snapshot.ready_tasks}") + lines.append(f"delayed_tasks {snapshot.delayed_tasks}") + for outcome, count in sorted(snapshot.attempt_outcomes.items()): + lines.append(f"attempts.{outcome} {count}") return "\n".join(lines) diff --git a/src/agent_backlog_runner/store.py b/src/agent_backlog_runner/store.py index 3f874ce..a577b46 100644 --- a/src/agent_backlog_runner/store.py +++ b/src/agent_backlog_runner/store.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os import sqlite3 import time import uuid @@ -12,7 +13,7 @@ from .models import TaskEvent, TaskRecord, TaskStatus from .safety import require_safe_slug, require_safe_task_id -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 SCHEMA = """ PRAGMA foreign_keys = ON; @@ -52,8 +53,40 @@ CREATE INDEX IF NOT EXISTS idx_tasks_status_priority ON tasks(status, priority DESC, created_at ASC); CREATE INDEX IF NOT EXISTS idx_events_task_id ON task_events(task_id, id DESC); +CREATE TABLE IF NOT EXISTS task_attempts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL REFERENCES tasks(id), + attempt_no INTEGER NOT NULL, + claim_token TEXT NOT NULL UNIQUE, + owner_pid INTEGER NOT NULL, + worker_pid INTEGER, + process_group INTEGER, + argv TEXT NOT NULL, + workdir TEXT, + started_at REAL NOT NULL, + finished_at REAL, + outcome TEXT NOT NULL DEFAULT 'running', + exit_code INTEGER, + stdout_path TEXT, + stderr_path TEXT, + stdout_bytes INTEGER NOT NULL DEFAULT 0, + stderr_bytes INTEGER NOT NULL DEFAULT 0, + stdout_truncated INTEGER NOT NULL DEFAULT 0, + stderr_truncated INTEGER NOT NULL DEFAULT 0, + UNIQUE(task_id, attempt_no) +); +CREATE INDEX IF NOT EXISTS idx_attempts_task ON task_attempts(task_id, attempt_no); """ +_TASK_COLUMNS = { + "command_argv": "TEXT", + "attempt_count": "INTEGER NOT NULL DEFAULT 0", + "max_attempts": "INTEGER NOT NULL DEFAULT 3", + "retry_backoff_seconds": "REAL NOT NULL DEFAULT 1", + "timeout_seconds": "REAL", + "not_before": "REAL NOT NULL DEFAULT 0", +} + def _now() -> int: return int(time.time()) @@ -74,6 +107,12 @@ def _task_from_row(row: sqlite3.Row) -> TaskRecord: created_at=row["created_at"], updated_at=row["updated_at"], heartbeat_at=row["heartbeat_at"], + command_argv=tuple(json.loads(row["command_argv"])) if row["command_argv"] else None, + attempt_count=row["attempt_count"], + max_attempts=row["max_attempts"], + retry_backoff_seconds=row["retry_backoff_seconds"], + timeout_seconds=row["timeout_seconds"], + not_before=row["not_before"], ) @@ -94,7 +133,7 @@ def __init__(self, path: str | Path) -> None: if self.path.exists() and self.path.is_dir(): raise ValueError("database path is a directory") self.path.parent.mkdir(parents=True, exist_ok=True) - self.conn = sqlite3.connect(self.path) + self.conn = sqlite3.connect(self.path, timeout=30) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA foreign_keys = ON") @@ -115,6 +154,11 @@ def init_schema(self) -> None: raise ValueError("database schema is newer than this package supports") with self.conn: self.conn.executescript(SCHEMA) + with self.transaction(): + columns = {row["name"] for row in self.conn.execute("PRAGMA table_info(tasks)")} + for name, definition in _TASK_COLUMNS.items(): + if name not in columns: + self.conn.execute(f"ALTER TABLE tasks ADD COLUMN {name} {definition}") self.conn.execute( """INSERT INTO schema_meta(key,value,updated_at) VALUES ('schema_version',?,?) ON CONFLICT(key) DO UPDATE SET @@ -160,12 +204,29 @@ def create_task( command: str | None = None, workdir: str | None = None, event_message: str = "created", + command_argv: tuple[str, ...] | None = None, + max_attempts: int = 3, + retry_backoff_seconds: float = 1, + timeout_seconds: float | None = None, ) -> TaskRecord: task_id = require_safe_task_id(task_id or self.new_task_id()) lane = require_safe_slug(lane, field="lane") role = require_safe_slug(role, field="role") for tag in tags: require_safe_slug(tag, field="tag") + if type(max_attempts) is not int or not 1 <= max_attempts <= 100: + raise ValueError("max_attempts must be an integer from 1 to 100") + if type(retry_backoff_seconds) not in (int, float) or not 0 <= retry_backoff_seconds <= 60: + raise ValueError("retry_backoff_seconds must be between 0 and 60") + if timeout_seconds is not None and ( + type(timeout_seconds) not in (int, float) or not 0 < timeout_seconds <= 86400 + ): + raise ValueError("timeout_seconds must be positive and at most 86400") + if command_argv is not None and ( + not command_argv + or not all(isinstance(arg, str) and "\0" not in arg for arg in command_argv) + ): + raise ValueError("command_argv must contain string arguments without NUL bytes") now = _now() with self.transaction(): self.conn.execute( @@ -189,6 +250,17 @@ def create_task( ), ) self._add_event_unlocked(task_id, "created", event_message, {}) + self.conn.execute( + """UPDATE tasks SET command_argv=?,max_attempts=?,retry_backoff_seconds=?, + timeout_seconds=? WHERE id=?""", + ( + json.dumps(command_argv) if command_argv else None, + max_attempts, + retry_backoff_seconds, + timeout_seconds, + task_id, + ), + ) task = self.get_task(task_id) assert task is not None return task @@ -227,7 +299,7 @@ def list_tasks(self, statuses: tuple[TaskStatus, ...] = ()) -> list[TaskRecord]: placeholders = ",".join("?" for _ in statuses) query = ( f"SELECT * FROM tasks WHERE status IN ({placeholders}) " - "ORDER BY priority DESC, created_at ASC" + "ORDER BY priority DESC, created_at ASC, rowid ASC" ) rows = self.conn.execute( query, @@ -235,7 +307,7 @@ def list_tasks(self, statuses: tuple[TaskStatus, ...] = ()) -> list[TaskRecord]: ).fetchall() else: rows = self.conn.execute( - "SELECT * FROM tasks ORDER BY priority DESC, created_at ASC" + "SELECT * FROM tasks ORDER BY priority DESC, created_at ASC, rowid ASC" ).fetchall() return [_task_from_row(row) for row in rows] @@ -304,6 +376,230 @@ def record_heartbeat(self, task_id: str, payload: dict[str, Any] | None = None) raise KeyError(task_id) self._add_event_unlocked(task_id, "heartbeat", "heartbeat", payload or {}) + def attempts(self, task_id: str | None = None) -> list[dict[str, Any]]: + if task_id is not None: + require_safe_task_id(task_id) + rows = self.conn.execute( + "SELECT * FROM task_attempts" + + (" WHERE task_id=?" if task_id else "") + + " ORDER BY id DESC", + (task_id,) if task_id else (), + ).fetchall() + return [{**dict(row), "argv": json.loads(row["argv"])} for row in rows] + + def attempt(self, token: str) -> dict[str, Any] | None: + row = self.conn.execute( + "SELECT * FROM task_attempts WHERE claim_token=?", (token,) + ).fetchone() + return None if row is None else {**dict(row), "argv": json.loads(row["argv"])} + + def claim_attempt( + self, task_id: str, max_running: int, argv: tuple[str, ...] + ) -> dict[str, Any] | None: + """Reserve capacity and a unique attempt in one write transaction.""" + require_safe_task_id(task_id) + with self.transaction(): + task = self.get_task(task_id) + if ( + task is None + or task.status != TaskStatus.TODO + or task.not_before > time.time() + or task.attempt_count >= task.max_attempts + ): + return None + if self.conn.execute( + "SELECT 1 FROM task_attempts WHERE task_id=? AND finished_at IS NULL", (task_id,) + ).fetchone(): + return None + # Cancelled but not yet stopped attempts still consume physical capacity. + running = self.conn.execute("""SELECT + (SELECT COUNT(*) FROM task_attempts WHERE finished_at IS NULL) + + (SELECT COUNT(*) FROM tasks t WHERE status='running' AND NOT EXISTS + (SELECT 1 FROM task_attempts a WHERE a.task_id=t.id AND a.finished_at IS NULL)) + """).fetchone()[0] + capacity = self.conn.execute( + "SELECT value FROM schema_meta WHERE key='dispatch_capacity'" + ).fetchone() + if running and capacity and int(capacity[0]) != max_running: + raise ValueError("active dispatchers must use the same --max-workers capacity") + if not running or capacity is None: + self.conn.execute( + """INSERT INTO schema_meta(key,value,updated_at) VALUES('dispatch_capacity',?,?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value, + updated_at=excluded.updated_at""", + (str(max_running), _now()), + ) + if running >= max_running: + return None + token = uuid.uuid4().hex + now = time.time() + self.conn.execute( + """INSERT INTO task_attempts + (task_id,attempt_no,claim_token,owner_pid,argv,workdir,started_at) + VALUES (?,?,?,?,?,?,?)""", + ( + task_id, + task.attempt_count + 1, + token, + os.getpid(), + json.dumps(argv), + task.workdir, + now, + ), + ) + self.conn.execute( + """UPDATE tasks SET status='running',attempt_count=attempt_count+1, + updated_at=?,heartbeat_at=? WHERE id=?""", + (int(now), int(now), task_id), + ) + self._add_event_unlocked( + task_id, "dispatched", "worker attempt claimed", {"claim_token": token} + ) + return self.attempt(token) + + def bind_attempt(self, token: str, **values: Any) -> None: + allowed = {"worker_pid", "process_group", "stdout_path", "stderr_path"} + if not values or set(values) - allowed: + raise ValueError("invalid attempt binding fields") + with self.transaction(): + cursor = self.conn.execute( + "UPDATE task_attempts SET " + + ",".join(f"{key}=?" for key in values) + + " WHERE claim_token=? AND finished_at IS NULL", + (*values.values(), token), + ) + if cursor.rowcount != 1: + raise ValueError("attempt is no longer active") + + def heartbeat_attempt(self, token: str) -> None: + with self.transaction(): + self.conn.execute( + """UPDATE tasks SET heartbeat_at=?,updated_at=? + WHERE status='running' AND id=(SELECT task_id FROM task_attempts + WHERE claim_token=? AND finished_at IS NULL)""", + (_now(), _now(), token), + ) + + def finish_attempt( + self, + token: str, + outcome: str, + message: str, + *, + exit_code: int | None = None, + stdout_bytes: int = 0, + stderr_bytes: int = 0, + stdout_truncated: bool = False, + stderr_truncated: bool = False, + ) -> bool: + if outcome not in { + "done", + "failed", + "timeout", + "cancelled", + "interrupted", + "abandoned", + "spawn_error", + }: + raise ValueError("invalid attempt outcome") + with self.transaction(): + attempt = self.attempt(token) + if attempt is None or attempt["finished_at"] is not None: + return False + self.conn.execute( + """UPDATE task_attempts SET outcome=?,finished_at=?,exit_code=?, + stdout_bytes=?,stderr_bytes=?,stdout_truncated=?,stderr_truncated=? + WHERE claim_token=?""", + ( + outcome, + time.time(), + exit_code, + stdout_bytes, + stderr_bytes, + int(stdout_truncated), + int(stderr_truncated), + token, + ), + ) + status = ( + TaskStatus.DONE + if outcome == "done" + else (TaskStatus.CANCELLED if outcome == "cancelled" else TaskStatus.FAILED) + ) + task = self.get_task(attempt["task_id"]) + if task and task.status == TaskStatus.RUNNING: + self.conn.execute( + "UPDATE tasks SET status=?,updated_at=? WHERE id=?", + (status.value, _now(), task.id), + ) + event_type = ( + "completed" + if outcome == "done" + else ("cancelled" if outcome == "cancelled" else "failed") + ) + self._add_event_unlocked( + attempt["task_id"], + event_type, + message, + { + "attempt": attempt["attempt_no"], + "outcome": outcome, + "exit_code": exit_code, + "stdout_path": attempt["stdout_path"], + "stderr_path": attempt["stderr_path"], + "stdout_truncated": stdout_truncated, + "stderr_truncated": stderr_truncated, + }, + ) + return True + + def retry_task(self, task_id: str) -> TaskRecord: + """Explicitly schedule one more attempt; never retry as a side effect of dispatch.""" + with self.transaction(): + task = self.get_task(task_id) + if task is None: + raise KeyError(task_id) + if task.status not in {TaskStatus.FAILED, TaskStatus.CANCELLED}: + raise ValueError("only failed or cancelled tasks can be retried") + if any(attempt["finished_at"] is None for attempt in self.attempts(task_id)): + raise ValueError( + "attempt is still active; wait for cancellation or recover it first" + ) + if task.attempt_count >= task.max_attempts: + raise ValueError("attempt budget exhausted") + delay = min(60, task.retry_backoff_seconds * 2 ** max(0, task.attempt_count - 1)) + due = time.time() + delay + self.conn.execute( + "UPDATE tasks SET status='todo',not_before=?,updated_at=? WHERE id=?", + (due, _now(), task_id), + ) + self._add_event_unlocked( + task_id, + "retry_scheduled", + f"retry available after {delay:g}s", + {"not_before": due, "next_attempt": task.attempt_count + 1}, + ) + result = self.get_task(task_id) + assert result is not None + return result + + def request_cancel( + self, task_id: str, message: str = "operator requested cancellation" + ) -> None: + with self.transaction(): + task = self.get_task(task_id) + if task is None: + raise KeyError(task_id) + if task.status not in {TaskStatus.TODO, TaskStatus.RUNNING, TaskStatus.BLOCKED}: + raise ValueError("task is already terminal") + if task.status == TaskStatus.RUNNING and not any( + attempt["finished_at"] is None for attempt in self.attempts(task_id) + ): + raise ValueError( + "running task has no managed attempt; inspect and recover it first" + ) + self.mark_task(task_id, TaskStatus.CANCELLED, message) + def active_depth(self, lanes: tuple[str, ...] = ()) -> int: params: list[Any] = [TaskStatus.TODO.value, TaskStatus.RUNNING.value] where = "status IN (?,?)" diff --git a/src/agent_backlog_runner/templates.py b/src/agent_backlog_runner/templates.py index c4140aa..188c9d0 100644 --- a/src/agent_backlog_runner/templates.py +++ b/src/agent_backlog_runner/templates.py @@ -15,6 +15,27 @@ _REQUIRED_FIELDS = {"slug", "title", "body"} +def _unique_mapping(pairs): + result = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate JSON field: {key}") + result[key] = value + return result + + +class _UniqueSafeLoader(yaml.SafeLoader): + def construct_mapping(self, node, deep=False): + mapping = super().construct_mapping(node, deep=deep) + seen = set() + for key_node, _ in node.value: + key = self.construct_object(key_node, deep=deep) + if key in seen: + raise ValueError(f"duplicate YAML field: {key}") + seen.add(key) + return mapping + + @dataclass(frozen=True) class TemplateCatalog: templates: tuple[TaskTemplate, ...] @@ -36,8 +57,8 @@ def select( def _load_data(path: Path) -> Any: text = path.read_text(encoding="utf-8") if path.suffix.lower() == ".json": - return json.loads(text) - return yaml.safe_load(text) + return json.loads(text, object_pairs_hook=_unique_mapping) + return yaml.load(text, Loader=_UniqueSafeLoader) def _as_list(data: Any) -> list[dict[str, Any]]: diff --git a/tests/test_batch_execution.py b/tests/test_batch_execution.py new file mode 100644 index 0000000..75c566b --- /dev/null +++ b/tests/test_batch_execution.py @@ -0,0 +1,604 @@ +from __future__ import annotations + +import json +import os +import signal +import sqlite3 +import subprocess +import sys +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Barrier, Timer + +import pytest + +from agent_backlog_runner.batches import enqueue_batch, load_batch_manifest +from agent_backlog_runner.cli import main +from agent_backlog_runner.dispatch import DispatchPolicy, dispatch_ready +from agent_backlog_runner.models import TaskStatus +from agent_backlog_runner.recovery import recover_task +from agent_backlog_runner.store import init_store + + +def _task(store, title, code, **kwargs): + return store.create_task( + title=title, body="", command_argv=(sys.executable, "-c", code), **kwargs + ) + + +def test_workers_must_overlap_to_complete(tmp_path): + """Each command waits for the other: a serial implementation cannot pass.""" + with init_store(tmp_path / "queue.sqlite3") as store: + for index in (0, 1): + _task( + store, + str(index), + "import time; from pathlib import Path; " + f"Path({str(tmp_path / str(index))!r}).write_text('ready'); " + f"other=Path({str(tmp_path / str(1 - index))!r}); " + "end=time.monotonic()+2\n" + "while not other.exists() and time.monotonic() time.time() + assert not dispatch_ready(store, DispatchPolicy(1, "dry-run")).started + dispatch_ready(store, DispatchPolicy(1, "subprocess", drain=True)) + assert store.get_task(task.id).attempt_count == 2 + assert [item["attempt_no"] for item in store.attempts(task.id)] == [2, 1] + with pytest.raises(ValueError, match="budget"): + store.retry_task(task.id) + + +def test_cancelled_attempt_keeps_capacity_until_coordinator_finishes(tmp_path): + with init_store(tmp_path / "queue.sqlite3") as store: + first = _task(store, "first", "pass") + second = _task(store, "second", "pass") + attempt = store.claim_attempt(first.id, 1, first.command_argv) + store.request_cancel(first.id) + assert store.claim_attempt(second.id, 1, second.command_argv) is None + with pytest.raises(ValueError, match="still active"): + store.retry_task(first.id) + store.finish_attempt(attempt["claim_token"], "cancelled", "stopped") + assert store.claim_attempt(second.id, 1, second.command_argv) is not None + + +def test_old_attempt_cannot_complete_a_new_attempt(tmp_path): + with init_store(tmp_path / "queue.sqlite3") as store: + task = _task(store, "tokens", "pass", retry_backoff_seconds=0) + first = store.claim_attempt(task.id, 1, task.command_argv) + store.finish_attempt(first["claim_token"], "failed", "failed") + store.retry_task(task.id) + second = store.claim_attempt(task.id, 1, task.command_argv) + assert not store.finish_attempt(first["claim_token"], "done", "stale completion") + assert store.get_task(task.id).status == TaskStatus.RUNNING + assert store.attempt(second["claim_token"])["finished_at"] is None + + +def test_claimed_dispatcher_crash_requires_explicit_unknown_acknowledgement(tmp_path): + db = tmp_path / "queue.sqlite3" + with init_store(db) as store: + task = _task(store, "crash window", "pass", retry_backoff_seconds=0) + code = ( + "import os; from agent_backlog_runner.store import init_store; " + f"s=init_store({str(db)!r}); " + f"s.claim_attempt({task.id!r},1,('unused',)); os._exit(0)" + ) + subprocess.run([sys.executable, "-c", code], check=True) + with init_store(db) as store: + with pytest.raises(ValueError, match="identity was not recorded"): + recover_task(store, task.id, stale_after=0) + recover_task(store, task.id, stale_after=0, acknowledge_unknown=True) + assert store.get_task(task.id).status == TaskStatus.FAILED + assert store.attempts(task.id)[0]["outcome"] == "abandoned" + store.retry_task(task.id) + assert dispatch_ready(store, DispatchPolicy(1, "subprocess")).completed == (task.id,) + assert store.get_task(task.id).attempt_count == 2 + + +def test_recovery_never_adopts_or_signals_a_live_pid(tmp_path): + with init_store(tmp_path / "queue.sqlite3") as store: + task = _task(store, "live owner", "pass") + store.claim_attempt(task.id, 1, task.command_argv) + with pytest.raises(ValueError, match="dispatcher PID still exists"): + recover_task(store, task.id, stale_after=0, acknowledge_unknown=True) + assert store.get_task(task.id).status == TaskStatus.RUNNING + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX inherited child signal disposition") +def test_inherited_ignored_sigchld_refused_before_claiming(tmp_path): + db = tmp_path / "queue.sqlite3" + with init_store(db) as store: + task = _task(store, "ignored child handler", "pass") + argv = [ + sys.executable, + "-m", + "agent_backlog_runner.cli", + "dispatch", + "--db", + str(db), + "--backend", + "subprocess", + ] + code = ( + "import os,signal; signal.signal(signal.SIGCHLD,signal.SIG_IGN); " + f"os.execv({sys.executable!r},{argv!r})" + ) + result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) + assert result.returncode == 1 + assert "default SIGCHLD" in result.stderr + with init_store(db) as store: + assert store.attempts(task.id) == [] + assert store.get_task(task.id).status == TaskStatus.TODO + + +@pytest.mark.skipif( + not all(hasattr(os, name) for name in ("waitid", "WNOWAIT", "P_PID")), + reason="requires retained-leader completion observation", +) +def test_external_reaper_never_allows_signalling_historical_group(tmp_path, monkeypatch, capsys): + import agent_backlog_runner.dispatch as dispatch + + reaped = [] + original_exit_ready = dispatch._exit_ready + + def reap(process): + os.waitpid(process.pid, 0) + reaped.append(process) + return original_exit_ready(process) + + def unsafe_signal(*_args): + pytest.fail("attempted to signal a reaped leader's historical group") + + monkeypatch.setattr(dispatch, "_exit_ready", reap) + monkeypatch.setattr(dispatch.os, "killpg", unsafe_signal) + with init_store(tmp_path / "queue.sqlite3") as store: + task = _task(store, "external reaper", "pass") + try: + with pytest.raises(ChildProcessError): + dispatch_ready(store, DispatchPolicy(1, "subprocess")) + assert store.attempts(task.id)[0]["finished_at"] is None + assert "already reaped" in capsys.readouterr().err + finally: + for process in reaped: + process.returncode = 0 # The test itself already reaped this child. + for pipe in (process.stdout, process.stderr): + if pipe: + pipe.close() + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process groups") +@pytest.mark.parametrize("cancel", [False, True]) +def test_timeout_and_cancellation_stop_child_process_group(tmp_path, cancel): + marker = tmp_path / "orphan.txt" + child = ( + "import time; from pathlib import Path; time.sleep(1); " + f"Path({str(marker)!r}).write_text('orphan')" + ) + code = ( + "import subprocess,sys,time; " + f"subprocess.Popen([sys.executable,'-c',{child!r}]); time.sleep(10)" + ) + db = tmp_path / "queue.sqlite3" + with init_store(db) as store: + task = _task(store, "group", code) + timer = None + if cancel: + + def request(): + with init_store(db) as other: + other.request_cancel(task.id) + + timer = Timer(0.2, request) + timer.start() + dispatch_ready(store, DispatchPolicy(1, "subprocess", timeout_seconds=0.3)) + if timer: + timer.join(timeout=5) + outcome = "cancelled" if cancel else "timeout" + assert store.attempts(task.id)[0]["outcome"] == outcome + time.sleep(1.1) + assert not marker.exists(), "worker's child survived process-group termination" + + +@pytest.mark.skipif(os.name != "posix", reason="SIGINT") +def test_graceful_interrupt_records_attempt_and_preserves_pending_queue(tmp_path): + db = tmp_path / "queue.sqlite3" + with init_store(db) as store: + first = _task( + store, "long", "import time; time.sleep(10)", priority=2, retry_backoff_seconds=0 + ) + second = _task(store, "pending", "print('remaining')") + process = subprocess.Popen( + [ + sys.executable, + "-m", + "agent_backlog_runner.cli", + "dispatch", + "--db", + str(db), + "--backend", + "subprocess", + "--drain", + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + try: + end = time.monotonic() + 5 + while time.monotonic() < end: + with init_store(db) as store: + attempts = store.attempts(first.id) + if attempts and attempts[0]["worker_pid"]: + break + time.sleep(0.02) + else: + pytest.fail("dispatcher did not start") + process.send_signal(signal.SIGINT) + process.communicate(timeout=5) + assert process.returncode == 130 + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + with init_store(db) as store: + assert store.attempts(first.id)[0]["outcome"] == "interrupted" + assert store.get_task(second.id).status == TaskStatus.TODO + result = dispatch_ready(store, DispatchPolicy(1, "subprocess", drain=True)) + assert result.completed == (second.id,) + + +def _manifest(tmp_path, **task_changes): + path = tmp_path / "batch.json" + task = {"id": "unit-tests", "argv": ["{python}", "-c", "print('ok')"], **task_changes} + path.write_text(json.dumps({"schema_version": 1, "name": "repo-checks", "tasks": [task]})) + return path + + +def test_manifest_load_is_idempotent_and_rejects_changed_run(tmp_path): + path = _manifest(tmp_path) + with init_store(tmp_path / "queue.sqlite3") as store: + manifest = load_batch_manifest(path) + first = enqueue_batch(store, manifest) + second = enqueue_batch(store, manifest) + assert first[0].id == second[0].id + assert len(store.list_tasks()) == 1 + changed = load_batch_manifest(_manifest(tmp_path, title="changed")) + with pytest.raises(ValueError, match="definition changed"): + enqueue_batch(store, changed) + assert len(enqueue_batch(store, changed, run_id="next-run")) == 1 + assert len(store.list_tasks()) == 2 + + +@pytest.mark.parametrize( + "changes", + [ + {"argv": "python"}, + {"workdir": ".."}, + {"max_attempts": True}, + {"timeout_seconds": 0}, + {"retry_backoff_seconds": -1}, + {"unknown": "typo"}, + ], +) +def test_manifest_rejects_unsafe_or_ambiguous_shapes(tmp_path, changes): + with pytest.raises(ValueError): + load_batch_manifest(_manifest(tmp_path, **changes)) + + +@pytest.mark.parametrize( + ("suffix", "content"), + [ + ("json", '{"schema_version":1,"schema_version":1,"name":"repo-checks","tasks":[]}'), + ("yaml", "schema_version: 1\nschema_version: 1\nname: repo-checks\ntasks: []\n"), + ], +) +def test_manifest_rejects_duplicate_fields_before_execution(tmp_path, suffix, content): + path = tmp_path / f"checks.{suffix}" + path.write_text(content) + with pytest.raises(ValueError, match="duplicate"): + load_batch_manifest(path) + + +def test_cli_failure_status_attempts_and_logs(tmp_path, capsys): + db = tmp_path / "queue.sqlite3" + with init_store(db) as store: + task = _task( + store, "failure", "import sys; print('helpful failure',file=sys.stderr); sys.exit(4)" + ) + assert main(["dispatch", "--db", str(db), "--backend", "subprocess"]) == 1 + capsys.readouterr() + assert main(["attempts", "--db", str(db), task.id, "--format", "json"]) == 0 + attempt = json.loads(capsys.readouterr().out)[0] + assert attempt["exit_code"] == 4 + assert main(["logs", "--db", str(db), task.id, "--stream", "stderr", "--tail", "10"]) == 0 + assert capsys.readouterr().out == "l failure\n" + + +def test_schema_one_migrates_without_changing_legacy_state(tmp_path): + db = tmp_path / "old.sqlite3" + conn = sqlite3.connect(db) + conn.executescript("""CREATE TABLE schema_meta( + key TEXT PRIMARY KEY,value TEXT,updated_at INTEGER); + INSERT INTO schema_meta VALUES('schema_version','1',0); + CREATE TABLE tasks(id TEXT PRIMARY KEY,title TEXT NOT NULL, + body TEXT NOT NULL,status TEXT NOT NULL, + priority INTEGER NOT NULL DEFAULT 0,lane TEXT NOT NULL DEFAULT 'default', + tags TEXT NOT NULL DEFAULT '[]', + role TEXT NOT NULL DEFAULT 'agent',workdir TEXT,command TEXT,created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL,heartbeat_at INTEGER); + INSERT INTO tasks(id,title,body,status,created_at,updated_at) VALUES + ('task_abcdef123456','legacy','body','running',0,0);""") + conn.close() + with init_store(db) as store: + assert store.schema_version() == 2 + assert store.get_task("task_abcdef123456").status == TaskStatus.RUNNING + assert store.attempts() == [] + with pytest.raises(ValueError, match="legacy running"): + recover_task(store, "task_abcdef123456", stale_after=0) + recover_task(store, "task_abcdef123456", stale_after=0, acknowledge_unknown=True) + + +def test_due_work_is_not_held_behind_a_delayed_retry(tmp_path): + with init_store(tmp_path / "queue.sqlite3") as store: + delayed = _task( + store, "delayed", "raise SystemExit(1)", priority=2, retry_backoff_seconds=2 + ) + dispatch_ready(store, DispatchPolicy(1, "subprocess")) + store.retry_task(delayed.id) + eligible = _task(store, "eligible", "pass") + result = dispatch_ready(store, DispatchPolicy(1, "subprocess")) + assert result.completed == (eligible.id,) + assert store.get_task(delayed.id).status == TaskStatus.TODO + + +def test_active_coordinators_cannot_disagree_about_capacity(tmp_path): + with init_store(tmp_path / "queue.sqlite3") as store: + first = _task(store, "first", "pass") + second = _task(store, "second", "pass") + attempt = store.claim_attempt(first.id, 1, first.command_argv) + with pytest.raises(ValueError, match="same --max-workers"): + store.claim_attempt(second.id, 2, second.command_argv) + store.finish_attempt(attempt["claim_token"], "done", "complete") + assert store.claim_attempt(second.id, 2, second.command_argv) + + +def test_capacity_is_reserved_when_legacy_running_work_already_exists(tmp_path): + with init_store(tmp_path / "queue.sqlite3") as store: + legacy = _task(store, "manual", "pass") + store.mark_task(legacy.id, TaskStatus.RUNNING) + first = _task(store, "first", "pass") + second = _task(store, "second", "pass") + assert store.claim_attempt(first.id, 2, first.command_argv) + with pytest.raises(ValueError, match="same --max-workers"): + store.claim_attempt(second.id, 3, second.command_argv) + + +def test_cleanup_failure_does_not_leave_other_owned_workers_running(tmp_path, monkeypatch, capsys): + import agent_backlog_runner.dispatch as dispatch + + workers = [] + original_spawn, original_stop, original_read = ( + dispatch._spawn, + dispatch._stop_owned_process, + dispatch._read_pipe, + ) + interrupted = False + failed_cleanup = False + + def spawn(*args, **kwargs): + worker = original_spawn(*args, **kwargs) + workers.append(worker) + return worker + + def read(*args, **kwargs): + nonlocal interrupted + if not interrupted: + interrupted = True + raise KeyboardInterrupt("test interruption") + return original_read(*args, **kwargs) + + def stop(process): + nonlocal failed_cleanup + if not failed_cleanup: + failed_cleanup = True + raise PermissionError("test cleanup denial") + return original_stop(process) + + monkeypatch.setattr(dispatch, "_spawn", spawn) + monkeypatch.setattr(dispatch, "_read_pipe", read) + monkeypatch.setattr(dispatch, "_stop_owned_process", stop) + with init_store(tmp_path / "queue.sqlite3") as store: + first = _task(store, "first", "import time; print('ready',flush=True); time.sleep(10)") + second = _task(store, "second", "import time; print('ready',flush=True); time.sleep(10)") + try: + with pytest.raises(KeyboardInterrupt, match="test interruption"): + dispatch_ready(store, DispatchPolicy(2, "subprocess")) + assert store.attempts(first.id)[0]["finished_at"] is None + assert store.attempts(second.id)[0]["outcome"] == "interrupted" + assert workers[1].process.returncode is not None + assert "test cleanup denial" in capsys.readouterr().err + finally: + for worker in workers: + original_stop(worker.process) + for pipe in (worker.process.stdout, worker.process.stderr): + if pipe and not pipe.closed: + pipe.close() + worker.stdout.file.close() + worker.stderr.file.close() + + +def test_spawn_cleanup_failure_preserves_unfinished_attempt(tmp_path, monkeypatch, capsys): + import agent_backlog_runner.dispatch as dispatch + + processes = [] + original_popen, original_stop = dispatch.subprocess.Popen, dispatch._stop_owned_process + + def popen(*args, **kwargs): + process = original_popen(*args, **kwargs) + processes.append(process) + return process + + def deny_stop(_process): + raise PermissionError("test launch cleanup denial") + + with init_store(tmp_path / "queue.sqlite3") as store: + task = _task(store, "launch", "import time; time.sleep(10)") + original_bind = store.bind_attempt + + def bind(token, **values): + original_bind(token, **values) + if "worker_pid" in values: + raise OSError("test launch binding failure") + + monkeypatch.setattr(store, "bind_attempt", bind) + monkeypatch.setattr(dispatch.subprocess, "Popen", popen) + monkeypatch.setattr(dispatch, "_stop_owned_process", deny_stop) + try: + with pytest.raises(OSError, match="test launch binding failure"): + dispatch_ready(store, DispatchPolicy(1, "subprocess")) + assert store.attempts(task.id)[0]["finished_at"] is None + assert store.get_task(task.id).status == TaskStatus.RUNNING + assert "test launch cleanup denial" in capsys.readouterr().err + finally: + for process in processes: + original_stop(process) + + +def test_interruption_during_spawn_stops_and_records_owned_process(tmp_path, monkeypatch): + with init_store(tmp_path / "queue.sqlite3") as store: + task = _task(store, "launch", "import time; time.sleep(10)") + original_bind = store.bind_attempt + + def bind(token, **values): + original_bind(token, **values) + if "worker_pid" in values: + raise KeyboardInterrupt("test launch interruption") + + monkeypatch.setattr(store, "bind_attempt", bind) + with pytest.raises(KeyboardInterrupt, match="test launch interruption"): + dispatch_ready(store, DispatchPolicy(1, "subprocess")) + attempt = store.attempts(task.id)[0] + assert attempt["outcome"] == "interrupted" + assert attempt["finished_at"] is not None + assert store.get_task(task.id).status == TaskStatus.FAILED + + +@pytest.mark.skipif( + not all(hasattr(os, name) for name in ("waitid", "WNOWAIT", "P_PID")), + reason="requires retained-leader completion observation", +) +def test_normal_exit_cleans_up_inherited_background_group(tmp_path): + marker = tmp_path / "background.txt" + child = f"import time; from pathlib import Path; time.sleep(.7); Path({str(marker)!r}).touch()" + code = f"import subprocess,sys; subprocess.Popen([sys.executable,'-c',{child!r}])" + with init_store(tmp_path / "queue.sqlite3") as store: + _task(store, "background", code) + result = dispatch_ready(store, DispatchPolicy(1, "subprocess")) + assert len(result.completed) == 1 + time.sleep(0.8) + assert not marker.exists() + + +def test_maintenance_walkthrough_checks_repairs_and_resumes(tmp_path): + from agent_backlog_runner.maintenance import run_demo + + result = run_demo(tmp_path / "demo") + assert { + key: result[key] + for key in ( + "checks", + "initial_completed", + "resume_failed", + "repaired_checks", + "attempts", + "final_done", + ) + } == { + "checks": 3, + "initial_completed": 1, + "resume_failed": 1, + "repaired_checks": 1, + "attempts": 4, + "final_done": 3, + } + report = Path(result["report"]).read_text() + assert "2/3" in report + assert "stderr" in report + before = json.loads((tmp_path / "demo" / "before-repair.json").read_text()) + assert before["summary"]["counts"]["failed"] == 1 diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index a8951dc..9048cc9 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -186,17 +186,25 @@ def test_non_utf8_worker_output_does_not_strand_batch(tmp_path): assert store.get_task(following.id).status == TaskStatus.DONE -def test_completion_preserves_operator_cancellation(tmp_path, monkeypatch): - import subprocess +def test_completion_preserves_operator_cancellation(tmp_path): + from threading import Timer - with init_store(tmp_path / "tasks.sqlite3") as store: - task = store.create_task(title="cancelled", body="", command="worker") + db = tmp_path / "tasks.sqlite3" + with init_store(db) as store: + task = store.create_task( + title="cancelled", + body="", + command_argv=(sys.executable, "-c", "import time; time.sleep(10)"), + ) - def complete_after_cancellation(*args, **kwargs): - store.mark_task(task.id, TaskStatus.CANCELLED, "operator cancelled") - return subprocess.CompletedProcess(args[0], returncode=0, stdout="ok", stderr="") + def cancel(): + with init_store(db) as other: + other.request_cancel(task.id) - monkeypatch.setattr(subprocess, "run", complete_after_cancellation) + timer = Timer(0.2, cancel) + timer.start() dispatch_ready(store, DispatchPolicy(backend="subprocess")) + timer.join(timeout=5) assert store.get_task(task.id).status == TaskStatus.CANCELLED + assert store.attempts(task.id)[0]["outcome"] == "cancelled" assert store.events(task.id)[0].event_type == "cancelled" diff --git a/tests/test_public_safety.py b/tests/test_public_safety.py index 833d480..5c14574 100644 --- a/tests/test_public_safety.py +++ b/tests/test_public_safety.py @@ -17,7 +17,19 @@ def test_public_files_avoid_private_or_sensitive_terms(): checked = [] for path in root.rglob("*"): if path.is_file() and path.suffix not in {".pyc"} and ".git" not in path.parts: - if any(part in {".venv", ".pytest_cache", ".ruff_cache"} for part in path.parts): + if any( + part + in { + ".venv", + ".pytest_cache", + ".ruff_cache", + ".agent-backlog", + ".wheel-smoke", + "dist", + "build", + } + for part in path.parts + ): continue text = path.read_text(encoding="utf-8", errors="ignore") checked.append(path) diff --git a/uv.lock b/uv.lock index 9cba82c..2efc1d3 100644 --- a/uv.lock +++ b/uv.lock @@ -4,7 +4,7 @@ requires-python = ">=3.11" [[package]] name = "agent-backlog-runner" -version = "0.2.0" +version = "0.3.0" source = { editable = "." } dependencies = [ { name = "pyyaml" },