From 769ceb6f0237850f79cc7092e85008a77a45a0c8 Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Tue, 15 Sep 2026 09:52:10 -0700 Subject: [PATCH 1/4] Add durable approval/decision primitive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Jobs can now ask a human a question and wait for the answer without holding a thread or interrupting the job. A decision is its own small state machine keyed by `decision_id` (dec_), kept in a `decisions` table so it never overloads `jobs.status`; the job keeps running (or stays queued) while it waits. - `job_request_decision(job_id, prompt, options=["approve","deny"], timeout_seconds=None)` inserts a pending decision and emits `decision_requested`. That reuses the ordinary event->wake-target->delivery path, so the owning thread is notified through the mechanism it already registered (targets must list `decision_requested` in their events). - `job_resolve(job_id, token, choice)` records one of the offered options. Resolving the same choice twice is idempotent; a different choice, an unknown token, or a choice outside the options is an error. The pending-> resolved transition is a guarded UPDATE, so concurrent resolutions cannot both win. - `job_withdraw_decision(job_id, token)` cancels a pending request. - `job_decisions(job_id=None, status=None, limit=50)` lists them. - `job_wait(job_id, ["decision_resolved"])` waits for the answer like any other event; `decision_requested` also ranks as an attention event in `job_view`. - Expiry: the existing maintenance loop sweeps overdue pending decisions, marks them `expired` and emits `decision_expired` (edge-triggered, one event per transition); an expired decision can no longer be resolved. Schema v16 adds the `decisions` table plus (job_id, created_at) and (status, expires_at) indexes. The migration is additive and existing databases still get the usual pre-migration backup. Wiring: JobManager methods, `@mcp.tool()` wrappers, and the daemon POST/GET routes (`/jobs//decision`, `...//resolve`, `...//withdraw`, `GET /decisions`). No CLI subcommand (none of the neighbouring job tools have one). Scoped to local jobs. Remote jobs keep their own database/events and the remote protocol has no decision methods; a cross-machine decision needs protocol + feed work and is deliberately out of scope. Tests: tests/test_decisions.py (13) — lifecycle, idempotency/conflict, withdraw, expiry sweep + resolve-after-expiry, input validation, terminal/ unknown job rejection, token/job mismatch, listing/filters, wake delivery enqueued, job_wait integration, v15->v16 migration, and the HTTP routes end to end. Full suite 802 passed / 6 skipped; go vet/test/build clean. --- CHANGELOG.md | 26 ++++ README.md | 19 +++ docs/agent-tools.md | 51 +++++++ src/vanth/daemon.py | 13 ++ src/vanth/migrations.py | 29 +++- src/vanth/server.py | 239 +++++++++++++++++++++++++++++++- tests/test_decisions.py | 296 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 670 insertions(+), 3 deletions(-) create mode 100644 tests/test_decisions.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ef09f68..ae2eecb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,32 @@ All notable changes to Vanth are documented here. +## Unreleased + +### Durable approval / decision requests + +Jobs can now ask a human a question and wait for the answer durably, without +holding a thread or killing the job. + +- `job_request_decision(job_id, prompt, options=["approve","deny"], + timeout_seconds=None)` records a decision (status `pending`) and emits + `decision_requested`, which reuses the wake-target delivery path so the + job's owning thread is notified. The job keeps running — its status is + untouched. +- `job_resolve(job_id, token, choice)` records one of the offered options + (idempotent for the same choice, an error for a different one); + `job_withdraw_decision(job_id, token)` cancels a pending request; + `job_decisions(...)` lists them. `job_wait(job_id, ["decision_resolved"])` + waits for the answer like any other event. +- An optional `timeout_seconds` expires the request; the maintenance loop + marks it `expired` and emits `decision_expired`, after which it can no + longer be resolved. +- Lifecycle events (`decision_requested` / `decision_resolved` / + `decision_withdrawn` / `decision_expired`) are the audit trail; + `decision_requested` also ranks as an attention event in `job_view`. +- Schema v16 adds the `decisions` table (additive; existing databases + migrate in place with the usual pre-migration backup). + ## 1.9.1 - 2026-09-11 ### POSIX Python CI + portability fixes diff --git a/README.md b/README.md index ac0f3cc..b2629af 100644 --- a/README.md +++ b/README.md @@ -435,6 +435,9 @@ events. | `job_delivery_attempts` | Attempt/lease history for one delivery | | `job_stop` | Stop a running job (terminate process tree) | | `job_pause` / `job_resume` | Hold / release a queued (pool or trigger) job | +| `job_request_decision` | Ask a human to decide something about a job; notifies its wake targets and returns a durable `decision_id` | +| `job_resolve` / `job_withdraw_decision` | Answer a pending decision with one of its options, or withdraw it | +| `job_decisions` | List decisions, filterable by `job_id` / `status` | | `pool_configure` / `pool_list` | Per-pool `max_parallel` + pause state, and live queue depths | | `schedule_create` | Create a cron or interval schedule that launches a job per fire | | `schedule_list` / `schedule_update` / `schedule_delete` | Manage schedules in place | @@ -515,6 +518,22 @@ returns immediately; the input is queued to the runner. Rejects jobs that are not interactive, not running, or unknown. `job_rerun` preserves the `interactive` flag. +### job_request_decision — ask a human and wait durably + +```text +job_request_decision(job_id="job_...", prompt="Ship the release?", + options=["approve", "deny"], timeout_seconds=3600) +job_wait(job_id="job_...", filters=["decision_resolved"]) # await the answer +job_resolve(job_id="job_...", token="dec_...", choice="approve") +``` + +Records a durable "needs a decision" request against a non-terminal job and +notifies the job's wake targets, so the owning thread learns a human is needed. +The job keeps running (its `status` is untouched) — the answer arrives as a +`decision_resolved` event. `timeout_seconds` expires the request (emitting +`decision_expired`); `job_withdraw_decision` cancels it; `job_decisions` lists +pending/answered requests. + ### job_status — see what a job is running ```text diff --git a/docs/agent-tools.md b/docs/agent-tools.md index df50048..f7a20cb 100644 --- a/docs/agent-tools.md +++ b/docs/agent-tools.md @@ -23,6 +23,7 @@ The current tool set is `job_start`, `job_rerun`, `job_status`, `job_metric_compare`, `job_duration_stats`, `job_run_summary`, `job_artifact_add`, `job_artifacts`, `job_dashboard`, `job_metric_ingest`, `job_artifact_read`, `job_add_wake_target`, `job_wake_now`, `job_cleanup_preview`, +`job_request_decision`, `job_resolve`, `job_withdraw_decision`, `job_decisions`, `pool_configure`, `pool_list`, `schedule_create`, `schedule_list`, `schedule_update`, `schedule_delete`, `schedule_next`. The wake tools (`job_add_wake_target` / `job_wake_now` / `daemon_wake`) and their `daemon_wake` / `job_wake_now` / `job_add_wake_target` @@ -501,6 +502,56 @@ Response: `{ "result": "ok", "job_id": "...", "paused": true|false }`. --- +## `job_request_decision` / `job_resolve` / `job_withdraw_decision` / `job_decisions` + +Ask a human to decide something about a **non-terminal** job and wait durably +for the answer. The request is its own state machine keyed by `decision_id` +(`dec_`); the job's `status` is not changed, so a running job keeps +running while it waits. + +`job_request_decision` parameters: + +| Param | Type | Default | Notes | +|---|---|---|---| +| `job_id` | `string` | required | A queued/running (non-terminal) job | +| `prompt` | `string` | required | The question shown to the human | +| `options` | `string[]` | `["approve", "deny"]` | Allowed choices (deduplicated) | +| `timeout_seconds` | `int` | none | Expire the request after N seconds | + +Requesting emits a `decision_requested` event, which reuses the wake-target +delivery path — the job's wake targets are notified, so the owning thread +learns a human is needed. Only targets whose `events` list includes +`decision_requested` are woken. Response: the decision object +(`status: "pending"`). + +```json +{ "result": "ok", "decision_id": "dec_1f2e...", "job_id": "job_ab12...", + "prompt": "Ship the release?", "options": ["approve", "deny"], + "choice": null, "status": "pending", "resolved_by": null, + "created_at": "2026-09-15T10:00:00Z", "expires_at": null, "resolved_at": null } +``` + +Wait for the answer with +`job_wait(job_id, ["decision_resolved"])`, or poll `job_decisions`. + +- `job_resolve(job_id, token, choice)` records one of the request's `options` + and emits `decision_resolved`. Resolving the same choice twice is + idempotent; a different choice is an error; a resolved/withdrawn/expired + decision cannot be resolved. +- `job_withdraw_decision(job_id, token)` cancels a pending request (emits + `decision_withdrawn`). +- When `timeout_seconds` elapses the daemon marks the decision `expired` + (emits `decision_expired`) and it can no longer be resolved. +- `job_decisions(job_id=None, status=None, limit=50)` lists decisions newest + first; `status` is one of `pending`, `resolved`, `withdrawn`, `expired`. + +```json +{ "decisions": [ { "decision_id": "dec_1f2e...", "status": "resolved", + "choice": "approve", "resolved_by": "user" } ], "count": 1 } +``` + +--- + ## `pool_configure` / `pool_list` `pool_configure(pool, max_parallel=0, paused=None)` upserts a concurrency pool. diff --git a/src/vanth/daemon.py b/src/vanth/daemon.py index 529ff58..54c258b 100644 --- a/src/vanth/daemon.py +++ b/src/vanth/daemon.py @@ -587,6 +587,12 @@ def do_GET(self) -> None: ok(self, get_manager().tail(parsed.path.split("/")[2], query.get("stream", ["stdout"])[0], int(query.get("max_bytes", ["8192"])[0]), int(query["offset"][0]) if "offset" in query else None, query.get("follow", ["false"])[0] == "true", float(query.get("timeout_seconds", ["5"])[0]), query.get("grep", [None])[0])) elif parsed.path == "/deliveries": ok(self, get_manager().deliveries(query.get("job_id", [None])[0], query.get("status", [None])[0], int(query.get("limit", ["20"])[0]))) + elif parsed.path == "/decisions": + ok(self, get_manager().list_decisions( + query.get("job_id", [None])[0], + query.get("status", [None])[0], + int(query.get("limit", ["50"])[0]), + )) elif parsed.path.startswith("/deliveries/") and parsed.path.endswith("/attempts"): ok(self, get_manager().delivery_attempts(parsed.path.split("/")[2], int(query.get("limit", ["20"])[0]))) elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/metrics"): @@ -748,6 +754,13 @@ def do_POST(self) -> None: ok(self, get_manager().job_pause(parsed.path.split("/")[2])) elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/resume"): ok(self, get_manager().job_resume(parsed.path.split("/")[2])) + elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/decision"): + ok(self, get_manager().request_decision(parsed.path.split("/")[2], **payload)) + elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/resolve"): + ok(self, get_manager().resolve_decision( + parsed.path.split("/")[2], parsed.path.split("/")[4], payload.get("choice", ""))) + elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/withdraw"): + ok(self, get_manager().withdraw_decision(parsed.path.split("/")[2], parsed.path.split("/")[4])) elif parsed.path == "/schedules": ok(self, get_manager().create_schedule(**payload)) elif parsed.path.startswith("/schedules/") and parsed.path.endswith("/update"): diff --git a/src/vanth/migrations.py b/src/vanth/migrations.py index aa381f5..4809e63 100644 --- a/src/vanth/migrations.py +++ b/src/vanth/migrations.py @@ -7,7 +7,7 @@ from datetime import datetime, timezone from pathlib import Path -LATEST_SCHEMA_VERSION = 15 +LATEST_SCHEMA_VERSION = 16 DEFAULT_BUSY_TIMEOUT_MS = 30000 @@ -125,7 +125,14 @@ def _create_latest_schema(db: sqlite3.Connection) -> None: CREATE INDEX IF NOT EXISTS idx_schedules_due ON schedules(enabled, next_fire_at); CREATE INDEX IF NOT EXISTS idx_jobs_schedule ON jobs(schedule_id); CREATE INDEX IF NOT EXISTS idx_jobs_pool_status ON jobs(pool, status); - PRAGMA user_version=15; + CREATE TABLE IF NOT EXISTS decisions ( + decision_id TEXT PRIMARY KEY, job_id TEXT NOT NULL, prompt TEXT NOT NULL, + options_json TEXT NOT NULL, choice TEXT, status TEXT NOT NULL, + resolved_by TEXT, created_at TEXT NOT NULL, expires_at TEXT, resolved_at TEXT + ); + CREATE INDEX IF NOT EXISTS idx_decisions_job ON decisions(job_id, created_at); + CREATE INDEX IF NOT EXISTS idx_decisions_pending ON decisions(status, expires_at); + PRAGMA user_version=16; """ ) @@ -313,6 +320,24 @@ def migrate(db: sqlite3.Connection, home: str | Path) -> Path | None: db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_pool_status ON jobs(pool, status)") db.execute("PRAGMA user_version=15") version = 15 + if version < 16: + # Durable approval/decision requests: a job's "needs a human" + # state machine, kept in its own table so it never overloads + # jobs.status. Lifecycle events (decision_requested/resolved/ + # withdrawn/expired) live in `events` for the audit trail. + db.execute( + """ + CREATE TABLE IF NOT EXISTS decisions ( + decision_id TEXT PRIMARY KEY, job_id TEXT NOT NULL, prompt TEXT NOT NULL, + options_json TEXT NOT NULL, choice TEXT, status TEXT NOT NULL, + resolved_by TEXT, created_at TEXT NOT NULL, expires_at TEXT, resolved_at TEXT + ) + """ + ) + db.execute("CREATE INDEX IF NOT EXISTS idx_decisions_job ON decisions(job_id, created_at)") + db.execute("CREATE INDEX IF NOT EXISTS idx_decisions_pending ON decisions(status, expires_at)") + db.execute("PRAGMA user_version=16") + version = 16 db.commit() except Exception: db.rollback() diff --git a/src/vanth/server.py b/src/vanth/server.py index 14f7d4d..f4aa9c8 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -328,8 +328,14 @@ def normalize_event_payload(payload: dict[str, Any]) -> dict[str, Any]: } -ATTENTION_EVENTS = {"needs_input", "permission_required", "blocked"} +ATTENTION_EVENTS = {"needs_input", "permission_required", "blocked", "decision_requested"} WAKE_TARGET_TYPES = {"local_command", "codex_cli_thread", "codex_thread", "codex_desktop", "opencode_thread", "webhook"} +# Durable approval/decision requests (roadmap Tier-1). A decision is its own +# small state machine keyed by ``decision_id``; the job row is untouched, so a +# job can keep running (or stay queued) while a human decides. +DECISION_PENDING = "pending" +DECISION_STATUSES = {DECISION_PENDING, "resolved", "withdrawn", "expired"} +DEFAULT_DECISION_OPTIONS = ["approve", "deny"] def resolve_wake_target_identity( @@ -787,6 +793,7 @@ def _dispatch_loop(self) -> None: self._dispatch_queued_jobs() self._watch_policies() self._maybe_auto_cleanup() + self._expire_decisions() self.relay_expire_stale(stale_after_seconds=int(os.environ.get("VANTH_RELAY_SUBSCRIPTION_TTL", "300"))) except Exception: self.logger.exception("maintenance iteration failed") @@ -5799,6 +5806,191 @@ def send_sync(self, job_id: str, input: str, eof: bool = False) -> dict[str, Any f.write(struct.pack(" dict[str, Any]: + return { + "decision_id": row["decision_id"], + "job_id": row["job_id"], + "prompt": row["prompt"], + "options": json.loads(row["options_json"] or "[]"), + "choice": row["choice"], + "status": row["status"], + "resolved_by": row["resolved_by"], + "created_at": row["created_at"], + "expires_at": row["expires_at"], + "resolved_at": row["resolved_at"], + } + + def _decision_isdue(self, decision: dict[str, Any], now: datetime | None = None) -> bool: + expires_at = decision.get("expires_at") + if not expires_at or decision["status"] != DECISION_PENDING: + return False + parsed = _parse_iso(str(expires_at)) + return parsed is not None and parsed <= (now or datetime.now(timezone.utc)) + + def request_decision( + self, + job_id: str, + prompt: str, + options: list[str] | None = None, + timeout_seconds: int | None = None, + ) -> dict[str, Any]: + """Persist a durable "needs a decision" request and wake the owner. + + Emitting ``decision_requested`` reuses the wake-target delivery path, so + whichever thread owns the job (``wake_targets``) is notified. The job + itself keeps running; nothing about its status changes. + """ + self._ensure_open() + if not isinstance(prompt, str) or not prompt.strip(): + raise ValueError("prompt must be a non-empty string") + choices = DEFAULT_DECISION_OPTIONS if options is None else options + if not isinstance(choices, list) or not choices: + raise ValueError("options must be a non-empty list of strings") + normalized: list[str] = [] + for option in choices: + if not isinstance(option, str) or not option.strip(): + raise ValueError("options must be non-empty strings") + if option not in normalized: + normalized.append(option) + if timeout_seconds is not None: + if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, int) or timeout_seconds < 1: + raise ValueError("timeout_seconds must be an integer >= 1") + row = self._row("SELECT status FROM jobs WHERE job_id=?", (job_id,)) + if not row: + raise ValueError(f"Unknown job_id: {job_id}") + if row["status"] in TERMINAL_STATUSES: + raise ValueError(f"job is already terminal: {row['status']}") + now = datetime.now(timezone.utc) + decision_id = "dec_" + uuid.uuid4().hex[:16] + created_at = now_iso() + expires_at = (now + timedelta(seconds=timeout_seconds)).isoformat().replace("+00:00", "Z") if timeout_seconds else None + with self.db_lock: + self.db.execute( + """ + INSERT INTO decisions(decision_id, job_id, prompt, options_json, choice, status, resolved_by, created_at, expires_at, resolved_at) + VALUES (?, ?, ?, ?, NULL, ?, NULL, ?, ?, NULL) + """, + (decision_id, job_id, prompt, json.dumps(normalized, separators=(",", ":")), DECISION_PENDING, created_at, expires_at), + ) + self.db.commit() + self._emit( + job_id, + "decision_requested", + message=prompt, + data={"decision_id": decision_id, "prompt": prompt, "options": normalized, "expires_at": expires_at}, + ) + decision = self._row("SELECT * FROM decisions WHERE decision_id=?", (decision_id,)) + return self._decision_dict(decision) + + def _decision_for(self, job_id: str, token: str) -> sqlite3.Row: + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string") + row = self._row("SELECT * FROM decisions WHERE decision_id=?", (token,)) + if not row or row["job_id"] != job_id: + raise ValueError(f"Unknown decision for job {job_id}: {token}") + return row + + def _expire_decision(self, row: sqlite3.Row) -> bool: + """CAS pending->expired; True when this call performed the transition.""" + with self.db_lock: + cursor = self.db.execute( + "UPDATE decisions SET status='expired', resolved_at=? WHERE decision_id=? AND status=?", + (now_iso(), row["decision_id"], DECISION_PENDING), + ) + self.db.commit() + return cursor.rowcount == 1 + + def resolve_decision(self, job_id: str, token: str, choice: str, actor: str = "user") -> dict[str, Any]: + """Record a human choice for a pending decision (idempotent per choice).""" + self._ensure_open() + row = self._decision_for(job_id, token) + if row["status"] == "resolved": + if row["choice"] == choice: + return self._decision_dict(row) + raise ValueError(f"decision already resolved as {row['choice']!r}") + if row["status"] == "expired" or self._decision_isdue(self._decision_dict(row)): + if row["status"] == DECISION_PENDING and self._expire_decision(row): + self._emit(job_id, "decision_expired", message="decision timed out", data={"decision_id": row["decision_id"]}) + raise ValueError("decision expired") + if row["status"] != DECISION_PENDING: + raise ValueError(f"decision is {row['status']}") + allowed = json.loads(row["options_json"] or "[]") + if choice not in allowed: + raise ValueError(f"choice must be one of {allowed}") + with self.db_lock: + cursor = self.db.execute( + "UPDATE decisions SET status='resolved', choice=?, resolved_by=?, resolved_at=? WHERE decision_id=? AND status=?", + (choice, actor, now_iso(), row["decision_id"], DECISION_PENDING), + ) + self.db.commit() + if cursor.rowcount != 1: + current = self._row("SELECT * FROM decisions WHERE decision_id=?", (row["decision_id"],)) + if current is not None and current["status"] == "resolved" and current["choice"] == choice: + return self._decision_dict(current) + raise ValueError("decision was resolved concurrently") + self._emit( + job_id, + "decision_resolved", + message=choice, + data={"decision_id": row["decision_id"], "choice": choice, "actor": actor, "prompt": row["prompt"]}, + ) + return self._decision_dict(self._row("SELECT * FROM decisions WHERE decision_id=?", (row["decision_id"],))) + + def withdraw_decision(self, job_id: str, token: str, actor: str = "user") -> dict[str, Any]: + """Cancel a pending decision (the requester no longer needs an answer).""" + self._ensure_open() + row = self._decision_for(job_id, token) + if row["status"] != DECISION_PENDING: + raise ValueError(f"decision is {row['status']}") + with self.db_lock: + cursor = self.db.execute( + "UPDATE decisions SET status='withdrawn', resolved_by=?, resolved_at=? WHERE decision_id=? AND status=?", + (actor, now_iso(), row["decision_id"], DECISION_PENDING), + ) + self.db.commit() + if cursor.rowcount != 1: + raise ValueError("decision was resolved concurrently") + self._emit( + job_id, + "decision_withdrawn", + message="decision withdrawn", + data={"decision_id": row["decision_id"], "actor": actor, "prompt": row["prompt"]}, + ) + return self._decision_dict(self._row("SELECT * FROM decisions WHERE decision_id=?", (row["decision_id"],))) + + def list_decisions(self, job_id: str | None = None, status: str | None = None, limit: int = 50) -> dict[str, Any]: + self._ensure_open() + if status is not None and status not in DECISION_STATUSES: + raise ValueError(f"status must be one of {sorted(DECISION_STATUSES)}") + sql = "SELECT * FROM decisions WHERE 1=1" + args: list[Any] = [] + if job_id: + sql += " AND job_id=?" + args.append(job_id) + if status: + sql += " AND status=?" + args.append(status) + sql += " ORDER BY created_at DESC, decision_id DESC LIMIT ?" + args.append(max(1, min(int(limit), 500))) + with self.db_lock: + rows = self.db.execute(sql, tuple(args)).fetchall() + return {"decisions": [self._decision_dict(row) for row in rows], "count": len(rows)} + + def _expire_decisions(self) -> None: + """Expire overdue pending decisions (maintenance loop; edge-triggered).""" + now = datetime.now(timezone.utc) + with self.db_lock: + rows = self.db.execute( + "SELECT * FROM decisions WHERE status=? AND expires_at IS NOT NULL", + (DECISION_PENDING,), + ).fetchall() + for row in rows: + if not self._decision_isdue(self._decision_dict(row), now): + continue + if self._expire_decision(row): + self._emit(row["job_id"], "decision_expired", message="decision timed out", + data={"decision_id": row["decision_id"], "prompt": row["prompt"]}) + client: VanthClient | None = None mcp = FastMCP("vanth") @@ -5956,6 +6148,51 @@ def job_send(job_id: str, input: str, eof: bool = False) -> dict[str, Any]: return get_client().post(f"/jobs/{job_id}/send", {"input": input, "eof": eof}) +@mcp.tool() +def job_request_decision( + job_id: str, + prompt: str, + options: list[str] | None = None, + timeout_seconds: int | None = None, +) -> dict[str, Any]: + """Ask a human to decide something about a job and wait durably for the answer. + + Creates a durable ``decision`` (status ``pending``) and notifies the job's + wake targets, so the owning thread learns a human is needed. The job keeps + running — its status is untouched. ``options`` defaults to + ``["approve", "deny"]``; ``timeout_seconds`` expires the request (the + daemon emits ``decision_expired`` and the decision can no longer be + resolved). Wait for the answer with + ``job_wait(job_id, ["decision_resolved"])`` or poll ``job_decisions``. + + Resolve with ``job_resolve(job_id, decision_id, choice)``; cancel an + unneeded request with ``job_withdraw_decision(job_id, decision_id)``. + """ + payload = {"prompt": prompt, "options": options, "timeout_seconds": timeout_seconds} + return get_client().post(f"/jobs/{job_id}/decision", payload) + + +@mcp.tool() +def job_resolve(job_id: str, token: str, choice: str) -> dict[str, Any]: + """Answer a pending decision with one of its ``options`` (see job_request_decision).""" + return get_client().post(f"/jobs/{job_id}/decision/{token}/resolve", {"choice": choice}) + + +@mcp.tool() +def job_withdraw_decision(job_id: str, token: str) -> dict[str, Any]: + """Withdraw a pending decision that no longer needs an answer.""" + return get_client().post(f"/jobs/{job_id}/decision/{token}/withdraw") + + +@mcp.tool() +def job_decisions(job_id: str | None = None, status: str | None = None, limit: int = 50) -> dict[str, Any]: + """List decisions (newest first), optionally filtered by job and/or status. + + ``status`` is one of ``pending``, ``resolved``, ``withdrawn``, ``expired``. + """ + return get_client().get("/decisions", {"job_id": job_id, "status": status, "limit": limit}) + + @mcp.tool() def job_list(status: list[str] | None = None, limit: int = 50, thread_id: str | None = None, name: str | None = None, tags: list[str] | None = None) -> dict[str, Any]: diff --git a/tests/test_decisions.py b/tests/test_decisions.py new file mode 100644 index 0000000..68bb6b5 --- /dev/null +++ b/tests/test_decisions.py @@ -0,0 +1,296 @@ +"""Tests for the durable approval/decision primitive. + +A decision is a small state machine (pending -> resolved/withdrawn/expired) +attached to a non-terminal job, kept out of ``jobs.status``. Requesting one +emits ``decision_requested`` (which reuses the wake-target delivery path so the +owning thread is notified); resolving emits ``decision_resolved``. Covers the +lifecycle, expiry (including the maintenance sweep), input validation, +idempotency, wake delivery, ``job_wait`` integration, and the v15->v16 +migration. +""" + +import asyncio +import os +import socket +import subprocess +import sys +import time + +import pytest + +from vanth.client import VanthClient +from vanth.server import JobManager + + +import shellcmd + + +def cmd(code: str) -> str: + return shellcmd.join([sys.executable, "-c", code]) + + +def running_job(manager: JobManager) -> str: + started = asyncio.run(manager.start(cmd("import time; time.sleep(20)"))) + job_id = started["job_id"] + asyncio.run(manager.wait(job_id, ["started"], timeout_seconds=10)) + return job_id + + +def event_types(manager: JobManager, job_id: str) -> list[str]: + rows = manager.db.execute("SELECT type FROM events WHERE job_id=? ORDER BY seq", (job_id,)).fetchall() + return [row["type"] for row in rows] + + +def decision_row(manager: JobManager, decision_id: str): + return manager.db.execute("SELECT * FROM decisions WHERE decision_id=?", (decision_id,)).fetchone() + + +def test_request_decision_is_pending_and_wakes_owner(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + manager.add_wake_target(job_id, { + "type": "local_command", + "command": cmd("pass"), + "events": ["decision_requested"], + "auto_dispatch": False, + }) + decision = manager.request_decision(job_id, "Ship the release?") + assert decision["status"] == "pending" + assert decision["options"] == ["approve", "deny"] + assert decision["choice"] is None + assert decision["expires_at"] is None + assert "decision_requested" in event_types(manager, job_id) + # The owner is notified through the ordinary delivery queue. + deliveries = manager.db.execute( + "SELECT target_type, status FROM deliveries WHERE job_id=?", (job_id,) + ).fetchall() + assert [row["target_type"] for row in deliveries] == ["local_command"] + # The job itself is untouched. + assert manager.status(job_id)["status"] == "running" + finally: + manager.close() + + +def test_request_decision_without_wake_target_still_records(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Proceed?", options=["yes", "no"]) + assert decision_row(manager, decision["decision_id"])["status"] == "pending" + assert manager.db.execute("SELECT COUNT(*) FROM deliveries").fetchone()[0] == 0 + finally: + manager.close() + + +def test_resolve_records_choice_and_is_waitable(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", options=["approve", "deny"]) + resolved = manager.resolve_decision(job_id, decision["decision_id"], "approve") + assert resolved["status"] == "resolved" + assert resolved["choice"] == "approve" + assert resolved["resolved_by"] == "user" + assert resolved["resolved_at"] is not None + assert event_types(manager, job_id)[-1] == "decision_resolved" + + waited = asyncio.run(manager.wait(job_id, ["decision_resolved"], timeout_seconds=5)) + assert waited["result"] == "event" + assert waited["event"]["type"] == "decision_resolved" + assert waited["event"]["data"]["choice"] == "approve" + finally: + manager.close() + + +def test_resolve_is_idempotent_for_same_choice_and_conflicts_otherwise(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", options=["approve", "deny"]) + first = manager.resolve_decision(job_id, decision["decision_id"], "deny") + again = manager.resolve_decision(job_id, decision["decision_id"], "deny") + assert again["choice"] == first["choice"] == "deny" + with pytest.raises(ValueError, match="already resolved"): + manager.resolve_decision(job_id, decision["decision_id"], "approve") + finally: + manager.close() + + +def test_withdraw_decision(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?") + withdrawn = manager.withdraw_decision(job_id, decision["decision_id"]) + assert withdrawn["status"] == "withdrawn" + assert event_types(manager, job_id)[-1] == "decision_withdrawn" + with pytest.raises(ValueError, match="withdrawn"): + manager.resolve_decision(job_id, decision["decision_id"], "approve") + finally: + manager.close() + + +def test_expired_decision_cannot_be_resolved_and_sweep_emits(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", timeout_seconds=3600) + assert decision["expires_at"] is not None + # Simulate the clock passing without sleeping. + manager.db.execute( + "UPDATE decisions SET expires_at=? WHERE decision_id=?", + ("2000-01-01T00:00:00Z", decision["decision_id"]), + ) + manager.db.commit() + manager._expire_decisions() + assert decision_row(manager, decision["decision_id"])["status"] == "expired" + assert "decision_expired" in event_types(manager, job_id) + with pytest.raises(ValueError, match="expired"): + manager.resolve_decision(job_id, decision["decision_id"], "approve") + finally: + manager.close() + + +def test_expiry_sweep_leaves_future_decisions_pending(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", timeout_seconds=3600) + manager._expire_decisions() + assert decision_row(manager, decision["decision_id"])["status"] == "pending" + finally: + manager.close() + + +def test_request_decision_validates_inputs(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + with pytest.raises(ValueError, match="non-empty"): + manager.request_decision(job_id, " ") + with pytest.raises(ValueError, match="options"): + manager.request_decision(job_id, "Deploy?", options=[]) + with pytest.raises(ValueError, match="options"): + manager.request_decision(job_id, "Deploy?", options=["ok", ""]) + with pytest.raises(ValueError, match="timeout_seconds"): + manager.request_decision(job_id, "Deploy?", timeout_seconds=0) + with pytest.raises(ValueError, match="Unknown job_id"): + manager.request_decision("job_missing", "Deploy?") + assert manager.db.execute("SELECT COUNT(*) FROM decisions").fetchone()[0] == 0 + finally: + manager.close() + + +def test_request_decision_rejects_terminal_job(tmp_path): + manager = JobManager(tmp_path / "state") + try: + started = asyncio.run(manager.start(cmd("print('done')"))) + job_id = started["job_id"] + asyncio.run(manager.wait(job_id, ["completed"], timeout_seconds=10)) + with pytest.raises(ValueError, match="terminal"): + manager.request_decision(job_id, "Too late?") + finally: + manager.close() + + +def test_resolve_rejects_unknown_choice_and_mismatched_token(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + other = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", options=["approve", "deny"]) + with pytest.raises(ValueError, match="choice must be one of"): + manager.resolve_decision(job_id, decision["decision_id"], "maybe") + with pytest.raises(ValueError, match="Unknown decision"): + manager.resolve_decision(other, decision["decision_id"], "approve") + with pytest.raises(ValueError, match="token"): + manager.resolve_decision(job_id, "", "approve") + finally: + manager.close() + + +def test_list_decisions_filters_by_job_and_status(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + other = running_job(manager) + first = manager.request_decision(job_id, "One") + manager.request_decision(job_id, "Two") + manager.request_decision(other, "Three") + manager.resolve_decision(job_id, first["decision_id"], "approve") + + assert manager.list_decisions(job_id=other)["count"] == 1 + assert manager.list_decisions(status="pending")["count"] == 2 + resolved = manager.list_decisions(job_id=job_id, status="resolved") + assert [item["decision_id"] for item in resolved["decisions"]] == [first["decision_id"]] + with pytest.raises(ValueError, match="status must be one of"): + manager.list_decisions(status="bogus") + finally: + manager.close() + + +def test_v15_database_upgrades_to_decisions_table(tmp_path): + home = tmp_path / "state" + manager = JobManager(home) + manager.db.execute("DROP TABLE decisions") + manager.db.execute("PRAGMA user_version=15") + manager.db.commit() + manager.close() + + reopened = JobManager(home) + try: + assert reopened.doctor()["schema_version"] == 16 + assert reopened.db.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='decisions'" + ).fetchone() is not None + finally: + reopened.close() + + +def free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def test_decisions_over_http(tmp_path): + """The daemon routes + client wrappers for the decision tools.""" + port = free_port() + proc = subprocess.Popen( + [sys.executable, "-m", "vanth.daemon"], + env={**os.environ, "VANTH_HOME": str(tmp_path / "state"), "VANTH_DAEMON_PORT": str(port)}, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + client = VanthClient(f"http://127.0.0.1:{port}", tmp_path / "state") + try: + deadline = time.monotonic() + 5 + while True: + try: + assert client.get("/health") == {"ok": True} + break + except Exception: + if time.monotonic() > deadline: + raise + time.sleep(0.05) + + job = client.post("/jobs", {"command": cmd("import time; time.sleep(20)")}) + client.post(f"/jobs/{job['job_id']}/wait", {"filters": ["started"], "timeout_seconds": 10}) + + decision = client.post(f"/jobs/{job['job_id']}/decision", {"prompt": "Deploy?", "options": ["approve", "deny"]}) + assert decision["status"] == "pending" + assert client.get("/decisions", {"job_id": job["job_id"]})["count"] == 1 + + resolved = client.post(f"/jobs/{job['job_id']}/decision/{decision['decision_id']}/resolve", {"choice": "approve"}) + assert resolved["status"] == "resolved" and resolved["choice"] == "approve" + + second = client.post(f"/jobs/{job['job_id']}/decision", {"prompt": "Again?", "options": ["yes"]}) + bad = client.post(f"/jobs/{job['job_id']}/decision/{second['decision_id']}/resolve", {"choice": "no"}) + assert bad["result"] == "error" + withdrawn = client.post(f"/jobs/{job['job_id']}/decision/{second['decision_id']}/withdraw") + assert withdrawn["status"] == "withdrawn" + assert client.get("/decisions", {"status": "withdrawn"})["count"] == 1 + finally: + proc.terminate() + proc.wait(timeout=5) From 6bcbf853d3d7548dcc754dc8e6e79933fcbe5fdd Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Tue, 15 Sep 2026 10:08:20 -0700 Subject: [PATCH 2/4] Bump Go mirrored schema version to 16 The Go state reader pins `LatestSchemaVersion` and the cross-language fixture conformance test asserts the generated fixture matches it. Schema v16 (decisions table) moved Python forward, so the mirrored constant and the committed fixture must follow or the go CI jobs fail. No Go logic change: the monitor/state readers do not touch the new table. --- internal/state/state.go | 2 +- testdata/state/jobs.sqlite | Bin 143360 -> 159744 bytes 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/state/state.go b/internal/state/state.go index 167a9f3..d168ca4 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -14,7 +14,7 @@ import ( const DefaultBusyTimeoutMS = 30000 // LatestSchemaVersion is the current schema version (matches migrations.py). -const LatestSchemaVersion = 15 +const LatestSchemaVersion = 16 // Open opens the database read-write and applies the shared connection policy. func Open(path string) (*sql.DB, error) { diff --git a/testdata/state/jobs.sqlite b/testdata/state/jobs.sqlite index 279e859ff1e4a8bad0b9efbd71b96cfcb118338d..6da121d8f104155821ff4d761c1e7a8e3f53f583 100644 GIT binary patch delta 490 zcmZp8z|ru4bAq(0Bm)D3IuOHvIFP{t;{a(!AQqUoP@Yk;F=1;0;;Uv z9E=`JjSLSM3MMWz(alp~7dMw@Y!uGSOG&NBOsR-ZNlnfy&dkp%jxR{fOUcYjhYP4r zcdTa=jzp7I@b^=IE72$}Nh~QX)=@~UD99{IEsjqt(M(fj7uN?`!Hva=to$UnUNx}s z;xOZZQpiS6pI^=>=3}PFE-o+6*kWChn3R(W*9vE-I0v~phPWz(I6C>bBAlh5!KDBP zFvZguD;Ray3X1Y`3rZ#nGD)%J7nA_~J^f-eqo{~Naz=h;a;idzYea~S0>oD!MchD7 z<>!>8ro<;zPQJ@6uaE3isA{;9#1gOw+-^-nrRjWSj54AyCltpU!kMbuw-hn@>|OLg KVbKQxAOZjtGn%ge delta 78 zcmZp8z}fJCV}i7*7y|=?G7!Un2#~=7;{a(!Am*RAP@Yk&F=1;0 Date: Tue, 15 Sep 2026 10:57:03 -0700 Subject: [PATCH 3/4] Fix review findings: atomic decision transitions, cap, deadline, bounds Adversarial review of the decision primitive surfaced three HIGH and four MEDIUM findings. All fixed: HIGH - state change and its event were not atomic. Every mutation committed the decision row and then emitted in a separate transaction, so a crash in between left a resolved decision with no `decision_resolved` event (a retry could not repair it, because the idempotent path short-circuits). `_emit` now takes an optional `mutate` callback that runs inside the write transaction after the cap check and before the event insert; `BEGIN IMMEDIATE` still covers the event and its deliveries. request/resolve/withdraw/expire and their events are now one transaction. A `_DecisionNoOp` abort keeps the idempotent same-choice retry from writing a duplicate event. HIGH - the per-job structured-event cap silently disabled the protocol: at the cap, `_emit` returned `persisted=False` with no event and no wake, while the tool still reported success. Decision lifecycle events are now exempt from the cap like terminal events (decisions are human-paced and bounded), and ordinary telemetry is still capped. HIGH - the deadline was checked on a stale read before the CAS, so lock ordering decided whether an overdue approval was accepted. Deadline, status and choice validation now all run inside the write transaction that performs the transition. Withdraw also enforces the deadline (it previously did not, letting an overdue request escape its expiry event). MEDIUM - cleanup left decisions behind, so a pending request could outlive its job and be resolved into an orphan event: cleanup now deletes a job's decisions with its other state. MEDIUM - unbounded prompt/options could exceed `max_event_bytes`, which replaces the whole data object and drops `decision_id`/`choice`, breaking the wake and wait paths. Input is now bounded (10000 chars, 50 options, 200 chars each) and options dedupe linearly. MEDIUM - the daemon matched decision routes by suffix only, so `/jobs//resolve` raised IndexError (HTTP 500) and `/jobs//not-a-decision//resolve` resolved successfully. Routes now match exact segment shapes via `_decision_route`. MEDIUM - job eligibility was read outside the insert transaction; it now runs in the mutate, so a job finishing mid-request cannot get a decision. Tests (13 -> 19): fault injection proving state+event roll back together, event-cap exemption, concurrent-resolve single-winner, withdraw-overdue rejection, input bounds, cleanup removing decisions, decision_requested event data and delivery linkage, exactly-one expiry event across sweeps, and malformed HTTP paths. Full suite 808 passed / 6 skipped; compileall clean. --- CHANGELOG.md | 17 ++- docs/agent-tools.md | 6 +- src/vanth/daemon.py | 33 ++++- src/vanth/server.py | 284 +++++++++++++++++++++++++++------------- tests/test_decisions.py | 155 +++++++++++++++++++++- 5 files changed, 393 insertions(+), 102 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae2eecb..79349eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,12 +21,25 @@ holding a thread or killing the job. waits for the answer like any other event. - An optional `timeout_seconds` expires the request; the maintenance loop marks it `expired` and emits `decision_expired`, after which it can no - longer be resolved. + longer be resolved. Deadline, status and choice are all validated **inside** + the write transaction that performs the resolution, so the deadline is + enforced at the authoritative transition rather than on a stale read. +- Each decision state change commits together with its lifecycle event and + wake deliveries in one transaction: a crash can never leave a resolved + decision with no `decision_resolved` event for a `job_wait` caller (a retry + could not repair that). Decision lifecycle events are also exempt from the + per-job structured-event cap, so a busy job at the cap still gets its wake. +- `prompt`/`options` are bounded (10000 chars, 50 options, 200 chars each) so + the lifecycle payload can never be truncated (which would drop the + `decision_id` and break the wake and wait paths). +- `job_cleanup` deletes a job's decisions with the rest of its state, so a + removed job leaves no actionable pending request behind. - Lifecycle events (`decision_requested` / `decision_resolved` / `decision_withdrawn` / `decision_expired`) are the audit trail; `decision_requested` also ranks as an attention event in `job_view`. - Schema v16 adds the `decisions` table (additive; existing databases - migrate in place with the usual pre-migration backup). + migrate in place with the usual pre-migration backup); the Go mirrored + `LatestSchemaVersion` and cross-language fixture move to 16 with it. ## 1.9.1 - 2026-09-11 diff --git a/docs/agent-tools.md b/docs/agent-tools.md index f7a20cb..e9b05bb 100644 --- a/docs/agent-tools.md +++ b/docs/agent-tools.md @@ -515,9 +515,13 @@ running while it waits. |---|---|---|---| | `job_id` | `string` | required | A queued/running (non-terminal) job | | `prompt` | `string` | required | The question shown to the human | -| `options` | `string[]` | `["approve", "deny"]` | Allowed choices (deduplicated) | +| `options` | `string[]` | `["approve", "deny"]` | Allowed choices (deduplicated; max 50, 200 chars each) | | `timeout_seconds` | `int` | none | Expire the request after N seconds | +`prompt` is limited to 10000 characters. The request, its `decision_requested` +event and any wake deliveries commit in one transaction; decision lifecycle +events are exempt from the per-job structured-event cap. + Requesting emits a `decision_requested` event, which reuses the wake-target delivery path — the job's wake targets are notified, so the owning thread learns a human is needed. Only targets whose `events` list includes diff --git a/src/vanth/daemon.py b/src/vanth/daemon.py index 54c258b..b29501d 100644 --- a/src/vanth/daemon.py +++ b/src/vanth/daemon.py @@ -479,6 +479,24 @@ def error(handler: BaseHTTPRequestHandler, message: str, status: int = 400) -> N ok(handler, {"result": "error", "error": message[:4096]}, status) +def _decision_route(path: str) -> tuple[str, str, str] | None: + """Match the exact decision routes, or None. + + Exact segment matching (rather than a loose endswith suffix) so malformed + paths like ``/jobs/x/resolve`` or ``/jobs/x/not-a-decision/t/resolve`` are + a clean 404 instead of an IndexError 500 or an unintended resolve. + """ + parts = path.split("/") + if len(parts) == 4 and parts[1] == "jobs" and parts[3] == "decision" and parts[2]: + return ("request", parts[2], "") + if len(parts) == 6 and parts[1] == "jobs" and parts[3] == "decision" and parts[2] and parts[4]: + if parts[5] == "resolve": + return ("resolve", parts[2], parts[4]) + if parts[5] == "withdraw": + return ("withdraw", parts[2], parts[4]) + return None + + class Handler(BaseHTTPRequestHandler): server_version = "vanthd/1" @@ -754,13 +772,14 @@ def do_POST(self) -> None: ok(self, get_manager().job_pause(parsed.path.split("/")[2])) elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/resume"): ok(self, get_manager().job_resume(parsed.path.split("/")[2])) - elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/decision"): - ok(self, get_manager().request_decision(parsed.path.split("/")[2], **payload)) - elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/resolve"): - ok(self, get_manager().resolve_decision( - parsed.path.split("/")[2], parsed.path.split("/")[4], payload.get("choice", ""))) - elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/withdraw"): - ok(self, get_manager().withdraw_decision(parsed.path.split("/")[2], parsed.path.split("/")[4])) + elif (decision_route := _decision_route(parsed.path)) is not None: + action, decision_job, token = decision_route + if action == "request": + ok(self, get_manager().request_decision(decision_job, **payload)) + elif action == "resolve": + ok(self, get_manager().resolve_decision(decision_job, token, payload.get("choice", ""))) + else: + ok(self, get_manager().withdraw_decision(decision_job, token)) elif parsed.path == "/schedules": ok(self, get_manager().create_schedule(**payload)) elif parsed.path.startswith("/schedules/") and parsed.path.endswith("/update"): diff --git a/src/vanth/server.py b/src/vanth/server.py index f4aa9c8..4f6d94f 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -20,7 +20,7 @@ import warnings from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any +from typing import Any, Callable from logging.handlers import RotatingFileHandler # mcp's FastMCP has a Settings model with a `lifespan` field whose annotation @@ -336,6 +336,21 @@ def normalize_event_payload(payload: dict[str, Any]) -> dict[str, Any]: DECISION_PENDING = "pending" DECISION_STATUSES = {DECISION_PENDING, "resolved", "withdrawn", "expired"} DEFAULT_DECISION_OPTIONS = ["approve", "deny"] +# Control events for the decision state machine. They are exempt from the +# per-job structured-event cap (like terminal events): the cap exists to bound +# telemetry, and dropping a decision event would break the durability contract +# (no wake, no waitable signal) with no way for a retry to repair it. +DECISION_EVENT_TYPES = {"decision_requested", "decision_resolved", "decision_withdrawn", "decision_expired"} +# Bound decision input so the lifecycle event payload can never hit +# ``max_event_bytes`` (which would strip decision_id/choice and break wake +# delivery and `job_wait`). Human-paced, so the limits are generous. +MAX_DECISION_PROMPT_CHARS = 10000 +MAX_DECISION_OPTIONS = 50 +MAX_DECISION_OPTION_CHARS = 200 + + +class _DecisionNoOp(Exception): + """Internal: a decision mutation found nothing to change (idempotent retry).""" def resolve_wake_target_identity( @@ -2087,6 +2102,7 @@ def _emit( data: dict[str, Any] | None = None, level: str = "info", source: str = "server", + mutate: Callable[[sqlite3.Connection], None] | None = None, ) -> dict[str, Any]: self._ensure_open() payload = normalize_event_payload({"type": event_type, "message": message, "data": data or {}, "level": level}) @@ -2100,7 +2116,7 @@ def _emit( for attempt in range(10): try: event = self._emit_transactional( - job_id, payload, data_json, event_type, level, source, message + job_id, payload, data_json, event_type, level, source, message, mutate ) break except sqlite3.OperationalError as exc: @@ -2125,10 +2141,19 @@ def _emit_transactional( level: str, source: str, message: str | None, + mutate: Callable[[sqlite3.Connection], None] | None = None, ) -> dict[str, Any]: + """Persist a state mutation and its event + deliveries in ONE transaction. + + ``mutate`` runs inside the write transaction, after the event-cap check + and before the event row is inserted, so a decision state change and the + event/wake it owes either both land or neither does (a crash can never + leave a resolved decision with no event, which a retry could not + repair). ``mutate`` may raise ``_DecisionNoOp`` to abort cleanly. + """ self.db.execute("BEGIN IMMEDIATE") try: - if event_type not in TERMINAL_STATUSES: + if event_type not in TERMINAL_STATUSES and event_type not in DECISION_EVENT_TYPES: count = self.db.execute("SELECT COUNT(*) FROM events WHERE job_id=?", (job_id,)).fetchone()[0] if count >= self.max_events_per_job: self.db.rollback() @@ -2147,6 +2172,24 @@ def _emit_transactional( "created_at": now_iso(), "persisted": False, } + if mutate is not None: + try: + mutate(self.db) + except _DecisionNoOp: + self.db.rollback() + return { + "event_id": None, + "job_id": job_id, + "seq": 0, + "type": event_type, + "level": level, + "message": message, + "data": payload["data"], + "source": source, + "created_at": now_iso(), + "persisted": False, + "noop": True, + } row = self.db.execute("SELECT COALESCE(MAX(seq), 0) + 1 AS seq FROM events WHERE job_id=?", (job_id,)).fetchone() seq = int(row["seq"]) created_at = now_iso() @@ -5393,6 +5436,7 @@ def cleanup(self, older_than_seconds: int, dry_run: bool = True) -> dict[str, An self.db.execute(f"DELETE FROM delivery_attempts WHERE delivery_id IN (SELECT delivery_id FROM deliveries WHERE job_id IN ({placeholders}))", job_ids) self.db.execute(f"DELETE FROM deliveries WHERE job_id IN ({placeholders})", job_ids) self.db.execute(f"DELETE FROM wake_targets WHERE job_id IN ({placeholders})", job_ids) + self.db.execute(f"DELETE FROM decisions WHERE job_id IN ({placeholders})", job_ids) self.db.execute(f"DELETE FROM events WHERE job_id IN ({placeholders})", job_ids) self.db.execute(f"DELETE FROM jobs WHERE job_id IN ({placeholders}) AND status!='running'", job_ids) self.db.commit() @@ -5821,11 +5865,13 @@ def _decision_dict(self, row: sqlite3.Row) -> dict[str, Any]: } def _decision_isdue(self, decision: dict[str, Any], now: datetime | None = None) -> bool: - expires_at = decision.get("expires_at") - if not expires_at or decision["status"] != DECISION_PENDING: - return False - parsed = _parse_iso(str(expires_at)) - return parsed is not None and parsed <= (now or datetime.now(timezone.utc)) + return _row_isdue(decision.get("expires_at"), decision.get("status"), now) + + def _reload_decision(self, decision_id: str) -> dict[str, Any]: + row = self._row("SELECT * FROM decisions WHERE decision_id=?", (decision_id,)) + if row is None: # pragma: no cover - the row was just written/committed + raise ValueError(f"Unknown decision: {decision_id}") + return self._decision_dict(row) def request_decision( self, @@ -5836,51 +5882,49 @@ def request_decision( ) -> dict[str, Any]: """Persist a durable "needs a decision" request and wake the owner. - Emitting ``decision_requested`` reuses the wake-target delivery path, so - whichever thread owns the job (``wake_targets``) is notified. The job - itself keeps running; nothing about its status changes. + The job-status eligibility check, the row insert, the + ``decision_requested`` event and its wake deliveries all commit in one + transaction, so a crash can never leave a pending decision with no wake + (which a retry could not repair). The job keeps running; nothing about + its status changes. """ self._ensure_open() if not isinstance(prompt, str) or not prompt.strip(): raise ValueError("prompt must be a non-empty string") - choices = DEFAULT_DECISION_OPTIONS if options is None else options - if not isinstance(choices, list) or not choices: - raise ValueError("options must be a non-empty list of strings") - normalized: list[str] = [] - for option in choices: - if not isinstance(option, str) or not option.strip(): - raise ValueError("options must be non-empty strings") - if option not in normalized: - normalized.append(option) - if timeout_seconds is not None: - if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, int) or timeout_seconds < 1: - raise ValueError("timeout_seconds must be an integer >= 1") - row = self._row("SELECT status FROM jobs WHERE job_id=?", (job_id,)) - if not row: - raise ValueError(f"Unknown job_id: {job_id}") - if row["status"] in TERMINAL_STATUSES: - raise ValueError(f"job is already terminal: {row['status']}") + if len(prompt) > MAX_DECISION_PROMPT_CHARS: + raise ValueError(f"prompt must be at most {MAX_DECISION_PROMPT_CHARS} characters") + normalized = _normalize_decision_options(options) + if timeout_seconds is not None and ( + isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, int) or timeout_seconds < 1 + ): + raise ValueError("timeout_seconds must be an integer >= 1") now = datetime.now(timezone.utc) decision_id = "dec_" + uuid.uuid4().hex[:16] created_at = now_iso() expires_at = (now + timedelta(seconds=timeout_seconds)).isoformat().replace("+00:00", "Z") if timeout_seconds else None - with self.db_lock: - self.db.execute( + + def mutate(db: sqlite3.Connection) -> None: + job = db.execute("SELECT status FROM jobs WHERE job_id=?", (job_id,)).fetchone() + if job is None: + raise ValueError(f"Unknown job_id: {job_id}") + if job["status"] in TERMINAL_STATUSES: + raise ValueError(f"job is already terminal: {job['status']}") + db.execute( """ INSERT INTO decisions(decision_id, job_id, prompt, options_json, choice, status, resolved_by, created_at, expires_at, resolved_at) VALUES (?, ?, ?, ?, NULL, ?, NULL, ?, ?, NULL) """, (decision_id, job_id, prompt, json.dumps(normalized, separators=(",", ":")), DECISION_PENDING, created_at, expires_at), ) - self.db.commit() + self._emit( job_id, "decision_requested", message=prompt, data={"decision_id": decision_id, "prompt": prompt, "options": normalized, "expires_at": expires_at}, + mutate=mutate, ) - decision = self._row("SELECT * FROM decisions WHERE decision_id=?", (decision_id,)) - return self._decision_dict(decision) + return self._reload_decision(decision_id) def _decision_for(self, job_id: str, token: str) -> sqlite3.Row: if not isinstance(token, str) or not token: @@ -5890,73 +5934,84 @@ def _decision_for(self, job_id: str, token: str) -> sqlite3.Row: raise ValueError(f"Unknown decision for job {job_id}: {token}") return row - def _expire_decision(self, row: sqlite3.Row) -> bool: - """CAS pending->expired; True when this call performed the transition.""" - with self.db_lock: - cursor = self.db.execute( - "UPDATE decisions SET status='expired', resolved_at=? WHERE decision_id=? AND status=?", - (now_iso(), row["decision_id"], DECISION_PENDING), - ) - self.db.commit() - return cursor.rowcount == 1 - def resolve_decision(self, job_id: str, token: str, choice: str, actor: str = "user") -> dict[str, Any]: - """Record a human choice for a pending decision (idempotent per choice).""" + """Record a human choice for a pending decision (idempotent per choice). + + Validation (deadline, status, choice) and the pending->resolved CAS all + run inside the write transaction, so the deadline is enforced at the + authoritative transition rather than on a stale read. + """ self._ensure_open() - row = self._decision_for(job_id, token) - if row["status"] == "resolved": - if row["choice"] == choice: - return self._decision_dict(row) - raise ValueError(f"decision already resolved as {row['choice']!r}") - if row["status"] == "expired" or self._decision_isdue(self._decision_dict(row)): - if row["status"] == DECISION_PENDING and self._expire_decision(row): - self._emit(job_id, "decision_expired", message="decision timed out", data={"decision_id": row["decision_id"]}) - raise ValueError("decision expired") - if row["status"] != DECISION_PENDING: - raise ValueError(f"decision is {row['status']}") - allowed = json.loads(row["options_json"] or "[]") - if choice not in allowed: - raise ValueError(f"choice must be one of {allowed}") - with self.db_lock: - cursor = self.db.execute( + if not isinstance(choice, str) or not choice: + raise ValueError("choice must be a non-empty string") + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string") + + def mutate(db: sqlite3.Connection) -> None: + row = db.execute("SELECT * FROM decisions WHERE decision_id=?", (token,)).fetchone() + if row is None or row["job_id"] != job_id: + raise ValueError(f"Unknown decision for job {job_id}: {token}") + if row["status"] == "resolved": + if row["choice"] == choice: + raise _DecisionNoOp() + raise ValueError(f"decision already resolved as {row['choice']!r}") + if row["status"] != DECISION_PENDING: + raise ValueError(f"decision is {row['status']}") + if _row_isdue(row["expires_at"], row["status"], datetime.now(timezone.utc)): + raise ValueError("decision expired") + allowed = json.loads(row["options_json"] or "[]") + if choice not in allowed: + raise ValueError(f"choice must be one of {allowed}") + cursor = db.execute( "UPDATE decisions SET status='resolved', choice=?, resolved_by=?, resolved_at=? WHERE decision_id=? AND status=?", - (choice, actor, now_iso(), row["decision_id"], DECISION_PENDING), + (choice, actor, now_iso(), token, DECISION_PENDING), ) - self.db.commit() - if cursor.rowcount != 1: - current = self._row("SELECT * FROM decisions WHERE decision_id=?", (row["decision_id"],)) - if current is not None and current["status"] == "resolved" and current["choice"] == choice: - return self._decision_dict(current) - raise ValueError("decision was resolved concurrently") - self._emit( + if cursor.rowcount != 1: + raise ValueError("decision changed concurrently") + + event = self._emit( job_id, "decision_resolved", message=choice, - data={"decision_id": row["decision_id"], "choice": choice, "actor": actor, "prompt": row["prompt"]}, + data={"decision_id": token, "choice": choice, "actor": actor}, + mutate=mutate, ) - return self._decision_dict(self._row("SELECT * FROM decisions WHERE decision_id=?", (row["decision_id"],))) + if event is not None and event.get("noop"): + row = self._decision_for(job_id, token) + if row["choice"] != choice: # pragma: no cover - noop only for equal choice + raise ValueError(f"decision already resolved as {row['choice']!r}") + return self._decision_dict(row) + return self._reload_decision(token) def withdraw_decision(self, job_id: str, token: str, actor: str = "user") -> dict[str, Any]: """Cancel a pending decision (the requester no longer needs an answer).""" self._ensure_open() - row = self._decision_for(job_id, token) - if row["status"] != DECISION_PENDING: - raise ValueError(f"decision is {row['status']}") - with self.db_lock: - cursor = self.db.execute( + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string") + + def mutate(db: sqlite3.Connection) -> None: + row = db.execute("SELECT * FROM decisions WHERE decision_id=?", (token,)).fetchone() + if row is None or row["job_id"] != job_id: + raise ValueError(f"Unknown decision for job {job_id}: {token}") + if row["status"] != DECISION_PENDING: + raise ValueError(f"decision is {row['status']}") + if _row_isdue(row["expires_at"], row["status"], datetime.now(timezone.utc)): + raise ValueError("decision expired") + cursor = db.execute( "UPDATE decisions SET status='withdrawn', resolved_by=?, resolved_at=? WHERE decision_id=? AND status=?", - (actor, now_iso(), row["decision_id"], DECISION_PENDING), + (actor, now_iso(), token, DECISION_PENDING), ) - self.db.commit() - if cursor.rowcount != 1: - raise ValueError("decision was resolved concurrently") + if cursor.rowcount != 1: + raise ValueError("decision changed concurrently") + self._emit( job_id, "decision_withdrawn", message="decision withdrawn", - data={"decision_id": row["decision_id"], "actor": actor, "prompt": row["prompt"]}, + data={"decision_id": token, "actor": actor}, + mutate=mutate, ) - return self._decision_dict(self._row("SELECT * FROM decisions WHERE decision_id=?", (row["decision_id"],))) + return self._reload_decision(token) def list_decisions(self, job_id: str | None = None, status: str | None = None, limit: int = 50) -> dict[str, Any]: self._ensure_open() @@ -5971,25 +6026,78 @@ def list_decisions(self, job_id: str | None = None, status: str | None = None, l sql += " AND status=?" args.append(status) sql += " ORDER BY created_at DESC, decision_id DESC LIMIT ?" - args.append(max(1, min(int(limit), 500))) + args.append(validate_limit(limit, "limit", maximum=500)) with self.db_lock: rows = self.db.execute(sql, tuple(args)).fetchall() return {"decisions": [self._decision_dict(row) for row in rows], "count": len(rows)} def _expire_decisions(self) -> None: - """Expire overdue pending decisions (maintenance loop; edge-triggered).""" + """Expire overdue pending decisions (maintenance loop; edge-triggered). + + Each expiry's state change and ``decision_expired`` event commit + together, so an expiry is never recorded without its event. + """ now = datetime.now(timezone.utc) with self.db_lock: rows = self.db.execute( - "SELECT * FROM decisions WHERE status=? AND expires_at IS NOT NULL", + "SELECT decision_id, job_id, expires_at FROM decisions WHERE status=? AND expires_at IS NOT NULL", (DECISION_PENDING,), ).fetchall() for row in rows: - if not self._decision_isdue(self._decision_dict(row), now): + if not _row_isdue(row["expires_at"], DECISION_PENDING, now): continue - if self._expire_decision(row): - self._emit(row["job_id"], "decision_expired", message="decision timed out", - data={"decision_id": row["decision_id"], "prompt": row["prompt"]}) + + def mutate(db: sqlite3.Connection, decision_id: str = row["decision_id"], expires_at: str = row["expires_at"]) -> None: + # Re-check under the write lock: a concurrent resolve/withdraw + # (or a resolver that beat the deadline by a hair) wins. + current = db.execute("SELECT expires_at, status FROM decisions WHERE decision_id=?", (decision_id,)).fetchone() + if current is None or not _row_isdue(current["expires_at"], current["status"], datetime.now(timezone.utc)): + raise _DecisionNoOp() + cursor = db.execute( + "UPDATE decisions SET status='expired', resolved_at=? WHERE decision_id=? AND status=?", + (now_iso(), decision_id, DECISION_PENDING), + ) + if cursor.rowcount != 1: + raise _DecisionNoOp() + + self._emit( + row["job_id"], + "decision_expired", + message="decision timed out", + data={"decision_id": row["decision_id"]}, + mutate=mutate, + ) + + +def _row_isdue(expires_at: Any, status: Any, now: datetime | None = None) -> bool: + """Whether a stored deadline has passed. Malformed/non-pending -> False.""" + if status != DECISION_PENDING or not expires_at: + return False + parsed = _parse_iso(str(expires_at)) + if parsed is None: + return False + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed <= (now or datetime.now(timezone.utc)) + + +def _normalize_decision_options(options: list[str] | None) -> list[str]: + choices = DEFAULT_DECISION_OPTIONS if options is None else options + if not isinstance(choices, list) or not choices: + raise ValueError("options must be a non-empty list of strings") + if len(choices) > MAX_DECISION_OPTIONS: + raise ValueError(f"options must contain at most {MAX_DECISION_OPTIONS} entries") + normalized: list[str] = [] + seen: set[str] = set() + for option in choices: + if not isinstance(option, str) or not option.strip(): + raise ValueError("options must be non-empty strings") + if len(option) > MAX_DECISION_OPTION_CHARS: + raise ValueError(f"each option must be at most {MAX_DECISION_OPTION_CHARS} characters") + if option not in seen: + seen.add(option) + normalized.append(option) + return normalized client: VanthClient | None = None diff --git a/tests/test_decisions.py b/tests/test_decisions.py index 68bb6b5..2f31fa8 100644 --- a/tests/test_decisions.py +++ b/tests/test_decisions.py @@ -10,10 +10,12 @@ """ import asyncio +import json import os import socket import subprocess import sys +import threading import time import pytest @@ -45,6 +47,13 @@ def decision_row(manager: JobManager, decision_id: str): return manager.db.execute("SELECT * FROM decisions WHERE decision_id=?", (decision_id,)).fetchone() +def event_rows(manager: JobManager, job_id: str, event_type: str) -> list[dict]: + rows = manager.db.execute( + "SELECT data_json FROM events WHERE job_id=? AND type=? ORDER BY seq", (job_id, event_type) + ).fetchall() + return [json.loads(row["data_json"]) for row in rows] + + def test_request_decision_is_pending_and_wakes_owner(tmp_path): manager = JobManager(tmp_path / "state") try: @@ -60,12 +69,24 @@ def test_request_decision_is_pending_and_wakes_owner(tmp_path): assert decision["options"] == ["approve", "deny"] assert decision["choice"] is None assert decision["expires_at"] is None - assert "decision_requested" in event_types(manager, job_id) - # The owner is notified through the ordinary delivery queue. + # The lifecycle event carries the identifiers a woken agent needs. + payload = event_rows(manager, job_id, "decision_requested") + assert len(payload) == 1 + assert payload[0] == { + "decision_id": decision["decision_id"], + "prompt": "Ship the release?", + "options": ["approve", "deny"], + "expires_at": None, + } + # The owner is notified through the ordinary delivery queue, linked to + # that event. deliveries = manager.db.execute( - "SELECT target_type, status FROM deliveries WHERE job_id=?", (job_id,) + "SELECT d.target_type, d.status, e.type FROM deliveries d JOIN events e ON e.event_id=d.event_id WHERE d.job_id=?", + (job_id,), ).fetchall() - assert [row["target_type"] for row in deliveries] == ["local_command"] + assert [(row["target_type"], row["status"], row["type"]) for row in deliveries] == [ + ("local_command", "pending", "decision_requested") + ] # The job itself is untouched. assert manager.status(job_id)["status"] == "running" finally: @@ -146,6 +167,9 @@ def test_expired_decision_cannot_be_resolved_and_sweep_emits(tmp_path): manager._expire_decisions() assert decision_row(manager, decision["decision_id"])["status"] == "expired" assert "decision_expired" in event_types(manager, job_id) + # A second sweep must not emit a duplicate expiry event. + manager._expire_decisions() + assert event_types(manager, job_id).count("decision_expired") == 1 with pytest.raises(ValueError, match="expired"): manager.resolve_decision(job_id, decision["decision_id"], "approve") finally: @@ -248,6 +272,121 @@ def test_v15_database_upgrades_to_decisions_table(tmp_path): reopened.close() +def test_decision_mutation_and_event_commit_together(tmp_path, monkeypatch): + """A failure after the state change must roll the state change back too.""" + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + + def boom(event): + raise RuntimeError("injected failure after mutation") + + monkeypatch.setattr(manager, "_enqueue_deliveries_uncommitted", boom) + with pytest.raises(RuntimeError, match="injected"): + manager.request_decision(job_id, "Deploy?") + # Neither half of the transaction survives. + assert manager.db.execute("SELECT COUNT(*) FROM decisions").fetchone()[0] == 0 + assert event_rows(manager, job_id, "decision_requested") == [] + finally: + manager.close() + + +def test_decision_events_survive_event_cap(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + manager.max_events_per_job = 1 # the job is already far past the cap + decision = manager.request_decision(job_id, "Deploy?") + assert decision["status"] == "pending" + assert len(event_rows(manager, job_id, "decision_requested")) == 1 + manager.resolve_decision(job_id, decision["decision_id"], "approve") + assert len(event_rows(manager, job_id, "decision_resolved")) == 1 + # Ordinary telemetry is still capped. + assert manager._emit(job_id, "progress", data={"current": 1})["persisted"] is False + finally: + manager.close() + + +def test_concurrent_resolve_has_single_winner(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", options=["approve", "deny"]) + results: list[str] = [] + errors: list[str] = [] + + def worker(choice: str) -> None: + try: + results.append(manager.resolve_decision(job_id, decision["decision_id"], choice)["choice"]) + except ValueError as exc: + errors.append(str(exc)) + + threads = [threading.Thread(target=worker, args=(choice,)) for choice in ("approve", "deny")] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert len(results) == 1, results + assert len(errors) == 1, errors + assert decision_row(manager, decision["decision_id"])["choice"] == results[0] + assert len(event_rows(manager, job_id, "decision_resolved")) == 1 + finally: + manager.close() + + +def test_withdraw_rejects_overdue_decision(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?", timeout_seconds=3600) + manager.db.execute( + "UPDATE decisions SET expires_at=? WHERE decision_id=?", + ("2000-01-01T00:00:00Z", decision["decision_id"]), + ) + manager.db.commit() + with pytest.raises(ValueError, match="expired"): + manager.withdraw_decision(job_id, decision["decision_id"]) + assert decision_row(manager, decision["decision_id"])["status"] == "pending" + finally: + manager.close() + + +def test_request_decision_bounds(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + with pytest.raises(ValueError, match="at most 10000"): + manager.request_decision(job_id, "x" * 10001) + with pytest.raises(ValueError, match="at most 50"): + manager.request_decision(job_id, "Deploy?", options=[f"opt{i}" for i in range(51)]) + with pytest.raises(ValueError, match="at most 200"): + manager.request_decision(job_id, "Deploy?", options=["y" * 201]) + # Duplicates are collapsed in order, not rejected. + assert manager.request_decision(job_id, "Deploy?", options=["a", "b", "a"])["options"] == ["a", "b"] + finally: + manager.close() + + +def test_cleanup_removes_decisions(tmp_path): + manager = JobManager(tmp_path / "state") + try: + started = asyncio.run(manager.start(cmd("print('done')"))) + job_id = started["job_id"] + asyncio.run(manager.wait(job_id, ["completed"], timeout_seconds=10)) + # Decisions can only be requested while non-terminal, so model a + # pending row left behind by inserting one directly. + manager.db.execute( + "INSERT INTO decisions(decision_id, job_id, prompt, options_json, choice, status, resolved_by, created_at, expires_at, resolved_at) " + "VALUES ('dec_left', ?, 'Deploy?', '[\"approve\"]', NULL, 'pending', NULL, ?, NULL, NULL)", + (job_id, "2026-09-15T00:00:00Z"), + ) + manager.db.commit() + manager.cleanup(0, dry_run=False) + assert manager.db.execute("SELECT COUNT(*) FROM decisions WHERE job_id=?", (job_id,)).fetchone()[0] == 0 + finally: + manager.close() + + def free_port() -> int: with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) @@ -291,6 +430,14 @@ def test_decisions_over_http(tmp_path): withdrawn = client.post(f"/jobs/{job['job_id']}/decision/{second['decision_id']}/withdraw") assert withdrawn["status"] == "withdrawn" assert client.get("/decisions", {"status": "withdrawn"})["count"] == 1 + + # Malformed paths are a clean error, never an unintended resolve/500. + third = client.post(f"/jobs/{job['job_id']}/decision", {"prompt": "Malformed?", "options": ["yes"]}) + assert client.post(f"/jobs/{job['job_id']}/resolve", {"choice": "yes"})["result"] == "error" + assert client.post( + f"/jobs/{job['job_id']}/not-a-decision/{third['decision_id']}/resolve", {"choice": "yes"} + )["result"] == "error" + assert client.get("/decisions", {"status": "pending"})["count"] == 1 finally: proc.terminate() proc.wait(timeout=5) From 3c429f303ce6dd299fefeaaf0abefcce53babb7c Mon Sep 17 00:00:00 2001 From: abhim-dv Date: Tue, 15 Sep 2026 11:39:25 -0700 Subject: [PATCH 4/4] Fix second-review findings: serialized payload bound, per-call cap exemption Verification review of the transactional fix confirmed 5 of 7 original defects resolved and found two remaining HIGH issues plus a routing gap: HIGH - character limits do not bound serialized bytes. `json.dumps` is ASCII-only, so 10000 astral chars serialize to ~120KB and blow past the 64KB event limit; `_emit` then replaces the whole data object with the truncation marker, dropping decision_id/choice and breaking the wake and wait paths while still committing the decision. `request_decision`/`resolve_decision`/ `withdraw_decision` now check the serialized lifecycle payload against `max_event_bytes` before the transaction and reject an over-budget request; `actor` is bounded too (it is only reachable from direct Python callers). HIGH - the cap exemption keyed on event type, but job stdout can emit any type via AGENT_EVENT, so a job printing `AGENT_EVENT {"type":"decision_requested"}` bypassed the per-job cap indefinitely (and fanned out wake deliveries). The exemption is now a per-CALL flag (`exempt_from_cap`) set only by the authoritative decision transitions, so forged telemetry is capped like everything else. MEDIUM - the decision matcher fixed the reported resolve paths but decision-looking paths still fell through to legacy suffix routes (`/jobs//decision//pause` paused the job). Decision routes are now matched before the suffix handlers, and any unmatched path in the decision namespace returns 404 instead of reaching another operation. Tests (19 -> 22): unicode/over-byte payload rejection (including a reduced byte limit), forged decision telemetry still capped, actor bounds, and a decision-namespace path not reaching another op (asserting the 404 message, not just an error). Full suite 811 passed / 6 skipped; compileall clean. --- CHANGELOG.md | 18 ++++++++---- docs/agent-tools.md | 5 ++-- src/vanth/daemon.py | 22 ++++++++------ src/vanth/server.py | 64 +++++++++++++++++++++++++++++++++-------- tests/test_decisions.py | 49 +++++++++++++++++++++++++++++++ 5 files changed, 131 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 79349eb..d0fd24d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,11 +27,19 @@ holding a thread or killing the job. - Each decision state change commits together with its lifecycle event and wake deliveries in one transaction: a crash can never leave a resolved decision with no `decision_resolved` event for a `job_wait` caller (a retry - could not repair that). Decision lifecycle events are also exempt from the - per-job structured-event cap, so a busy job at the cap still gets its wake. -- `prompt`/`options` are bounded (10000 chars, 50 options, 200 chars each) so - the lifecycle payload can never be truncated (which would drop the - `decision_id` and break the wake and wait paths). + could not repair that). Authoritative decision transitions are also exempt + from the per-job structured-event cap, so a busy job at the cap still gets + its wake. The exemption is per-call rather than per event type, because job + stdout can emit any event type via `AGENT_EVENT` — keying on the type would + let a job forge `decision_requested` lines and bypass the cap. +- `prompt`/`options` are bounded (10000 chars, 50 options, 200 chars each) and + the *serialized* lifecycle payload is checked against `max_event_bytes` + before commit, so the payload can never be truncated (truncation replaces + the whole data object, dropping the `decision_id` and breaking the wake and + wait paths). +- Decision routes match exact segment shapes, and an unmatched + decision-looking path is a 404 rather than falling through to another job + operation (e.g. `.../decision//pause` cannot pause the job). - `job_cleanup` deletes a job's decisions with the rest of its state, so a removed job leaves no actionable pending request behind. - Lifecycle events (`decision_requested` / `decision_resolved` / diff --git a/docs/agent-tools.md b/docs/agent-tools.md index e9b05bb..d1a088b 100644 --- a/docs/agent-tools.md +++ b/docs/agent-tools.md @@ -519,8 +519,9 @@ running while it waits. | `timeout_seconds` | `int` | none | Expire the request after N seconds | `prompt` is limited to 10000 characters. The request, its `decision_requested` -event and any wake deliveries commit in one transaction; decision lifecycle -events are exempt from the per-job structured-event cap. +event and any wake deliveries commit in one transaction; the serialized +payload is checked against the event byte limit, and authoritative decision +transitions are exempt from the per-job structured-event cap. Requesting emits a `decision_requested` event, which reuses the wake-target delivery path — the job's wake targets are notified, so the owning thread diff --git a/src/vanth/daemon.py b/src/vanth/daemon.py index b29501d..cb7170b 100644 --- a/src/vanth/daemon.py +++ b/src/vanth/daemon.py @@ -726,6 +726,7 @@ def do_POST(self) -> None: error(self, "Unauthorized", 401) return parsed = urllib.parse.urlparse(self.path) + decision_route = _decision_route(parsed.path) if parsed.path == "/jobs": remote_id = payload.pop("remote_id", None) if remote_id: @@ -733,6 +734,19 @@ def do_POST(self) -> None: else: payload.pop("idempotency_key", None) ok(self, asyncio.run(get_manager().start(**payload))) + elif decision_route is not None: + action, decision_job, token = decision_route + if action == "request": + ok(self, get_manager().request_decision(decision_job, **payload)) + elif action == "resolve": + ok(self, get_manager().resolve_decision(decision_job, token, payload.get("choice", ""))) + else: + ok(self, get_manager().withdraw_decision(decision_job, token)) + elif "decision" in parsed.path.split("/"): + # A decision-looking path that did not match one of the exact + # shapes must not fall through to another job operation (e.g. + # .../decision//pause must not pause the job). + error(self, "Not found", 404) elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/rerun"): remote_id = payload.pop("remote_id", None) if remote_id: @@ -772,14 +786,6 @@ def do_POST(self) -> None: ok(self, get_manager().job_pause(parsed.path.split("/")[2])) elif parsed.path.startswith("/jobs/") and parsed.path.endswith("/resume"): ok(self, get_manager().job_resume(parsed.path.split("/")[2])) - elif (decision_route := _decision_route(parsed.path)) is not None: - action, decision_job, token = decision_route - if action == "request": - ok(self, get_manager().request_decision(decision_job, **payload)) - elif action == "resolve": - ok(self, get_manager().resolve_decision(decision_job, token, payload.get("choice", ""))) - else: - ok(self, get_manager().withdraw_decision(decision_job, token)) elif parsed.path == "/schedules": ok(self, get_manager().create_schedule(**payload)) elif parsed.path.startswith("/schedules/") and parsed.path.endswith("/update"): diff --git a/src/vanth/server.py b/src/vanth/server.py index 4f6d94f..5091295 100644 --- a/src/vanth/server.py +++ b/src/vanth/server.py @@ -336,15 +336,17 @@ def normalize_event_payload(payload: dict[str, Any]) -> dict[str, Any]: DECISION_PENDING = "pending" DECISION_STATUSES = {DECISION_PENDING, "resolved", "withdrawn", "expired"} DEFAULT_DECISION_OPTIONS = ["approve", "deny"] -# Control events for the decision state machine. They are exempt from the -# per-job structured-event cap (like terminal events): the cap exists to bound -# telemetry, and dropping a decision event would break the durability contract -# (no wake, no waitable signal) with no way for a retry to repair it. -DECISION_EVENT_TYPES = {"decision_requested", "decision_resolved", "decision_withdrawn", "decision_expired"} -# Bound decision input so the lifecycle event payload can never hit -# ``max_event_bytes`` (which would strip decision_id/choice and break wake +# Authoritative decision transitions are exempt from the per-job structured +# event cap: the cap bounds telemetry, and dropping a decision event would +# break the durability contract (no wake, no waitable signal) with no way for +# a retry to repair it. The exemption is per-CALL (`exempt_from_cap`), NOT per +# event type: job stdout can emit any event type via AGENT_EVENT, so keying on +# the type would let a job forge `decision_requested` lines and bypass the cap. +# Bound decision input so the lifecycle event payload cannot be truncated by +# `max_event_bytes` (which would strip decision_id/choice and break wake # delivery and `job_wait`). Human-paced, so the limits are generous. MAX_DECISION_PROMPT_CHARS = 10000 +MAX_DECISION_ACTOR_CHARS = 200 MAX_DECISION_OPTIONS = 50 MAX_DECISION_OPTION_CHARS = 200 @@ -2103,6 +2105,7 @@ def _emit( level: str = "info", source: str = "server", mutate: Callable[[sqlite3.Connection], None] | None = None, + exempt_from_cap: bool = False, ) -> dict[str, Any]: self._ensure_open() payload = normalize_event_payload({"type": event_type, "message": message, "data": data or {}, "level": level}) @@ -2116,7 +2119,7 @@ def _emit( for attempt in range(10): try: event = self._emit_transactional( - job_id, payload, data_json, event_type, level, source, message, mutate + job_id, payload, data_json, event_type, level, source, message, mutate, exempt_from_cap ) break except sqlite3.OperationalError as exc: @@ -2142,6 +2145,7 @@ def _emit_transactional( source: str, message: str | None, mutate: Callable[[sqlite3.Connection], None] | None = None, + exempt_from_cap: bool = False, ) -> dict[str, Any]: """Persist a state mutation and its event + deliveries in ONE transaction. @@ -2150,10 +2154,11 @@ def _emit_transactional( event/wake it owes either both land or neither does (a crash can never leave a resolved decision with no event, which a retry could not repair). ``mutate`` may raise ``_DecisionNoOp`` to abort cleanly. + ``exempt_from_cap`` is set only by authoritative decision transitions. """ self.db.execute("BEGIN IMMEDIATE") try: - if event_type not in TERMINAL_STATUSES and event_type not in DECISION_EVENT_TYPES: + if event_type not in TERMINAL_STATUSES and not exempt_from_cap: count = self.db.execute("SELECT COUNT(*) FROM events WHERE job_id=?", (job_id,)).fetchone()[0] if count >= self.max_events_per_job: self.db.rollback() @@ -5917,15 +5922,31 @@ def mutate(db: sqlite3.Connection) -> None: (decision_id, job_id, prompt, json.dumps(normalized, separators=(",", ":")), DECISION_PENDING, created_at, expires_at), ) + event_data = {"decision_id": decision_id, "prompt": prompt, "options": normalized, "expires_at": expires_at} + if not self._decision_event_fits(event_data): + raise ValueError("prompt/options are too large for the decision event payload; shorten them") + self._emit( job_id, "decision_requested", message=prompt, - data={"decision_id": decision_id, "prompt": prompt, "options": normalized, "expires_at": expires_at}, + data=event_data, mutate=mutate, + exempt_from_cap=True, ) return self._reload_decision(decision_id) + def _decision_event_fits(self, data: dict[str, Any]) -> bool: + """Whether a decision lifecycle payload survives ``max_event_bytes``. + + Character limits do not bound the *serialized* size (``json.dumps`` is + ASCII-only, so one astral char is 12 bytes). Truncation would replace the + whole data object, dropping ``decision_id``/``choice`` and breaking the + wake and wait paths, so an oversized payload is rejected up front. + """ + payload = normalize_event_payload({"type": "decision_event", "message": None, "data": data, "level": "info"}) + return len(json.dumps(payload["data"], separators=(",", ":")).encode()) <= self.max_event_bytes + def _decision_for(self, job_id: str, token: str) -> sqlite3.Row: if not isinstance(token, str) or not token: raise ValueError("token must be a non-empty string") @@ -5946,6 +5967,10 @@ def resolve_decision(self, job_id: str, token: str, choice: str, actor: str = "u raise ValueError("choice must be a non-empty string") if not isinstance(token, str) or not token: raise ValueError("token must be a non-empty string") + actor = _normalize_decision_actor(actor) + event_data = {"decision_id": token, "choice": choice, "actor": actor} + if not self._decision_event_fits(event_data): + raise ValueError("decision payload is too large for the event; shorten the choice") def mutate(db: sqlite3.Connection) -> None: row = db.execute("SELECT * FROM decisions WHERE decision_id=?", (token,)).fetchone() @@ -5973,8 +5998,9 @@ def mutate(db: sqlite3.Connection) -> None: job_id, "decision_resolved", message=choice, - data={"decision_id": token, "choice": choice, "actor": actor}, + data=event_data, mutate=mutate, + exempt_from_cap=True, ) if event is not None and event.get("noop"): row = self._decision_for(job_id, token) @@ -5988,6 +6014,10 @@ def withdraw_decision(self, job_id: str, token: str, actor: str = "user") -> dic self._ensure_open() if not isinstance(token, str) or not token: raise ValueError("token must be a non-empty string") + actor = _normalize_decision_actor(actor) + event_data = {"decision_id": token, "actor": actor} + if not self._decision_event_fits(event_data): + raise ValueError("decision payload is too large for the event") def mutate(db: sqlite3.Connection) -> None: row = db.execute("SELECT * FROM decisions WHERE decision_id=?", (token,)).fetchone() @@ -6008,8 +6038,9 @@ def mutate(db: sqlite3.Connection) -> None: job_id, "decision_withdrawn", message="decision withdrawn", - data={"decision_id": token, "actor": actor}, + data=event_data, mutate=mutate, + exempt_from_cap=True, ) return self._reload_decision(token) @@ -6066,6 +6097,7 @@ def mutate(db: sqlite3.Connection, decision_id: str = row["decision_id"], expire message="decision timed out", data={"decision_id": row["decision_id"]}, mutate=mutate, + exempt_from_cap=True, ) @@ -6081,6 +6113,14 @@ def _row_isdue(expires_at: Any, status: Any, now: datetime | None = None) -> boo return parsed <= (now or datetime.now(timezone.utc)) +def _normalize_decision_actor(actor: Any) -> str: + if not isinstance(actor, str) or not actor.strip(): + raise ValueError("actor must be a non-empty string") + if len(actor) > MAX_DECISION_ACTOR_CHARS: + raise ValueError(f"actor must be at most {MAX_DECISION_ACTOR_CHARS} characters") + return actor + + def _normalize_decision_options(options: list[str] | None) -> list[str]: choices = DEFAULT_DECISION_OPTIONS if options is None else options if not isinstance(choices, list) or not choices: diff --git a/tests/test_decisions.py b/tests/test_decisions.py index 2f31fa8..eb77214 100644 --- a/tests/test_decisions.py +++ b/tests/test_decisions.py @@ -387,6 +387,52 @@ def test_cleanup_removes_decisions(tmp_path): manager.close() +def test_request_decision_rejects_payload_over_event_bytes(tmp_path): + """Character limits do not bound serialized bytes: an over-budget payload + must be rejected, never committed and then silently truncated (which would + drop decision_id/choice and break the wake and wait paths).""" + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + # 10000 astral chars serialize to ~120KB (json.dumps is ASCII-only), + # over the default 64KB limit despite passing the char check. + with pytest.raises(ValueError, match="too large"): + manager.request_decision(job_id, "\U0001f600" * 10000) + manager.max_event_bytes = 64 + with pytest.raises(ValueError, match="too large"): + manager.request_decision(job_id, "x" * 50) + assert manager.db.execute("SELECT COUNT(*) FROM decisions").fetchone()[0] == 0 + finally: + manager.close() + + +def test_forged_decision_events_do_not_bypass_cap(tmp_path): + """The cap exemption is per-call, not per event type: job stdout can emit + AGENT_EVENT lines of any type, so a forged decision_requested must still be + capped.""" + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + manager.max_events_per_job = 1 + assert manager._emit(job_id, "decision_requested", data={"spam": True})["persisted"] is False + finally: + manager.close() + + +def test_resolve_rejects_bad_actor(tmp_path): + manager = JobManager(tmp_path / "state") + try: + job_id = running_job(manager) + decision = manager.request_decision(job_id, "Deploy?") + with pytest.raises(ValueError, match="actor"): + manager.resolve_decision(job_id, decision["decision_id"], "approve", actor="x" * 201) + with pytest.raises(ValueError, match="actor"): + manager.resolve_decision(job_id, decision["decision_id"], "approve", actor="") + assert decision_row(manager, decision["decision_id"])["status"] == "pending" + finally: + manager.close() + + def free_port() -> int: with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) @@ -438,6 +484,9 @@ def test_decisions_over_http(tmp_path): f"/jobs/{job['job_id']}/not-a-decision/{third['decision_id']}/resolve", {"choice": "yes"} )["result"] == "error" assert client.get("/decisions", {"status": "pending"})["count"] == 1 + # A decision-namespace path must not fall through to another job op. + assert client.post(f"/jobs/{job['job_id']}/decision/{third['decision_id']}/pause")["error"] == "Not found" + assert client.get("/decisions", {"status": "pending"})["count"] == 1 finally: proc.terminate() proc.wait(timeout=5)