fix(approval): reduce prompt friction and make approval cards answerable forever

Approval friction (harmless commands kept prompting):
- Persist "Allow for this session" grants to approval_allowlist.yaml under a
  new sessions scope (capped LRU), hydrated lazily, so they survive `opc ui`
  restarts and re-entering the session instead of living only in memory.
- Safe-prefix matching now accepts compound read-only commands: every segment
  must match a safe prefix, and fd-duplication / /dev/null redirections
  (2>&1, 2>/dev/null) no longer disqualify a command; real write redirections
  (>, >>, <) still do. Default safe prefixes gain common read-only commands
  (cd, cat, head, grep, git log, ...).
- First-use approval now gates only MEDIUM+ risk; heuristically LOW actions
  proceed without a card.
- Shell-substitution detection flags eval/source only at command position of a
  segment (no more false positives on `grep source file`); $(...) and
  backticks still flag anywhere.

Approval card timeout redesign (deferred decisions):
- The card's structured approval context (action, allowlist patterns, scopes)
  now travels through the escalation event into the persisted card metadata.
- Timeout without a default action no longer marks the card timed out, and the
  session-detail reconciler no longer stales deferred-capable cards: the card
  stays pending and clickable indefinitely, including across restarts.
- Clicking after the inline wait expired applies the allowlist grant
  (approve-once grants the exact command at session scope), resolves the card,
  and rewrites the reply to target the parked AWAITING_HUMAN checkpoint so the
  task resumes through the normal message pipeline and the retried command
  auto-approves. With no parked checkpoint the grant still lands and a helper
  reply explains the state.

Verified: approval engine suite (40) incl. new deferred-decision and
compound-command tests, ws_handler + runtime suites green, real escalated
commands from project 999 replayed against the user's config now auto-approve
while pip install / $(...) / rm -rf / write redirects still prompt.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

opc/core/config.py

@@ -994,6 +994,12 @@ class AutonomyConfig(BaseModel):
     safe_command_prefixes: list[str] = Field(default_factory=lambda: [
         "ls", "pwd", "echo", "rg", "find", "git status", "git diff", "python -V",
         "python3 -V", "node -v", "npm -v", "curl", "wget", "yt-dlp", "aria2c", "ffmpeg",
+        # Read-only commands agents chain constantly; each segment of a compound
+        # command must match one of these for the whole command to stay LOW risk.
+        "cd", "cat", "head", "tail", "grep", "wc", "sort", "uniq", "cut", "tr",
+        "stat", "file", "which", "date", "du", "df", "tree", "basename", "dirname",
+        "realpath", "readlink", "uname", "nproc", "whoami", "hostname", "git log",
+        "git show", "git rev-parse",
     ])
     permissions_v2: PermissionsV2Config = Field(default_factory=PermissionsV2Config)
 

opc/layer2_organization/approval.py

@@ -40,6 +40,10 @@ from opc.llm.retry import LLMRetryError, call_llm_json_with_retry
 
 
 _SHELL_CONTROL_TOKENS = {"&&", "||", ";", "|", "&"}
+# Redirections that cannot write to a real file: fd duplication (2>&1, >&2)
+# and discarding output into /dev/null. Everything else keeps counting as a
+# redirection for the safe-prefix check.
+_SAFE_REDIRECTION_RE = re.compile(r"(?:\d?>>?\s*/dev/null\b|\d?>&\d|&>>?\s*/dev/null\b)")
 _LOW_RISK_SHELL_PREFIXES = set(ACQUISITION_SHELL_PREFIXES)
 _EXTERNAL_AGENT_DIRECT_HUMAN_MARKERS = (
     "--dangerously-bypass-approvals-and-sandbox",
@@ -495,7 +499,11 @@ class ApprovalEngine:
             metadata=metadata,
         )
 
-        if tool_requires_allowlist:
+        if tool_requires_allowlist and heuristic.risk_level != RiskLevel.LOW:
+            # First-use approval exists to catch unfamiliar, potentially risky
+            # actions. Actions the heuristic already classified LOW (read-only
+            # safe-prefix shell commands, clean tool arguments) proceed without
+            # a card; MEDIUM and above still require the human gate.
             decision = self._force_first_use_approval(heuristic)
         elif external_direct_prompt_reason:
             decision = ApprovalDecision(
@@ -666,7 +674,20 @@ class ApprovalEngine:
         session_scope_id = self._approval_session_scope_id(task)
         if not session_scope_id:
             return None
-        scope = self._session_allowlist.get(session_scope_id, {})
+        scope = self._session_allowlist.get(session_scope_id)
+        if scope is None:
+            # Hydrate from the persisted allowlist so "Allow for this session"
+            # grants survive `opc ui` restarts and re-entering the session.
+            scope = {}
+            if self.allowlist:
+                try:
+                    scope = self.allowlist.session_scope(session_scope_id)
+                except Exception:
+                    logger.opt(exception=True).debug(
+                        "Failed to hydrate persisted session allowlist; using empty scope"
+                    )
+                    scope = {}
+            self._session_allowlist[session_scope_id] = scope
         patterns = ApprovalAllowlistManager._scope_patterns(scope, action_kind, action_name)
         if not patterns:
             return None
@@ -716,7 +737,21 @@ class ApprovalEngine:
         action_name: str,
         patterns: list[str],
     ) -> list[str]:
-        session_scope_id = self._approval_session_scope_id(task)
+        return self._add_session_patterns_by_scope(
+            session_scope_id=self._approval_session_scope_id(task),
+            action_kind=action_kind,
+            action_name=action_name,
+            patterns=patterns,
+        )
+
+    def _add_session_patterns_by_scope(
+        self,
+        *,
+        session_scope_id: str,
+        action_kind: str,
+        action_name: str,
+        patterns: list[str],
+    ) -> list[str]:
         if not session_scope_id:
             return []
         normalized_patterns = ApprovalAllowlistManager._normalize_pattern_list(patterns)
@@ -735,6 +770,13 @@ class ApprovalEngine:
             existing.append(pattern)
             added.append(pattern)
         action_bucket[action_name] = existing
+        if added and self.allowlist:
+            try:
+                self.allowlist.add_session_patterns(session_scope_id, action_kind, action_name, added)
+            except Exception:
+                logger.opt(exception=True).debug(
+                    "Failed to persist session allowlist patterns; grant remains in-memory only"
+                )
         return added
 
     def _tool_requires_first_use_approval(
@@ -875,29 +917,48 @@ class ApprovalEngine:
         text = str(command or "")
         if "$(" in text or "`" in text:
             return True
-        # ``eval`` / ``source`` let a "safe" prefix execute an arbitrary follow-up arg.
-        tokens = text.split()
-        if tokens and tokens[0] in {"eval", "source", "."}:
-            return True
-        return any(tok in {"eval", "source"} for tok in tokens)
+        # ``eval`` / ``source`` let a "safe" prefix execute an arbitrary follow-up
+        # arg, but only when they are the command itself. As ordinary arguments
+        # (``grep source config.py``) they are inert; flagging them there only
+        # produces false approval prompts.
+        for tokens in self._split_shell_command_segments(text):
+            if tokens and tokens[0] in {"eval", "source", "."}:
+                return True
+        return False
 
     def _command_matches_safe_prefix(self, command: str, prefixes: list[str]) -> bool:
         cleaned = " ".join(str(command or "").split()).strip()
-        if not cleaned or self._command_has_redirection(cleaned):
+        if not cleaned:
             return False
-        if self._command_has_shell_substitution(cleaned):
+        # Discarding stderr or duplicating fds writes nothing, and agents
+        # habitually append `2>&1` / `2>/dev/null` to read-only commands; strip
+        # those before the redirection check so they alone do not disqualify a
+        # command. Anything else touching `>`/`>>`/`<` still fails.
+        sanitized = _SAFE_REDIRECTION_RE.sub(" ", cleaned)
+        if self._command_has_redirection(sanitized):
             return False
-        commands, command_prefixes = self._extract_shell_command_targets(cleaned)
-        if len(commands) != 1 or len(command_prefixes) != 1:
+        if self._command_has_shell_substitution(sanitized):
             return False
-        prefix = command_prefixes[0].casefold()
-        for item in prefixes:
-            candidate = str(item or "").strip().casefold()
-            if not candidate:
-                continue
-            if prefix == candidate or prefix.startswith(f"{candidate} "):
-                return True
-        return False
+        commands, command_prefixes = self._extract_shell_command_targets(sanitized)
+        if not commands or not command_prefixes:
+            return False
+        candidates = [
+            candidate
+            for candidate in (str(item or "").strip().casefold() for item in prefixes)
+            if candidate
+        ]
+        if not candidates:
+            return False
+        # A compound command (`ls a && echo --- && ls b`, `git status | head`)
+        # qualifies only when every segment independently matches a safe prefix.
+        for prefix in command_prefixes:
+            normalized_prefix = prefix.casefold()
+            if not any(
+                normalized_prefix == candidate or normalized_prefix.startswith(f"{candidate} ")
+                for candidate in candidates
+            ):
+                return False
+        return True
 
     def _is_low_risk_shell_first_use_exempt(self, action_name: str, metadata: dict[str, Any]) -> bool:
         if action_name != "shell_exec":
@@ -1503,11 +1564,25 @@ class ApprovalEngine:
                 {"id": "always_project", "label": "Always allow for this project"},
                 {"id": "always_global", "label": "Always allow globally"},
             ])
+        approval_context = {
+            "action_kind": action_kind,
+            "action_name": action_name,
+            "project_id": str(task.project_id or "") if task else "",
+            "session_scope_id": self._approval_session_scope_id(task),
+            "allowlist_enabled": allowlist_enabled,
+            "allowlist_patterns": list(allowlist_patterns),
+            "candidates": self._build_allowlist_candidates(
+                action_kind=action_kind,
+                action_name=action_name,
+                metadata=metadata,
+            ),
+        }
         reply = await self.escalation.escalate_decision(
             task,
             question,
             options,
             default_action=None,
+            context=approval_context,
         )
         if reply is None:
             return False, ApprovalDecision(
@@ -1574,6 +1649,104 @@ class ApprovalEngine:
             metadata=result_metadata,
         )
 
+    def apply_deferred_escalation_decision(
+        self,
+        reply: str,
+        context: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Apply a decision clicked on an approval card after its inline wait
+        expired (the blocked task has parked on AWAITING_HUMAN by then).
+
+        Persists the same allowlist grant the live path would have applied, so
+        the re-run of the blocked action passes automatically. ``context`` is
+        the ``approval_context`` the card was created with. Returns a summary
+        {approved, scope, patterns} for UI messaging.
+        """
+        normalized_reply = str(reply or "").strip()
+        context = dict(context or {})
+        action_kind = str(context.get("action_kind", "") or "").strip()
+        action_name = str(context.get("action_name", "") or "").strip()
+        project_id = str(context.get("project_id", "") or "").strip() or None
+        session_scope_id = str(context.get("session_scope_id", "") or "").strip()
+        allowlist_enabled = bool(context.get("allowlist_enabled", False))
+        allowlist_patterns = [
+            str(item).strip() for item in list(context.get("allowlist_patterns", []) or [])
+            if str(item).strip()
+        ]
+        exact_candidates = [
+            str(item).strip() for item in list(context.get("candidates", []) or [])
+            if str(item).strip()
+        ]
+        if not allowlist_enabled and normalized_reply in {"approve_session", "always_project", "always_global"}:
+            normalized_reply = "approve_once"
+        approved = normalized_reply in {"approve_once", "approve_session", "always_project", "always_global"}
+
+        saved_patterns: list[str] = []
+        scope: str | None = None
+        if normalized_reply == "approve_session" and session_scope_id and allowlist_patterns:
+            saved_patterns = self._add_session_patterns_by_scope(
+                session_scope_id=session_scope_id,
+                action_kind=action_kind,
+                action_name=action_name,
+                patterns=allowlist_patterns,
+            )
+            scope = f"session:{session_scope_id}"
+        elif normalized_reply == "approve_once" and session_scope_id:
+            # No one-shot grant store exists; the narrowest durable equivalent
+            # is a session grant for the exact blocked command(s), so the
+            # resumed run passes without widening approval to the whole family.
+            once_patterns = exact_candidates or allowlist_patterns
+            if once_patterns:
+                saved_patterns = self._add_session_patterns_by_scope(
+                    session_scope_id=session_scope_id,
+                    action_kind=action_kind,
+                    action_name=action_name,
+                    patterns=once_patterns,
+                )
+                scope = f"session:{session_scope_id}"
+        elif normalized_reply == "always_project" and self.allowlist and project_id and allowlist_patterns:
+            saved_patterns = self.allowlist.add_patterns(
+                action_kind=action_kind,
+                action_name=action_name,
+                patterns=allowlist_patterns,
+                project_id=project_id,
+            )
+            scope = f"project:{project_id}"
+        elif normalized_reply == "always_global" and self.allowlist and allowlist_patterns:
+            saved_patterns = self.allowlist.add_patterns(
+                action_kind=action_kind,
+                action_name=action_name,
+                patterns=allowlist_patterns,
+                project_id=None,
+            )
+            scope = "global"
+
+        if action_name:
+            try:
+                self.preferences.record_autonomy_feedback(
+                    action_name=action_name,
+                    approved=approved,
+                    project_id=project_id if normalized_reply == "always_project" else None,
+                    explicit=normalized_reply in {"approve_session", "always_project", "always_global"},
+                    notes=(
+                        "User approved via deferred escalation card."
+                        if approved
+                        else "User denied via deferred escalation card."
+                    ),
+                )
+            except Exception:
+                logger.opt(exception=True).debug(
+                    "Failed to record autonomy feedback for deferred escalation decision"
+                )
+
+        return {
+            "approved": approved,
+            "reply": normalized_reply,
+            "scope": scope,
+            "patterns": saved_patterns,
+            "action_name": action_name,
+        }
+
     async def _record(
         self,
         task: Task | None,

opc/layer2_organization/escalation.py

@@ -37,10 +37,14 @@ class EscalationEngine:
         message: str,
         options: list[dict[str, str]] | None = None,
         default_action: str | None = None,
+        context: dict[str, Any] | None = None,
     ) -> str | None:
         """Escalate to the user and wait for a reply.
 
         Returns the user's reply or the default action on timeout.
+        ``context`` carries structured approval data (action, allowlist
+        patterns, scopes) into the UI card so a decision can still be applied
+        after this inline wait has expired.
         """
         # Use a unique escalation id per prompt so repeated approvals for the
         # same task do not alias to older UI cards or stale pending state.
@@ -55,6 +59,7 @@ class EscalationEngine:
                 "message": message,
                 "options": options or [],
                 "default_action": default_action,
+                "approval_context": dict(context or {}),
             },
         ))
 
@@ -97,6 +102,7 @@ class EscalationEngine:
         question: str,
         options: list[dict[str, str]],
         default_action: str | None = None,
+        context: dict[str, Any] | None = None,
     ) -> str | None:
         metadata = dict(getattr(task, "metadata", {}) or {})
         execution_mode = str(metadata.get("execution_mode", "") or "").strip()
@@ -118,6 +124,7 @@ class EscalationEngine:
             message=f"[DECISION NEEDED] Task: {task_label}\n{question}",
             options=options,
             default_action=default_action,
+            context=context,
         )
 
     async def escalate_risk(self, task: Task, risk_description: str) -> str | None:

opc/layer5_memory/approval_allowlist.py

@@ -23,6 +23,7 @@ def _empty_payload() -> dict[str, Any]:
         "version": 1,
         "global": _empty_scope(),
         "projects": {},
+        "sessions": {},
     }
 
 
@@ -111,6 +112,58 @@ class ApprovalAllowlistManager:
             self.save(payload)
         return added
 
+    # "Allow for this session" grants used to live only in ApprovalEngine
+    # memory, so a `opc ui` restart or re-entering the session re-prompted for
+    # commands the user had already approved. They are now persisted here,
+    # keyed by the session scope id, capped to the most recent entries.
+    _MAX_SESSION_SCOPES = 200
+
+    def session_scope(self, session_id: str) -> dict[str, dict[str, list[str]]]:
+        key = str(session_id or "").strip()
+        if not key:
+            return _empty_scope()
+        payload = self.load()
+        return self._normalize_scope(payload["sessions"].get(key, {}))
+
+    def add_session_patterns(
+        self,
+        session_id: str,
+        action_kind: str,
+        action_name: str,
+        patterns: list[str],
+    ) -> list[str]:
+        key = str(session_id or "").strip()
+        normalized_patterns = [
+            self._normalize_pattern(pattern)
+            for pattern in patterns
+            if self._normalize_pattern(pattern)
+        ]
+        if not key or not normalized_patterns:
+            return []
+
+        payload = self.load()
+        sessions = payload["sessions"]
+        # Re-inserting moves an active session to the newest position so the
+        # recency cap below always evicts the longest-idle session first.
+        scope = self._normalize_scope(sessions.pop(key, {}))
+        sessions[key] = scope
+
+        action_bucket = scope.setdefault(action_kind, {})
+        existing = self._normalize_pattern_list(action_bucket.get(action_name, []))
+        added: list[str] = []
+        for pattern in normalized_patterns:
+            if pattern in existing:
+                continue
+            existing.append(pattern)
+            added.append(pattern)
+        action_bucket[action_name] = existing
+
+        while len(sessions) > self._MAX_SESSION_SCOPES:
+            sessions.pop(next(iter(sessions)))
+        if added:
+            self.save(payload)
+        return added
+
     def reset(self, project_id: str | None = None) -> None:
         payload = self.load()
         if project_id:
@@ -180,6 +233,14 @@ class ApprovalAllowlistManager:
                 if not key:
                     continue
                 normalized["projects"][key] = ApprovalAllowlistManager._normalize_scope(scope)
+
+        sessions = data.get("sessions", {})
+        if isinstance(sessions, dict):
+            for session_id, scope in sessions.items():
+                key = str(session_id).strip()
+                if not key:
+                    continue
+                normalized["sessions"][key] = ApprovalAllowlistManager._normalize_scope(scope)
         return normalized
 
     @staticmethod

opc/plugins/office_ui/ws_handler.py

@@ -4889,6 +4889,12 @@ class WSHandler:
             "project_id": pid,
             "approval_group_key": esc_record.get("approval_group_key"),
         }
+        approval_context = dict(p.get("approval_context") or {})
+        if approval_context:
+            # Persisted with the card so a click AFTER the inline wait expired
+            # (or after a restart) can still apply the same allowlist grant and
+            # resume the parked task.
+            esc_meta["approval_context"] = approval_context
         if is_task_mode:
             esc_meta["execution_mode"] = "task_mode"
             esc_meta["permission_group_key"] = esc_record.get("approval_group_key")
@@ -4942,6 +4948,137 @@ class WSHandler:
                 return True
         return False
 
+    async def _find_pending_approval_park_checkpoint(
+        self,
+        engine: Any,
+        task_id: str,
+        project_id: str,
+    ) -> Any | None:
+        """Locate the pending checkpoint a tool-approval timeout parked on.
+
+        When an approval card's inline wait expires, the blocked runtime task
+        returns AWAITING_HUMAN and the engine saves a durable pause checkpoint
+        (task mode: ``task_user_input``; company mode: ``company_work_item_gate``).
+        A later click on the card resumes execution through that checkpoint.
+        """
+        source_task_id = str(task_id or "").strip()
+        if not source_task_id:
+            return None
+        store = getattr(engine, "store", None)
+        getter = getattr(store, "get_pending_checkpoints", None)
+        if not callable(getter):
+            return None
+        try:
+            pending = await getter(project_id=project_id)
+        except Exception:
+            logger.opt(exception=True).debug(
+                "Failed to load pending checkpoints for deferred escalation resume"
+            )
+            return None
+        candidates = []
+        for checkpoint in pending or []:
+            if str(getattr(checkpoint, "checkpoint_type", "") or "") not in {
+                "task_user_input",
+                "company_work_item_gate",
+            }:
+                continue
+            payload = dict(getattr(checkpoint, "payload", {}) or {})
+            linked_ids = {
+                str(payload.get("task_id") or "").strip(),
+                str(payload.get("waiting_task_id") or "").strip(),
+                str(getattr(checkpoint, "task_id", "") or "").strip(),
+            }
+            linked_ids.update(str(item or "").strip() for item in list(payload.get("task_ids", []) or []))
+            if source_task_id in linked_ids:
+                candidates.append(checkpoint)
+        if not candidates:
+            return None
+
+        def _checkpoint_timestamp(checkpoint: Any) -> float:
+            created = getattr(checkpoint, "created_at", None)
+            try:
+                return float(created.timestamp())
+            except (AttributeError, TypeError, ValueError, OSError):
+                return 0.0
+
+        return max(candidates, key=_checkpoint_timestamp)
+
+    async def _resolve_deferred_escalation_click(
+        self,
+        *,
+        engine: Any,
+        project_id: str,
+        channel_id: str,
+        checkpoint_id: str,
+        card_meta: dict[str, Any],
+        option_id: str,
+    ) -> dict[str, Any]:
+        """Apply a decision clicked on an approval card whose inline wait has
+        expired: persist the allowlist grant, resolve the card, and hand back
+        either a flow-through rewrite (resume the parked task through the
+        normal message pipeline) or a helper reply when nothing is parked."""
+        approval_context = dict(card_meta.get("approval_context") or {})
+        summary: dict[str, Any] = {
+            "approved": option_id in {"approve_once", "approve_session", "always_project", "always_global"},
+            "scope": None,
+        }
+        approval_engine = getattr(engine, "approval_engine", None)
+        apply_decision = getattr(approval_engine, "apply_deferred_escalation_decision", None)
+        if callable(apply_decision):
+            try:
+                summary = apply_decision(option_id, approval_context)
+            except Exception:
+                logger.opt(exception=True).warning(
+                    "Deferred approval grant failed; resuming the parked task without a new allowlist entry"
+                )
+        await self._mark_human_escalation_checkpoint_status(
+            checkpoint_id,
+            status="resolved",
+            project_id=project_id,
+            channel_id=channel_id,
+            reply=option_id,
+            reason="deferred_decision",
+        )
+
+        source_task_id = str(
+            card_meta.get("source_task_id") or card_meta.get("task_id") or ""
+        ).strip()
+        park_checkpoint = await self._find_pending_approval_park_checkpoint(
+            engine, source_task_id, project_id
+        )
+        action_name = str(approval_context.get("action_name", "") or "").strip() or "action"
+        approved = bool(summary.get("approved"))
+        scope = str(summary.get("scope") or "").strip()
+        if park_checkpoint is None:
+            return {
+                "action": "reply",
+                "text": (
+                    f"Decision `{option_id}` recorded"
+                    + (f"; allowlist updated ({scope})" if approved and scope else "")
+                    + ". No parked task is currently waiting on this approval — if the runtime "
+                    "is still transitioning, the grant applies on its next attempt."
+                ),
+            }
+        if approved:
+            scope_note = f" (allowlisted: {scope})" if scope else ""
+            crafted = (
+                f"Approval decision: {option_id}. The previously blocked `{action_name}` action "
+                f"is now permitted{scope_note}. Re-run it and continue the task."
+            )
+        else:
+            crafted = (
+                f"Approval decision: deny. Do not run the blocked `{action_name}` action; "
+                "choose an alternative approach or report the limitation to your manager."
+            )
+        return {
+            "action": "flow_through",
+            "content": crafted,
+            "reply_metadata": {
+                "response_to_checkpoint_id": str(getattr(park_checkpoint, "checkpoint_id", "") or ""),
+                "response_to_checkpoint_type": str(getattr(park_checkpoint, "checkpoint_type", "") or ""),
+            },
+        }
+
     async def _mark_human_escalation_checkpoint_status(
         self,
         escalation_id: str,
@@ -4997,11 +5134,18 @@ class WSHandler:
         if not escalation_id:
             return
         if event.event_type == "escalation_timeout":
+            default_action = str(payload.get("default_action", "") or "").strip() or None
+            if default_action is None:
+                # No default was applied on timeout — the decision is still the
+                # user's to make. The task parks on AWAITING_HUMAN and the card
+                # stays pending; clicking it later applies the decision and
+                # resumes the parked task (deferred approval path).
+                return
             await self._mark_human_escalation_checkpoint_status(
                 escalation_id,
                 status="timeout",
                 project_id=project_id,
-                default_action=str(payload.get("default_action", "") or "").strip() or None,
+                default_action=default_action,
                 reason="timeout",
             )
             return
@@ -5060,6 +5204,11 @@ class WSHandler:
                 project_id=project_id,
             ):
                 continue
+            if isinstance(metadata.get("approval_context"), dict) and metadata.get("approval_context"):
+                # Deferred-capable approval card: it stays answerable after the
+                # inline wait expired or across restarts, so a missing pending
+                # future does NOT make it stale.
+                continue
             updated = await self._mark_human_escalation_checkpoint_status(
                 escalation_id,
                 status="stale",
@@ -5625,6 +5774,7 @@ class WSHandler:
             and bool(explicit_checkpoint_id or explicit_escalation_id)
         )
         if stale_human_escalation:
+            handled_as_deferred = False
             if _looks_like_escalation_reply(content):
                 stale_checkpoint_id = explicit_escalation_id or explicit_checkpoint_id
                 # Duplicate clicks on an approval card that was JUST resolved
@@ -5645,6 +5795,7 @@ class WSHandler:
                     )
                 card_meta = dict((card or {}).get("metadata", {}) or {})
                 card_status = str(card_meta.get("checkpoint_status", "") or "").strip().lower()
+                helper_text: str | None = None
                 if card_status in {"resolved", "responded"}:
                     resolution_reply = str(
                         card_meta.get("checkpoint_resolution_reply", "") or ""
@@ -5655,37 +5806,70 @@ class WSHandler:
                         + ". No further action is needed."
                     )
                 else:
-                    await self._mark_human_escalation_checkpoint_status(
-                        stale_checkpoint_id,
-                        status="stale",
-                        project_id=pid,
+                    approval_context = card_meta.get("approval_context")
+                    deferred_option = (
+                        _normalize_escalation_reply(content, list(card_meta.get("options") or []))
+                        if isinstance(approval_context, dict) and approval_context
+                        else None
+                    )
+                    if deferred_option:
+                        # The inline wait expired (or the server restarted), but
+                        # the decision is still the user's to make: apply the
+                        # grant, resolve the card, and resume the parked task.
+                        outcome = await self._resolve_deferred_escalation_click(
+                            engine=run_engine,
+                            project_id=pid,
+                            channel_id=channel_id,
+                            checkpoint_id=stale_checkpoint_id,
+                            card_meta=card_meta,
+                            option_id=deferred_option,
+                        )
+                        if outcome.get("action") == "flow_through":
+                            content = str(outcome.get("content") or content)
+                            for key in (
+                                "response_to_checkpoint_id",
+                                "response_to_checkpoint_type",
+                                "response_to_escalation_id",
+                            ):
+                                reply_metadata.pop(key, None)
+                            reply_metadata.update(dict(outcome.get("reply_metadata") or {}))
+                            handled_as_deferred = True
+                        else:
+                            helper_text = str(outcome.get("text") or "Decision recorded.")
+                    else:
+                        await self._mark_human_escalation_checkpoint_status(
+                            stale_checkpoint_id,
+                            status="stale",
+                            project_id=pid,
+                            channel_id=channel_id,
+                            reason="reply_to_inactive_escalation",
+                        )
+                        helper_text = (
+                            "That approval request is no longer active. "
+                            "The approval card has been marked inactive in the session history."
+                        )
+                if helper_text is not None:
+                    if await self._recent_identical_helper_exists(
+                        channel_id, helper_text, project_id=pid
+                    ):
+                        return
+                    helper = await self.chat_store.insert_message(
                         channel_id=channel_id,
-                        reason="reply_to_inactive_escalation",
+                        sender="assistant",
+                        sender_name="OPC",
+                        content=helper_text,
+                        project_id=pid,
+                        metadata={"type": "system"},
                     )
-                    helper_text = (
-                        "That approval request is no longer active. "
-                        "The approval card has been marked inactive in the session history."
-                    )
-                if await self._recent_identical_helper_exists(
-                    channel_id, helper_text, project_id=pid
-                ):
+                    await self.broadcast({"type": "session_message", "payload": helper})
                     return
-                helper = await self.chat_store.insert_message(
-                    channel_id=channel_id,
-                    sender="assistant",
-                    sender_name="OPC",
-                    content=helper_text,
-                    project_id=pid,
-                    metadata={"type": "system"},
-                )
-                await self.broadcast({"type": "session_message", "payload": helper})
-                return
-            for key in (
-                "response_to_checkpoint_id",
-                "response_to_checkpoint_type",
-                "response_to_escalation_id",
-            ):
-                reply_metadata.pop(key, None)
+            if not handled_as_deferred:
+                for key in (
+                    "response_to_checkpoint_id",
+                    "response_to_checkpoint_type",
+                    "response_to_escalation_id",
+                ):
+                    reply_metadata.pop(key, None)
 
         if (
             explicit_checkpoint_type == "company_delivery_feedback"

tests/test_approval_engine.py

@@ -157,6 +157,47 @@ class ApprovalEngineHeuristicTests(unittest.TestCase):
             self.assertFalse(self.engine._command_has_shell_substitution(payload))
             self.assertTrue(self.engine._command_matches_safe_prefix(payload, prefixes))
 
+    def test_compound_readonly_command_matches_safe_prefix(self) -> None:
+        # Agents habitually chain read-only commands and discard stderr; that
+        # alone must not disqualify the command from LOW risk.
+        prefixes = list(self.engine.config.safe_command_prefixes)
+        payloads = [
+            'ls -la /a 2>&1 && echo "---" && ls -la /b 2>/dev/null',
+            "cd /repo && git status --short 2>&1 | head -20",
+            "git log --oneline -5 | head -3",
+            "grep -rn pattern src | wc -l",
+        ]
+        for payload in payloads:
+            self.assertTrue(
+                self.engine._command_matches_safe_prefix(payload, prefixes),
+                f"compound read-only command must stay safe: {payload}",
+            )
+
+    def test_write_redirection_or_unsafe_segment_still_not_safe(self) -> None:
+        prefixes = list(self.engine.config.safe_command_prefixes)
+        payloads = [
+            "ls -la /a > out.txt",
+            "echo hi >> log.txt",
+            "cat notes.md | tee copy.md",
+            "ls /tmp && rm -rf /tmp/x",
+            "sort data.txt < input.txt",
+        ]
+        for payload in payloads:
+            self.assertFalse(
+                self.engine._command_matches_safe_prefix(payload, prefixes),
+                f"unsafe command must not match a safe prefix: {payload}",
+            )
+
+    def test_source_eval_flag_only_at_command_position(self) -> None:
+        # As arguments these words are inert; flagging them produced false
+        # approval prompts (e.g. `grep source config.py`).
+        self.assertFalse(self.engine._command_has_shell_substitution("grep source config.py"))
+        self.assertFalse(self.engine._command_has_shell_substitution("echo eval"))
+        # At command position they still count, in any segment.
+        self.assertTrue(self.engine._command_has_shell_substitution("source ./env.sh"))
+        self.assertTrue(self.engine._command_has_shell_substitution("ls && source ./env.sh"))
+        self.assertTrue(self.engine._command_has_shell_substitution("eval $CMD"))
+
     def test_external_prompt_text_still_escalates_for_destructive_command(self) -> None:
         metadata = {
             "prompt_text": "Approve command: rm -rf /tmp/demo",
@@ -220,11 +261,13 @@ class _EscalationStub:
         self.reply = reply
         self.calls: list[tuple[str, list[dict]]] = []
         self.default_actions: list[str | None] = []
+        self.contexts: list[dict | None] = []
 
-    async def escalate_decision(self, task, question, options, default_action=None):
+    async def escalate_decision(self, task, question, options, default_action=None, context=None):
         _ = task
         self.calls.append((question, options))
         self.default_actions.append(default_action)
+        self.contexts.append(context)
         return self.reply
 
 
@@ -379,25 +422,25 @@ class ApprovalEngineAllowlistTests(unittest.IsolatedAsyncioTestCase):
                 escalation=escalation,
                 config=AutonomyConfig(),
             )
-            task = Task(title="Write config", project_id="demo")
+            task = Task(title="Install deps", project_id="demo")
 
             approved, decision = await engine.authorize_tool_call(
                 task=task,
-                tool_name="file_write",
-                arguments={"path": "/tmp/demo.txt", "content": "hello"},
+                tool_name="shell_exec",
+                arguments={"command": "pip install requests"},
             )
 
             self.assertTrue(approved)
             self.assertEqual(decision.policy_source, "human_escalation")
             self.assertEqual(len(escalation.calls), 1)
 
-            rules = ApprovalAllowlistManager(opc_home).list_patterns("tool", "file_write", project_id="demo")
-            self.assertEqual(rules, ["*"])
+            rules = ApprovalAllowlistManager(opc_home).list_patterns("tool", "shell_exec", project_id="demo")
+            self.assertEqual(rules, ["pip install"])
 
             approved, decision = await engine.authorize_tool_call(
                 task=task,
-                tool_name="file_write",
-                arguments={"path": "/tmp/another.txt", "content": "world"},
+                tool_name="shell_exec",
+                arguments={"command": "pip install flask"},
             )
 
             self.assertTrue(approved)
@@ -471,19 +514,19 @@ class ApprovalEngineAllowlistTests(unittest.IsolatedAsyncioTestCase):
             approved, decision = await engine.authorize_tool_call(
                 task=task,
                 tool_name="shell_exec",
-                arguments={"command": "git status --short"},
+                arguments={"command": "git commit -m demo"},
             )
 
             self.assertTrue(approved)
             self.assertEqual(decision.policy_source, "human_escalation")
 
             rules = ApprovalAllowlistManager(opc_home).list_patterns("tool", "shell_exec")
-            self.assertEqual(rules, ["git status"])
+            self.assertEqual(rules, ["git commit"])
 
             approved, decision = await engine.authorize_tool_call(
                 task=task,
                 tool_name="shell_exec",
-                arguments={"command": "git status -sb"},
+                arguments={"command": "git commit -m again"},
             )
 
             self.assertTrue(approved)
@@ -528,6 +571,154 @@ class ApprovalEngineAllowlistTests(unittest.IsolatedAsyncioTestCase):
             self.assertEqual(decision.policy_source, "heuristic")
             self.assertEqual(len(escalation.calls), 0)
 
+    async def test_low_risk_readonly_command_skips_first_use_prompt(self) -> None:
+        with _workspace_tempdir() as opc_home:
+            prefs = PreferenceManager(opc_home)
+            escalation = _EscalationStub("approve_once")
+            engine = ApprovalEngine(
+                llm=_LLMStub(),
+                store=_StoreStub(),
+                preferences=prefs,
+                memory=_MemoryStub(),
+                escalation=escalation,
+                config=AutonomyConfig(),
+            )
+            task = Task(title="Inspect repo", project_id="demo")
+
+            approved, decision = await engine.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "cd /repo && git status --short 2>&1 | head -20"},
+            )
+
+            self.assertTrue(approved)
+            self.assertEqual(decision.action, ApprovalAction.AUTO_APPROVE)
+            self.assertEqual(decision.risk_level, RiskLevel.LOW)
+            self.assertEqual(len(escalation.calls), 0)
+
+    async def test_session_allowlist_persists_across_engine_restart(self) -> None:
+        with _workspace_tempdir() as opc_home:
+            prefs = PreferenceManager(opc_home)
+            escalation = _EscalationStub("approve_session")
+            config = AutonomyConfig()
+            engine = ApprovalEngine(
+                llm=_LLMStub(),
+                store=_StoreStub(),
+                preferences=prefs,
+                memory=_MemoryStub(),
+                escalation=escalation,
+                config=config,
+            )
+            task = Task(title="Check repo", project_id="demo", session_id="sess-persist")
+
+            approved, decision = await engine.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "git commit -m demo"},
+            )
+
+            self.assertTrue(approved)
+            self.assertEqual(decision.policy_source, "human_escalation")
+            self.assertEqual(len(escalation.calls), 1)
+
+            # A fresh engine over the same OPC home simulates an `opc ui`
+            # restart: the session grant must survive, not re-prompt.
+            engine_restarted = ApprovalEngine(
+                llm=_LLMStub(),
+                store=_StoreStub(),
+                preferences=prefs,
+                memory=_MemoryStub(),
+                escalation=escalation,
+                config=config,
+            )
+            approved, decision = await engine_restarted.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "git commit -m again"},
+            )
+
+            self.assertTrue(approved)
+            self.assertEqual(decision.policy_source, "session_approval")
+            self.assertEqual(len(escalation.calls), 1)
+
+    async def test_deferred_escalation_decision_applies_session_grant(self) -> None:
+        with _workspace_tempdir() as opc_home:
+            prefs = PreferenceManager(opc_home)
+            escalation = _EscalationStub(None)  # inline wait times out
+            engine = ApprovalEngine(
+                llm=_LLMStub(),
+                store=_StoreStub(),
+                preferences=prefs,
+                memory=_MemoryStub(),
+                escalation=escalation,
+                config=AutonomyConfig(),
+            )
+            task = Task(title="Install deps", project_id="demo", session_id="sess-deferred")
+
+            approved, decision = await engine.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "pip install requests"},
+            )
+            self.assertFalse(approved)
+            self.assertEqual(decision.action, ApprovalAction.REQUIRE_INPUT)
+            # The card carries the approval context needed for a late decision.
+            context = escalation.contexts[-1]
+            self.assertIsInstance(context, dict)
+            self.assertEqual(context["action_name"], "shell_exec")
+            self.assertEqual(context["session_scope_id"], "sess-deferred")
+            self.assertIn("pip install", context["allowlist_patterns"])
+
+            # The user clicks the card minutes later: the grant persists and
+            # the retried command auto-approves without a new prompt.
+            summary = engine.apply_deferred_escalation_decision("approve_session", context)
+            self.assertTrue(summary["approved"])
+            self.assertEqual(summary["scope"], "session:sess-deferred")
+
+            prompts_before = len(escalation.calls)
+            approved, decision = await engine.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "pip install flask"},
+            )
+            self.assertTrue(approved)
+            self.assertEqual(decision.policy_source, "session_approval")
+            self.assertEqual(len(escalation.calls), prompts_before)
+
+    async def test_deferred_escalation_deny_grants_nothing(self) -> None:
+        with _workspace_tempdir() as opc_home:
+            prefs = PreferenceManager(opc_home)
+            escalation = _EscalationStub(None)
+            engine = ApprovalEngine(
+                llm=_LLMStub(),
+                store=_StoreStub(),
+                preferences=prefs,
+                memory=_MemoryStub(),
+                escalation=escalation,
+                config=AutonomyConfig(),
+            )
+            task = Task(title="Install deps", project_id="demo", session_id="sess-deny")
+
+            await engine.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "pip install requests"},
+            )
+            context = escalation.contexts[-1]
+
+            summary = engine.apply_deferred_escalation_decision("deny", context)
+            self.assertFalse(summary["approved"])
+            self.assertIsNone(summary["scope"])
+
+            prompts_before = len(escalation.calls)
+            approved, _ = await engine.authorize_tool_call(
+                task=task,
+                tool_name="shell_exec",
+                arguments={"command": "pip install requests"},
+            )
+            self.assertFalse(approved)
+            self.assertEqual(len(escalation.calls), prompts_before + 1)
+
     async def test_download_command_outside_acquisition_work_item_does_not_skip_first_use_prompt(self) -> None:
         with _workspace_tempdir() as opc_home:
             prefs = PreferenceManager(opc_home)