diff --git a/CHANGELOG.md b/CHANGELOG.md index ef09f68..d0fd24d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,53 @@ 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. 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). 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` / + `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); the Go mirrored + `LatestSchemaVersion` and cross-language fixture move to 16 with it. + ## 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..d1a088b 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,61 @@ 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; 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; 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 +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/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/src/vanth/daemon.py b/src/vanth/daemon.py index 529ff58..cb7170b 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" @@ -587,6 +605,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"): @@ -702,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: @@ -709,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: 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..5091295 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 @@ -328,8 +328,31 @@ 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"] +# 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 + + +class _DecisionNoOp(Exception): + """Internal: a decision mutation found nothing to change (idempotent retry).""" def resolve_wake_target_identity( @@ -787,6 +810,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") @@ -2080,6 +2104,8 @@ def _emit( data: dict[str, Any] | None = None, 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}) @@ -2093,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 + job_id, payload, data_json, event_type, level, source, message, mutate, exempt_from_cap ) break except sqlite3.OperationalError as exc: @@ -2118,10 +2144,21 @@ def _emit_transactional( level: str, 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. + + ``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. + ``exempt_from_cap`` is set only by authoritative decision transitions. + """ self.db.execute("BEGIN IMMEDIATE") try: - if event_type not in TERMINAL_STATUSES: + 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() @@ -2140,6 +2177,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() @@ -5386,6 +5441,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() @@ -5799,6 +5855,290 @@ 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: + 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, + 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. + + 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") + 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 + + 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), + ) + + 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=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") + 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 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). + + 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() + 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") + 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() + 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(), token, DECISION_PENDING), + ) + if cursor.rowcount != 1: + raise ValueError("decision changed concurrently") + + event = self._emit( + job_id, + "decision_resolved", + message=choice, + 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) + 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() + 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() + 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(), token, DECISION_PENDING), + ) + if cursor.rowcount != 1: + raise ValueError("decision changed concurrently") + + self._emit( + job_id, + "decision_withdrawn", + message="decision withdrawn", + data=event_data, + mutate=mutate, + exempt_from_cap=True, + ) + 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() + 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(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). + + 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 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 _row_isdue(row["expires_at"], DECISION_PENDING, now): + continue + + 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, + exempt_from_cap=True, + ) + + +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_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: + 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 mcp = FastMCP("vanth") @@ -5956,6 +6296,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/testdata/state/jobs.sqlite b/testdata/state/jobs.sqlite index 279e859..6da121d 100644 Binary files a/testdata/state/jobs.sqlite and b/testdata/state/jobs.sqlite differ diff --git a/tests/test_decisions.py b/tests/test_decisions.py new file mode 100644 index 0000000..eb77214 --- /dev/null +++ b/tests/test_decisions.py @@ -0,0 +1,492 @@ +"""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 json +import os +import socket +import subprocess +import sys +import threading +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 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: + 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 + # 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 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"], 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: + 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) + # 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: + 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 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 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)) + 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 + + # 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 + # 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)