diff --git a/opc/core/config.py b/opc/core/config.py index 4ea120d..8ce73ba 100644 --- a/opc/core/config.py +++ b/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) diff --git a/opc/layer2_organization/approval.py b/opc/layer2_organization/approval.py index 4ab51e9..a2ecea0 100644 --- a/opc/layer2_organization/approval.py +++ b/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, diff --git a/opc/layer2_organization/escalation.py b/opc/layer2_organization/escalation.py index 7c1c3ed..42ef7b3 100644 --- a/opc/layer2_organization/escalation.py +++ b/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: diff --git a/opc/layer5_memory/approval_allowlist.py b/opc/layer5_memory/approval_allowlist.py index 8b54074..ce33d85 100644 --- a/opc/layer5_memory/approval_allowlist.py +++ b/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 diff --git a/opc/plugins/office_ui/ws_handler.py b/opc/plugins/office_ui/ws_handler.py index 2302ce6..099bd80 100644 --- a/opc/plugins/office_ui/ws_handler.py +++ b/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" diff --git a/tests/test_approval_engine.py b/tests/test_approval_engine.py index 213e1dd..0e8595b 100644 --- a/tests/test_approval_engine.py +++ b/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)