vrr/ouroboros
2
0
Fork 0
mirror of https://github.com/razzant/ouroboros.git synced 2026-10-02 19:58:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Let a read-only child read Deliverables and its lineage's task files

Browse source 
A root task put files into its own task_drive and sent four read-only
children to check them; all four were refused with "outside selected
root". The only recorded reason for hiding the orchestrator roots from
children was sibling-project independence, which covers subagent_projects
alone. Owner decision T4=A (#1105): the local_readonly_subagent matrix
gains the Deliverables row, and one lineage rule (lineage_task_ids /
lineage_read_base) lets an actor READ the task_drive/artifact_store of its
own id, its parent_task_id and its root_task_id, anchored on the canonical
data root while the child runs on a child or headless drive. The two
cross-prefix carve-outs fold into it: the own-prefix delegated-capture
anchor is the lineage rule's own-id case, and the orphan capture rule
keeps its custody authority. Lineage roots are not content roots, so
secret-named files in a parent's drive stay denied by name.

The read-only child's schema enums now come from the same matrix the
dispatcher enforces instead of a second hard-coded list, refusals and the
roots hint share one vocabulary (operation_roots), and the parent at
schedule time and the child on its first context line see what the child
can read and what stays invisible. The prompt sentence is replaced under
owner 6A (165 of 168 bytes) and the ARCH 06 node byte-negative.
This commit is contained in:
Ouroboros 2026-09-21 12:34:37 +03:00
parent 35fbf7616c
commit 7b0f649061
9 changed files with 424 additions and 74 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/architecture/06-agent-core.md
View file

@ -80,7 +80,7 @@ Tool API v2 exposes neutral canonical names directly (`read_file`, `list_files`,
Filesystem tool output is self-locating: results use canonical `root:path` labels and `run_command`/`run_script` echo the resolved `cwd`. A direct Project room selects one active physical folder for reads, writes, editing, process cwd, VCS and delegation; governance stays at `system_repo`, and a missing folder keeps its selected address with an availability note rather than falling back to Ouroboros source. Plain folders support ordinary file/process work and directory delegation (`delegate_directory.py`); Git-only snapshot and integration paths retain typed failures when their selected target lacks Git. Physical file resolution preserves the requested address: an absolute path inside any selected base normalizes to that base, an outside absolute path is refused before `safe_relpath` can turn it into a similarly named file, and the repo basename-prefix and canonical delegated-artifact read redirects sit on the same resolver guards and handlers use (`ToolContext.repo_path`/`drive_path`). A Docker workspace's mapped backend absolute address (`workspace_executor.map_backend_path`) is accepted only inside `active_workspace`; other roots and unmapped absolute paths keep their confinement.
`user_files` is the first-class root for user-visible files under the owner's home (the Ouroboros repo and runtime control-plane are rejected); `task_drive` is task-scoped scratch; `artifact_store` is task-scoped under `data/task_results/artifacts/<task_id>/`, created lazily by a write or output registration, where external deliverables written through `user_files` or declared process `outputs` are copied for audit, and a rewritten user-visible file keeps its previous copy under `task_results/artifact_versions/<task_id>/` (§1 tree). Two READ-ONLY orchestrator roots, `subagent_projects` and `deliverables`, grant `read`/`list`/`search` only to orchestrator profiles (parent synthesis); a top-level task still writes the physical Deliverables container through `user_files`. Process admission preserves the original argv and the prepared physical target, and that one resource binding is reused for every other authorized destination; command words, script examples and unknown interpreter effects do not establish write intent — the selected Safety Supervisor receives the full task source (§6 Safety and runtime mode) — and the post-execution shell audit is observational: no replay, rollback, interpreter or attribution proof. Invalid path-like prose contributes no finding. A failed audit adds a bounded diagnostic (typed results also record its exception class in `output_audit_unavailable`) while preserving stdout/stderr and completed exit, signal, timeout and runtime facts; it cannot invent a spawn failure or undeclared output. Computed targets remain a disclosed parser limit, not grounds for a new semantic scanner.
`user_files` is the first-class root for user-visible files under the owner's home (the Ouroboros repo and runtime control plane are refused); `task_drive` is task-scoped scratch; `artifact_store` is task-scoped under `data/task_results/artifacts/<task_id>/`, created lazily on a write or output registration, where deliverables written through `user_files` or declared process `outputs` are copied for audit, and a rewritten user-visible file keeps its prior copy under `task_results/artifact_versions/<task_id>/` (§1 tree). `subagent_projects` and `deliverables` grant `read`/`list`/`search` only; a read-only child reads `deliverables` and its lineage's (parent/root) `task_drive`/`artifact_store`, never a sibling's; a top-level task writes the Deliverables container through `user_files`. Process admission keeps the original argv and the prepared physical target, and that one binding serves every other authorized destination; command words, script examples and unknown interpreter effects establish no write intent — the selected Safety Supervisor receives the full task source (§6 Safety and runtime mode) — and the post-execution shell audit is observational: no replay, rollback, interpreter or attribution proof. Invalid path-like prose yields no finding. A failed audit adds a bounded diagnostic (typed results record its exception class in `output_audit_unavailable`) while keeping stdout/stderr and completed exit, signal, timeout and runtime facts; it cannot invent a spawn failure or undeclared output. Computed targets stay a disclosed parser limit, not grounds for a new semantic scanner.
#### Credential fence and byte masking

140
ouroboros/tool_access.py
View file

@ -20,6 +20,7 @@ from ouroboros.tool_capabilities import ACTING_SUBAGENT_MODE, LOCAL_READONLY_SUB
from ouroboros.contracts.task_constraint import VALID_WRITE_SURFACES, normalize_task_constraint # noqa: F401 — historical facade surface
from ouroboros import deliverables_paths as _deliverables_paths
from ouroboros.shell_parse import is_absolute_path_text
from ouroboros.task_results import validate_task_id
from ouroboros.utils import safe_relpath
_deliverables_root_lexical = _deliverables_paths._deliverables_root_lexical
@ -78,11 +79,10 @@ from ouroboros.tool_access_user_files import ( # noqa: F401 — re-exported mov
)
# Deferral 1: orchestrator-visible READ-ONLY roots — durable subagent (genesis) projects
# and the unnamed-deliverables container. Only ever granted {read,list,search}; NEVER
# write/edit/shell/vcs (no mutation, no shell-cwd — deliberately absent from
# resolve_shell_cwd candidates) and NEVER to acting/readonly subagents (a child must not
# read sibling projects). operator_control is capped to read-only on these too.
# Orchestrator READ-ONLY roots (subagent projects, the Deliverables container): only ever
# {read,list,search}, never a shell cwd, read-only even for operator_control. A child never
# reads `subagent_projects` (parallel candidates stay independent); it reads owner-visible
# `deliverables` and, by the lineage rule below, its parent's and root's task files (T4=A).
def summarize_subagent_profile(profile: ToolProfile, *, effective_lane: str = "") -> str:
"""Compact, human-readable summary of a subagent's EFFECTIVE tool profile
@ -90,10 +90,13 @@ def summarize_subagent_profile(profile: ToolProfile, *, effective_lane: str = ""
_POLICY matrix (the same SSOT active_tool_profile resolves), so the parent sees
at schedule time (and the child sees first line of its context) what the child
CAN and CANNOT do. Prevents the wasted rounds where a prober child hit
workspace_blocked on run_script because neither side knew shell was off."""
workspace_blocked on run_script because neither side knew shell was off.
The second line names what the child can READ (incl. its parent's and root's
task files, never a sibling's) and which roots stay invisible to it."""
matrix = _POLICY.get(_effective_policy_profile(profile), {})
shell_roots = sorted(root for root, ops in matrix.items() if "shell" in ops)
write_roots = sorted(root for root, ops in matrix.items() if ops & {"write", "edit"})
read_roots = sorted(root for root, ops in matrix.items() if "read" in ops)
has_shell = bool(shell_roots)
bits = [
f"profile={profile}",
@ -103,7 +106,66 @@ def summarize_subagent_profile(profile: ToolProfile, *, effective_lane: str = ""
lane = str(effective_lane or "").strip()
if lane:
bits.append(f"model_lane={lane}")
return "child capabilities — " + " · ".join(bits)
lineage = (" (task_drive/artifact_store: its own, its parent's and its root task's files,"
" never a sibling's)" if "task_drive" in read_roots else "")
return (
"child capabilities — " + " · ".join(bits)
+ f"\nreadable={', '.join(read_roots) or 'none'}{lineage}"
+ f" · unreadable={', '.join(sorted(_ALL_ROOTS - set(read_roots))) or 'none'}"
)
def lineage_task_ids(ctx: Any) -> tuple[str, ...]:
"""Task ids whose ``task_drive``/``artifact_store`` this actor may READ: its own,
then ``parent_task_id`` and ``root_task_id`` from its own lineage fields (T4=A,
#1105) — never a sibling's, nothing found by walking the disk, malformed ids dropped."""
meta = getattr(ctx, "task_metadata", None)
meta = meta if isinstance(meta, dict) else {}
ids = [task_id_for_artifacts(ctx)]
for key in ("parent_task_id", "root_task_id"):
try:
candidate = validate_task_id(meta.get(key))
except ValueError:
continue
if candidate not in ids:
ids.append(candidate)
return tuple(ids)
def _task_root_drives(ctx: Any) -> list[pathlib.Path]:
"""The data drives a task's own task roots are enumerated on."""
meta = getattr(ctx, "task_metadata", {})
meta = meta if isinstance(meta, dict) else {}
drives: list[pathlib.Path] = []
for raw in (getattr(ctx, "drive_root", ""), *(meta.get(key) for key in (
"drive_root", "child_drive_root", "headless_child_drive_root"))):
if not raw:
continue
drive = pathlib.Path(raw).resolve(strict=False)
if drive not in drives:
drives.append(drive)
return drives
def lineage_read_base(ctx: Any, root: ResourceRoot, target: pathlib.Path) -> pathlib.Path | None:
"""The lineage ``task_drive``/``artifact_store`` base containing ``target``, or None:
``lineage_task_ids`` on the canonical data root (where a parent's task files live
while the child runs on a child or headless drive) and on the task's own drives.
Physical containment only; the caller keeps the READ-only gate."""
if root not in {"task_drive", "artifact_store"} or not hasattr(ctx, "drive_root"):
return None
candidate = pathlib.Path(target).resolve(strict=False)
drives = [canonical_data_root(ctx)]
drives += [drive for drive in _task_root_drives(ctx) if drive not in drives]
for drive in drives:
for task_id in lineage_task_ids(ctx):
base = (
drive / "task_drives" / task_id if root == "task_drive"
else task_artifact_dir_path(drive, task_id, create=False)
).resolve(strict=False)
if path_is_relative_to(candidate, base):
return base
return None
def _effective_policy_profile(profile: ToolProfile) -> ToolProfile:
@ -121,6 +183,12 @@ def _effective_policy_profile(profile: ToolProfile) -> ToolProfile:
return profile
def operation_roots(profile: ToolProfile, operation: Operation) -> str:
"""The roots ``profile`` may ``operation`` — the one vocabulary every refusal names."""
matrix = _POLICY.get(_effective_policy_profile(profile), {})
return ", ".join(sorted(root for root, ops in matrix.items() if operation in ops)) or "(none)"
def decide_tool_access(
*,
profile: ToolProfile,
@ -131,10 +199,10 @@ def decide_tool_access(
allowed = operation in _POLICY.get(effective_profile, {}).get(root, set())
if allowed:
return ToolAccessDecision(True, guard=f"{effective_profile}:{root}:{operation}")
allowed_roots = ", ".join(sorted(r for r, ops in _POLICY.get(effective_profile, {}).items() if operation in ops)) or "(none)"
return ToolAccessDecision(
False,
reason=f"profile={effective_profile} cannot {operation} root={root}. Roots your profile can {operation}: {allowed_roots}.",
reason=f"profile={effective_profile} cannot {operation} root={root}. "
f"Roots your profile can {operation}: {operation_roots(profile, operation)}.",
guard=f"{effective_profile}:{root}:{operation}",
)
@ -209,14 +277,8 @@ def _process_root_candidates(
])
if hasattr(ctx, "drive_root"):
_add_task_roots(pathlib.Path(ctx.drive_root).resolve(strict=False))
meta = getattr(ctx, "task_metadata", {})
meta = meta if isinstance(meta, dict) else {}
for key in ("drive_root", "child_drive_root", "headless_child_drive_root"):
if not meta.get(key):
continue
_add_task_roots(pathlib.Path(meta[key]).resolve(strict=False))
if hasattr(ctx, "drive_root"):
for drive in _task_root_drives(ctx):
_add_task_roots(drive)
candidates.append(("user_files", resource_root_path(ctx, "user_files"), "user_files", ""))
if include_skill:
base, source, selected_name = _skill_payload_base(
@ -614,25 +676,13 @@ def _resolve_target_in_selected_base(
try:
path_text = candidate.relative_to(resolved_base).as_posix()
except ValueError:
if root == "artifact_store" and operation in _READ_OPS:
canonical = task_artifact_dir_path(
canonical_data_root(ctx), task_id_for_artifacts(ctx), create=False,
)
try:
relative = candidate.relative_to(canonical).as_posix()
except ValueError:
relative = ""
if relative:
anchored = delegated_capture_read_target(
canonical_data_root(ctx), task_id_for_artifacts(ctx), relative, resolved_base,
)
if anchored is not None:
return anchored
else:
# An ORPHAN's capture lives under ANOTHER task's prefix, so
# `relative` is empty and control would fall straight to the
# raise below. Read-only, prefix-confined, no new root: the
# orphan disposition rule is the only authority consulted.
if operation in _READ_OPS:
# Cross-prefix READS: the lineage rule (own id on every drive, the
# parent's and the root's task files), then the ORPHAN capture rule,
# which the custody authority alone answers.
if lineage_read_base(ctx, root, candidate) is not None:
return candidate
if root == "artifact_store":
from ouroboros.delegate_shared import orphan_capture_read_target
anchored = orphan_capture_read_target(ctx, candidate)
@ -802,20 +852,24 @@ def build_resolved_resource_binding(
path=path,
operation=operation,
)
# An absolute user_files target may intentionally land in the configured
# Deliverables container outside the user's ordinary home. The binding's
# physical base must follow that selected container; otherwise an exact
# Presence path-prefix check treats a valid deliverable as outside the
# binding merely because the default user_files home is a sibling.
# The physical base follows the container holding the target: an absolute
# user_files target may land in the configured Deliverables container outside the
# home (else an exact Presence path-prefix check calls a valid deliverable outside
# the binding), and a lineage READ lands in the parent's or root's task root.
logical_base_path = None
container = None
if normalized == "user_files":
try:
deliverables = _deliverables_root()
if path_is_relative_to(target, deliverables) or _path_is_relative_to_casefold(target, deliverables):
logical_base_path = pathlib.Path(base).resolve(strict=False)
base = deliverables
container = deliverables
except (OSError, TypeError, ValueError, RuntimeError):
pass
elif operation in _READ_OPS and not path_is_relative_to(target, base):
container = lineage_read_base(ctx, normalized, target)
if container is not None:
logical_base_path = pathlib.Path(base).resolve(strict=False)
base = container
return ResolvedResourceBinding(
profile=profile,
root=normalized,

3
ouroboros/tool_access_types.py
View file

@ -146,6 +146,9 @@ _POLICY: dict[str, dict[str, set[str]]] = {
# (data/skills/...); grants/secrets live in data/state/skills, which
# stays invisible to this profile.
"skill_payload": {"read", "list", "search"},
# Owner T4=A (#1105): the owner-visible Deliverables container is readable
# by a read-only child; `subagent_projects` stays top-level only.
"deliverables": {"read", "list", "search"},
},
# Top-level preset names remain observable, but workspace focus never narrows
# the ordinary principal. Independent path/credential/child/runtime guards

8
ouroboros/tools/core_file_tools.py
View file

@ -546,11 +546,9 @@ def _profile_roots_hint(ctx: ToolContext, operation: str) -> str:
model turns a dead-end error into a self-correcting retry instead of a
probe loop over blocked roots (v6.70.0)."""
try:
from ouroboros.tool_access import _POLICY, _effective_policy_profile
from ouroboros.tool_access import operation_roots
policy = _POLICY.get(_effective_policy_profile(active_tool_profile(ctx)), {})
visible = sorted(root for root, ops in policy.items() if operation in ops)
return f" Roots your profile can {operation}: {', '.join(visible) or '(none)'}."
return f" Roots your profile can {operation}: {operation_roots(active_tool_profile(ctx), operation)}."
except Exception:
return ""
@ -871,7 +869,7 @@ def _list_files(
if is_restricted_subagent_profile(ctx):
if normalized == "system_repo":
items = _filter_subagent_secret_repo_listing(items, binding.base_path, ctx=ctx)
elif normalized in {"task_drive", "skill_payload", "artifact_store", "user_files"}:
elif normalized in {"task_drive", "skill_payload", "artifact_store", "user_files", "deliverables"}:
items = _filter_subagent_secret_listing(items, binding.base_path, ctx=ctx)
return json.dumps(items, ensure_ascii=False, indent=2)
except _ListingMiss as exc:

35
ouroboros/tools/registry_core.py
View file

@ -501,16 +501,9 @@ class ToolRegistry:
parameters["properties"]["contract_kind"]["enum"] = ["delegation_zero_run"]
parameters["required"] = ["contract_kind", "zero_run_decision", "zero_run_basis"]
elif entry.name in {"read_file", "list_files", "search_code", "query_code"}:
schema = copy.deepcopy(schema)
root_schema = schema.get("parameters", {}).get("properties", {}).get("root", {})
if entry.name == "search_code":
allowed = {"active_workspace", "system_repo", "skill_payload"}
elif entry.name == "query_code":
# query_code itself rejects non-repo roots — do not advertise more.
allowed = {"active_workspace", "system_repo"}
else:
allowed = {"active_workspace", "system_repo", "runtime_data", "task_drive", "skill_payload", "artifact_store"}
if isinstance(root_schema.get("enum"), list): root_schema["enum"] = [root for root in root_schema["enum"] if root in allowed]
# The advertised roots are the matrix's answer for this profile and
# operation (the same SSOT the dispatcher enforces), never a second list.
schema = self._schema_with_matrix_roots(entry)
elif entry.name in {"browse_page", "browser_action"}:
schema = copy.deepcopy(entry.schema)
if entry.name == "browse_page":
@ -557,16 +550,22 @@ class ToolRegistry:
elif (entry.name in tool_resolution._ROOT_ARG_REPO_WRITE_TOOLS
or entry.name in _GENERIC_VCS_TARGET_TOOLS
or entry.name in {"read_file", "list_files", "search_code", "query_code"}):
schema = copy.deepcopy(schema)
root_schema = schema.get("parameters", {}).get("properties", {}).get("root", {})
operation = _target_binding_operation(entry.name, {})
if isinstance(root_schema.get("enum"), list) and operation:
root_schema["enum"] = [root for root in root_schema["enum"]
if decide_tool_access(profile=active_tool_profile(self._ctx), root=root,
operation=operation).allow
and (entry.name != "query_code" or root in {"active_workspace", "system_repo"})]
schema = self._schema_with_matrix_roots(entry)
return {"type": "function", "function": schema}
def _schema_with_matrix_roots(self, entry: ToolEntry) -> Dict[str, Any]:
"""A copy of the schema whose ``root`` enum is what the matrix grants this
profile for the tool's operation; query_code stays repo-only by contract."""
schema = copy.deepcopy(entry.schema)
root_schema = schema.get("parameters", {}).get("properties", {}).get("root", {})
operation = _target_binding_operation(entry.name, {})
if isinstance(root_schema.get("enum"), list) and operation:
root_schema["enum"] = [root for root in root_schema["enum"]
if decide_tool_access(profile=active_tool_profile(self._ctx), root=root,
operation=operation).allow
and (entry.name != "query_code" or root in {"active_workspace", "system_repo"})]
return schema
def _schemas_for_entry(self, entry: ToolEntry) -> List[Dict[str, Any]]:
return [self._schema_for_entry(entry)]

6
prompts/SYSTEM.md
View file

@ -163,9 +163,9 @@ active profile permits it, `task_drive` for task scratch, `artifact_store` for
canonical deliverables, `skill_payload` for reviewed skill payloads, and
`user_files` for user-visible files under the owner's home (a bare filename
lands in the visible Deliverables folder, not the home root).
`subagent_projects` and `deliverables` are read-only orchestrator roots for
inspecting children's work — never written, never a shell cwd, never handed to
a subagent.
`subagent_projects` and `deliverables` are read-only (never written or a
shell cwd); a subagent reads `deliverables` and its lineage's task files, never
a sibling's.
My cognitive memory has first-class tools — `update_identity`,
`update_scratchpad`, `knowledge_write` — and I never reach for

8
tests/test_filesystem_root_observability.py
View file

@ -264,7 +264,8 @@ def test_search_skips_non_regular_files(tmp_path):
def test_new_readonly_roots_access_policy():
"""v6.40: subagent_projects/deliverables are READ-ONLY orchestrator roots — read/list/search
where granted, never write/edit/shell, and never to subagents."""
where granted, never write/edit/shell. Owner T4=A (#1105): a read-only child reads the
owner-visible Deliverables root; sibling projects (subagent_projects) stay top-level only."""
from ouroboros.tool_access import _POLICY, _READONLY_RESOURCE_ROOTS, decide_tool_access
roots = ("subagent_projects", "deliverables")
@ -280,5 +281,6 @@ def test_new_readonly_roots_access_policy():
for op in ("write", "edit", "shell"):
assert not decide_tool_access(profile=profile, root=root, operation=op).allow, (profile, root, op)
for profile in ("acting_subagent", "local_readonly_subagent"):
for root in roots:
assert not decide_tool_access(profile=profile, root=root, operation="read").allow, (profile, root)
assert not decide_tool_access(profile=profile, root="subagent_projects", operation="read").allow, profile
assert decide_tool_access(profile="local_readonly_subagent", root="deliverables", operation="read").allow
assert not decide_tool_access(profile="acting_subagent", root="deliverables", operation="read").allow

3
tests/test_tool_capabilities_subagent_scheduling.py
View file

@ -161,7 +161,8 @@ def test_local_readonly_subagent_initial_schemas_are_allowlisted(tmp_path):
for tool_name in ("read_file", "list_files", "search_code"):
root_enum = schemas[tool_name]["parameters"]["properties"]["root"]["enum"]
assert "user_files" not in root_enum
assert set(schemas["search_code"]["parameters"]["properties"]["root"]["enum"]) == {"active_workspace", "system_repo", "skill_payload"}
# Owner T4=A (#1105): the Deliverables root is readable/searchable by a read-only child.
assert set(schemas["search_code"]["parameters"]["properties"]["root"]["enum"]) == {"active_workspace", "system_repo", "skill_payload", "deliverables"}
action_schema = schemas["browser_action"]["parameters"]["properties"]["action"]
assert "evaluate" not in action_schema["enum"]
assert "send_photo" not in schemas["browse_page"]["description"]

293
tests/test_tool_roots_lineage_read.py Normal file
View file

@ -0,0 +1,293 @@
"""A read-only child reads what its parent points it to (owner T4=A, #1105).
Measured on a live install: a root task put files into its own ``task_drive``
and sent four read-only children to check them; all four were refused with
``outside selected root``. The recorded reason for hiding the orchestrator
roots from children (``tool_access.py``: "a child must not read sibling
projects") covers ``subagent_projects`` only. Now a child reads the
owner-visible Deliverables root and the ``task_drive``/``artifact_store`` of
its OWN lineage (parent and root ids from its own lineage fields), anchored on
the canonical data root while the child itself runs on a headless drive. A
sibling's or a stranger's task files stay refused; secret-named files in a
parent's drive stay denied by name; ``subagent_projects`` stays top-level only.
"""
from __future__ import annotations
import json
import pathlib
from types import SimpleNamespace
import pytest
from ouroboros.contracts.task_constraint import TaskConstraint
from ouroboros.tool_access import (
_POLICY,
_TOP_LEVEL_PRINCIPAL_POLICY,
decide_tool_access,
summarize_subagent_profile,
)
from ouroboros.tools.registry import ToolContext, ToolRegistry
PARENT = "p07499dc017c01f83"
ROOT = "r5173b7c3c15d4c0b"
CHILD = "c11ae4fd0aa111111"
SIBLING = "s339e97de0b222222"
STRANGER = "x8b8af5e9cc333333"
@pytest.fixture
def geometry(tmp_path, monkeypatch):
"""The parent's task files live on the CANONICAL data root; the child runs
on its own headless drive; the owner home is a fake tmp home."""
home = tmp_path / "home"
repo = tmp_path / "repo"
canonical = tmp_path / "data"
headless = tmp_path / "headless"
for path in (home, repo, canonical, headless):
path.mkdir()
monkeypatch.setattr(pathlib.Path, "home", lambda: home)
monkeypatch.setenv("OUROBOROS_USER_FILES_ROOT", str(home))
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
monkeypatch.setenv("OUROBOROS_SAFETY_MODE", "off")
(repo / "README.md").write_text("repo readme\n", encoding="utf-8")
parent_drive = canonical / "task_drives" / PARENT
(parent_drive / "source" / "ouroboros").mkdir(parents=True)
(parent_drive / "source" / "ouroboros" / "update_letter.py").write_text(
"PARENT_DRIVE_BYTES = 1\n", encoding="utf-8")
(parent_drive / "triage-draft.json").write_text('{"triage": "draft"}\n', encoding="utf-8")
(parent_drive / ".env").write_text("SECRET_TOKEN=1\n", encoding="utf-8")
(parent_drive / "settings.json").write_text('{"OPENAI_API_KEY": "sk-secret"}\n', encoding="utf-8")
root_artifacts = canonical / "task_results" / "artifacts" / ROOT
root_artifacts.mkdir(parents=True)
(root_artifacts / "report.txt").write_text("ROOT_ARTIFACT_BYTES\n", encoding="utf-8")
sibling_drive = canonical / "task_drives" / SIBLING
sibling_drive.mkdir(parents=True)
(sibling_drive / "notes.txt").write_text("SIBLING_BYTES\n", encoding="utf-8")
stranger_artifacts = canonical / "task_results" / "artifacts" / STRANGER
stranger_artifacts.mkdir(parents=True)
(stranger_artifacts / "out.txt").write_text("STRANGER_BYTES\n", encoding="utf-8")
deliverables = home / "Deliverables"
deliverables.mkdir()
(deliverables / "answer.txt").write_text("DELIVERABLE_BYTES needle\n", encoding="utf-8")
return SimpleNamespace(
home=home, repo=repo, canonical=canonical, headless=headless,
parent_drive=parent_drive, root_artifacts=root_artifacts,
sibling_drive=sibling_drive, stranger_artifacts=stranger_artifacts,
deliverables=deliverables,
)
def child_registry(geo, *, drive=None, acting=False):
"""A delegated child of PARENT under ROOT, through the real registry."""
ctx = ToolContext(repo_dir=geo.repo, drive_root=drive or geo.headless, task_id=CHILD)
ctx.budget_drive_root = str(geo.canonical)
ctx.task_metadata = {
"delegation_role": "subagent",
"parent_task_id": PARENT,
"root_task_id": ROOT,
"budget_drive_root": str(geo.canonical),
}
if acting:
work = geo.home / "work"
work.mkdir(exist_ok=True)
ctx.workspace_root = work
ctx.workspace_mode = "external"
ctx.task_constraint = TaskConstraint(
mode="acting_subagent", allow_enable=False, surface="external_workspace")
else:
ctx.task_constraint = TaskConstraint(mode="local_readonly_subagent", allow_enable=False)
registry = ToolRegistry(repo_dir=geo.repo, drive_root=ctx.drive_root)
registry.set_context(ctx)
return registry, ctx
# --- the lineage read: parent's and root's task files, never a sibling's ------
def test_child_reads_its_parents_task_drive_from_a_headless_drive(geometry):
registry, ctx = child_registry(geometry)
target = geometry.parent_drive / "source" / "ouroboros" / "update_letter.py"
out = registry.execute("read_file", {"root": "task_drive", "path": str(target)})
assert "PARENT_DRIVE_BYTES" in out, out
assert out.startswith("# task_drive:"), out
assert ctx.last_read_view["opened_root"] == "task_drive"
assert ctx.last_read_view["target"] == str(target.resolve())
def test_child_reads_the_root_tasks_artifact(geometry):
registry, _ctx = child_registry(geometry)
target = geometry.root_artifacts / "report.txt"
out = registry.execute("read_file", {"root": "artifact_store", "path": str(target)})
assert "ROOT_ARTIFACT_BYTES" in out, out
assert out.startswith("# artifact_store:"), out
def test_single_drive_child_reads_the_parent_drive_too(geometry):
registry, _ctx = child_registry(geometry, drive=geometry.canonical)
out = registry.execute(
"read_file", {"root": "task_drive", "path": str(geometry.parent_drive / "triage-draft.json")})
assert '"triage": "draft"' in out, out
def test_an_acting_child_shares_the_lineage_read(geometry):
registry, _ctx = child_registry(geometry, acting=True)
out = registry.execute(
"read_file", {"root": "task_drive", "path": str(geometry.parent_drive / "triage-draft.json")})
assert '"triage": "draft"' in out, out
def test_a_siblings_drive_and_a_strangers_artifacts_stay_refused(geometry):
registry, _ctx = child_registry(geometry)
sibling = registry.execute(
"read_file", {"root": "task_drive", "path": str(geometry.sibling_drive / "notes.txt")})
stranger = registry.execute(
"read_file", {"root": "artifact_store", "path": str(geometry.stranger_artifacts / "out.txt")})
assert "SIBLING_BYTES" not in sibling and "outside selected root=task_drive" in sibling, sibling
assert "STRANGER_BYTES" not in stranger and "outside selected root=artifact_store" in stranger, stranger
def test_lineage_is_read_only_even_for_a_top_level_parent_drive(geometry):
"""The rule is a READ rule: an acting child never writes into its parent's
drive through the same path, and its own task_drive stays the write target
the matrix says (none for an acting child)."""
registry, _ctx = child_registry(geometry, acting=True)
target = geometry.parent_drive / "triage-draft.json"
before = target.read_text(encoding="utf-8")
out = registry.execute("write_file", {"root": "task_drive", "path": str(target), "content": "x"})
assert out.startswith("⚠️"), out
assert target.read_text(encoding="utf-8") == before
# --- secrets in a parent's drive stay denied by NAME -------------------------
@pytest.mark.parametrize("name", [".env", "settings.json"])
def test_secret_named_files_in_the_parents_drive_stay_denied(geometry, name):
registry, _ctx = child_registry(geometry)
out = registry.execute("read_file", {"root": "task_drive", "path": str(geometry.parent_drive / name)})
assert "READ_FILE_BLOCKED" in out and "secret" in out, out
assert "SECRET_TOKEN" not in out and "sk-secret" not in out
def test_child_lists_the_parents_drive_with_secret_names_hidden(geometry):
registry, _ctx = child_registry(geometry)
out = registry.execute("list_files", {"root": "task_drive", "path": str(geometry.parent_drive)})
items = json.loads(out)
assert "triage-draft.json" in items and "source/" in items, items
assert ".env" not in items and "settings.json" not in items, items
assert any("hidden from this subagent" in item for item in items), items
# --- the pure lineage function ------------------------------------------------
def test_lineage_task_ids_are_own_parent_and_root_and_nothing_else(geometry):
from ouroboros.tool_access import lineage_task_ids
_registry, ctx = child_registry(geometry)
assert lineage_task_ids(ctx) == (CHILD, PARENT, ROOT)
ctx.task_metadata["root_task_id"] = PARENT # parent IS the root: no duplicate
assert lineage_task_ids(ctx) == (CHILD, PARENT)
ctx.task_metadata["parent_task_id"] = "../escape" # malformed ids are dropped, not guessed
ctx.task_metadata["root_task_id"] = ""
assert lineage_task_ids(ctx) == (CHILD,)
top = ToolContext(repo_dir=geometry.repo, drive_root=geometry.canonical, task_id=ROOT)
assert lineage_task_ids(top) == (ROOT,)
def test_lineage_read_base_names_the_containing_lineage_root(geometry):
from ouroboros.tool_access import lineage_read_base
_registry, ctx = child_registry(geometry)
parent_file = geometry.parent_drive / "triage-draft.json"
root_file = geometry.root_artifacts / "report.txt"
assert lineage_read_base(ctx, "task_drive", parent_file) == geometry.parent_drive.resolve()
assert lineage_read_base(ctx, "artifact_store", root_file) == geometry.root_artifacts.resolve()
# The label must match the physical kind: a task_drive path is not an artifact base.
assert lineage_read_base(ctx, "artifact_store", parent_file) is None
assert lineage_read_base(ctx, "task_drive", geometry.sibling_drive / "notes.txt") is None
assert lineage_read_base(ctx, "runtime_data", parent_file) is None
# The child's OWN task root on its headless drive is a lineage base as well.
own = geometry.headless / "task_drives" / CHILD / "scratch.txt"
assert lineage_read_base(ctx, "task_drive", own) == (geometry.headless / "task_drives" / CHILD).resolve()
# --- Deliverables: a read-only child reads, never writes ---------------------
def test_deliverables_row_reads_only_and_only_for_the_readonly_child():
for op in ("read", "list", "search"):
assert decide_tool_access(profile="local_readonly_subagent", root="deliverables", operation=op).allow, op
for op in ("write", "edit", "shell", "vcs", "service", "review", "delegate"):
assert not decide_tool_access(profile="local_readonly_subagent", root="deliverables", operation=op).allow, op
for profile in ("acting_subagent", "local_readonly_subagent"):
assert not decide_tool_access(profile=profile, root="subagent_projects", operation="read").allow, profile
assert not decide_tool_access(profile="acting_subagent", root="deliverables", operation="read").allow
# Top-level principals are untouched: one shared matrix object, unchanged rows.
for profile in ("workspace_task", "external_workspace_task", "self_modification"):
assert _POLICY[profile] is _TOP_LEVEL_PRINCIPAL_POLICY
assert _TOP_LEVEL_PRINCIPAL_POLICY["deliverables"] == {"read", "list", "search"}
assert _TOP_LEVEL_PRINCIPAL_POLICY["subagent_projects"] == {"read", "list", "search"}
def test_child_reads_lists_and_searches_deliverables_but_cannot_touch_them(geometry):
registry, _ctx = child_registry(geometry)
answer = geometry.deliverables / "answer.txt"
read = registry.execute("read_file", {"root": "deliverables", "path": "answer.txt"})
listing = registry.execute("list_files", {"root": "deliverables", "path": "."})
search = registry.execute("search_code", {"root": "deliverables", "query": "needle"})
assert "DELIVERABLE_BYTES" in read and read.startswith("# deliverables:answer.txt"), read
assert "answer.txt" in json.loads(listing), listing
assert "deliverables:answer.txt:1:" in search, search
write = registry.execute("write_file", {"root": "deliverables", "path": "answer.txt", "content": "x"})
edit = registry.execute("edit_text", {"root": "deliverables", "path": "answer.txt",
"old_str": "needle", "new_str": "x"})
shell = registry.execute("run_command", {"command": "ls", "cwd": "deliverables"})
for out in (write, edit, shell):
assert out.startswith("⚠️"), out
assert answer.read_text(encoding="utf-8") == "DELIVERABLE_BYTES needle\n"
assert "⚠️" in registry.execute("list_files", {"root": "subagent_projects", "path": "."})
def test_readonly_child_schema_enums_follow_the_matrix(geometry):
registry, _ctx = child_registry(geometry)
def enum(name):
return registry.get_schema_by_name(name)["function"]["parameters"]["properties"]["root"]["enum"]
for name in ("read_file", "list_files"):
assert "deliverables" in enum(name), name
assert "subagent_projects" not in enum(name) and "user_files" not in enum(name), name
assert set(enum("search_code")) == {"active_workspace", "system_repo", "skill_payload", "deliverables"}
assert enum("query_code") == ["active_workspace", "system_repo"]
# --- both sides see what the child can read -----------------------------------
def test_profile_summary_names_readable_and_unreadable_roots(monkeypatch):
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
readonly = summarize_subagent_profile("local_readonly_subagent", effective_lane="light").splitlines()
assert len(readonly) == 2, readonly
assert readonly[0].startswith("child capabilities — ") and "model_lane=light" in readonly[0]
readable, unreadable = readonly[1].split(" · unreadable=")
assert readable.startswith("readable=") and "deliverables" in readable and "task_drive" in readable
assert "parent" in readable and "sibling" in readable, readable
assert unreadable == "subagent_projects, user_files", unreadable
acting = summarize_subagent_profile("acting_subagent").splitlines()
assert len(acting) == 2, acting
acting_readable, acting_unreadable = acting[1].split(" · unreadable=")
assert "deliverables" not in acting_readable and "deliverables" in acting_unreadable