diff --git a/docs/mcp-servers.md b/docs/mcp-servers.md index ebecc147c..013ac096b 100644 --- a/docs/mcp-servers.md +++ b/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 diff --git a/scripts/wo_pending.py b/scripts/wo_pending.py new file mode 100755 index 000000000..5ea3ba541 --- /dev/null +++ b/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 [--by NAME] + wo_pending.py deny [--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()) diff --git a/src/servers/wo/governance.py b/src/servers/wo/governance.py new file mode 100644 index 000000000..3686c0932 --- /dev/null +++ b/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") diff --git a/src/servers/wo/main.py b/src/servers/wo/main.py index 55029f858..93cb5d7e0 100644 --- a/src/servers/wo/main.py +++ b/src/servers/wo/main.py @@ -9,6 +9,11 @@ Env (AssetOpsBench-compatible): COUCHDB_URL, COUCHDB_USERNAME, COUCHDB_PASSWORD, WO_DBNAME, FAILURE_CODE_DBNAME. Set AOB_READONLY=1 to expose only the read tools. + +Set AOB_REQUIRE_APPROVAL=1 to gate approve_workorder/close_workorder/cancel_workorder +behind an operator pending-approval step instead of running them immediately (off by +default; see governance.py and scripts/wo_pending.py). Uses WO_PENDING_DBNAME +(default "workorder_pending") for the pending-action records. """ import logging @@ -19,6 +24,7 @@ from mcp.server.fastmcp import FastMCP from mcp.types import ToolAnnotations +from . import governance as gov from . import workorders as wo from .couch import CouchClient from .models import ( @@ -52,6 +58,7 @@ _db: Optional[CouchClient] = None _fcc_db: Optional[CouchClient] = None +_pending_db: Optional[CouchClient] = None def db() -> CouchClient: @@ -78,6 +85,18 @@ def failure_code_db() -> CouchClient: return _fcc_db +def pending_db() -> CouchClient: + global _pending_db + if _pending_db is None: + _pending_db = 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"), + ) + return _pending_db + + # --------------------------------------------------------------------------- # # MCP-facing tool functions: call the Maximo-derived logic (dict envelope), then # convert to the typed Pydantic result model (AssetOpsBench output convention). @@ -90,6 +109,26 @@ def _failed(res: Dict[str, Any]) -> Optional[ErrorResult]: return None +async def _check_approval( + tool: str, wonum: str, site_id: str, args: Dict[str, Any] +) -> Optional[ErrorResult]: + """Gate a status-transition tool behind AOB_REQUIRE_APPROVAL. + + Returns None when the call may proceed (gate off, or this exact call was + already approved and is being consumed now). Returns an ErrorResult when + it must not — including when the pending store is unreachable, which + fails closed rather than letting the call through. + """ + if not gov.approval_required(): + return None + result = await gov.PendingStore(pending_db()).try_consume( + tool, wonum, site_id, args + ) + if result["ok"]: + return None + return _failed(gov.pending_response(tool, result["action_id"], result["status"])) + + def _mutation( res: Dict[str, Any], verb: str ) -> Union[WorkOrderMutationResult, ErrorResult]: @@ -335,7 +374,10 @@ async def update_workorder( async def approve_workorder( wonum: str, site_id: str ) -> Union[WorkOrderMutationResult, ErrorResult]: - """Approve a work order (-> APPR).""" + """Approve a work order (-> APPR). Gated by AOB_REQUIRE_APPROVAL; see governance.py.""" + gate = await _check_approval("approve_workorder", wonum, site_id, {}) + if gate is not None: + return gate return _mutation( await wo.approve_workorder(db(), wonum, site_id), "approved (APPR)" ) @@ -363,7 +405,16 @@ async def close_workorder( failure_code: Optional[str] = None, resolution_notes: Optional[str] = None, ) -> Union[WorkOrderMutationResult, ErrorResult]: - """Close a work order (-> COMP) with actuals and resolution.""" + """Close a work order (-> COMP) with actuals and resolution. + Gated by AOB_REQUIRE_APPROVAL; see governance.py.""" + args = { + "actual_hours": actual_hours, + "failure_code": failure_code, + "resolution_notes": resolution_notes, + } + gate = await _check_approval("close_workorder", wonum, site_id, args) + if gate is not None: + return gate res = await wo.close_workorder( db(), wonum, site_id, actual_hours, failure_code, resolution_notes ) @@ -373,7 +424,10 @@ async def close_workorder( async def cancel_workorder( wonum: str, site_id: str, reason: Optional[str] = None ) -> Union[WorkOrderMutationResult, ErrorResult]: - """Cancel a work order (-> CAN).""" + """Cancel a work order (-> CAN). Gated by AOB_REQUIRE_APPROVAL; see governance.py.""" + gate = await _check_approval("cancel_workorder", wonum, site_id, {"reason": reason}) + if gate is not None: + return gate return _mutation( await wo.cancel_workorder(db(), wonum, site_id, reason), "cancelled (CAN)" ) @@ -411,7 +465,13 @@ async def cancel_workorder( destructiveHint=False, idempotentHint=True, openWorldHint=True, - ) + ), + # Status-transition tools: irreversible under the current data model (no + # status-transition guard — see docs/mcp-servers.md), so clients that honor + # MCP's destructiveHint can choose to confirm with a human before calling. + approve_workorder: ToolAnnotations(destructiveHint=True, idempotentHint=False), + close_workorder: ToolAnnotations(destructiveHint=True, idempotentHint=False), + cancel_workorder: ToolAnnotations(destructiveHint=True, idempotentHint=False), } for _fn, _title in _TOOLS: mcp.tool(title=_title, annotations=_TOOL_ANNOTATIONS.get(_fn))(_fn) diff --git a/src/servers/wo/tests/test_governance.py b/src/servers/wo/tests/test_governance.py new file mode 100644 index 000000000..0fc67c2be --- /dev/null +++ b/src/servers/wo/tests/test_governance.py @@ -0,0 +1,221 @@ +"""Unit tests for the optional pending-approval gate (governance.py). + +Uses the same in-memory fake pattern as test_workorders.py so no CouchDB is +needed. These tests cover the one-shot approval contract directly; the +request/response wiring through main.py's gated tools is exercised in +test_workorders.py's existing mutation tests with AOB_REQUIRE_APPROVAL unset +(default path, unaffected by this change). +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from servers.wo import governance as gov +from servers.wo import main +from servers.wo import workorders as wo +from servers.wo.couch import DatabaseUnavailable +from servers.wo.models import ErrorResult, WorkOrderMutationResult + + +class FakeCouch: + """Minimal get/put/find fake — the subset governance.py actually uses.""" + + def __init__(self): + self.docs = {} + + async def get(self, doc_id): + d = self.docs.get(doc_id) + return dict(d) if d else None + + async def put(self, doc): + self.docs[doc["_id"]] = dict(doc) + return {"id": doc["_id"], "rev": "1-x"} + + async def find(self, selector, **kw): + return [ + dict(d) + for d in self.docs.values() + if all(d.get(k) == v for k, v in selector.items()) + ] + + +class UnavailableCouch: + """A store that always raises, to exercise the fail-closed path.""" + + async def get(self, doc_id): + raise DatabaseUnavailable() + + async def put(self, doc): + raise DatabaseUnavailable() + + +def test_approval_required_is_off_by_default(monkeypatch): + monkeypatch.delenv(gov.REQUIRE_APPROVAL_ENV, raising=False) + assert gov.approval_required() is False + + +def test_approval_required_honors_env(monkeypatch): + monkeypatch.setenv(gov.REQUIRE_APPROVAL_ENV, "1") + assert gov.approval_required() is True + + +def test_digest_changes_with_any_argument(): + base = gov.args_digest("cancel_workorder", "1000045", "MAIN", {"reason": "dup"}) + diff_wonum = gov.args_digest( + "cancel_workorder", "1000046", "MAIN", {"reason": "dup"} + ) + diff_reason = gov.args_digest( + "cancel_workorder", "1000045", "MAIN", {"reason": "x"} + ) + assert base != diff_wonum + assert base != diff_reason + + +@pytest.mark.anyio +async def test_first_call_creates_pending_and_refuses(): + store = gov.PendingStore(FakeCouch()) + result = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + assert result["ok"] is False + assert result["status"] == gov.PENDING + assert len(result["action_id"]) == 16 + + +@pytest.mark.anyio +async def test_repeat_call_before_decision_stays_pending_no_duplicate(): + db = FakeCouch() + store = gov.PendingStore(db) + first = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + second = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + assert first["action_id"] == second["action_id"] + assert second["status"] == gov.PENDING + assert len(db.docs) == 1 # one record, not re-created on every retry + + +@pytest.mark.anyio +async def test_approve_then_matching_retry_consumes_once(): + db = FakeCouch() + store = gov.PendingStore(db) + pending = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + decided = await store.decide(pending["action_id"], approve=True, decided_by="alice") + assert decided["status"] == gov.APPROVED + + first_retry = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + assert first_retry["ok"] is True + assert first_retry["status"] == gov.CONSUMED + + # Same call again: already consumed, not approved anymore -> refused again. + second_retry = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + assert second_retry["ok"] is False + assert second_retry["status"] == gov.CONSUMED + + +@pytest.mark.anyio +async def test_approval_does_not_cover_a_different_call(): + """Approving $500 must not approve a later call for $5,000,000.""" + db = FakeCouch() + store = gov.PendingStore(db) + pending = await store.try_consume( + "close_workorder", "1000045", "MAIN", {"actual_hours": 2.0} + ) + await store.decide(pending["action_id"], approve=True) + + different_call = await store.try_consume( + "close_workorder", "1000045", "MAIN", {"actual_hours": 200.0} + ) + assert different_call["ok"] is False + assert different_call["status"] == gov.PENDING # a brand new decision, not denied + + +@pytest.mark.anyio +async def test_deny_refuses_and_does_not_consume(): + db = FakeCouch() + store = gov.PendingStore(db) + pending = await store.try_consume( + "cancel_workorder", "1000045", "MAIN", {"reason": "x"} + ) + await store.decide(pending["action_id"], approve=False) + + retry = await store.try_consume( + "cancel_workorder", "1000045", "MAIN", {"reason": "x"} + ) + assert retry["ok"] is False + assert retry["status"] == gov.DENIED + + +@pytest.mark.anyio +async def test_decide_on_unknown_action_returns_none(): + store = gov.PendingStore(FakeCouch()) + assert await store.decide("not-a-real-id", approve=True) is None + + +@pytest.mark.anyio +async def test_unreachable_store_fails_closed(): + store = gov.PendingStore(UnavailableCouch()) + result = await store.try_consume("approve_workorder", "1000045", "MAIN", {}) + assert result["ok"] is False + assert result["status"] == gov.UNAVAILABLE + + +def test_pending_response_shapes_match_error_envelope(): + resp = gov.pending_response("approve_workorder", "abc123", gov.PENDING) + assert resp["success"] is False + assert resp["error_code"] == "APPROVAL_REQUIRED" + assert "abc123" in resp["error"] + + +# --------------------------------------------------------------------------- # +# End-to-end through main.py's actual tools, not just the governance module +# in isolation — proves the gate really blocks and really unblocks a real +# approve_workorder call, and that it is a no-op when the env var is unset. +# --------------------------------------------------------------------------- # +T0 = datetime(2020, 4, 28, 9, 0, 0, tzinfo=timezone.utc) + + +async def _seeded_wo_db() -> FakeCouch: + db = FakeCouch() + await wo.create_workorder( + db, + description="Investigate Chiller 6 anomaly", + asset_num="CHILLER6", + site_id="MAIN", + wonum="1000045", + now=T0, + ) + return db + + +@pytest.mark.anyio +async def test_gate_is_a_noop_when_env_var_unset(monkeypatch): + monkeypatch.delenv(gov.REQUIRE_APPROVAL_ENV, raising=False) + monkeypatch.setattr(main, "_db", await _seeded_wo_db()) + + result = await main.approve_workorder("1000045", "MAIN") + + assert isinstance(result, WorkOrderMutationResult) + assert result.status == "APPR" + + +@pytest.mark.anyio +async def test_gate_blocks_then_unblocks_a_real_approve_call(monkeypatch): + monkeypatch.setenv(gov.REQUIRE_APPROVAL_ENV, "1") + monkeypatch.setattr(main, "_db", await _seeded_wo_db()) + monkeypatch.setattr(main, "_pending_db", FakeCouch()) + + first = await main.approve_workorder("1000045", "MAIN") + assert isinstance(first, ErrorResult) + assert "action_id" in first.error or "requires operator approval" in first.error + + # The work order must still be WAPPR — the gated call never ran wo.approve_workorder. + still_open = await wo.get_workorder(main._db, "1000045", "MAIN") + assert still_open["data"]["status"] == "WAPPR" + + store = gov.PendingStore(main._pending_db) + [pending] = await store.list_pending() + await store.decide(pending["action_id"], approve=True, decided_by="dhaval") + + second = await main.approve_workorder("1000045", "MAIN") + assert isinstance(second, WorkOrderMutationResult) + assert second.status == "APPR"