Add human-in-the-loop approval for destructive tool calls

- Introduce ApprovalRequest dataclass and ApprovalHandler callback type
- In prompt permission mode, pause the run with status
  "pending_approval" when no handler is configured, recording the
  approval request in result metadata for resume
- Add --hitl-approval CLI flag that prompts y/N on stderr for each
  unapproved destructive call
- TUI always attaches an approval handler that prompts inline and
  records the approved call_id for the session
- Emit tool_call_pending_approval, tool_call_approved, tool_call_denied,
  and run_paused audit events through the lifecycle
- Track pending_approval status in RunStore heartbeat and summarize

Author: johnteee

README.md

@@ -70,7 +70,7 @@ export OPENCODEZEN_API_KEY=...
 |------|----------|
 | `read-only` | Blocks all destructive tools |
 | `workspace-write` | Allows file writes; blocks shell mutation |
-| `prompt` | Destructive tools require approval token |
+| `prompt` | Destructive tools pause for HITL approval or require an approval token |
 | `allow` | Allows destructive tools for the session |
 | `danger-full-access` | Full access; reserve for trusted automation |
 

docs/cli.md

@@ -384,6 +384,14 @@ teaagent agent run gpt "Create a TODO.md summary" --approve-call-id write-todo-1
 
 The model decision must use the approved `call_id` for that exact destructive tool call. Other destructive calls remain blocked.
 
+For interactive HITL approval during a CLI run, use:
+
+```bash
+teaagent agent run gpt "Create a TODO.md summary" --hitl-approval
+```
+
+Without `--hitl-approval`, an unapproved destructive tool in `prompt` mode returns `pending_approval` with the required `call_id`. Re-run with `--approve-call-id <call_id>` or use `agent resume` with the same approval token.
+
 Prefer explicit permission modes for regular use:
 
 ```bash
@@ -397,7 +405,7 @@ Permission modes:
 
 - `read-only`: blocks every destructive tool.
 - `workspace-write`: allows file write/patch/hash-edit tools, blocks shell mutation.
-- `prompt`: destructive tools require an approval token.
+- `prompt`: destructive tools pause for HITL approval or require an approval token.
 - `allow`: allows destructive tools for the session.
 - `danger-full-access`: allows destructive tools; reserve for trusted automation.
 
@@ -431,6 +439,7 @@ teaagent agent resume gpt <run_id> --root /path/to/repo --approve-call-id write-
 Inside TUI:
 
 ```text
+ask write TODO.md       # prompts y/N when a destructive call is proposed
 approve write-todo-1
 approvals
 unapprove write-todo-1

teaagent/__init__.py

@@ -55,7 +55,13 @@
 from teaagent.rag import Document, InMemoryRetriever, agentic_retrieve
 from teaagent.readiness import ReadinessReport, assess_managed_agent_readiness
 from teaagent.run_store import RunStore, RunSummary
-from teaagent.runner import AgentRunner, Decision, FinalAnswer, ToolRequest
+from teaagent.runner import (
+    AgentRunner,
+    ApprovalRequest,
+    Decision,
+    FinalAnswer,
+    ToolRequest,
+)
 from teaagent.skill_review import SkillReviewResult, review_skill
 from teaagent.stateless_mcp import (
     StatelessMCPRequest,
@@ -75,6 +81,7 @@
     "AIBOMManifest",
     "AgentRunner",
     "ApprovalPolicy",
+    "ApprovalRequest",
     "AuditLogger",
     "ChatAgentConfig",
     "ClarificationResult",

teaagent/chat_agent.py

@@ -19,7 +19,7 @@
     parse_model_decision,
 )
 from teaagent.run_store import RunStore
-from teaagent.runner import AgentRunner, Decision, RunResult
+from teaagent.runner import AgentRunner, ApprovalHandler, Decision, RunResult
 from teaagent.tools import ToolAnnotations, ToolRegistry
 from teaagent.workspace_tools import build_workspace_tool_registry
 
@@ -39,6 +39,7 @@ class ChatAgentConfig:
     heartbeat_seconds: float = 0.0
     stream: bool = False
     on_chunk: Optional[Callable[[str], None]] = None
+    approval_handler: Optional[ApprovalHandler] = None
 
     @classmethod
     def from_root(cls, root: str | Path, **kwargs) -> "ChatAgentConfig":
@@ -121,6 +122,7 @@ def run_chat_agent(
             allow_all_destructive=config.allow_destructive,
             permission_mode=config.permission_mode,
         ),
+        approval_handler=config.approval_handler,
         compactor=ContextCompactor(memory_keys=("task_spec", "memories")),
     )
     run_id = uuid4().hex

teaagent/cli.py

@@ -2,6 +2,7 @@
 
 import argparse
 import json
+import sys
 from typing import Any, Optional
 
 from teaagent import __version__
@@ -25,6 +26,7 @@
 from teaagent.policy import PermissionMode, parse_permission_mode
 from teaagent.preflight import preflight
 from teaagent.run_store import RunStore
+from teaagent.runner import ApprovalRequest, RunResult
 from teaagent.tui import run_tui
 from teaagent.ultrawork import UltraworkStore
 from teaagent.workspace_tools import build_workspace_tool_registry
@@ -104,6 +106,11 @@ def build_parser() -> argparse.ArgumentParser:
         default=[],
         help="Approve one exact destructive tool call id. Can be repeated.",
     )
+    agent_run.add_argument(
+        "--hitl-approval",
+        action="store_true",
+        help="Prompt before executing unapproved destructive tool calls in prompt permission mode.",
+    )
     agent_run.add_argument(
         "--permission-mode",
         choices=[mode.value for mode in PermissionMode],
@@ -166,6 +173,7 @@ def build_parser() -> argparse.ArgumentParser:
         default=[],
         help="Approve one exact destructive tool call id. Can be repeated.",
     )
+    agent_resume.add_argument("--hitl-approval", action="store_true", help="Prompt before unapproved destructive tool calls.")
     agent_resume.add_argument(
         "--permission-mode",
         choices=[mode.value for mode in PermissionMode],
@@ -364,6 +372,7 @@ def _execute_agent_task(args: argparse.Namespace, task: str, *, resumed_from: Op
     adapter = create_llm_adapter(args.provider, model=selected_model)
     store = RunStore(args.root)
     audit = store.audit_logger()
+    approval_handler = cli_approval_handler if args.hitl_approval else None
     result = run_chat_agent(
         task=task,
         adapter=adapter,
@@ -378,24 +387,42 @@ def _execute_agent_task(args: argparse.Namespace, task: str, *, resumed_from: Op
             enable_subagent=args.subagent,
             max_subagent_depth=args.max_subagent_depth,
             heartbeat_seconds=args.heartbeat,
+            approval_handler=approval_handler,
         ),
         audit=audit,
         task_spec=task_spec,
     )
     store.logger_for_result(result, audit)
+    payload = run_result_payload(result, routing=routing.to_dict() if routing else None)
+    if resumed_from:
+        payload["resumed_from"] = resumed_from
+        payload["task"] = task
+    print_json(payload)
+    return 0 if result.status == "completed" else 1
+
+
+def run_result_payload(result: RunResult, *, routing: Optional[dict[str, Any]]) -> dict[str, Any]:
     payload: dict[str, Any] = {
         "run_id": result.run_id,
         "status": result.status,
         "iterations": result.iterations,
         "tool_calls": result.tool_calls,
-        "routing": routing.to_dict() if routing else None,
+        "routing": routing,
         "final_answer": result.final_answer.content if result.final_answer else None,
     }
-    if resumed_from:
-        payload["resumed_from"] = resumed_from
-        payload["task"] = task
-    print_json(payload)
-    return 0 if result.status == "completed" else 1
+    if "approval" in result.metadata:
+        payload["approval"] = result.metadata["approval"]
+    return payload
+
+
+def cli_approval_handler(request: ApprovalRequest) -> bool:
+    print(
+        json.dumps({"status": "approval_required", "approval": request.to_dict()}, sort_keys=True),
+        file=sys.stderr,
+    )
+    print(f"Approve destructive tool call {request.call_id} ({request.tool_name})? [y/N] ", end="", file=sys.stderr)
+    answer = input()
+    return answer.strip().lower() in {"y", "yes"}
 
 
 def agent_preflight_command(args: argparse.Namespace) -> int:

teaagent/run_store.py

@@ -83,6 +83,8 @@ def heartbeat_for_run(self, run_id: str) -> dict[str, Any]:
                 last_heartbeat = event
             elif event_type in {"run_completed", "run_failed"}:
                 terminal_status = "completed" if event_type == "run_completed" else f"failed:{event.get('payload', {}).get('category', 'unknown')}"
+            elif event_type == "run_paused":
+                terminal_status = event.get("payload", {}).get("status", "paused")
         return {
             "run_id": run_id,
             "status": terminal_status or "running",
@@ -118,6 +120,8 @@ def summarize(self, path: Path) -> Optional[RunSummary]:
                 final_answer = payload.get("answer")
             elif event_type == "run_failed":
                 status = f"failed:{payload.get('category', 'unknown')}"
+            elif event_type == "run_paused":
+                status = payload.get("status", "paused")
         return RunSummary(
             run_id=run_id,
             task=task,

teaagent/runner.py

@@ -14,7 +14,7 @@
     ErrorCategory,
     ToolPermissionError,
 )
-from teaagent.policy import ApprovalPolicy
+from teaagent.policy import ApprovalPolicy, PermissionMode
 from teaagent.tools import ToolRegistry
 
 
@@ -35,13 +35,35 @@ class FinalAnswer:
 DecisionFn = Callable[[dict[str, Any]], Decision]
 
 
+@dataclass(frozen=True)
+class ApprovalRequest:
+    call_id: str
+    tool_name: str
+    arguments: dict[str, Any]
+    reason: str
+    annotations: dict[str, bool]
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "call_id": self.call_id,
+            "tool_name": self.tool_name,
+            "arguments": self.arguments,
+            "reason": self.reason,
+            "annotations": self.annotations,
+        }
+
+
+ApprovalHandler = Callable[[ApprovalRequest], bool]
+
+
 @dataclass(frozen=True)
 class RunResult:
     run_id: str
     final_answer: Optional[FinalAnswer]
     iterations: int
     tool_calls: int
     status: str
+    metadata: dict[str, Any] = field(default_factory=dict)
 
 
 class AgentRunner:
@@ -52,6 +74,7 @@ def __init__(
         audit: AuditLogger,
         budget: Optional[RunBudget] = None,
         approval_policy: Optional[ApprovalPolicy] = None,
+        approval_handler: Optional[ApprovalHandler] = None,
         compactor: Optional[ContextCompactor] = None,
         compact_after_observations: int = 20,
     ) -> None:
@@ -60,6 +83,7 @@ def __init__(
         self.budget = budget or RunBudget()
         self.budget.validate()
         self.approval_policy = approval_policy or ApprovalPolicy()
+        self.approval_handler = approval_handler
         self.compactor = compactor
         self.compact_after_observations = compact_after_observations
 
@@ -103,38 +127,76 @@ def run(self, *, task: str, decide: DecisionFn, run_id: Optional[str] = None) ->
                     raise BudgetExceededError("tool-call budget exceeded")
 
                 tool = self.registry.get(decision.tool_name)
+                annotations = {
+                    "read_only": tool.annotations.read_only,
+                    "destructive": tool.annotations.destructive,
+                    "idempotent": tool.annotations.idempotent,
+                }
                 try:
                     self.approval_policy.assert_allowed(
                         tool_name=decision.tool_name,
                         call_id=decision.call_id,
                         destructive=tool.annotations.destructive,
                     )
                 except ToolPermissionError as exc:
-                    self.audit.record(
-                        "tool_call_blocked",
-                        current_run_id,
+                    approval_request = ApprovalRequest(
                         call_id=decision.call_id,
                         tool_name=decision.tool_name,
                         arguments=decision.arguments,
                         reason=str(exc),
-                        annotations={
-                            "read_only": tool.annotations.read_only,
-                            "destructive": tool.annotations.destructive,
-                            "idempotent": tool.annotations.idempotent,
-                        },
+                        annotations=annotations,
                     )
-                    raise
+                    if self._can_request_approval(tool.annotations.destructive):
+                        self.audit.record(
+                            "tool_call_pending_approval",
+                            current_run_id,
+                            **approval_request.to_dict(),
+                        )
+                        if self.approval_handler is None:
+                            self.audit.record(
+                                "run_paused",
+                                current_run_id,
+                                status="pending_approval",
+                                approval=approval_request.to_dict(),
+                                cost_cents=cost_cents,
+                            )
+                            return RunResult(
+                                run_id=current_run_id,
+                                final_answer=None,
+                                iterations=iterations,
+                                tool_calls=tool_calls,
+                                status="pending_approval",
+                                metadata={"approval": approval_request.to_dict()},
+                            )
+                        if self.approval_handler(approval_request):
+                            self.audit.record(
+                                "tool_call_approved",
+                                current_run_id,
+                                call_id=decision.call_id,
+                                tool_name=decision.tool_name,
+                            )
+                        else:
+                            self.audit.record(
+                                "tool_call_denied",
+                                current_run_id,
+                                call_id=decision.call_id,
+                                tool_name=decision.tool_name,
+                            )
+                            raise
+                    else:
+                        self.audit.record(
+                            "tool_call_blocked",
+                            current_run_id,
+                            **approval_request.to_dict(),
+                        )
+                        raise
                 self.audit.record(
                     "tool_call_started",
                     current_run_id,
                     call_id=decision.call_id,
                     tool_name=decision.tool_name,
                     arguments=decision.arguments,
-                    annotations={
-                        "read_only": tool.annotations.read_only,
-                        "destructive": tool.annotations.destructive,
-                        "idempotent": tool.annotations.idempotent,
-                    },
+                    annotations=annotations,
                 )
                 result = self.registry.execute(decision.tool_name, decision.arguments)
                 tool_calls += 1
@@ -196,3 +258,6 @@ def run(self, *, task: str, decide: DecisionFn, run_id: Optional[str] = None) ->
             tool_calls=tool_calls,
             status=f"failed:{ErrorCategory.MODEL_LOGIC}",
         )
+
+    def _can_request_approval(self, destructive: bool) -> bool:
+        return destructive and self.approval_policy.permission_mode == PermissionMode.PROMPT