Add durable approval/decision primitive

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/<token>/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

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

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_<hex>`); 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.

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) {

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,13 +726,27 @@ 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:
                     ok(self, _remote_submit(remote_id, "job.start", _remote_payload(payload)))
                 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/<token>/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:

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()