diff --git a/examples/claude-goalmode-hook-policy-smoke.py b/examples/claude-goalmode-hook-policy-smoke.py index c89759246e..13cfe9b17f 100644 --- a/examples/claude-goalmode-hook-policy-smoke.py +++ b/examples/claude-goalmode-hook-policy-smoke.py @@ -125,6 +125,59 @@ def capture_health_failure(command, **_kwargs): assert decision("Bash", True, command="make test") == "allow", "non-destructive bash allowed" assert decision("Read", True, file_path="/anywhere") == "allow", "read-only allowed" + # An event with no session cwd (what an unreadable event becomes in main) + # cannot be attributed to an armed project, so the gate must refuse it + # rather than defer: rc=1 with no decision is a non-blocking hook error to + # Claude Code, which then runs the tool with no gate at all. + assert goal_policy.decide({}).get("hookSpecificOutput", {}).get("permissionDecision") == "deny", \ + "a cwd-less/unreadable event must be DENIED, not deferred" + + # A non-read-only tool whose tool_input is not an object is unreadable the + # same way: decide must refuse it rather than crash on `ti.get` (rc=1 with + # no decision, which Claude Code reports and then runs the tool ungated). + goal_policy.active_context = lambda cwd: CTX + goal_policy.should_run = lambda *a, **k: True + non_object_input = goal_policy.decide( + {"cwd": "/proj", "tool_name": "Bash", "tool_input": "not-an-object"}) + assert non_object_input.get("hookSpecificOutput", {}).get("permissionDecision") == "deny", \ + "a non-object tool_input must be DENIED, not crash the gate (rc=1 fail-open)" + assert goal_policy.decide( + {"cwd": "/proj", "tool_name": "Read", "tool_input": "not-an-object"} + ).get("hookSpecificOutput", {}).get("permissionDecision") == "allow", \ + "read-only tools keep their unconditional allow" + + # A decoded cwd can still be unusable as a path: an embedded NUL raises + # ValueError inside active_context, which would escape decide as the same + # rc=1/no-decision fail-open. The gate must refuse it instead. + assert goal_policy.decide( + {"cwd": "/bad\u0000cwd", "tool_name": "Bash", "tool_input": {"command": "ls"}} + ).get("hookSpecificOutput", {}).get("permissionDecision") == "deny", \ + "a session cwd that cannot be resolved as a path must be DENIED" + + # `tool_input` is type-checked before any default: only absent/null gets {}; + # a present falsy non-object (`[]`, `false`, `0`, `""`) is unreadable and + # refused, so an open gate cannot allow it as if it carried no input. + for falsy_input in ([], False, 0, ""): + assert goal_policy.decide( + {"cwd": "/proj", "tool_name": "Bash", "tool_input": falsy_input} + ).get("hookSpecificOutput", {}).get("permissionDecision") == "deny", \ + f"falsy non-object tool_input {falsy_input!r} must be DENIED, not coerced" + assert goal_policy.decide( + {"cwd": "/proj", "tool_name": "Bash"} + ).get("hookSpecificOutput", {}).get("permissionDecision") == "allow", \ + "an absent tool_input keeps the documented empty-input default" + + # A decoded value the gate cannot use is refused the same way instead of + # crashing on `.lower()` / set membership or silently matching the scope. + assert goal_policy.decide( + {"cwd": "/proj", "tool_name": ["Bash"], "tool_input": {}} + ).get("hookSpecificOutput", {}).get("permissionDecision") == "deny", \ + "a non-string tool_name must be DENIED, not crash the gate" + assert goal_policy.decide( + {"cwd": "/proj", "tool_name": "Bash", "tool_input": {"command": ["rm", "-rf", "/"]}} + ).get("hookSpecificOutput", {}).get("permissionDecision") == "deny", \ + "a non-string command must be DENIED, not crash the gate" + print("claude-goalmode-hook-policy-smoke ok") return 0 diff --git a/loopx/claude_goal_mode/README.md b/loopx/claude_goal_mode/README.md index fd3c99b9cf..b5eb721071 100644 --- a/loopx/claude_goal_mode/README.md +++ b/loopx/claude_goal_mode/README.md @@ -52,6 +52,12 @@ armed, for each tool call it: before consulting the gate; - otherwise consults `should_run` and **denies** the (non-read-only) tool when `should_run` is `false`, or when the probe is unreachable (fail-closed); +- **denies** an event it cannot read at all (undecodable stdin, malformed or + non-object JSON, a missing or unusable session `cwd`, a missing/non-string + tool name, or - for a non-read-only tool - a non-object `tool_input` or a + `command` / `file_path` / `notebook_path` field that is present but not a + string) rather than exiting with no decision, which Claude Code reports as a + non-blocking hook error and follows with an ungated tool call; - when `should_run` is `true`: **allows** `Edit` / `Write` whose path is inside `write_scope` and **denies** them outside it; gates `Bash` only by a **denylist** of destructive commands (everything else is allowed); and **defers diff --git a/loopx/claude_goal_mode/hooks/goal_policy.py b/loopx/claude_goal_mode/hooks/goal_policy.py index dc1ccf59a9..19dbe868dc 100644 --- a/loopx/claude_goal_mode/hooks/goal_policy.py +++ b/loopx/claude_goal_mode/hooks/goal_policy.py @@ -19,7 +19,11 @@ Behavior (only while goal-mode is ARMED for the event's project): - no goal here / goal-mode off -> {} (no-op; normal permission flow) +- event unreadable -> deny (FAIL-CLOSED) + (no usable session cwd, no usable tool_name) - read-only/safe tools -> allow +- non-read-only tool, unreadable event value -> deny (FAIL-CLOSED) + (non-object tool_input, non-string command/file path) - should_run == false (quota/gate closed) -> deny - should_run probe unavailable (error) -> deny for non-read-only tools (FAIL-CLOSED) - should_run == true: @@ -31,6 +35,13 @@ Fail-closed: if the deterministic ``should_run`` probe cannot be reached, a non-read-only tool is denied rather than allowed, so a flaky/broken control plane pauses delivery instead of running ungated. Read-only tools stay allowed. +An event this hook cannot read joins that rule: ``rc=1`` with no decision is a +non-blocking hook error to Claude Code, which then continues the tool call with +no gate at all, so an unreadable event (undecodable stdin, malformed or +non-object JSON, no session cwd or one that cannot be resolved as a path, no +usable tool name, or - for a non-read-only tool - a non-object ``tool_input`` or +a non-string command/file path field) is answered with an explicit deny instead +of dying. Read-only tools keep their unconditional allow. Identity-aware: when the active goal registered an agent (loopx ``coordination.registered_agents``), the context carries ``agent_id`` and this @@ -83,6 +94,22 @@ def within(path: str, root: str) -> bool: return False +def _usable_cwd(value: str) -> bool: + """True if a decoded session cwd can be resolved as a filesystem path. + + A string can still be unusable as a path: an embedded NUL makes + ``Path.resolve`` raise ``ValueError``, and a lone surrogate raises + ``UnicodeEncodeError`` (a ``ValueError``) against the filesystem encoding. + ``active_context`` must never receive such a value, or the exception escapes + ``decide`` as the rc=1/no-decision hook error the module docstring refuses. + """ + try: + Path(value).resolve() + except (OSError, RuntimeError, ValueError): + return False + return True + + def _gh_prefix(): _exe = shutil.which("loopx") return [_exe] if _exe else [__import__("sys").executable, "-m", "loopx.cli"] @@ -152,11 +179,38 @@ def decide(ev: dict) -> dict: (or {} for no-op). Reused by the CLI hook AND the Agent SDK in-process hook. The hook is the whole gate (no OS sandbox): should_run (fail-closed) + - Edit/Write write_scope + a destructive-Bash denylist.""" + Edit/Write write_scope + a destructive-Bash denylist. An event value this + gate cannot use - no usable session cwd or tool name, or a non-object + tool_input / non-string command or file path on a non-read-only tool - is + refused rather than deferred, because the event cannot be classified (the + module docstring states the fail-closed rule).""" + def d(decision, reason): + return {"hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": decision, + "permissionDecisionReason": reason}} + # Claude Code feeds the session's working directory on the event; resolve the # goal from THAT project's registry, and gate only while goal-mode is armed # (the project's .claude/loop.md exists). Unrelated/unarmed projects -> no-op. - cwd = ev.get("cwd") or (ev.get("workspace") or {}).get("current_dir") + cwd = ev.get("cwd") + workspace = ev.get("workspace") + if not isinstance(cwd, str) or not cwd: + cwd = workspace.get("current_dir") if isinstance(workspace, dict) else None + if not isinstance(cwd, str) or not cwd: + # Every real PreToolUse event carries the session cwd. Without one this + # hook cannot tell an armed project from an unrelated one, so it cannot + # decide to stay out of the way: refusing is the same fail-closed rule + # the unreachable-should_run branch below applies. Unreadable events + # (see main) arrive here as an empty dict. + return d("deny", "unreadable PreToolUse event (no session cwd) - failing closed under goal-mode") + if not _usable_cwd(cwd): + # A decoded cwd can still be unusable as a path (embedded NUL, lone + # surrogate, OS-level resolution failure). active_context cannot + # attribute the event to a project then, so the event stays unreadable + # for this gate and is refused instead of raising out of decide. + return d("deny", "unreadable PreToolUse event (session cwd is not a usable path) - failing closed under goal-mode") + ctx = active_context(cwd) if not ctx: return {} # no goal here, or goal-mode off -> defer to normal flow @@ -165,18 +219,43 @@ def decide(ev: dict) -> dict: registry = ctx.get("registry") agent_id = ctx.get("agent_id") scope = ctx.get("write_scope") or [] - tool = ev.get("tool_name", "") - ti = ev.get("tool_input", {}) or {} - - def d(decision, reason): - return {"hookSpecificOutput": { - "hookEventName": "PreToolUse", - "permissionDecision": decision, - "permissionDecisionReason": reason}} + tool = ev.get("tool_name") + if not isinstance(tool, str) or not tool: + # Every real PreToolUse event names its tool; without a usable name the + # gate cannot apply its read-only/gated classification (and the set + # membership tests below would raise on an unhashable value). + return d("deny", "unreadable PreToolUse event (tool_name is not a usable string) - failing closed under goal-mode") + ti = ev.get("tool_input") + if ti is None: + # Only an absent/null field takes the empty-input default. Other falsy + # values (`[]`, `false`, `0`, `""`) are present non-objects and must + # stay visible to the check below instead of being hidden by `or {}`. + ti = {} if tool in READONLY_TOOLS: return d("allow", "read-only/safe tool under goal-mode") + if not isinstance(ti, dict): + # A tool_input that is present but not an object cannot be read: refuse + # the non-read-only tool instead of crashing on `ti.get` (same + # fail-closed rule as an unreachable `should_run` probe). + return d("deny", "unreadable PreToolUse event (tool_input is not an object) - failing closed under goal-mode") + + # Every value the gate reads below must be readable before the probe is + # consulted, so an unreadable event is refused whether or not the gate is + # open (an absent/null field keeps its empty default). + if tool in WRITE_TOOLS: + file_path = ti.get("file_path") + notebook_path = ti.get("notebook_path") + if file_path is not None and not isinstance(file_path, str): + return d("deny", "unreadable PreToolUse event (tool_input.file_path is not a string) - failing closed under goal-mode") + if notebook_path is not None and not isinstance(notebook_path, str): + return d("deny", "unreadable PreToolUse event (tool_input.notebook_path is not a string) - failing closed under goal-mode") + elif tool == "Bash": + command = ti.get("command") + if command is not None and not isinstance(command, str): + return d("deny", "unreadable PreToolUse event (tool_input.command is not a string) - failing closed under goal-mode") + sr = should_run(registry, goal_id, agent_id) # True / False / None if sr is False: return d("deny", f"goal '{goal_id}' should_run=false (quota/gate closed)") @@ -200,10 +279,15 @@ def d(decision, reason): def main(): - raw = sys.stdin.read() or "{}" try: - ev = json.loads(raw) - except Exception: + # `ValueError` also covers `UnicodeDecodeError`: the strict stdin pin + # (stdio_utf8) never repairs a bad byte into a different event, so an + # event that cannot be read becomes the explicit deny in decide() + # instead of a traceback with no decision. + ev = json.loads(sys.stdin.read()) + except ValueError: + ev = {} + if not isinstance(ev, dict): ev = {} result = decide(ev) sys.stdout.write(json.dumps(result) if result else "{}") diff --git a/tests/test_claude_goal_mode_hook_fail_closed.py b/tests/test_claude_goal_mode_hook_fail_closed.py new file mode 100644 index 0000000000..1427965849 --- /dev/null +++ b/tests/test_claude_goal_mode_hook_fail_closed.py @@ -0,0 +1,411 @@ +"""The Claude Code goal-mode gate must refuse an event it cannot read. + +``hooks/goal_policy.py`` documents itself as fail-closed around the +deterministic ``should_run`` probe: when the probe cannot be reached, a +non-read-only tool is denied so the gate never silently disappears. The event +read was the one boundary outside that rule. With the strict stdin pin +(``test_claude_goal_mode_stdio_utf8.py``) an undecodable byte raises +``UnicodeDecodeError`` at the read, and a decodable-but-malformed payload raised +inside ``active_context``: the hook exited ``rc=1`` with a traceback and printed +no decision. Claude Code treats a hook that fails that way as a non-blocking +error and continues the tool call, so the gate vanished exactly when the hook +could not verify anything. #6166 named this as its deliberately deferred +successor. + +The same read boundary also covers values that decode but cannot be used: a +session ``cwd`` string that is not a resolvable path (an embedded NUL raises +``ValueError`` inside ``active_context``), a missing/non-string ``tool_name``, +a falsy non-object ``tool_input`` (``[]`` / ``false`` / ``0`` / ``""``, which +the old ``or {}`` coerced before the type check), and a non-string +``command`` / ``file_path`` / ``notebook_path`` on a gated tool. Those are +refused the same way; the read-only exception and the absent-``tool_input`` +default are pinned so the refusal does not widen onto valid events. + +These cases run the shipped script the way Claude Code invokes it - own process, +event on stdin, answer read back as JSON - and assert the rc/decision contract, +so the regression fails on the real entry point rather than on an in-process +approximation. The decision-level cases drive ``decide`` directly against an +armed context with the gate open (``should_run`` stubbed true), which is what +made the falsy-coercion failure visible. The unarmed case pins the other +direction: the refusal must not widen the gate onto projects where goal-mode is +off. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +HOOK = REPO_ROOT / "loopx/claude_goal_mode/hooks/goal_policy.py" + +GOAL_ID = "hook-fail-closed-goal" + + +def _project(root: Path, *, armed: bool) -> Path: + """A project with one registry goal; armed also carries `.claude/loop.md`.""" + + registry = root / ".loopx" / "registry.json" + registry.parent.mkdir(parents=True, exist_ok=True) + registry.write_text( + json.dumps( + { + "schema_version": "0.1", + "goals": [ + { + "id": GOAL_ID, + "repo": str(root), + "coordination": {"registered_agents": ["cc"]}, + } + ], + } + ), + encoding="utf-8", + ) + if armed: + (root / ".claude").mkdir(parents=True, exist_ok=True) + (root / ".claude" / "loop.md").write_text("# loop\n", encoding="utf-8") + return root + + +def _run( + project: Path, payload: bytes, *, force_armed: bool +) -> subprocess.CompletedProcess[bytes]: + environment = {**os.environ} + if force_armed: + environment["LOOPX_GOAL_FORCE"] = "1" + else: + environment.pop("LOOPX_GOAL_FORCE", None) + return subprocess.run( + [sys.executable, str(HOOK)], + cwd=project, + env=environment, + input=payload, + capture_output=True, + timeout=120, + ) + + +UNREADABLE_EVENT_PAYLOADS = [ + pytest.param(b"\xff\xfe{", id="undecodable-stdin"), + pytest.param(b"{not json", id="malformed-json"), + pytest.param(b"[]", id="json-not-an-object"), + pytest.param(b"", id="empty-stdin"), + pytest.param( + json.dumps({"tool_name": "Bash", "tool_input": {"command": "rm -rf /"}}).encode( + "utf-8" + ), + id="object-without-cwd", + ), + pytest.param( + json.dumps({"tool_name": "Bash", "workspace": "not-an-object"}).encode("utf-8"), + id="non-object-workspace", + ), +] + +UNUSABLE_CWD_EVENTS = [ + pytest.param( + { + "tool_name": "Bash", + "cwd": "/bad\u0000cwd", + "tool_input": {"command": "ls"}, + }, + id="cwd-nul-byte", + ), + pytest.param( + { + "tool_name": "Bash", + "workspace": {"current_dir": "/bad\u0000cwd"}, + "tool_input": {"command": "ls"}, + }, + id="workspace-current-dir-nul-byte", + ), +] + +FALSY_NON_OBJECT_TOOL_INPUTS = [ + pytest.param([], id="empty-list"), + pytest.param(False, id="false"), + pytest.param(0, id="zero"), + pytest.param("", id="empty-string"), +] + +UNREADABLE_EVENT_VALUES = [ + pytest.param( + {"tool_name": ["Bash"], "tool_input": {"command": "ls"}}, + "tool_name", + id="non-string-tool-name", + ), + pytest.param( + {"tool_name": "Bash", "tool_input": {"command": ["rm", "/"]}}, + "command", + id="non-string-command", + ), + pytest.param( + {"tool_name": "Write", "tool_input": {"file_path": ["/proj/notes.md"]}}, + "file_path", + id="non-string-file-path", + ), + pytest.param( + { + "tool_name": "NotebookEdit", + "tool_input": {"notebook_path": {"p": "/proj/n.ipynb"}}, + }, + "notebook_path", + id="non-string-notebook-path", + ), +] + +ARMED_CTX = { + "goal_id": GOAL_ID, + "registry": "/tech/registry.json", + "agent_id": "cc", + "write_scope": ["/proj"], + "project_root": "/proj", +} + + +@pytest.mark.parametrize("payload", UNREADABLE_EVENT_PAYLOADS) +def test_unreadable_event_is_refused_with_an_explicit_deny( + tmp_path: Path, payload: bytes +) -> None: + project = _project(tmp_path / "armed", armed=True) + + result = _run(project, payload, force_armed=True) + + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + assert b"Traceback" not in result.stderr, result.stderr.decode("utf-8", "replace") + decision = json.loads(result.stdout.decode("utf-8")) + output = decision["hookSpecificOutput"] + assert output["hookEventName"] == "PreToolUse" + assert output["permissionDecision"] == "deny" + assert output["permissionDecisionReason"] + + +@pytest.mark.parametrize("event", UNUSABLE_CWD_EVENTS) +def test_unusable_cwd_is_refused_with_an_explicit_deny( + tmp_path: Path, event: dict +) -> None: + """A decoded cwd string can still be unusable as a path. + + ``Path.resolve`` on a string with an embedded NUL raises ``ValueError`` + inside ``active_context``, so the pre-fix hook exited rc=1 with no decision + and a traceback - the same fail-open signature the unreadable-event refusal + removes. The reason substring pins the refusal to the usable-path rule, so + a deny from another branch (for example the closed gate) cannot satisfy the + regression. + """ + + project = _project(tmp_path / "armed", armed=True) + + result = _run(project, json.dumps(event).encode("utf-8"), force_armed=True) + + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + assert b"Traceback" not in result.stderr, result.stderr.decode("utf-8", "replace") + output = json.loads(result.stdout.decode("utf-8"))["hookSpecificOutput"] + assert output["permissionDecision"] == "deny" + assert "usable path" in output["permissionDecisionReason"] + + +def test_non_object_tool_input_for_a_gated_tool_is_refused(tmp_path: Path) -> None: + """A truthy non-object `tool_input` is an unreadable event, not a crash. + + ``decide`` reads the tool input with ``ti.get`` for the gated tools, so on + the pre-fix hook a Bash event carrying a non-object input died with a + traceback (rc=1, no decision) as soon as ``should_run`` returned true. The + reason substring pins the refusal to the unreadable-input rule: a deny from + the probe path carries a different reason, so the regression fails against + the old hook whichever branch the probe takes. + """ + + project = _project(tmp_path / "armed", armed=True) + event = json.dumps( + { + "tool_name": "Bash", + "cwd": str(project), + "tool_input": "not-an-object", + } + ).encode("utf-8") + + result = _run(project, event, force_armed=True) + + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + assert b"Traceback" not in result.stderr, result.stderr.decode("utf-8", "replace") + output = json.loads(result.stdout.decode("utf-8"))["hookSpecificOutput"] + assert output["permissionDecision"] == "deny" + assert "tool_input" in output["permissionDecisionReason"] + + # Read-only tools keep their unconditional allow even then: the refusal + # must not widen the gate past the tools it was meant to gate. + read_event = json.dumps( + { + "tool_name": "Read", + "cwd": str(project), + "tool_input": "not-an-object", + } + ).encode("utf-8") + read_result = _run(project, read_event, force_armed=True) + assert read_result.returncode == 0, read_result.stderr.decode("utf-8", "replace") + read_output = json.loads(read_result.stdout.decode("utf-8"))["hookSpecificOutput"] + assert read_output["permissionDecision"] == "allow" + + +@pytest.mark.parametrize("tool_input", FALSY_NON_OBJECT_TOOL_INPUTS) +def test_falsy_non_object_tool_input_is_refused_not_coerced( + tmp_path: Path, tool_input +) -> None: + """``tool_input`` must be type-checked before any default is applied. + + The pre-review hook read ``ev.get("tool_input", {}) or {}``, so a present + falsy non-object (``[]`` / ``false`` / ``0`` / ``""``) became ``{}`` before + the non-object check; with an open gate the pre-review ``decide`` answered + "allow / bash permitted under goal policy" for it. The reason substring + proves the deny comes from the unreadable-input rule rather than the + closed-gate branch. + """ + + project = _project(tmp_path / "armed", armed=True) + event = json.dumps( + {"tool_name": "Bash", "cwd": str(project), "tool_input": tool_input} + ).encode("utf-8") + + result = _run(project, event, force_armed=True) + + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + assert b"Traceback" not in result.stderr, result.stderr.decode("utf-8", "replace") + output = json.loads(result.stdout.decode("utf-8"))["hookSpecificOutput"] + assert output["permissionDecision"] == "deny" + assert "tool_input" in output["permissionDecisionReason"] + + +@pytest.mark.parametrize("partial,field", UNREADABLE_EVENT_VALUES) +def test_unreadable_event_value_is_refused_with_an_explicit_deny( + tmp_path: Path, partial: dict, field: str +) -> None: + """A decoded event value the gate cannot use is refused, never a traceback. + + A non-string ``tool_name`` crashed the set membership tests, a non-string + ``command`` crashed ``.lower()``, and a non-string file path silently + resolved to the project root - all variations of the same fail-open event + that the gate must refuse instead. + """ + + project = _project(tmp_path / "armed", armed=True) + event = json.dumps({"cwd": str(project), **partial}).encode("utf-8") + + result = _run(project, event, force_armed=True) + + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + assert b"Traceback" not in result.stderr, result.stderr.decode("utf-8", "replace") + output = json.loads(result.stdout.decode("utf-8"))["hookSpecificOutput"] + assert output["permissionDecision"] == "deny" + assert field in output["permissionDecisionReason"] + + +@pytest.fixture() +def hook_policy(): + """The shipped hook module, imported in-process for ``decide`` cases.""" + + hooks_dir = REPO_ROOT / "loopx" / "claude_goal_mode" / "hooks" + if str(hooks_dir) not in sys.path: + sys.path.insert(0, str(hooks_dir)) + import goal_policy + + return goal_policy + + +@pytest.fixture() +def armed_decide(hook_policy, monkeypatch): + """``decide`` against an armed project with the gate OPEN (should_run=true).""" + + monkeypatch.setattr(hook_policy, "active_context", lambda cwd: ARMED_CTX) + monkeypatch.setattr(hook_policy, "should_run", lambda *a, **k: True) + return hook_policy.decide + + +def test_decide_refuses_an_unusable_cwd_before_resolving_the_project( + hook_policy, monkeypatch +) -> None: + def never_active(cwd): + pytest.fail(f"active_context received an unusable cwd: {cwd!r}") + + monkeypatch.setattr(hook_policy, "active_context", never_active) + + output = hook_policy.decide( + {"cwd": "/bad\u0000cwd", "tool_name": "Bash", "tool_input": {"command": "ls"}} + )["hookSpecificOutput"] + + assert output["permissionDecision"] == "deny" + assert "usable path" in output["permissionDecisionReason"] + + +def test_decide_keeps_the_empty_input_default_for_an_absent_tool_input( + armed_decide, +) -> None: + """Only an absent/null field takes the permitted ``{}`` default.""" + + output = armed_decide({"cwd": "/proj", "tool_name": "Bash"})["hookSpecificOutput"] + + assert output["permissionDecision"] == "allow" + assert output["permissionDecisionReason"] == "bash permitted under goal policy" + + +@pytest.mark.parametrize("tool_input", FALSY_NON_OBJECT_TOOL_INPUTS) +def test_decide_refuses_falsy_non_object_tool_input_when_the_gate_is_open( + armed_decide, tool_input +) -> None: + """With should_run=true the pre-review ``decide`` allowed these (P2).""" + + output = armed_decide( + {"cwd": "/proj", "tool_name": "Bash", "tool_input": tool_input} + )["hookSpecificOutput"] + + assert output["permissionDecision"] == "deny" + assert "tool_input" in output["permissionDecisionReason"] + + +@pytest.mark.parametrize("partial,field", UNREADABLE_EVENT_VALUES) +def test_decide_refuses_unreadable_event_values_when_the_gate_is_open( + armed_decide, partial: dict, field: str +) -> None: + output = armed_decide({"cwd": "/proj", **partial})["hookSpecificOutput"] + + assert output["permissionDecision"] == "deny" + assert field in output["permissionDecisionReason"] + + +def test_decide_keeps_the_read_only_exception_for_unreadable_input( + armed_decide, +) -> None: + """The gated-tool refusal must not capture read-only tools.""" + + output = armed_decide( + {"cwd": "/proj", "tool_name": "Read", "tool_input": []} + )["hookSpecificOutput"] + + assert output["permissionDecision"] == "allow" + assert output["permissionDecisionReason"] == "read-only/safe tool under goal-mode" + + +def test_unarmed_project_still_defers_to_the_normal_flow(tmp_path: Path) -> None: + project = _project(tmp_path / "unarmed", armed=False) + + # A valid event in a project with a registry goal but no `.claude/loop.md` + # (goal-mode off): its write tool must still defer to Claude Code's normal + # permission flow, not be captured by the new refusal. + event = json.dumps( + { + "tool_name": "Edit", + "cwd": str(project), + "tool_input": {"file_path": str(project / "notes.txt")}, + } + ).encode("utf-8") + result = _run(project, event, force_armed=False) + + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + # `{}` is Claude Code's "defer to the normal permission flow". + assert result.stdout == b"{}"