feat: AI command intent analysis on approval prompts

coworker/agent.py

@@ -167,6 +167,8 @@ def build_engine(
     connector_filter: Optional[set[str]] = None,
     # A set (static snapshot) or a zero-arg callable (live, re-evaluated per load_skill).
     skill_filter: Optional[set[str] | Callable[[], set[str]]] = None,
+    # Optional AI intent analyzer (dependency injection): None = feature off.
+    intent_analyzer: Optional[Callable] = None,
 ) -> TurnEngine:
     ws = Path(workspace).expanduser().resolve() if workspace else None
     if agent.needs_workspace and ws is None:
@@ -395,6 +397,7 @@ def context_provider() -> str:
         directory_requester=directory_requester,
         plan_approver=plan_approver,
         question_asker=question_asker,
+        intent_analyzer=intent_analyzer,
     )
     engine.executor = executor  # type: ignore[attr-defined]
     engine.todo = todo  # type: ignore[attr-defined]

coworker/engine.py

@@ -41,6 +41,7 @@ class PermissionRequest:
     metadata: Any
     reason: str
     tool_call_id: Optional[str] = None  # for durable resume (idempotent inbox item)
+    intent: Optional[str] = None  # AI-generated consequence summary, surfaced to the approver
 
 
 Approver = Callable[[PermissionRequest], Awaitable[ApprovalOutcome]]
@@ -77,8 +78,16 @@ def __init__(
         # Called (thread-safe, best-effort) when the user stops the turn — e.g. the
         # executor's kill for a running shell command.
         interrupt_hooks: Optional[list[Callable[[], None]]] = None,
+        # Optional AI intent analyzer (dependency injection). None = feature off
+        # (upstream behavior unchanged). Signature: (tool_call, provider, model) -> str | None.
+        intent_analyzer: Optional[Callable] = None,
+        # Analyzer timeout in seconds (default 20s covers cloud-model tail latency;
+        # tests inject smaller values to avoid real waits).
+        intent_analyzer_timeout: float = 20.0,
     ) -> None:
         self.provider = provider
+        self.intent_analyzer = intent_analyzer  # None = feature off
+        self.intent_analyzer_timeout = intent_analyzer_timeout
         self.registry = registry
         self.permissions = permissions
         self.model = model
@@ -695,6 +704,40 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
             )
 
         if not allowed and decision.needs_user:
+            # AI intent analysis: synchronously generate a consequence summary before
+            # surfacing the approval prompt, so the card appears already annotated.
+            # None = feature off (or analysis failed/timed out/stopped) → no annotation.
+            intent: Optional[str] = None
+            if self.intent_analyzer:
+                async def _do_analyze():
+                    return await asyncio.wait_for(
+                        asyncio.to_thread(
+                            self.intent_analyzer, tool_call, self.provider, self.model
+                        ),
+                        timeout=self.intent_analyzer_timeout,
+                    )
+                # try/except is required: wait_for re-raises TimeoutError via task.result(),
+                # and _interruptible does not swallow it — without this the approval flow
+                # would crash on timeout.
+                try:
+                    intent = await self._interruptible(_do_analyze(), interrupted=None)
+                except Exception:
+                    intent = None
+
+            # If the user stopped during analysis, don't surface a card at all — go
+            # straight to the interrupted path (avoids a flash-then-deny flicker).
+            if self._cancel.is_set():
+                self.messages.append(_tool_error_message(tool_call, "interrupted by user"))
+                self._audit(
+                    tool_call, stage="finished", status="interrupted", reason="user stop"
+                )
+                yield Event(
+                    EventType.TOOL_FINISHED,
+                    {"name": tool_call.name, "status": "interrupted", "reason": "stopped"},
+                )
+                yield False
+                return
+
             yield Event(
                 EventType.PERMISSION_REQUIRED,
                 {
@@ -711,6 +754,7 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
                         metadata,
                         self.permissions.risk_overrides,
                     ),
+                    "intent": intent,
                 },
             )
             self._audit(tool_call, stage="approval_requested", reason=decision.reason)
@@ -722,6 +766,7 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
                         metadata=metadata,
                         reason=decision.reason,
                         tool_call_id=tool_call.id,
+                        intent=intent,
                     )
                 ),
                 interrupted=ApprovalOutcome.DENY,

coworker/intent_analysis/analyzer.py

@@ -0,0 +1,69 @@
+"""AI command intent analysis — generate a consequence summary before approval.
+
+Uses a single blocking provider.complete call (tools disabled, no side effects),
+mirroring the compaction.summarize_span pattern. Returns None on failure/empty
+so the engine simply omits the annotation.
+
+Positional signature: the engine calls
+    asyncio.to_thread(self.intent_analyzer, tool_call, self.provider, self.model)
+so positional args must align.
+"""
+import json
+from typing import Any, Optional, Protocol
+
+from .prompts import build_system_prompt, build_user_prompt
+
+
+class _ToolCallLike(Protocol):
+    name: str
+    arguments: dict[str, Any]
+
+
+# Aligned with coworker/risk.py WRITE_TOOLS (workspace-mutating tools whose
+# path argument is the meaningful thing to show the user).
+_WRITE_TOOLS = {"write_file", "replace_in_file", "apply_patch", "apply_unified_diff"}
+_SEND_TOOLS = {"send_message", "send_file"}
+
+
+def extract_input(tool_call: _ToolCallLike) -> str:
+    """Extract a structured 'operation description' from the tool call."""
+    name = tool_call.name
+    args = tool_call.arguments or {}
+    if name == "run_shell" and args.get("command"):
+        return str(args["command"])
+    if name in _WRITE_TOOLS:
+        path = args.get("path", "")
+        return f"Operation: {name}\nPath: {path}"
+    if name in _SEND_TOOLS:
+        target = args.get("target") or args.get("destination") or args.get("channel") or ""
+        content = args.get("text") or args.get("content") or ""
+        return f"Operation: {name}\nTarget: {target}\nContent: {content}"
+    return f"Operation: {name}\nArgs: {json.dumps(args, ensure_ascii=False)}"
+
+
+def _clean(text: str) -> Optional[str]:
+    """Keep only '• '/'- ' bullet lines, strip code fences and leading labels."""
+    if not text:
+        return None
+    lines = []
+    for line in text.splitlines():
+        line = line.strip().strip("`")
+        if line.startswith("• ") or line.startswith("- "):
+            lines.append(line)
+    return "\n".join(lines) if lines else None
+
+
+def analyze(tool_call, provider, model) -> Optional[str]:
+    """Generate the consequence summary. Synchronous blocking (the engine
+    wraps it in asyncio.to_thread). Returns None on failure/empty."""
+    try:
+        messages = [
+            {"role": "system", "content": build_system_prompt(2)},
+            {"role": "user", "content": build_user_prompt(extract_input(tool_call))},
+        ]
+        turn = provider.complete(
+            model=model, messages=messages, tools=None, max_tokens=300
+        )
+        return _clean(getattr(turn, "text", None))
+    except Exception:
+        return None

coworker/intent_analysis/prompts.py

@@ -0,0 +1,56 @@
+"""Prompts for AI command intent analysis."""
+
+MAX_INPUT_CHARS = 2000
+MAX_BULLETS = 6
+
+_EN_SYSTEM = """You are an expert at explaining operation consequences to users. The user is about to approve an operation. Your job is to clearly explain the consequences using bullet points so they can make an informed decision.
+
+# Rules
+1. Format: Use bullet points (starting with "• "). Each point describes one specific consequence. 1-{n} bullet points total.
+2. Language: STRICTLY output in English.
+3. Emphasis: Wrap the most critical keywords with **double asterisks** for bold. Only bold the most important terms (1-2 per bullet), such as action verbs and irreversible consequences.
+4. Focus: Each bullet should answer one of: What will happen? What will be affected? What is the risk/consequence?
+5. CRITICAL - Dangerous operations: If the operation involves destructive or risky actions (rm, delete, drop, kill, format, overwrite, force push, reset --hard, chmod 777, truncate, revoke, clear, etc.), you MUST emphasize severity and irreversibility.
+6. Non-dangerous operations: Still use bullet points, but in a neutral helpful tone without severity emphasis. Do not call tools. Do not include greetings, analysis, code fences, or extra explanation. Output only the bullet points.
+
+# Examples
+Operation: run_shell
+Command: rm ~/Desktop/test.sh
+Intent:
+• The rm command will **permanently delete** the file, this action is irreversible
+• The file **cannot be recovered** from Trash after deletion
+
+Operation: run_shell
+Command: git push --force origin main
+Intent:
+• Will **forcefully overwrite** the remote main branch history
+• Other people's code on this branch **may be lost**
+• This action **cannot be easily undone**
+
+Operation: write_file
+Path: /etc/config
+Intent:
+• Will **overwrite** the file's current contents
+• The original contents **cannot be recovered**, confirm you don't need to keep them
+
+Operation: send_message
+Target: slack:#ops-channel
+Intent:
+• Will post a message to **#ops-channel**, visible to everyone in the channel
+• The message **cannot be unsent** once delivered
+"""
+
+
+def build_system_prompt(max_bullets: int) -> str:
+    """Build the system prompt. max_bullets is clamped to 1..MAX_BULLETS."""
+    n = max(1, min(MAX_BULLETS, max_bullets))
+    return _EN_SYSTEM.format(n=n)
+
+
+def build_user_prompt(operation_input: str) -> str:
+    """Build the user prompt. Long inputs are truncated to MAX_INPUT_CHARS."""
+    operation_input = operation_input or ""
+    truncated = operation_input[:MAX_INPUT_CHARS]
+    if len(operation_input) > MAX_INPUT_CHARS:
+        truncated += "..."
+    return f"# Input\n{truncated}\n\n# Output\nReturn only the bullet points."

coworker/server/app.py

@@ -1434,6 +1434,10 @@ def settings_set_context_bar(body: dict) -> dict[str, Any]:
         # Composer: show the context-window fill bar, or just the popover (owner ask).
         return manager.set_context_bar((body or {}).get("context_bar", True))
 
+    @app.post("/v1/settings/intent-analysis")
+    def settings_set_intent_analysis(body: dict) -> dict[str, Any]:
+        return manager.set_intent_analysis(bool((body or {}).get("enabled", True)))
+
     @app.post("/v1/settings/pdf")
     def settings_set_pdf(body: dict) -> dict[str, Any]:
         # Token savings (owner ask, 2026-07-17): fallback mode for models without native

coworker/server/manager.py

@@ -434,6 +434,11 @@ def get_engine(
                 if Path(str(r.get("path", ""))).is_dir()
             ]
             roots = [{"path": ws, "writable": True, "label": "scratch"}, *extra]
+        # AI intent analysis: inject when the pref is on (opt-in; default off).
+        from ..intent_analysis.analyzer import analyze
+        intent_analyzer = None
+        if self._prefs.get("intent_analysis", False):
+            intent_analyzer = analyze
         engine = build_engine(
             agent=ag,
             workspace=ws,
@@ -467,6 +472,7 @@ def get_engine(
             # Per-session skill menu, LIVE (SKILLS-SPEC §3): a callable so load_skill sees
             # disables/new skills immediately; the catalog snapshot is taken at build.
             skill_filter=lambda sid=session_id, w=ws: self.effective_skill_names(sid, w),
+            intent_analyzer=intent_analyzer,
         )
         # An automation run rebuilt here (manual "Run now" over WS, durable resume) still
         # carries its task's standing allowances — the rules live on the task record.
@@ -1851,6 +1857,7 @@ def _selectable(m: str) -> bool:
             "nav_layout": self._nav_layout(),
             "sessions_peek": self.sessions_peek(),
             "context_bar": self.context_bar(),
+            "intent_analysis": self._prefs.get("intent_analysis", False),
             "scratch_base": self._prefs.get("scratch_base")
             or self.DEFAULT_SCRATCH_BASE,
             # Real on-disk secrets location, so the UI shows the OS-native path instead of a
@@ -1920,6 +1927,11 @@ def set_context_bar(self, shown: Any) -> dict[str, Any]:
         self._save_prefs()
         return {"ok": True, "context_bar": self.context_bar()}
 
+    def set_intent_analysis(self, enabled: bool) -> dict[str, Any]:
+        self._prefs["intent_analysis"] = bool(enabled)
+        self._save_prefs()
+        return {"ok": True, "intent_analysis": bool(enabled)}
+
     # -- PDF attachments / token savings (owner ask, 2026-07-17) ----------------
     DEFAULT_PDF_MAX_PAGES = 20
     DEFAULT_PDF_MAX_MB = 10
@@ -2604,6 +2616,9 @@ def approval_prompt_data(self, session_id: str, request) -> dict[str, Any]:
             "tool": request.tool_name,
             "arguments": getattr(request, "arguments", None) or {},
         }
+        # Carry the AI intent annotation into the Inbox snapshot.
+        if getattr(request, "intent", None):
+            data["intent"] = request.intent
         task = self.task_store.task_for_run_session(session_id)
         if task is None:
             return data
@@ -2719,6 +2734,11 @@ def _seed_task_permissions(self, engine: TurnEngine, task) -> None:
     def _build_task_engine(self, task, *, session_id: str) -> TurnEngine:
         ag = get_agent(task.agent)
         Path(task.workspace).mkdir(parents=True, exist_ok=True)
+        # AI intent analysis: inject when the pref is on (opt-in; default off).
+        from ..intent_analysis.analyzer import analyze
+        intent_analyzer = None
+        if self._prefs.get("intent_analysis", False):
+            intent_analyzer = analyze
         engine = build_engine(
             agent=ag,
             workspace=task.workspace,
@@ -2740,6 +2760,7 @@ def _build_task_engine(self, task, *, session_id: str) -> TurnEngine:
             skill_filter=lambda sid=session_id, w=task.workspace: (
                 self.effective_skill_names(sid, w)
             ),
+            intent_analyzer=intent_analyzer,
         )
         self._seed_task_permissions(engine, task)
         return engine

pyproject.toml

@@ -62,3 +62,6 @@ coworker = ["personas/builtin/*.md"]
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 asyncio_mode = "auto"
+markers = [
+    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
+]

surfaces/gui/src/App.tsx

@@ -671,6 +671,7 @@ export function App() {
               reason: d.reason,
               category: d.category,
               standingTarget: d.standing_target || undefined,
+              intent: d.intent || undefined,
             },
           ]);
           break;

surfaces/gui/src/api.ts

@@ -697,6 +697,9 @@ export interface ModelSettings {
   // Composer: show the context-window fill bar (default FALSE; absent → the chip shows
   // the session total). The usage popover keeps both numbers regardless.
   context_bar?: boolean;
+  // AI command intent analysis (default true): annotate approval prompts with a
+  // short LLM-generated consequence summary before the user decides.
+  intent_analysis?: boolean;
   // Curated-matrix display names ({full id → "GLM-5.2 · via Together"}); custom models absent.
   model_labels?: Record<string, string>;
   // {full id → context window in tokens}, verified matrix entries only — drives the
@@ -775,6 +778,18 @@ export async function setContextBar(
   return res.json();
 }
 
+/** Toggle AI command intent analysis on approval prompts. */
+export async function setIntentAnalysis(
+  enabled: boolean,
+): Promise<{ ok: boolean; intent_analysis?: boolean; error?: string }> {
+  const res = await fetch(`${httpBase()}/v1/settings/intent-analysis`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ enabled }),
+  });
+  return res.json();
+}
+
 /** Persist how many sessions a sidebar group shows before "Show more". */
 export async function setSessionsPeek(
   n: number,

surfaces/gui/src/components/ApprovalCard.test.tsx

@@ -304,3 +304,39 @@ describe("InboxItemCard — parked save_skill proposals (SKILLS-SPEC §5.2)", ()
     expect(onResolve).toHaveBeenCalledWith("i9", "deny");
   });
 });
+
+describe("ApprovalCard — AI intent block", () => {
+  it("renders the intent block when item.intent is present", () => {
+    render(
+      <ApprovalCard
+        item={sendApproval({ intent: "• dangerous op\n• irreversible" })}
+        onApprove={vi.fn()}
+      />,
+    );
+    expect(screen.getByText(/dangerous op/)).toBeTruthy();
+    expect(screen.getByText(/irreversible/)).toBeTruthy();
+  });
+
+  it("does not render an intent block when item.intent is absent", () => {
+    const { container } = render(<ApprovalCard item={sendApproval()} onApprove={vi.fn()} />);
+    expect(container.querySelector(".approval-intent")).toBeNull();
+  });
+
+  it("renders the intent for file_write (forces the full card, not the compact row)", () => {
+    // Routine file writes normally render as a compact one-line row; when an intent is
+    // present they must fall back to the full card so the annotation is visible.
+    render(
+      <ApprovalCard
+        item={{
+          kind: "approval",
+          name: "write_file",
+          args: { path: "/x", content: "y" },
+          reason: "test",
+          intent: "• overwrites the file",
+        }}
+        onApprove={vi.fn()}
+      />,
+    );
+    expect(screen.getByText(/overwrites the file/)).toBeTruthy();
+  });
+});