Stop fencing a top-level task on a command word; name the reason and the roots

The direct shell write fence guessed a write from the command word alone
(`touch`, `rm`, `mkdir`, `tee`, `sort`, `uniq`, `gzip`, with options
skipped) and refused a top-level task when that guessed target lay outside
every root it may write. The guess was self-inconsistent — `mkdir -p
/tmp/x` and `git clone ... /tmp/x` passed, `rm -rf /tmp/x` was refused —
and it contradicted the documented rule that command words do not
establish write intent (DEVELOPMENT §2, ARCHITECTURE §6). Owner decision
5A (2026-09-21): drop the word guess for TOP-LEVEL principals, keep it for
subordinate subagents, and make every remaining refusal name the real
reason and the writable directories of the task.

`direct_utility_target_rows(certain_only=True)` gives the view without the
guesses; the guard diffs it against the full view and lets a top-level
principal's guessed target through the outside-every-root fence. The
guess still meets the surviving guards for everyone: light mode (repo and
runtime data read-only, now said so), protected repo paths (now the
existing CORE_PROTECTION_BLOCKED refusal instead of a generic one) and
skill trust state. An acting or read-only child keeps its write
confinement, so the guess still fences it. Explicit shell syntax — a
redirect, cp/mv/ln, dd of=, sed -i, tar -C, rsync — is evidence and is
fenced as before.

Every refusal now names the reason and lists the roots this task may write
with their physical paths (task_drive, artifact_store, user_files, ...),
plus what user_files or the Presence policy said when it declined; an
acting child sees only its own workspace instead of file-tool roots it
cannot write. The unused typed carrier of the old acting message is gone.

Capability widening, disclosed: in advanced/pro mode a top-level task's
bare `touch`/`rm`/`mkdir`/... into the runtime data drive or an owner
credential leaf is no longer pre-refused on the word (a redirect or cp
there still is, light mode still refuses it, BIBLE/identity deletion keep
their dedicated guard); `python -c`, `mkdir -p`, `git clone` and every
unlisted spelling already reached those places, so the marginal change
is small and the Safety Supervisor remains the semantic authority.

Two pins rewritten to the new truth: a top-level pro task's `touch
../data/state/state.json` is no longer refused on the word (a redirect
there still is, naming the roots); a top-level pro task's `rm
<identity.md>` is refused by the identity guard, not the word fence.
This commit is contained in:
Ouroboros 2026-09-21 12:58:24 +03:00
parent 3c52c5ad2f
commit b94d544576
5 changed files with 243 additions and 48 deletions

View file

@ -595,17 +595,6 @@ def _executor_backend_candidate_path(ctx: Any, candidate: str) -> pathlib.Path |
return None
def _workspace_write_block_outside_root_result(
path_text: Any = "", work_dir: Any = "", spelled: Any = "",
) -> ToolResult:
"""Typed carrier of the Guard-B outside-root denial (same bytes in text)."""
return ToolResult(
status="blocked",
code="WORKSPACE_BLOCKED",
text=_workspace_write_block_outside_root_message(path_text, work_dir, spelled),
)
def _git_protected_roots(self) -> list:
"""Ouroboros runtime roots the target-aware git resolver protects, by
enumeration: the system repo + EVERY data drive the task touches (parent
@ -656,6 +645,7 @@ def _direct_shell_write_block(self, raw_cmd: Any, work_dir: pathlib.Path, runtim
"""Apply existing resource authority to certain direct writes, never mentions."""
from dataclasses import replace
from ouroboros.tool_access import _process_root_candidates, _resolve_target_in_selected_base, decide_tool_access, path_is_relative_to
from ouroboros.tool_access_user_files import UserFilesPathBlockedError
from ouroboros.tools.deliverables_shell import _command_path
from ouroboros.tools.shell_guards import direct_utility_target_rows, directory_destination_pairs
from ouroboros.tools.core import _binding_skill_control_plane_path, is_skill_control_plane_path
@ -666,25 +656,38 @@ def _direct_shell_write_block(self, raw_cmd: Any, work_dir: pathlib.Path, runtim
rows = direct_utility_target_rows(raw_cmd)
if not any(row[1] for row in rows):
return None
# A command word alone (touch, rm, mkdir, ...) is a guess, not evidence of a
# write (DEVELOPMENT §2: permission is never reconstructed from command
# words). For a top-level principal the guess still meets light mode,
# protected paths and skill trust state below, never the outside-every-root
# fence (owner decision 5A); a subordinate child keeps its write confinement,
# so the guess still fences it.
guess_fences = self._is_acting_subagent() or self._is_local_readonly_subagent()
certain_rows = direct_utility_target_rows(raw_cmd, certain_only=True)
items = _registry()._binding_items(binding)
selected = items[0] if items else _registry().build_resolved_resource_binding(
self._ctx, operation="shell", process_cwd=str(work_dir))
roots = list(dict.fromkeys([(selected.root, selected.base_path, selected.source, selected.skill_name),
*_process_root_candidates(self._ctx, "shell")]))
system_repo = pathlib.Path(getattr(self._ctx, "system_repo_dir", None) or self._ctx.repo_dir)
roots = [row for row in dict.fromkeys([(selected.root, selected.base_path, selected.source, selected.skill_name),
*_process_root_candidates(self._ctx, "shell")])
if decide_tool_access(profile=selected.profile, root=row[0], operation="write").allow]
system_repo = pathlib.Path(getattr(self._ctx, "system_repo_dir", None) or self._ctx.repo_dir).resolve(strict=False)
def _refuse_write(target: pathlib.Path, token: str) -> ToolResult:
if self._is_acting_subagent():
return _workspace_write_block_outside_root_result(target.resolve(strict=False), work_dir, token)
light_internal = runtime_mode == "light" and any(
path_is_relative_to(target, root) for root in _git_protected_roots(self))
code = "LIGHT_MODE_BLOCKED" if light_internal else "WORKSPACE_BLOCKED"
prefix = "LIGHT_MODE_BLOCKED" if light_internal else "WORKSPACE_SHELL_BLOCKED"
return ToolResult(status="blocked", code=code, text=(
f"⚠️ {prefix}: explicit write target {target} is outside the resources this task may write. "
f"Selected process root: {work_dir}. The process was not started."))
def _refuse_write(target: pathlib.Path, token: str, declined: list, light_internal: bool) -> ToolResult:
# The refusal names the real reason and every root this task may write.
writable = ", ".join(dict.fromkeys(
f"{root}={pathlib.Path(base).resolve(strict=False)}" for root, base, _source, _skill in roots
if not (light_internal and pathlib.Path(base).resolve(strict=False) == system_repo))) or "(none)"
reason = ("runtime_mode=light keeps the Ouroboros repository and runtime data read-only"
if light_internal else "explicit write target is outside every root this task may write")
return ToolResult(
status="blocked", code="LIGHT_MODE_BLOCKED" if light_internal else "WORKSPACE_BLOCKED",
text=(f"⚠️ {'LIGHT_MODE_BLOCKED' if light_internal else 'WORKSPACE_SHELL_BLOCKED'}: {reason}."
+ _blocked_path_note(target, token) + f" Writable roots: {writable}."
+ "".join(f" {why}" for why in declined) + " The process was not started."))
for (argv, targets, _inline, _unknown), cwd in zip(rows, _registry().sequential_effective_cwds(rows, work_dir)):
for (argv, targets, _inline, _unknown), (_argv, certain, _i, _u), cwd in zip(
rows, certain_rows, _registry().sequential_effective_cwds(rows, work_dir)):
guessed = set(targets) - set(certain)
for command, destination, source in directory_destination_pairs(argv):
directory = _command_path(self._ctx, cwd, destination)
child = directory_destination_child_name(command, argv, source)
@ -701,10 +704,9 @@ def _direct_shell_write_block(self, raw_cmd: Any, work_dir: pathlib.Path, runtim
# install resolves user_files to the whole host, which would admit a
# repository target under that name for a light-capped task.
if runtime_mode == "light" and path_is_relative_to(target, system_repo):
return _refuse_write(target, token)
return _refuse_write(target, token, [], True)
declined: list[str] = []
for root, base, source, skill in roots:
if not decide_tool_access(profile=selected.profile, root=root, operation="write").allow:
continue
try:
resolved = _resolve_target_in_selected_base(
self._ctx, root=root, base_path=base, path=str(target), operation="write")
@ -716,21 +718,28 @@ def _direct_shell_write_block(self, raw_cmd: Any, work_dir: pathlib.Path, runtim
f"⚠️ SKILL_PAYLOAD_BLOCKED: explicit write target {resolved} is skill control-plane state. "
"Edit user-authored payload files instead. The process was not started."))
if _registry().binding_targets_system_repo(self._ctx, target_binding):
if runtime_mode == "light" or (protected_paths_in([resolved.relative_to(base).as_posix()])
and not mode_allows_protected_write(runtime_mode)):
if runtime_mode == "light":
continue
protected = protected_paths_in([resolved.relative_to(base).as_posix()])
if protected and not mode_allows_protected_write(runtime_mode):
return _registry()._protected_write_block_result(
path=protected[0].path, runtime_mode=runtime_mode, action="let run_command write")
# Process output uses the existing shell/write grant owner,
# including a remapped Deliverables logical path prefix.
if root == "user_files":
if not _presence_allows_user_output(self._ctx, resolved):
continue
elif not _registry()._presence_binding_allowed(self._ctx, target_binding):
if (not _presence_allows_user_output(self._ctx, resolved) if root == "user_files"
else not _registry()._presence_binding_allowed(self._ctx, target_binding)):
declined.append(f"{root} declines it under the Presence output policy.")
continue
break
except UserFilesPathBlockedError as exc:
declined.append(f"user_files declines it: {str(exc).removeprefix('user_files path blocked: ')}")
except (OSError, ValueError, RuntimeError):
continue
else:
return _refuse_write(target, token)
light_internal = runtime_mode == "light" and any(
path_is_relative_to(target, root) for root in _git_protected_roots(self))
if guess_fences or light_internal or token not in guessed:
return _refuse_write(target, token, declined, light_internal)
return None

View file

@ -969,9 +969,9 @@ def direct_shell_rows(raw_cmd: Any) -> List[tuple]:
return rows
def direct_utility_target_rows(raw_cmd: Any) -> List[tuple]:
"""Certain utility writes plus real output redirects; input reads are independent."""
return [(argv, [*_writer_target_tokens_single(argv, direct_only=True, parse_redirects=False), *writes], (), False)
def direct_utility_target_rows(raw_cmd: Any, *, certain_only: bool = False) -> List[tuple]:
"""Certain utility writes and real output redirects (reads stay independent); ``certain_only`` drops word-guessed targets."""
return [(argv, [*_writer_target_tokens_single(argv, direct_only=True, parse_redirects=False, certain_only=certain_only), *writes], (), False)
for argv, _reads, writes, _shell in direct_shell_rows(raw_cmd)]
@ -1195,7 +1195,7 @@ def directory_destination_pairs(argv: List[str]) -> List[tuple[str, str, str]]:
def _writer_target_tokens_single(
argv: List[str], *, include_inline: bool = True,
direct_only: bool = False, parse_redirects: bool = True,
direct_only: bool = False, parse_redirects: bool = True, certain_only: bool = False,
) -> List[str]:
if not argv:
return []
@ -1224,7 +1224,9 @@ def _writer_target_tokens_single(
return redirect_targets
if cmd == "rsync":
return [argv[-1], *redirect_targets] if len(argv) == 3 and all(not a.startswith("-") for a in argv[1:]) and ":" not in argv[-1] else redirect_targets
if cmd not in {"touch", "rm", "mkdir", "tee", "sort", "uniq", "gzip"}:
# Below here a WRITE is guessed from the command word alone; a top-level
# principal is not fenced by that guess (owner 5A), subordinate children are.
if certain_only or cmd not in {"touch", "rm", "mkdir", "tee", "sort", "uniq", "gzip"}:
return redirect_targets
if cmd in {"touch", "mkdir", "uniq", "gzip"} and any(arg.startswith("-") and arg != "--" for arg in argv[1:]):
return redirect_targets # Option operands need their own concrete role; do not guess.
@ -1232,11 +1234,9 @@ def _writer_target_tokens_single(
# output operand (`uniq - OUT` writes OUT) from every consumer (sol-max r2).
operands = [arg for arg in argv[1:] if arg and (arg == "-" or not arg.startswith("-"))]
targets: List[str] = []
if cmd == "cp":
targets.extend(operands[-1:] if len(operands) >= 2 else [])
elif cmd == "ln":
# The LINK NAME is the write target; the SOURCE is only pointed at, and
# symlink-following reads are containment-checked at resolve time anyway.
if cmd in {"cp", "ln"}:
# The last operand is written (for ln the LINK NAME; the SOURCE is only
# pointed at, and symlink-following reads are containment-checked at resolve time).
targets.extend(operands[-1:] if len(operands) >= 2 else [])
elif cmd in {"chmod", "chown"}:
targets.extend(operands[1:] if len(operands) >= 2 else [])

View file

@ -403,10 +403,15 @@ def test_workspace_run_shell_allows_absolute_cwd_under_workspace_and_child_drive
"advanced",
)
assert git_escape and "WORKSPACE_GIT_BLOCKED" in git_escape, git_escape
# Owner 5A: a bare command word no longer fences a TOP-LEVEL task (a guess is
# not write evidence); explicit shell syntax into the runtime drive still is,
# and the refusal names every root this task may write.
assert _shell_guard_text(registry, {"cmd": ["touch", "../data/state/state.json"]}, "pro") is None
protected_escape = _shell_guard_text(registry,
{"cmd": ["touch", "../data/state/state.json"]}, "pro",
{"cmd": ["sh", "-c", "printf x > ../data/state/state.json"]}, "pro",
)
assert "WORKSPACE_SHELL_BLOCKED" in protected_escape
assert "WORKSPACE_SHELL_BLOCKED" in protected_escape and "outside every root this task may write" in protected_escape
assert f"task_drive={parent_task_dir.parent.resolve()}" in protected_escape, protected_escape
task_drive_write = registry.execute("run_command", {"cmd": ["touch", "output.txt"], "cwd": str(child_dir)})
assert "WORKSPACE_SHELL_BLOCKED" not in task_drive_write
assert (child_dir / "output.txt").is_file()

View file

@ -1111,7 +1111,7 @@ def test_runtime_identity_deletion_uses_the_selected_mode(tmp_path, monkeypatch,
if mode == "cyber_pro":
assert "exit_code=0" in result and not identity.exists(), result
else:
assert "WORKSPACE_SHELL_BLOCKED" in result
assert "IDENTITY_DELETE_BLOCKED" in result, result # owner 5A: the bare word no longer fences a root; the identity guard refuses
assert identity.read_text() == "identity\n"

View file

@ -0,0 +1,181 @@
"""Owner decision 5A (2026-09-21): a TOP-LEVEL principal's shell write is fenced by
explicit evidence only — a real output redirect or a utility with a certain
destination grammar (cp/mv/ln, dd of=, sed -i, tar -C, rsync). The word guess
(``touch``/``rm``/``mkdir``/``tee``/``sort``/``uniq``/``gzip`` + option skipping)
that refused ``rm -rf /tmp/x`` while ``mkdir -p /tmp/x`` and ``git clone … /tmp/x``
passed is gone for top-level tasks and stays for subordinate subagents, whose
write confinement is their contract.
Both directions, through the surviving guards (CHECKLISTS item 21):
top-level word-guessed writes outside every root → allowed; the same argv from an
acting or read-only subagent → refused, naming the reason and the writable roots;
protected repo paths and light mode → refused for everyone; explicit shell syntax
(a redirect) into a place no root admits → still refused with the real reason.
"""
from __future__ import annotations
import pathlib
import shlex
import pytest
from ouroboros import config, safety
from ouroboros.contracts.task_constraint import TaskConstraint
from ouroboros.tools.registry import ToolContext, ToolRegistry
pytestmark = pytest.mark.serial
@pytest.fixture
def principal(tmp_path, monkeypatch):
"""A NON-external top-level principal (a direct owner turn) whose user_files root
is a jailed home: OS scratch outside it is exactly the /tmp class of 5A."""
home, system, data, scratch = [tmp_path / name for name in ("home", "system", "data", "scratch")]
for path in (home / "project", system, data, scratch, system / "prompts", system / "ouroboros"):
path.mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
monkeypatch.setenv("OUROBOROS_USER_FILES_ROOT", str(home))
monkeypatch.setenv("OUROBOROS_DATA_DIR", str(data))
monkeypatch.setattr(pathlib.Path, "home", classmethod(lambda cls: home))
monkeypatch.setattr(config, "DATA_DIR", data)
monkeypatch.setattr(config, "SETTINGS_PATH", data / "settings.json")
monkeypatch.setattr(safety, "check_safety", lambda *_a, **_kw: (True, ""))
ctx = ToolContext(repo_dir=system, system_repo_dir=system, drive_root=data,
task_id="fence-5a", is_direct_chat=True)
registry = ToolRegistry(repo_dir=system, drive_root=data)
registry.set_context(ctx)
return registry, ctx, home, system, data, scratch
def _acting(ctx: ToolContext, workspace: pathlib.Path) -> None:
"""Turn the context into an acting child confined to ``workspace``."""
workspace.mkdir(exist_ok=True)
ctx.workspace_root, ctx.workspace_mode = workspace, "external"
ctx.task_constraint = TaskConstraint(mode="acting_subagent", surface="external_workspace", write_root=str(workspace))
def _mode(monkeypatch, name):
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", name)
monkeypatch.setattr(config, "get_runtime_mode", lambda: name)
def _word_commands(scratch: pathlib.Path) -> dict:
victim = scratch / "victim"
victim.mkdir(exist_ok=True)
(victim / "f.txt").write_text("x", encoding="utf-8")
return {
"rm": (["rm", "-rf", str(victim)], victim, False),
"mkdir": (["mkdir", str(scratch / "made")], scratch / "made", True),
"touch": (["touch", str(scratch / "touched.txt")], scratch / "touched.txt", True),
}
@pytest.mark.parametrize("utility", ["rm", "mkdir", "touch"])
@pytest.mark.parametrize("runtime", ["light", "advanced"])
def test_top_level_word_guessed_write_outside_every_root_runs(principal, monkeypatch, utility, runtime):
registry, _ctx, home, _system, _data, scratch = principal
_mode(monkeypatch, runtime)
cmd, path, exists_after = _word_commands(scratch)[utility]
result = registry.execute_result("run_command", {"cmd": cmd, "cwd": str(home / "project")})
assert result.status == "ok", result.text
assert path.exists() is exists_after
@pytest.mark.parametrize("utility", ["rm", "mkdir", "touch"])
def test_acting_subagent_keeps_the_word_fence_and_learns_where_it_may_write(principal, monkeypatch, utility):
registry, ctx, home, _system, _data, scratch = principal
_mode(monkeypatch, "advanced")
workspace = home / "child-ws"
_acting(ctx, workspace)
cmd, path, exists_after = _word_commands(scratch)[utility]
result = registry.execute_result("run_command", {"cmd": cmd})
assert result.status == "blocked", result.text
assert path.exists() is (not exists_after) # nothing ran
assert str(path) in result.text
assert "outside every root this task may write" in result.text
assert f"active_workspace={workspace.resolve()}" in result.text
assert "task_drive=" not in result.text and "user_files=" not in result.text # not writable for it
# The same child still writes inside its own root: the confinement, not a blanket.
assert registry.execute_result("run_command", {"cmd": ["touch", "inside.txt"]}).status == "ok"
@pytest.mark.parametrize("utility", ["rm", "mkdir", "touch"])
def test_read_only_subagent_has_no_shell_at_all(principal, monkeypatch, utility):
registry, ctx, _home, _system, _data, scratch = principal
_mode(monkeypatch, "advanced")
ctx.task_constraint = TaskConstraint(mode="local_readonly_subagent")
cmd, path, exists_after = _word_commands(scratch)[utility]
result = registry.execute_result("run_command", {"cmd": cmd})
assert result.status == "blocked", result.text
assert path.exists() is (not exists_after)
assert "LOCAL_READONLY_SUBAGENT_BLOCKED" in result.text and "shell" in result.text
@pytest.mark.parametrize("rel", ["BIBLE.md", "prompts/SAFETY.md", "ouroboros/safety.py"])
def test_protected_paths_stay_refused_for_a_top_level_principal(principal, monkeypatch, rel):
registry, _ctx, _home, system, _data, _scratch = principal
_mode(monkeypatch, "advanced")
target = system / rel
result = registry.execute_result("run_command", {"cmd": ["touch", str(target)], "cwd": str(system)})
assert result.status == "blocked", result.text
assert not target.exists()
assert "protected" in result.text.lower() and rel in result.text
@pytest.mark.parametrize("rel", ["BIBLE.md", "prompts/SAFETY.md"])
def test_protected_paths_stay_refused_for_an_acting_child(principal, monkeypatch, rel):
registry, ctx, home, system, _data, _scratch = principal
_mode(monkeypatch, "advanced")
_acting(ctx, home / "ws")
result = registry.execute_result("run_command", {"cmd": ["touch", str(system / rel)]})
assert result.status == "blocked", result.text
assert not (system / rel).exists()
def test_light_mode_still_keeps_the_repository_read_only_for_a_top_level_principal(principal, monkeypatch):
registry, _ctx, home, system, _data, _scratch = principal
_mode(monkeypatch, "light")
target = system / "ordinary.py"
result = registry.execute_result("run_command", {"cmd": ["touch", str(target)], "cwd": str(home / "project")})
assert result.status == "blocked" and result.code == "LIGHT_MODE_BLOCKED", result.text
assert not target.exists()
assert "light" in result.text.lower()
def test_runtime_data_drive_word_guess_is_light_only_for_a_top_level_principal(principal, monkeypatch):
"""The runtime data drive: light mode still refuses a word-guessed write (the
light contract keeps repo and runtime read-only); in advanced mode the bare
word no longer fences a top-level task, while explicit shell syntax into the
drive is still refused with the real reason and the writable roots."""
registry, _ctx, home, _system, data, _scratch = principal
(data / "logs").mkdir()
target = data / "logs" / "planted.jsonl"
_mode(monkeypatch, "light")
result = registry.execute_result("run_command", {"cmd": ["touch", str(target)], "cwd": str(home / "project")})
assert result.status == "blocked" and result.code == "LIGHT_MODE_BLOCKED", result.text
assert not target.exists() and "runtime_mode=light" in result.text
_mode(monkeypatch, "advanced")
result = registry.execute_result(
"run_command", {"cmd": ["sh", "-c", f"printf x > {shlex.quote(str(target))}"], "cwd": str(home / "project")})
assert result.status == "blocked" and "outside every root this task may write" in result.text, result.text
assert not target.exists() and "user_files declines it" in result.text
result = registry.execute_result("run_command", {"cmd": ["touch", str(target)], "cwd": str(home / "project")})
assert result.status == "ok" and target.exists(), result.text
def test_explicit_redirect_outside_every_root_is_still_fenced_and_names_the_writable_roots(principal, monkeypatch):
"""The surviving top-level fence is EVIDENCE-based: a real output redirect is
explicit shell syntax, not a guess. Its refusal names the reason and every
root this task may write (task_drive / artifact_store paths included)."""
registry, _ctx, home, _system, data, scratch = principal
_mode(monkeypatch, "advanced")
target = scratch / "redirected.txt"
result = registry.execute_result(
"run_command", {"cmd": ["sh", "-c", f"printf x > {shlex.quote(str(target))}"], "cwd": str(home / "project")})
assert result.status == "blocked", result.text
assert not target.exists()
assert "outside every root this task may write" in result.text
assert f"task_drive={data.resolve() / 'task_drives' / 'fence-5a'}" in result.text
assert "artifact_store=" in result.text and f"user_files={home.resolve()}" in result.text
assert "outside the user_files home" in result.text # the real reason user_files declined it