diff --git a/emrg/sandbox/escalation.py b/emrg/sandbox/escalation.py index 6958fa5c..bc19a602 100644 --- a/emrg/sandbox/escalation.py +++ b/emrg/sandbox/escalation.py @@ -49,6 +49,7 @@ __all__ = [ "APPROVAL_TIMEOUT_SECONDS", + "ESCALATION_PARAMETERS", "ESCALATION_TARGETS", "ApprovalOutcome", "EscalationRefused", @@ -69,6 +70,41 @@ #: what is reachable from where it actually stands. ESCALATION_TARGETS: tuple[str, ...] = ("workspace-write", "danger-full-access") +#: The two fields a tier-limited tool advertises, **as one object**. +#: +#: ``bash``, ``pwsh``, ``write`` and ``edit`` each consume the tier (``write`` +#: and ``edit`` the writable roots with it, ``_ROOT_CONSUMING_TOOLS``), so all +#: four may ask for one hop and all four must say so in the same words: a model +#: reading two schemas that describe one mechanism two ways has to decide which +#: one it is. Four hand-written copies were what this replaced — **equal by +#: value**, which is precisely why a value comparison could not tell them apart +#: from one source (issue #2734). Each tool spreads this dict into its +#: ``properties``, so the dicts that reach the wire *are* these dicts and a fifth +#: copy is a failure at the moment it is written rather than an equivalence +#: somebody has to notice later. +#: +#: ``enum`` is the whole target set, not the current tier's row: the blueprint +#: advertises what is reachable in principle and lets :func:`validate_hop` decide +#: what *this* call may reach (``escalation.ts:41``). +ESCALATION_PARAMETERS: dict[str, dict] = { + "sandbox_permissions": { + "type": "string", + "enum": list(ESCALATION_TARGETS), + "description": "Ask for a one-hop wider sandbox tier for THIS " + "call only. Must be sent together with `justification`, and it " + "is refused unless the host approves: it never changes the " + "session's default tier, and it reaches only the tiers the " + "call's current one lists (from `read-only`: `workspace-write` " + "or `danger-full-access`).", + }, + "justification": { + "type": "string", + "description": "Why this call needs the wider tier — a non-empty " + "sentence the host reads before approving. Required with " + "`sandbox_permissions`.", + }, +} + #: The strict-wider table, verbatim from ``escalation.ts:28-31``. A mode with #: no entry has nowhere to escalate to, and an entry listing no mode is the same #: statement — both are read through :func:`hops_from` so there is one table. diff --git a/emrg/sandbox/fence.py b/emrg/sandbox/fence.py index 7e4b82f9..f3007d1c 100644 --- a/emrg/sandbox/fence.py +++ b/emrg/sandbox/fence.py @@ -50,6 +50,7 @@ import os +from emrg.sandbox.escalation import with_retry_hint from emrg.sandbox.grants import GrantKindNotEnforced from emrg.sandbox.policy import DANGER_FULL_ACCESS, SandboxPolicy from emrg.sandbox.roots import canonical_path, named_grants, writable_roots @@ -132,10 +133,26 @@ def file_refusal(file_path: str, policy: SandboxPolicy) -> str | None: # the same file. Compared as the canonical identity both sides of the # comparison use, so one assertion holds on either platform. resolved = f" (resolves to {target!r})" if target != file_path else "" - return ( + refusal = ( f"{policy.mode} sandbox: blocked file write to {file_path!r}{resolved} — " f"outside every root this tier grants ({granted})" ) + # A file refusal that names no way out reads as a dead end, and a dead end + # invites the workaround rather than the question: the model would echo the + # file from `bash`, which is the tool that *can* ask — the escalation moving + # from the front door to the side door, with the host told in neither case + # (issue #2734). So the sentence carries the shell's own next step, through + # the one producer of it; `with_retry_hint` appends nothing at a tier with no + # hop, which is why a `danger-full-access` refusal stays as it was — it + # already returned above, and an instruction that could only fail is not + # advice. + # + # The two branches above keep no hint, and the reason is the cause rather + # than the shape: a protected daemon file is refused for being the host's own + # state, and an unenforceable grant for a platform that cannot apply it. + # Neither is a root boundary, so "ask for one wider tier" is not their + # remedy — one remedy for every cause is false for most of them. + return with_retry_hint(refusal, mode=policy.mode) def _is_within_root(target: str, root: str) -> bool: diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index 4dc76a4c..1981365b 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -4497,7 +4497,12 @@ async def _apply_escalation( :returns: ``None`` to run the call, or the refusal text to report instead. """ - if tc_name not in SHELL_TOOL_NAMES: + # The set that may ask is the set the tier and the roots reach, not the + # shell's names: `write` and `edit` consume both, so a refusal that told + # them apart from a command was an answer about the tool rather than + # about the boundary (issue #2734). One name for the set, read here and + # at the injection site, so the two cannot drift. + if tc_name not in _ROOT_CONSUMING_TOOLS: return None raw_target = args.get("sandbox_permissions") raw_reason = args.get("justification") @@ -4523,7 +4528,7 @@ async def _apply_escalation( # The command does NOT run at the wider tier, and saying so is the # whole point: a silent "no" reads as a broken command. return ( - f"⛔ escalation refused ({outcome.reason}): this command runs at " + f"⛔ escalation refused ({outcome.reason}): this call runs at " f"{hop.from_mode!r}. Nothing was widened, and the session's " "default tier is unchanged." ) diff --git a/emrg/tools/bash_tool_v2.py b/emrg/tools/bash_tool_v2.py index 3503bed3..dbe9b2a2 100644 --- a/emrg/tools/bash_tool_v2.py +++ b/emrg/tools/bash_tool_v2.py @@ -62,7 +62,7 @@ from emrg.server.redaction import redact_argv from emrg.server.tool_types import ToolDefinition, ToolResult from emrg.tools import command_scan -from emrg.sandbox.escalation import ESCALATION_TARGETS +from emrg.sandbox.escalation import ESCALATION_PARAMETERS from emrg.sandbox.escalation import hops_from as escalation_hops from emrg.tools.base import ToolExecutor @@ -674,22 +674,7 @@ def definition(self) -> ToolDefinition: "caller's to set: it is also the writable boundary, and a " "value passed here does not widen it.", }, - "sandbox_permissions": { - "type": "string", - "enum": list(ESCALATION_TARGETS), - "description": "Ask for a one-hop wider sandbox tier for THIS " - "call only. Must be sent together with `justification`, and it " - "is refused unless the host approves: it never changes the " - "session's default tier, and it reaches only the tiers the " - "call's current one lists (from `read-only`: `workspace-write` " - "or `danger-full-access`).", - }, - "justification": { - "type": "string", - "description": "Why this call needs the wider tier — a non-empty " - "sentence the host reads before approving. Required with " - "`sandbox_permissions`.", - }, + **ESCALATION_PARAMETERS, "intent": { "type": "string", "description": "The purpose of this call: why you are invoking it and what you want to achieve. " diff --git a/emrg/tools/edit_tool.py b/emrg/tools/edit_tool.py index c47ce3a1..36ee6658 100644 --- a/emrg/tools/edit_tool.py +++ b/emrg/tools/edit_tool.py @@ -6,6 +6,7 @@ import os from pathlib import Path +from emrg.sandbox.escalation import ESCALATION_PARAMETERS from emrg.sandbox.fence import file_refusal from emrg.sandbox.policy import grants_from_arguments, resolve_policy from emrg.server.tool_types import ToolDefinition, ToolResult @@ -70,6 +71,7 @@ def definition(self) -> ToolDefinition: "edit writes to." ), }, + **ESCALATION_PARAMETERS, "intent": { "type": "string", "description": "The purpose of this call: why you are invoking it and what you want to achieve. " diff --git a/emrg/tools/pwsh_tool_v2.py b/emrg/tools/pwsh_tool_v2.py index bd1ff35d..902772e5 100644 --- a/emrg/tools/pwsh_tool_v2.py +++ b/emrg/tools/pwsh_tool_v2.py @@ -73,7 +73,7 @@ from emrg.server.redaction import redact_argv from emrg.server.tool_types import ToolDefinition, ToolResult from emrg.tools import command_scan -from emrg.sandbox.escalation import ESCALATION_TARGETS +from emrg.sandbox.escalation import ESCALATION_PARAMETERS from emrg.sandbox.escalation import hops_from as escalation_hops from emrg.tools.base import ToolExecutor from emrg.tools.shell_env import confined_env, runner_import_env @@ -792,22 +792,7 @@ def definition(self) -> ToolDefinition: "caller's to set: it is also the writable boundary, and a " "value passed here does not widen it.", }, - "sandbox_permissions": { - "type": "string", - "enum": list(ESCALATION_TARGETS), - "description": "Ask for a one-hop wider sandbox tier for THIS " - "call only. Must be sent together with `justification`, and it " - "is refused unless the host approves: it never changes the " - "session's default tier, and it reaches only the tiers the " - "call's current one lists (from `read-only`: `workspace-write` " - "or `danger-full-access`).", - }, - "justification": { - "type": "string", - "description": "Why this call needs the wider tier — a non-empty " - "sentence the host reads before approving. Required with " - "`sandbox_permissions`.", - }, + **ESCALATION_PARAMETERS, "intent": { "type": "string", "description": "The purpose of this call: why you are invoking it and what you want to achieve. " diff --git a/emrg/tools/write_tool.py b/emrg/tools/write_tool.py index f36ae0ea..8e6dba97 100644 --- a/emrg/tools/write_tool.py +++ b/emrg/tools/write_tool.py @@ -6,6 +6,7 @@ import os from pathlib import Path +from emrg.sandbox.escalation import ESCALATION_PARAMETERS from emrg.sandbox.fence import file_refusal from emrg.sandbox.policy import grants_from_arguments, resolve_policy from emrg.server.tool_types import ToolDefinition, ToolResult @@ -119,6 +120,7 @@ def definition(self) -> ToolDefinition: "type": "string", "description": "The complete content to write to the file.", }, + **ESCALATION_PARAMETERS, "intent": { "type": "string", "description": "The purpose of this call: why you are invoking it and what you want to achieve. " diff --git a/tests/test_sandbox_escalation.py b/tests/test_sandbox_escalation.py index a00a81b6..c0741649 100644 --- a/tests/test_sandbox_escalation.py +++ b/tests/test_sandbox_escalation.py @@ -25,6 +25,7 @@ import asyncio import json +import os import tempfile from pathlib import Path @@ -33,7 +34,7 @@ from emrg.config import LlmConfig from emrg.protocol import TaskRequest from emrg.sandbox import escalation -from emrg.sandbox.policy import DANGER_FULL_ACCESS +from emrg.sandbox.policy import DANGER_FULL_ACCESS, resolve_policy from emrg.server import daemon as daemon_mod from emrg.server.daemon import EmrgServer from emrg.server.tool_types import ToolDefinition, ToolResult @@ -55,11 +56,14 @@ async def send(self, data) -> None: class _StubTool: - """A shell-tool stand-in: records the arguments the daemon handed it. - - Registered under a real shell tool's name on purpose — `_apply_escalation` - keys on `SHELL_TOOL_NAMES`, so a stand-in under a made-up name would test - the branch that skips escalation instead of the one that takes it. + """A stand-in for one of the tools the daemon gives a tier to: it records + the arguments the daemon handed it. + + Registered under a real tool's name on purpose — `_apply_escalation` keys on + `_ROOT_CONSUMING_TOOLS`, the set the tier and the writable roots are injected + into, so a stand-in under a made-up name would test the branch that skips + escalation instead of the one that takes it. The name is not restricted to + the shell's: `write` and `edit` are in that set too (issue #2734). """ def __init__(self, name: str) -> None: @@ -86,7 +90,15 @@ def _subscribe(server: EmrgServer, session: Session) -> _FakeWs: return ws -async def _answer_next_question(server: EmrgServer, ws: _FakeWs, answer=object()): +#: The "this test answers nothing" sentinel. A bare `object()` cannot be one: +#: `answer is not object()` constructs a fresh object each time it is evaluated +#: and so answers `True` for every value, which made "nobody answered" a state +#: this helper could not express — it always sent *some* answer, and an unreadable +#: one at that. +_NO_ANSWER = object() + + +async def _answer_next_question(server: EmrgServer, ws: _FakeWs, answer=_NO_ANSWER): """Answer the first approval question the daemon broadcasts. Mirrors what a client does rather than reaching into the future: the test @@ -98,7 +110,7 @@ async def _answer_next_question(server: EmrgServer, ws: _FakeWs, answer=object() continue frame["_answered"] = True payload: dict = {"request_id": frame.get("request_id")} - if answer is not object(): + if answer is not _NO_ANSWER: payload["approved"] = answer server._resolve_approval(payload) return frame @@ -620,7 +632,7 @@ async def scenario(): def _drive_shell_call( - tmp_path, monkeypatch, *, args: dict, answer=object(), sandbox="read-only", + tmp_path, monkeypatch, *, args: dict, answer=_NO_ANSWER, sandbox="read-only", tool_name=None, ): """Run one tool loop whose single tool call carries `args`. @@ -668,7 +680,7 @@ async def fake_stream(messages, tools=None): async def scenario(): task = asyncio.create_task(server._run_tool_loop(req, None, session)) - if answer is not object(): + if answer is not _NO_ANSWER: await _answer_next_question(server, ws, answer) await asyncio.wait_for(task, 10) @@ -738,16 +750,40 @@ def test_a_shell_call_without_the_pair_is_untouched(tmp_path, monkeypatch): assert stub.calls and stub.calls[0].get("sandbox") == "read-only", stub.calls -def test_a_tool_the_tier_does_not_reach_cannot_escalate(tmp_path, monkeypatch): - """`read` is not a shell tool: the daemon injects no tier for it, so the - field is a stray argument rather than a request — and no question is asked.""" +#: The three tools that stay outside the escalation surface, with arguments that +#: would otherwise run. They are in no tier's reach (`_ROOT_CONSUMING_TOOLS` omits +#: them) because they consume no root, so there is nothing to widen and the field +#: is a stray argument rather than a request. +_CLAMPED_CALLS = { + "read": {"path": "README.md"}, + "glob": {"pattern": "*.py"}, + "grep": {"pattern": "def "}, +} + + +@pytest.mark.parametrize("tool_name", sorted(_CLAMPED_CALLS)) +def test_a_tool_the_tier_does_not_reach_cannot_escalate( + tmp_path, monkeypatch, tool_name, +): + """No root, no tier, no hop: the pair is inert here. + + Requirement 4 of issue #2734, and the guard is the *pair* of assertions — + no question was asked, and the call ran exactly as it would have without the + two arguments. Asserting only the first would pass a build that silently + dropped the call. + """ stub, ws, _session = _drive_shell_call( - tmp_path, monkeypatch, tool_name="read", - args={"path": "README.md", "sandbox_permissions": "workspace-write", - "justification": "I want to read"}, + tmp_path, monkeypatch, tool_name=tool_name, + args={**_CLAMPED_CALLS[tool_name], + "sandbox_permissions": "workspace-write", + "justification": "I want a wider tier"}, ) assert not [f for f in ws.sent if f.get("type") == "approval_request"] assert stub.calls, "the tool did not run" + # The tier the daemon injected is the session's own, not the one the call + # named — a field that changed the tier without a question would be the + # defect this test exists for, and `read` has no tier injected at all. + assert stub.calls[0].get("sandbox") in (None, "read-only"), stub.calls # ── requirement 5: the hint line ──────────────────────────────────────────── @@ -882,3 +918,295 @@ async def fake_run_command(command, *, policy, workdir, timeout): {"command": "echo x > /tmp/nope", "sandbox": "read-only", "workspace": str(tmp_path)}, )) assert escalation.RETRY_HINT in denied.content, denied.content + + +# ── the file tools may ask for the same one hop ───────────────────────────── +# +# The shell was the only tool whose call could escalate, and the refusal a +# `write` got named no way out — so a model blocked on a file path had one move +# left: echo the content from `bash`, the tool that *can* ask. The escalation +# then happened through the side door, with the host told in neither case +# (issue #2734). Three things have to hold at once and each is asserted on its +# own: the gate recognizes the file tools, their schema says the field exists, +# and the refusal names the same next step the shell's does. + +_FILE_CALL_ARGS = { + "write": {"path": "out.txt", "content": "x"}, + "edit": {"path": "out.txt", "old_string": "a", "new_string": "b"}, +} + + +#: Every tool the daemon hands a tier to, by the name the daemon injects under. +#: The set on the left is checked against `_ROOT_CONSUMING_TOOLS` below, so a +#: carrier added to the daemon and forgotten here is a red test rather than a +#: tool that may ask and does not say so. +_ADVERTISING_TOOLS: dict[str, type] = {} + + +def _advertising_tools() -> dict[str, type]: + """The tool class behind each tier-limited name, imported lazily.""" + from emrg.tools.bash_tool_v2 import BashToolV2 + from emrg.tools.edit_tool import EditTool + from emrg.tools.pwsh_tool_v2 import PwshToolV2 + from emrg.tools.write_tool import WriteTool + + return { + "bash": BashToolV2, "pwsh": PwshToolV2, + "write": WriteTool, "edit": EditTool, + } + + +def test_every_tier_limited_tool_advertises_the_pair_from_one_source(): + """Requirement 2, read at the wire: the model has to know it may ask. + + A gate that accepted the pair while the schema advertised nothing would grant + a hop nobody can request. Two things are asserted, and the second is the one + a value comparison could not make: **identity**, not equality. Four + hand-written copies of this block were equal to each other, so any test + comparing values stayed green while a fifth copy was written — the drift this + replaces (issue #2734, and the rule that one rule with several carriers needs + a producer rather than a comparison). A hand-copied literal is a different + object, and it reddens here. + """ + classes = _advertising_tools() + assert set(classes) == set(daemon_mod._ROOT_CONSUMING_TOOLS), ( + "the daemon and this test disagree about which tools may escalate" + ) + for name, tool_class in classes.items(): + properties = tool_class().definition().parameters["properties"] + assert properties["sandbox_permissions"] is ( + escalation.ESCALATION_PARAMETERS["sandbox_permissions"] + ), name + assert properties["justification"] is ( + escalation.ESCALATION_PARAMETERS["justification"] + ), name + # …and the enum is the module's own list, so the advertised set cannot age + # away from the table `validate_hop` reads. + assert set(escalation.ESCALATION_PARAMETERS["sandbox_permissions"]["enum"]) == set( + escalation.ESCALATION_TARGETS + ) + + +@pytest.mark.parametrize("tool_name", ("read", "glob", "grep")) +def test_a_tool_outside_the_surface_advertises_no_hop(tool_name): + """The other side of the line: no root, no tier, nothing to widen. + + "One remedy for every cause" is the defect this keeps out — a `grep` told it + may ask for a wider tier would be told to do something that cannot change + its answer. + """ + from emrg.tools.glob_tool import GlobTool + from emrg.tools.grep_tool import GrepTool + from emrg.tools.read_tool import ReadTool + + tool_class = {"read": ReadTool, "glob": GlobTool, "grep": GrepTool}[tool_name] + properties = tool_class().definition().parameters["properties"] + assert "sandbox_permissions" not in properties, tool_name + assert "justification" not in properties, tool_name + + +@pytest.mark.parametrize("tool_name", ("write", "edit")) +def test_a_file_tool_call_that_names_a_hop_asks_before_it_runs( + tmp_path, monkeypatch, tool_name, +): + """Requirement 1: the gate recognizes the file tools. + + `_apply_escalation` keyed on `SHELL_TOOL_NAMES` alone, so the identical + request from `write` was a stray argument — no question, no widening, and + nothing said. The stub is registered under the real tool's name for the + same reason the shell stand-in is: the gate keys on a name set. + """ + stub, ws, _session = _drive_shell_call( + tmp_path, monkeypatch, tool_name=tool_name, + args={**_FILE_CALL_ARGS[tool_name], + "sandbox_permissions": "workspace-write", + "justification": "the target sits outside every root this tier grants"}, + answer=True, + ) + + asked = [f for f in ws.sent if f.get("type") == "approval_request"] + assert len(asked) == 1, "the daemon widened nothing without asking" + assert "read-only" in asked[0]["question"] + assert stub.calls, "an approved escalation did not run the call" + assert stub.calls[0]["sandbox"] == "workspace-write" + + +@pytest.mark.parametrize("tool_name", ("write", "edit")) +@pytest.mark.parametrize( + "answer, why", + ( + (False, "the host said no"), + (None, "an answer that maps to no boolean"), + (_NO_ANSWER, "a client that never replies"), + ), + ids=("refused", "unreadable", "silent"), +) +def test_a_file_tool_that_was_not_approved_does_not_run( + tmp_path, monkeypatch, tool_name, answer, why, +): + """Fail closed, three ways, with the semantics the shell already has. + + All three are refusals, and testing all three is not repetition: they reach + the refusal through different code — the host's explicit no, an answer nobody + can map to a boolean, and a client that never replies. A tool that escalated + correctly but ran on the "no" path would be worse than one that never + escalated, because the model would believe it had been granted something. + + The question itself is asserted too: a refusal the model is never told about + is indistinguishable from a broken tool. + """ + # The wait the silent case would otherwise spend 120 s on; the other two + # answer immediately and do not reach it. + monkeypatch.setattr(escalation, "APPROVAL_TIMEOUT_SECONDS", 0.05) + stub, ws, session = _drive_shell_call( + tmp_path, monkeypatch, tool_name=tool_name, + args={**_FILE_CALL_ARGS[tool_name], + "sandbox_permissions": "workspace-write", + "justification": "needs one hop"}, + answer=answer, + ) + + assert [f for f in ws.sent if f.get("type") == "approval_request"], why + assert stub.calls == [], f"{why}: the call ran anyway" + results = [m for m in session.get_messages_for_llm() if m.get("role") == "tool"] + assert any("escalation refused" in r["content"] for r in results), results + # The sentence is the shell's own, which is why it says "call" and no longer + # "command" — one branch, one wording, whichever tool was refused. + assert any("this call runs at" in r["content"] for r in results), results + + +def _drive_two_file_calls(tmp_path, monkeypatch, *, first: dict, second: dict): + """Two rounds, two `write` calls, one question — the widening is the first + call's alone.""" + monkeypatch.setattr(daemon_mod, "_PLANTED_FIRE_MARKER_PATH", tmp_path / "marker") + monkeypatch.setattr( + daemon_mod, "_PLANTED_FIRE_ROUND_COMPLETE_PATH", tmp_path / "round-complete" + ) + server = _server() + session = Session.create_with_id("esc-two-calls", tmp_path) + ws = _subscribe(server, session) + stub = _StubTool("write") + server.tools._tools["write"] = stub + + payloads = [first, second] + round_no = 0 + + async def fake_stream(messages, tools=None): + nonlocal round_no + if round_no < len(payloads): + index = round_no + round_no += 1 + yield { + "content": "", "finish_reason": "tool_calls", + "usage": {"prompt_tokens": 10, "completion_tokens": 1}, + "tool_calls": [{ + "index": 0, "id": f"call_{index + 1}", "type": "function", + "function": { + "name": "write", + "arguments": json.dumps(payloads[index]), + }, + }], + } + else: + yield {"content": "done", "tool_calls": None, "finish_reason": "stop", + "usage": {"prompt_tokens": 10, "completion_tokens": 1}} + + server.llm.chat_stream = fake_stream + req = TaskRequest(id="req-esc-two", session_id=session.session_id, + prompt="go", sandbox="read-only") + + async def scenario(): + task = asyncio.create_task(server._run_tool_loop(req, None, session)) + await _answer_next_question(server, ws, True) + await asyncio.wait_for(task, 10) + + asyncio.run(scenario()) + return stub, ws, session + + +def test_the_widening_is_that_one_call_and_the_next_goes_back(tmp_path, monkeypatch): + """The requirement's second half: the next call runs at the session's own + tier again. A grant that stuck would be a session silently widened by one + approved call — the default written back through the side door.""" + stub, _ws, session = _drive_two_file_calls( + tmp_path, monkeypatch, + first={**_FILE_CALL_ARGS["write"], + "sandbox_permissions": "workspace-write", + "justification": "one call needs the wider tier"}, + second={**_FILE_CALL_ARGS["write"], "path": "second.txt"}, + ) + + assert len(stub.calls) == 2, stub.calls + assert stub.calls[0]["sandbox"] == "workspace-write" + assert stub.calls[1]["sandbox"] == "read-only", ( + "the widening stuck to the next call" + ) + assert session.session_id == "esc-two-calls" + + +#: A path no tier grants: outside every root, so the refusal is about the +#: boundary rather than about a cause of its own. Not under `/tmp`, which the +#: temp grant makes writable at `workspace-write`. +_OUTSIDE_EVERY_ROOT = os.path.join(os.sep, "not-a-granted-root", "notes.txt") + + +@pytest.mark.parametrize( + "mode, reachable, unreachable", + ( + ("read-only", ("workspace-write", DANGER_FULL_ACCESS), ()), + ("workspace-write", (DANGER_FULL_ACCESS,), ("workspace-write",)), + ), +) +def test_a_root_refusal_names_what_that_tier_can_reach( + mode, reachable, unreachable, tmp_path, +): + """Requirement 3: a blocked file tool is offered the shell's own next step. + + And offered *its own* step — the hint is rendered from the tier the call + stands at, so a `workspace-write` call is not told about `workspace-write`. + Both halves are asserted, because a hint that names the whole target set + would send a model to a tier `validate_hop` will refuse. + """ + from emrg.sandbox import fence + + policy = resolve_policy(mode=mode, workspace_root=str(tmp_path)) + refusal = fence.file_refusal(_OUTSIDE_EVERY_ROOT, policy) + + assert refusal is not None, mode + assert escalation.RETRY_HINT in refusal, refusal + for target in reachable: + assert target in refusal, (mode, target, refusal) + for target in unreachable: + assert f"available here: {target}" not in refusal, (mode, refusal) + + +def test_the_tier_with_nowhere_to_go_refuses_nothing_and_hints_nothing(tmp_path): + """`danger-full-access` checks nothing, so there is no root refusal to hint + at — and an instruction that could only fail is not advice.""" + from emrg.sandbox import fence + + policy = resolve_policy(mode=DANGER_FULL_ACCESS, workspace_root=str(tmp_path)) + assert fence.file_refusal(_OUTSIDE_EVERY_ROOT, policy) is None + + +def test_a_protected_file_refusal_carries_no_hint(tmp_path, monkeypatch): + """The cause decides the remedy, and this cause has a different one. + + A protected daemon file is refused for being the host's own state, not for + crossing a root, so "ask for one wider tier" would be advice that cannot + help — one remedy for every cause is false for most of them. + """ + from emrg.sandbox import fence + from emrg.tools import file_policy + + victim = tmp_path / "protected.toml" + victim.write_text("host state", encoding="utf-8") + monkeypatch.setattr( + file_policy, "protected_paths", lambda: [os.path.realpath(victim)] + ) + policy = resolve_policy(mode="workspace-write", workspace_root=str(tmp_path)) + + refusal = fence.file_refusal(str(victim), policy) + assert refusal is not None + assert "protected daemon file" in refusal + assert escalation.RETRY_HINT not in refusal, refusal