feat(wo): add optional pending-approval gate for work-order writes

docs/mcp-servers.md

@@ -161,6 +161,11 @@ Priorities are integers `1`–`5` (`wopriority`), 1 being most urgent.
   `cancel_workorder` apply their target status from *any* current state; they do not enforce
   `WAPPR → APPR → COMP`. Scenarios that need to test illegal transitions must assert on the
   sequence themselves.
+- Separately, these same three tools can be gated behind an operator approval step: set
+  `AOB_REQUIRE_APPROVAL=1` and the first call to any of them returns an `action_id` instead of
+  running, until an operator approves it with `scripts/wo_pending.py` and the agent repeats the
+  identical call. Off by default; see `governance.py`. This does not add the status-transition
+  check above — it adds a human checkpoint in front of whatever transition the agent requests.
 - Every status change stamps `status_date`; `close_workorder` additionally sets `actlabhrs` and
   `actfinish`.
 - `notes` (create/update), `reason` (cancel), and `resolution_notes` (close) all write the same

scripts/wo_pending.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Operator CLI for the WO server's pending-approval gate (AOB_REQUIRE_APPROVAL).
+
+Lists, approves, or denies pending approve_workorder / close_workorder /
+cancel_workorder calls. Talks directly to CouchDB using the same
+COUCHDB_URL / COUCHDB_USERNAME / COUCHDB_PASSWORD / WO_PENDING_DBNAME env vars
+the WO server itself uses — see src/servers/wo/governance.py.
+
+Usage:
+    wo_pending.py list
+    wo_pending.py approve <action_id> [--by NAME]
+    wo_pending.py deny <action_id> [--by NAME]
+
+Exit code 0 = command completed, 1 = action not found / nothing to do,
+2 = the command itself failed (bad args, store unreachable).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
+
+from servers.wo.couch import CouchClient, DatabaseUnavailable
+from servers.wo.governance import PendingStore
+
+
+def _pending_db() -> CouchClient:
+    return CouchClient(
+        base_url=os.environ.get("COUCHDB_URL", "http://localhost:5984"),
+        db=os.environ.get("WO_PENDING_DBNAME", "workorder_pending"),
+        username=os.environ.get("COUCHDB_USERNAME", "admin"),
+        password=os.environ.get("COUCHDB_PASSWORD", "password"),
+    )
+
+
+async def _list(store: PendingStore) -> int:
+    rows = await store.list_pending()
+    if not rows:
+        print("No pending actions.")
+        return 0
+    for row in rows:
+        print(
+            f"{row['action_id']}  {row['tool']:<18} wonum={row['wonum']} "
+            f"site={row['site_id']} args={row.get('args')} "
+            f"created={row['created_at']}"
+        )
+    return 0
+
+
+async def _decide(
+    store: PendingStore, action_id: str, approve: bool, by: str | None
+) -> int:
+    doc = await store.decide(action_id, approve=approve, decided_by=by)
+    if doc is None:
+        print(f"No pending action '{action_id}' (already decided, or no such id).")
+        return 1
+    verb = "approved" if approve else "denied"
+    print(
+        f"{verb}: {action_id} ({doc['tool']} on wonum={doc['wonum']} site={doc['site_id']})"
+    )
+    if approve:
+        print("The agent must repeat the identical call once to consume this approval.")
+    return 0
+
+
+async def _run(args: argparse.Namespace) -> int:
+    db = _pending_db()
+    store = PendingStore(db)
+    try:
+        if args.command == "list":
+            return await _list(store)
+        if args.command == "approve":
+            return await _decide(store, args.action_id, approve=True, by=args.by)
+        if args.command == "deny":
+            return await _decide(store, args.action_id, approve=False, by=args.by)
+    except DatabaseUnavailable:
+        print("Pending-action store is unreachable (CouchDB down or misconfigured).")
+        return 2
+    finally:
+        await db.aclose()
+    return 2
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    sub.add_parser("list", help="List all pending actions.")
+
+    approve = sub.add_parser("approve", help="Approve a pending action.")
+    approve.add_argument("action_id")
+    approve.add_argument("--by", default=None, help="Operator identity to record.")
+
+    deny = sub.add_parser("deny", help="Deny a pending action.")
+    deny.add_argument("action_id")
+    deny.add_argument("--by", default=None, help="Operator identity to record.")
+
+    args = parser.parse_args()
+    return asyncio.run(_run(args))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/servers/wo/governance.py

@@ -0,0 +1,175 @@
+"""Optional pending-approval gate for destructive work-order actions.
+
+Off by default — set ``AOB_REQUIRE_APPROVAL=1`` to turn it on (same opt-in
+convention as ``AOB_READONLY``). When enabled, ``approve_workorder``,
+``close_workorder``, and ``cancel_workorder`` do not execute on the first
+call. That call is recorded as a *pending* action keyed by a SHA-256 digest
+of the tool name and its arguments, and the tool returns an ``action_id``
+instead of a result. An operator approves or denies the pending action out
+of band (see ``scripts/wo_pending.py``); the agent must then repeat the
+identical call. The matching digest consumes the approval exactly once and
+the action runs. A retry with different arguments (a different `wonum`, a
+different `reason`) is a new digest and a new pending decision — approving
+one call never silently approves a different one.
+
+If the pending store itself is unreachable, the gate fails closed (denies)
+rather than letting the call through — the same posture ``AOB_READONLY``
+already takes for reads.
+
+This adds no new storage layer: pending records are plain CouchDB documents
+in a dedicated database, read/written through the same ``CouchClient`` duck
+type ``workorders.py`` already uses, so unit tests bind the same kind of
+in-memory fake as ``test_workorders.py``.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from .couch import CouchClient, DatabaseUnavailable
+from .envelope import error
+
+PENDING = "pending"
+APPROVED = "approved"
+DENIED = "denied"
+CONSUMED = "consumed"
+UNAVAILABLE = "unavailable"
+
+REQUIRE_APPROVAL_ENV = "AOB_REQUIRE_APPROVAL"
+
+# The three status-transition tools docs/mcp-servers.md documents as having
+# no guard today. Scoped deliberately to these; read tools and additive
+# writes (assign_technician, update_workorder) are left alone.
+GATED_TOOLS = ("approve_workorder", "close_workorder", "cancel_workorder")
+
+
+def approval_required() -> bool:
+    return os.environ.get(REQUIRE_APPROVAL_ENV) == "1"
+
+
+def args_digest(tool: str, wonum: str, site_id: str, args: Dict[str, Any]) -> str:
+    """Stable SHA-256 over the tool name and every argument of the call.
+
+    Same call, same digest. Change any argument and the digest changes, so an
+    approval granted for one call can never cover a different one.
+    """
+    payload = json.dumps(
+        {"tool": tool, "wonum": wonum, "site_id": site_id, "args": args},
+        sort_keys=True,
+        default=str,
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _doc_id(action_id: str) -> str:
+    return f"pending:{action_id}"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@dataclass
+class PendingStore:
+    """CouchDB-backed pending/approved/denied/consumed state for gated tools."""
+
+    db: CouchClient
+
+    async def request(
+        self, tool: str, wonum: str, site_id: str, args: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Create, or return the existing, pending record for this exact call."""
+        digest = args_digest(tool, wonum, site_id, args)
+        action_id = digest[:16]
+        doc = await self.db.get(_doc_id(action_id))
+        if doc is None:
+            doc = {
+                "_id": _doc_id(action_id),
+                "action_id": action_id,
+                "tool": tool,
+                "wonum": wonum,
+                "site_id": site_id,
+                "args": args,
+                "args_digest": digest,
+                "status": PENDING,
+                "created_at": _now(),
+                "updated_at": _now(),
+            }
+            await self.db.put(doc)
+        return doc
+
+    async def decide(
+        self, action_id: str, approve: bool, decided_by: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """Operator approves or denies a pending action.
+
+        Returns the updated record, or None if no such action is pending
+        (already decided, consumed, or never existed).
+        """
+        doc = await self.db.get(_doc_id(action_id))
+        if doc is None or doc.get("status") != PENDING:
+            return None
+        doc["status"] = APPROVED if approve else DENIED
+        doc["decided_by"] = decided_by
+        doc["updated_at"] = _now()
+        await self.db.put(doc)
+        return doc
+
+    async def list_pending(self) -> list[Dict[str, Any]]:
+        return await self.db.find({"status": PENDING})
+
+    async def try_consume(
+        self, tool: str, wonum: str, site_id: str, args: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Check whether this exact call has a one-shot approval, and consume it.
+
+        Returns ``{"ok": True, ...}`` when the caller may proceed, or
+        ``{"ok": False, "action_id": ..., "status": ...}`` otherwise — including
+        when the store itself is unreachable, which fails closed instead of
+        letting the call through silently.
+        """
+        digest = args_digest(tool, wonum, site_id, args)
+        action_id = digest[:16]
+        try:
+            doc = await self.db.get(_doc_id(action_id))
+        except DatabaseUnavailable:
+            return {"ok": False, "action_id": action_id, "status": UNAVAILABLE}
+
+        if doc is None:
+            created = await self.request(tool, wonum, site_id, args)
+            return {"ok": False, "action_id": created["action_id"], "status": PENDING}
+
+        if doc.get("status") != APPROVED:
+            return {"ok": False, "action_id": action_id, "status": doc.get("status")}
+
+        doc["status"] = CONSUMED
+        doc["updated_at"] = _now()
+        await self.db.put(doc)
+        return {"ok": True, "action_id": action_id, "status": CONSUMED}
+
+
+def pending_response(tool: str, action_id: str, status: str) -> Dict[str, Any]:
+    """The ``{success: False, ...}`` envelope returned instead of executing."""
+    if status == PENDING:
+        msg = (
+            f"'{tool}' requires operator approval before it can run "
+            f"(action_id={action_id}). Ask an operator to review it with "
+            "scripts/wo_pending.py, then repeat this exact call once approved."
+        )
+    elif status == DENIED:
+        msg = f"'{tool}' was denied by an operator (action_id={action_id})."
+    elif status == UNAVAILABLE:
+        msg = (
+            f"'{tool}' requires operator approval, but the approval store is "
+            "unreachable; refusing to run rather than skipping the check."
+        )
+    else:
+        msg = (
+            f"'{tool}' is not approved to run (action_id={action_id}, status={status})."
+        )
+    return error(msg, "APPROVAL_REQUIRED")