From 43882d61ed8cd6de25e26710addf1e8d052e736a Mon Sep 17 00:00:00 2001 From: 0xcardinal <0xcardinal@gmail.com> Date: Wed, 4 Nov 2026 23:09:25 +0530 Subject: [PATCH] feat(rules): block or ask first before Claude Code and Cursor run a tool Rules can now stop a tool call instead of only recording it. A rule whose action is "deny" (Block) or "ask" (Ask first) is checked from the agent's pre-tool hook, and the agent either refuses the call or asks the person, showing the rule's reason. Collector - ENFORCEMENT_ENABLED is on. Blocking rules must be match rules over details known before the tool runs, scoped to an agent that can block (Claude Code and Cursor for now, ENFORCEMENT_SUPPORT). - POST /v1/decide/{source} answers from the in-memory rule cache and never waits on a DB write. Blocks and asks go to the audit log off-thread. - PUT /v1/enforcement pauses every blocking rule at once. - cot's own bridge commands are never blocked. Bridge - Pre-tool hooks (Claude PreToolUse, Cursor preToolUse) ask /v1/decide with a 0.5s timeout and allow the call on any failure. The check ignores the collector-down backoff: one slow ingest post used to mark the collector down and let blocked calls through for ten seconds. - The decision maps onto Claude's permissionDecision and Cursor's {permission, user_message, agent_message} replies, and is recorded on the stored event as cot_decision. Dashboard - The rule sentence opens with "Tell me when / Ask me first when / Block it when"; blocking hides what can't be decided before a run (spend, "and it fails", repeat limits, agents that can't block). - The rules list shows block / ask first chips and a Blocking on switch; previews read "Would have blocked". --- backend/app/main.py | 49 ++++++++ backend/app/user_rules.py | 116 ++++++++++++++---- backend/tests/test_bridge_decide.py | 97 +++++++++++++++ backend/tests/test_user_rules.py | 91 +++++++++++++- bridge/cot | 56 ++++++++- .../dashboard/RuleSentenceEditor.tsx | 41 +++++-- src/components/dashboard/RulesView.tsx | 37 +++++- src/forest.css | 2 + src/lib/api.ts | 11 +- src/lib/ruleSentence.test.ts | 20 ++- src/lib/ruleSentence.ts | 30 ++++- 11 files changed, 497 insertions(+), 53 deletions(-) create mode 100644 backend/tests/test_bridge_decide.py diff --git a/backend/app/main.py b/backend/app/main.py index 9ca46b4..cbc3f0a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2186,6 +2186,55 @@ def delete_rule(rule_id: str) -> dict[str, Any]: return {"ok": True} +class EnforcementBody(BaseModel): + paused: bool + + +@app.put("/v1/enforcement") +def set_rule_enforcement(body: EnforcementBody) -> dict[str, Any]: + """The global switch that pauses every Block / Ask first rule at once.""" + user_rules.set_enforcement_paused(body.paused) + db.record_audit_event("rules.enforcement", target="rules", detail={"paused": body.paused}) + return {"paused": body.paused} + + +def _record_block(source: str, body: dict[str, Any], d: user_rules.Decision) -> None: + try: + db.record_audit_event( + f"rules.{d.action}", + target=d.rule_id, + detail={ + "source": source, + "rule": d.rule_name, + "session_id": body.get("session_id"), + "tool": body.get("tool_name"), + }, + ) + except Exception: + pass + + +@app.post("/v1/decide/{source}") +async def decide_hook(source: str, request: Request) -> dict[str, Any]: + """Allow, ask or deny a pending tool call. The bridge calls this from an + agent's pre-tool hook with a short timeout and allows the call on any + failure, so this must answer from memory and never wait on a DB write.""" + try: + body = await request.json() + except Exception: + return {"decision": "allow"} + if not isinstance(body, dict): + return {"decision": "allow"} + try: + d = await asyncio.to_thread(user_rules.decide_hook, source, body) + except Exception: + return {"decision": "allow"} + if d is None: + return {"decision": "allow"} + threading.Thread(target=_record_block, args=(source, body, d), daemon=True).start() + return {"decision": d.action, "rule_id": d.rule_id, "rule_name": d.rule_name, "reason": d.reason} + + class AnalyzeRequest(BaseModel): days: int = 30 diff --git a/backend/app/user_rules.py b/backend/app/user_rules.py index 4afd4f8..d95980e 100644 --- a/backend/app/user_rules.py +++ b/backend/app/user_rules.py @@ -33,10 +33,13 @@ "field": , "op": "gte"|"gt"|..., "value": }`` and fires once per group whose aggregate passes. -## Built for enforcement later -Today cot only observes — the bridge posts hooks fire-and-forget. Blocking a -tool call needs a synchronous ``PreToolUse`` round trip, and the pieces it -needs are already here so that step is wiring, not a rewrite: +## Enforcement +A rule whose action is ``ask`` or ``deny`` can stop a tool call before it runs. +The bridge asks ``POST /v1/decide/{source}`` from the agent's pre-tool hook, +:func:`decide_hook` answers from the in-memory rule cache, and the bridge maps +the answer onto that agent's hook reply. Only agents in +:data:`ENFORCEMENT_SUPPORT` can carry a decision today; every other agent keeps +observing. If the collector is slow or down the bridge allows the call. * **One evaluator, two callers.** Conditions run over a flat *facts* mapping, not over SQLite rows. Stored events become facts via :func:`facts_from_event`; @@ -48,9 +51,9 @@ (``post``: status, duration, tokens, cost). A rule is *enforceable* only if it is a ``match`` rule over ``pre`` fields — the API reports this per rule so the UI can show which rules could block once enforcement ships. -* **Actions.** ``action`` is ``"finding"`` today; ``"ask"`` and ``"deny"`` are - reserved and rejected by validation until :data:`ENFORCEMENT_ENABLED` flips. - :func:`decide` already resolves a hook to the strictest matching decision. +* **Actions.** ``"finding"`` records an insight; ``"ask"`` and ``"deny"`` also + block (they must be enforceable). :func:`decide` resolves a hook to the + strictest matching decision. * **Bounded cost.** Specs are size-capped, regexes are compiled once at load and matched against a capped prefix of the value, and compiled rules are cached — the hot path must stay well inside a hook's timeout. @@ -87,11 +90,22 @@ "subagent", "compaction", "lifecycle", "notification", ) -# "finding" records an insight. "ask"/"deny" are the enforcement actions a -# synchronous PreToolUse hook will return; validation rejects them until then. +# "finding" records an insight. "ask"/"deny" are returned from the agent's +# pre-tool hook (and still record findings over history). ACTIONS = ("finding", "ask", "deny") ENFORCEMENT_ACTIONS = ("ask", "deny") -ENFORCEMENT_ENABLED = False +ENFORCEMENT_ENABLED = True + +# Agents whose pre-tool hook reply can carry a decision, and which actions each +# honours. The bridge holds the matching reply shapes (``_hook_response``). +ENFORCEMENT_SUPPORT: dict[str, tuple[str, ...]] = { + "claude": ("deny", "ask"), + "cursor": ("deny", "ask"), +} +# The hook each agent fires before a tool runs. +PRE_TOOL_HOOKS = {"claude": "PreToolUse", "cursor": "preToolUse"} +# Settings key for the global "pause all blocks" switch. +PAUSED_KEY = "rules_enforcement_paused" ID_PREFIX = "user." @@ -505,9 +519,27 @@ def validate(spec: Any, *, allow_enforcement: bool | None = None) -> dict[str, A out["threshold"] = _validate_threshold(spec.get("threshold")) if out["action"] in ENFORCEMENT_ACTIONS: raise SpecError("action", "only match rules can block") + if out["action"] in ENFORCEMENT_ACTIONS: + _check_blockable(out) return out +def _check_blockable(spec: Mapping[str, Any]) -> None: + """A blocking rule must be decidable before the tool runs, on an agent that can block.""" + late = sorted(f for f in _fields_used(spec["when"]) if FIELDS[f].phase == "post") + if late: + labels = ", ".join(FIELDS[f].label for f in late) + raise SpecError("action", f"can't block on {labels}: it's only known after the tool runs") + sources = spec["scope"]["sources"] + if sources and not any(s in ENFORCEMENT_SUPPORT for s in sources): + raise SpecError("scope.sources", f"blocking works for {_support_labels()} today") + + +def _support_labels() -> str: + labels = {"claude": "Claude Code", "cursor": "Cursor"} + return " and ".join(labels.get(s, s) for s in ENFORCEMENT_SUPPORT) + + def _fields_used(node: Mapping[str, Any]) -> set[str]: if "all" in node or "any" in node: return set().union(*(_fields_used(c) for c in node.get("all") or node.get("any"))) @@ -759,7 +791,7 @@ def compile_rule(rule_id: str, spec: dict[str, Any], enabled: bool = True) -> Co ) -# --- enforcement (not wired to the bridge yet) ------------------------------------- +# --- enforcement --------------------------------------------------------------------- @dataclass(frozen=True) @@ -773,13 +805,17 @@ class Decision: _ACTION_RANK = {"deny": 0, "ask": 1} +def _reason(action: str, r: CompiledRule) -> str: + """What the agent and the person see: which rule, and the rule's advice.""" + lead = f"Blocked by cot rule “{r.name}”" if action == "deny" else f"cot rule “{r.name}” asks first" + rec = (r.spec.get("recommendation") or "").strip() + return f"{lead}. {rec}" if rec else f"{lead}." + + def decide(facts: Mapping[str, Any], rules: list[CompiledRule] | None = None) -> Decision | None: """The strictest enforcement decision for a pending tool call, or None to allow. Only enabled, enforceable rules with an ``ask``/``deny`` action take part. - A future synchronous ``PreToolUse`` endpoint calls this with - :func:`facts_from_hook`; the bridge maps the decision onto each agent's - hook response (Claude ``permissionDecision``, Cursor ``permission``, …). """ pool = rules if rules is not None else compiled_rules() best: Decision | None = None @@ -789,17 +825,49 @@ def decide(facts: Mapping[str, Any], rules: list[CompiledRule] | None = None) -> continue if not r.matches(facts): continue - d = Decision( - action=action, - rule_id=r.id, - rule_name=r.name, - reason=r.spec.get("recommendation") or f"Blocked by cot rule “{r.name}”.", - ) + d = Decision(action=action, rule_id=r.id, rule_name=r.name, reason=_reason(action, r)) if best is None or _ACTION_RANK[action] < _ACTION_RANK[best.action]: best = d return best +# cot's own bridge runs as a shell command from some agents' tools; blocking it +# could cut off the hooks that do the blocking. +_COT_SELF = re.compile(r"(^|[\s/])\.cot/bin/cot(\s|$)") + + +def decide_hook(source: str, body: Mapping[str, Any]) -> Decision | None: + """The decision for a live pre-tool hook body, or None to allow. + + Reads only the in-memory rule cache once it is warm, so it stays well + inside the bridge's short decide timeout. + """ + supported = ENFORCEMENT_SUPPORT.get(source) + if not supported or enforcement_paused(): + return None + hook = body.get("hook_event_name") or body.get("hook") or body.get("event") + if hook != PRE_TOOL_HOOKS.get(source): + return None + if not any(r.enabled and r.spec.get("action") in ENFORCEMENT_ACTIONS for r in compiled_rules()): + return None + facts = facts_from_hook(source, body) + if _COT_SELF.search(_as_text(facts.get("target"))): + return None + d = decide(facts) + return d + + +def enforcement_paused() -> bool: + if _cache["paused"] is None: + _cache["paused"] = db.get_setting(PAUSED_KEY) == "1" + return bool(_cache["paused"]) + + +def set_enforcement_paused(paused: bool) -> None: + db.set_setting(PAUSED_KEY, "1" if paused else "0") + _cache["paused"] = paused + + # --- storage & cache -------------------------------------------------------------- # The collector is the only writer of ``user_rules`` (every write goes through @@ -808,11 +876,11 @@ def decide(facts: Mapping[str, Any], rules: list[CompiledRule] | None = None) -> # because a read there can queue for seconds behind a long findings pass plus a # pending writer (rollback-journal mode). Keyed by DB path: tests and the replay # store swap databases under the same process. -_cache: dict[str, Any] = {"path": None, "rows": None, "compiled": None} +_cache: dict[str, Any] = {"path": None, "rows": None, "compiled": None, "paused": None} def _invalidate() -> None: - _cache.update(path=None, rows=None, compiled=None) + _cache.update(path=None, rows=None, compiled=None, paused=None) # Findings computed with the old rule set are shared for a few seconds; # drop them so a saved rule shows up straight away. from .insights import forget_shared_insights # insights imports this module @@ -824,7 +892,7 @@ def _rows() -> list[dict[str, Any]]: path = str(db.db_path()) if _cache["path"] != path or _cache["rows"] is None: rows = [dict(r) for r in db.list_user_rules()] - _cache.update(path=path, rows=rows, compiled=None) + _cache.update(path=path, rows=rows, compiled=None, paused=None) return _cache["rows"] @@ -953,6 +1021,8 @@ def schema() -> dict[str, Any]: "severities": list(SEVERITIES), "actions": [a for a in ACTIONS if ENFORCEMENT_ENABLED or a not in ENFORCEMENT_ACTIONS], "enforcement_enabled": ENFORCEMENT_ENABLED, + "enforcement_sources": {s: list(a) for s, a in ENFORCEMENT_SUPPORT.items()}, + "enforcement_paused": enforcement_paused(), "sources": list(SOURCES), "fields": [ {"name": f.name, "label": f.label, "kind": f.kind, "phase": f.phase, diff --git a/backend/tests/test_bridge_decide.py b/backend/tests/test_bridge_decide.py new file mode 100644 index 0000000..5a78eb8 --- /dev/null +++ b/backend/tests/test_bridge_decide.py @@ -0,0 +1,97 @@ +"""The bridge's side of Block / Ask first: asking /v1/decide and replying per agent.""" + +from __future__ import annotations + +import importlib.machinery +import importlib.util +import io +import json +import os +import sys +import urllib.error + +_REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + + +def _load_bridge(): + path = os.path.join(_REPO, "bridge", "cot") + loader = importlib.machinery.SourceFileLoader("cot_bridge_decide_under_test", path) + spec = importlib.util.spec_from_loader("cot_bridge_decide_under_test", loader) + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + return module + + +bridge = _load_bridge() + +_DENY = {"decision": "deny", "rule_id": "user.x", "rule_name": "No prod", "reason": "Blocked by cot rule “No prod”."} + + +class _Resp(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + +def _fake_collector(monkeypatch, answer): + calls = [] + + def urlopen(req, timeout=None): + calls.append((req.full_url, timeout)) + if isinstance(answer, Exception): + raise answer + return _Resp(json.dumps(answer).encode()) + + monkeypatch.setattr(bridge, "_collector_reachable", lambda: True) + monkeypatch.setattr(bridge.urllib.request, "urlopen", urlopen) + return calls + + +def test_claude_deny_reply(): + out = bridge._hook_response("claude", {"hook_event_name": "PreToolUse"}, _DENY) + assert out == {"hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": _DENY["reason"], + }} + + +def test_cursor_ask_reply(): + out = bridge._hook_response("cursor", {"hook_event_name": "preToolUse"}, {**_DENY, "decision": "ask"}) + assert out == {"permission": "ask", "user_message": _DENY["reason"], "agent_message": _DENY["reason"]} + + +def test_no_decision_keeps_passive_replies(): + assert bridge._hook_response("claude", {"hook_event_name": "PreToolUse"}) == { + "continue": True, "suppressOutput": True} + assert bridge._hook_response("cursor", {"hook_event_name": "preToolUse"}) == {"continue": True} + assert bridge._hook_response("codex", {"hook_event_name": "PreToolUse"}) is None + + +def test_decide_asks_only_on_pre_tool_hooks(monkeypatch): + calls = _fake_collector(monkeypatch, _DENY) + assert bridge._decide("claude", {"hook_event_name": "PreToolUse"}) == _DENY + assert calls[0][0].endswith("/v1/decide/claude") + assert calls[0][1] == bridge._DECIDE_TIMEOUT_S + assert bridge._decide("claude", {"hook_event_name": "PostToolUse"}) is None + assert bridge._decide("codex", {"hook_event_name": "PreToolUse"}) is None + assert len(calls) == 1 + + +def test_decide_allows_on_failure(monkeypatch): + _fake_collector(monkeypatch, TimeoutError()) + assert bridge._decide("claude", {"hook_event_name": "PreToolUse"}) is None + _fake_collector(monkeypatch, urllib.error.URLError("down")) + assert bridge._decide("cursor", {"hook_event_name": "preToolUse"}) is None + _fake_collector(monkeypatch, {"decision": "allow"}) + assert bridge._decide("claude", {"hook_event_name": "PreToolUse"}) is None + + +def test_decide_still_asks_inside_the_down_backoff(monkeypatch): + # A slow ingest post marks the collector down; blocking must not lapse with it. + calls = _fake_collector(monkeypatch, _DENY) + monkeypatch.setattr(bridge, "_collector_reachable", lambda: False) + assert bridge._decide("claude", {"hook_event_name": "PreToolUse"}) == _DENY + assert len(calls) == 1 diff --git a/backend/tests/test_user_rules.py b/backend/tests/test_user_rules.py index 60e4c62..b83a798 100644 --- a/backend/tests/test_user_rules.py +++ b/backend/tests/test_user_rules.py @@ -3,7 +3,7 @@ Covers spec validation, the condition evaluator as a pure function over facts, match/threshold compilation against a real store, the insights integration (lifecycle + cleanup on delete), the HTTP API, and the enforcement seam -(``decide`` over hook facts) that isn't wired to the bridge yet. +(``decide`` over hook facts) and the ``/v1/decide`` endpoint the bridge calls. """ from __future__ import annotations @@ -99,11 +99,23 @@ def test_validate_rejects_with_path(bad, path): assert exc.value.path == path -def test_blocking_actions_reserved_until_enforcement_ships(): +def test_blocking_actions_validate_when_enforceable(): + assert user_rules.validate(_spec(action="deny"))["action"] == "deny" with pytest.raises(user_rules.SpecError) as exc: - user_rules.validate(_spec(action="deny")) + user_rules.validate(_spec(action="deny"), allow_enforcement=False) assert exc.value.path == "action" - assert user_rules.validate(_spec(action="deny"), allow_enforcement=True)["action"] == "deny" + + +def test_blocking_rule_rejects_post_fields_and_unsupported_agents(): + late = _spec(action="deny", when={"field": "status", "op": "eq", "value": "error"}) + with pytest.raises(user_rules.SpecError) as exc: + user_rules.validate(late) + assert exc.value.path == "action" and "Result" in str(exc.value) + with pytest.raises(user_rules.SpecError) as exc: + user_rules.validate(_spec(action="ask", scope={"sources": ["codex"]})) + assert exc.value.path == "scope.sources" + # A mixed scope is fine: it blocks where it can. + assert user_rules.validate(_spec(action="ask", scope={"sources": ["codex", "claude"]})) def test_condition_count_is_capped(): @@ -299,7 +311,9 @@ def client(): def test_api_crud_preview_and_errors(client): _event("s1", category="shell", target="psql -h prod-db", status="ok") schema = client.get("/v1/rules/schema").json() - assert "deny" not in schema["actions"] and schema["enforcement_enabled"] is False + assert "deny" in schema["actions"] and schema["enforcement_enabled"] is True + assert schema["enforcement_sources"]["claude"] == ["deny", "ask"] + assert schema["enforcement_paused"] is False assert any(f["name"] == "target" and f["phase"] == "pre" for f in schema["fields"]) preview = client.post("/v1/rules/preview", json={"spec": _spec(), "days": 30}).json() @@ -455,3 +469,70 @@ def test_saving_a_rule_drops_shared_findings(): insights.compute_insights_shared(30) # cached without the rule rule = user_rules.create_rule(_spec()) assert any(f["id"] == rule["id"] for f in insights.compute_insights_shared(30)["insights"]) + + +# --- live decisions (/v1/decide) --------------------------------------------------- + +_CLAUDE_PRE = { + "hook_event_name": "PreToolUse", + "session_id": "s1", + "cwd": "/work/app", + "tool_name": "Bash", + "tool_input": {"command": "psql -h prod-db -c 'select 1'"}, +} +_CURSOR_PRE = { + "hook_event_name": "preToolUse", + "conversation_id": "c1", + "workspace_roots": ["/work/app"], + "tool_name": "Shell", + "tool_input": {"command": "psql -h prod-db"}, +} + + +def test_decide_hook_blocks_pre_tool_calls_only(): + user_rules.create_rule(_spec(action="deny", recommendation="Use the replica.")) + d = user_rules.decide_hook("claude", _CLAUDE_PRE) + assert d is not None and d.action == "deny" + assert d.reason == "Blocked by cot rule “Prod database access”. Use the replica." + assert user_rules.decide_hook("claude", {**_CLAUDE_PRE, "hook_event_name": "PostToolUse"}) is None + assert user_rules.decide_hook("codex", _CLAUDE_PRE) is None # can't carry a decision + safe = {**_CLAUDE_PRE, "tool_input": {"command": "ls"}} + assert user_rules.decide_hook("claude", safe) is None + + +def test_decide_hook_reads_cursor_payloads(): + user_rules.create_rule(_spec(action="ask")) + d = user_rules.decide_hook("cursor", _CURSOR_PRE) + assert d is not None and d.action == "ask" and "asks first" in d.reason + + +def test_decide_hook_ignores_finding_and_disabled_rules(): + user_rules.create_rule(_spec()) + user_rules.create_rule(_spec(action="deny"), enabled=False) + assert user_rules.decide_hook("claude", _CLAUDE_PRE) is None + + +def test_decide_hook_never_blocks_cot_itself(): + user_rules.create_rule(_spec(action="deny", when={"field": "category", "op": "eq", "value": "shell"})) + own = {**_CLAUDE_PRE, "tool_input": {"command": "$HOME/.cot/bin/cot status"}} + assert user_rules.decide_hook("claude", own) is None + assert user_rules.decide_hook("claude", _CLAUDE_PRE) is not None + + +def test_pause_switch_allows_everything(client): + user_rules.create_rule(_spec(action="deny")) + assert client.post("/v1/decide/claude", json=_CLAUDE_PRE).json()["decision"] == "deny" + assert client.put("/v1/enforcement", json={"paused": True}).json() == {"paused": True} + assert client.post("/v1/decide/claude", json=_CLAUDE_PRE).json() == {"decision": "allow"} + assert client.get("/v1/rules/schema").json()["enforcement_paused"] is True + client.put("/v1/enforcement", json={"paused": False}) + assert client.post("/v1/decide/claude", json=_CLAUDE_PRE).json()["decision"] == "deny" + + +def test_decide_endpoint_answers_and_allows_on_bad_input(client): + rule = user_rules.create_rule(_spec(action="deny")) + r = client.post("/v1/decide/claude", json=_CLAUDE_PRE).json() + assert r["decision"] == "deny" and r["rule_id"] == rule["id"] + assert client.post("/v1/decide/claude", content=b"not json").json() == {"decision": "allow"} + assert client.post("/v1/decide/claude", json=[1, 2]).json() == {"decision": "allow"} + assert client.post("/v1/decide/nope", json=_CLAUDE_PRE).json() == {"decision": "allow"} diff --git a/bridge/cot b/bridge/cot index 9637a10..4f82e93 100755 --- a/bridge/cot +++ b/bridge/cot @@ -1542,8 +1542,54 @@ def _emit_codex_events(body: dict, source: str) -> None: ]) -def _hook_response(source: str, body: dict) -> dict | None: +# The hook each agent fires before a tool runs, where a cot Block / Ask first +# rule can answer. Mirrors ENFORCEMENT_SUPPORT in backend/app/user_rules.py. +DECIDE_HOOKS = {"claude": "PreToolUse", "cursor": "preToolUse"} +# The agent waits on this before the tool runs. On a timeout or any error the +# call is allowed: a slow collector must never stall or block an agent. +_DECIDE_TIMEOUT_S = 0.5 + + +def _decide(source: str, body: dict) -> dict | None: + """A cot rule's ``deny``/``ask`` decision for a pending tool call, or None to allow.""" hook = body.get("hook_event_name") or body.get("hook") or body.get("event") or "" + # Always ask, even inside the down backoff: one slow ingest post marks the + # collector down for a few seconds, and skipping here would let blocked + # calls through. A collector that is really down refuses at once; a hung + # one costs at most the short timeout below. + if DECIDE_HOOKS.get(source) != hook: + return None + req = urllib.request.Request( + f"{COT_ENDPOINT}/v1/decide/{source}", + data=json.dumps(body, ensure_ascii=False, default=str).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=_DECIDE_TIMEOUT_S) as resp: + data = json.loads(resp.read().decode("utf-8") or "{}") + except Exception: + return None + if not isinstance(data, dict) or data.get("decision") not in ("deny", "ask"): + return None + return data + + +def _hook_response(source: str, body: dict, decision: dict | None = None) -> dict | None: + hook = body.get("hook_event_name") or body.get("hook") or body.get("event") or "" + if decision is not None: + action = decision["decision"] + reason = str(decision.get("reason") or "Blocked by a cot rule.") + if source == "cursor": + return {"permission": action, "user_message": reason, "agent_message": reason} + if source == "claude": + return { + "hookSpecificOutput": { + "hookEventName": hook, + "permissionDecision": action, + "permissionDecisionReason": reason, + }, + } if source == "cursor": if hook in CURSOR_ALLOW_HOOKS: return {"permission": "allow"} @@ -8972,6 +9018,12 @@ def main() -> int: body = {"_raw": raw} hook = body.get("hook_event_name") or body.get("hook") or body.get("event") or "" + # Ask before the tool runs, then record the answer on the stored event. + decision = _decide(source, body) + if decision is not None: + body["cot_decision"] = { + k: decision.get(k) for k in ("decision", "rule_id", "rule_name", "reason") + } # Fold the sandbox/approval posture onto the live Codex SessionStart so it # matches what the importer records (Codex keeps it out of the hook payload). if source == "codex" and hook in ("SessionStart", "sessionStart"): @@ -9038,7 +9090,7 @@ def main() -> int: # responses in now, so active sessions don't wait for cot watch/import. _reconcile_cursor_subagents(body) - response = _hook_response(source, body) + response = _hook_response(source, body, decision) if response is not None: print(json.dumps(response)) return 0 diff --git a/src/components/dashboard/RuleSentenceEditor.tsx b/src/components/dashboard/RuleSentenceEditor.tsx index 89f811e..9204716 100644 --- a/src/components/dashboard/RuleSentenceEditor.tsx +++ b/src/components/dashboard/RuleSentenceEditor.tsx @@ -3,6 +3,7 @@ import { draftRuleFromText, getSettings, RuleSpecError, + type RuleAction, type RuleEventDraft, type RuleLeaf, type RulePreview, @@ -10,7 +11,10 @@ import { type RuleSpec, } from '../../lib/api'; import { + ACTION_LEADS, autoName, + blocks, + fitForBlocking, fromSpec, isSpecific, opsFor, @@ -30,7 +34,7 @@ import { Icon } from '../forest/icons'; import { Seg } from '../forest/ui'; const PREVIEW_DAYS = 30; -const PRE_RUN_HINT = 'Uses only details known before a tool runs, so this rule can switch to blocking once cot supports it.'; +const PRE_RUN_HINT = 'Uses only details known before a tool runs, so this rule can block or ask first.'; /** The friendly editor: a rule written as one sentence, with real examples underneath. */ export function RuleSentenceEditor({ schema, title, initial, draft, onCancel, onSave, onAdvanced }: { @@ -68,6 +72,11 @@ export function RuleSentenceEditor({ schema, title, initial, draft, onCancel, on limit: verb === 'spend' ? null : s.limit, }); }; + const canBlock = (agent: string) => !agent || agent in schema.enforcement_sources; + const blocking = blocks(s.action); + const pickAction = (action: RuleAction) => set({ action, ...(blocks(action) ? fitForBlocking(s, canBlock) : {}) }); + const blockingAgents = Object.keys(schema.enforcement_sources).map(agentLabel).join(' and '); + const pickChoice = (i: number) => { if (!draft) return; setPicked(i); @@ -89,7 +98,8 @@ export function RuleSentenceEditor({ schema, title, initial, draft, onCancel, on } }; - const agents = [{ value: '', label: 'any agent' }, ...schema.sources.map((a) => ({ value: a, label: agentLabel(a) }))]; + const agents = [{ value: '', label: 'any agent' }, ...schema.sources.filter((a) => !blocking || canBlock(a)).map((a) => ({ value: a, label: agentLabel(a) }))]; + const verbs = VERBS.filter((x) => !blocking || x.key !== 'spend'); const perOptions = Object.entries(PER) .filter(([k]) => k !== 'mcp_server' || s.verb === 'mcp') .filter(([k]) => k !== 'tool' || s.verb === 'mcp' || s.verb === 'any') @@ -140,9 +150,9 @@ export function RuleSentenceEditor({ schema, title, initial, draft, onCancel, on
- Tell me when + pickAction(a as RuleAction)} options={ACTION_LEADS.map((x) => ({ value: x.action, label: x.label }))} /> set({ agent })} options={agents} /> - pickVerb(k as VerbKey)} options={VERBS.map((x) => ({ value: x.key, label: x.label }))} /> + pickVerb(k as VerbKey)} options={verbs.map((x) => ({ value: x.key, label: x.label }))} /> {s.verb === 'spend' && ( <> $ @@ -179,9 +189,9 @@ export function RuleSentenceEditor({ schema, title, initial, draft, onCancel, on {s.verb !== 'spend' && (
{v.about.length > 0 && } - + {!blocking && } {!s.projects.length && !details && } - {s.limit ? ( + {blocking ? null : s.limit ? ( at least
)} + {blocking && ( +

+ {s.action === 'deny' ? 'The agent stops before the tool runs' : 'The agent waits for your OK before the tool runs'} and sees your reason. + {!s.agent && ` Works in ${blockingAgents}. Other agents keep recording it.`} +

+ )} {needsDetail &&

Say what to look for first, for example what the {v.noun} contains. Without it, this rule would flag every {v.noun}.

} {problem &&

{problem.message}

}
@@ -231,8 +247,9 @@ export function RuleSentenceEditor({ schema, title, initial, draft, onCancel, on
- -