mirror of
https://github.com/razzant/ouroboros.git
synced 2026-10-03 04:07:04 +00:00
Keep helper starts coherent through unavailable focus and admission
This commit is contained in:
parent
456fdbde42
commit
0bc6d45d78
15 changed files with 194 additions and 30 deletions
|
|
@ -21,6 +21,11 @@ def capture_parent_workspace(ctx: Any) -> dict[str, str]:
|
|||
from ouroboros.tools.tool_resolution import active_repo_dir_for
|
||||
|
||||
access = _access()
|
||||
meta = getattr(ctx, "task_metadata", None) or {}
|
||||
if (getattr(ctx, "is_direct_chat", False) and not getattr(ctx, "workspace_root", None)
|
||||
and isinstance(meta, dict) and meta.get("_project_room_note")):
|
||||
return {"root": str(meta.get("_project_room_dir") or ""), "mode": "", "source": "project_room",
|
||||
"availability": "unavailable", "detail": str(meta["_project_room_note"])}
|
||||
if getattr(ctx, "workspace_root", None):
|
||||
source = "active_workspace"
|
||||
elif access.project_room_lens_dir(ctx) is not None:
|
||||
|
|
@ -122,7 +127,8 @@ def admit_child_start_folder(ctx: Any, value: Any, params: dict) -> str:
|
|||
raise ValueError("workspace_root cannot select a starting folder for genesis, which creates its own empty project")
|
||||
folder = readonly_start_folder(value)
|
||||
write_root = str(params.get("write_root") or "").strip()
|
||||
if write_root and pathlib.Path(write_root).expanduser().resolve(strict=False) != pathlib.Path(folder):
|
||||
if (str(params.get("write_surface") or "").strip().lower() == "external_workspace" and write_root
|
||||
and pathlib.Path(write_root).expanduser().resolve(strict=False) != pathlib.Path(folder)):
|
||||
raise ValueError("workspace_root and write_root name different folders; select one acting workspace")
|
||||
from ouroboros.tools.tool_resolution import _root_containing_absolute_path
|
||||
|
||||
|
|
|
|||
|
|
@ -120,12 +120,12 @@ def user_files_path_block_reason(
|
|||
) -> str:
|
||||
"""Return a block reason when candidate is not an external user file.
|
||||
|
||||
Location checks (outside-home, control-plane overlap) apply to every
|
||||
operation. Root reads are location-authorized with byte masking at egress.
|
||||
Ordinary reads follow location checks and return unchanged bytes. Children
|
||||
inherit their parent's read reach; their write/action ceilings remain separate.
|
||||
Mutations additionally protect known credential leaves and physical owner
|
||||
stores through credential_shapes; ordinary .config/Library/settings files
|
||||
and the exact SSH config are not rejected as credential stores by name.
|
||||
Children never hold a user_files grant in the profile matrix.
|
||||
Cyber Pro follows the existing per-operation agency exemption below.
|
||||
"""
|
||||
|
||||
resolved = pathlib.Path(candidate).expanduser().resolve(strict=False)
|
||||
|
|
@ -143,9 +143,10 @@ def user_files_path_block_reason(
|
|||
# path, so the Ouroboros repo/data drive stays protected even when home
|
||||
# confinement is lifted.
|
||||
from ouroboros.tool_access_reads import read_allows_outside_home
|
||||
outside_home_allowed = read_allows_outside_home(ctx) if read_only else _tool_access().is_external_workspace(ctx)
|
||||
if outside_home and not outside_home_allowed:
|
||||
return f"path is outside user home {home}"
|
||||
if outside_home:
|
||||
outside_home_allowed = read_allows_outside_home(ctx) if read_only else _tool_access().is_external_workspace(ctx)
|
||||
if not outside_home_allowed:
|
||||
return f"path is outside user home {home}"
|
||||
|
||||
# The Ouroboros runtime/control surface is the system repo PLUS every data
|
||||
# drive the task touches: the parent drive (ctx.drive_root) and any child /
|
||||
|
|
@ -287,8 +288,7 @@ def resolve_user_file_path(
|
|||
read_only = operation in _tool_access()._READ_OPS
|
||||
cyber = mode_has_unrestricted_agency(get_runtime_mode()) and (read_only or active_tool_profile(ctx) != "local_readonly_subagent")
|
||||
from ouroboros.tool_access_reads import read_allows_outside_home
|
||||
outside_home_allowed = read_allows_outside_home(ctx) if read_only else _tool_access().is_external_workspace(ctx)
|
||||
if not allow_outside_home and not cyber and not outside_home_allowed:
|
||||
if not allow_outside_home and not cyber:
|
||||
home_resolved = home.resolve(strict=False)
|
||||
# Case-insensitive-platform parity with the user_files_path_block_reason
|
||||
# authority: a differently-cased safe home path must not be rejected
|
||||
|
|
@ -305,7 +305,10 @@ def resolve_user_file_path(
|
|||
) or _tool_access()._path_is_relative_to_casefold(candidate, deliverables_resolved)
|
||||
except (OSError, ValueError):
|
||||
inside_deliverables = False
|
||||
outside_home_allowed = True
|
||||
if not inside_home and not inside_deliverables:
|
||||
outside_home_allowed = read_allows_outside_home(ctx) if read_only else _tool_access().is_external_workspace(ctx)
|
||||
if not outside_home_allowed:
|
||||
raise UserFilesPathBlockedError(
|
||||
"user_files path blocked: absolute path "
|
||||
f"{raw_text!r} is outside the user_files home ({home_resolved}). "
|
||||
|
|
|
|||
|
|
@ -664,6 +664,9 @@ def _child_workspace(ctx, metadata, params):
|
|||
if selected_folder:
|
||||
try:
|
||||
selected_path = Path(selected_folder).expanduser()
|
||||
if not selected_path.is_absolute() and (not parent_workspace["root"]
|
||||
or parent_workspace.get("availability") == "unavailable"):
|
||||
raise ValueError("relative workspace_root needs an available parent folder; name an absolute readable folder")
|
||||
workspace_root = admit_child_start_folder(ctx,
|
||||
selected_path if selected_path.is_absolute() else Path(parent_workspace["root"]) / selected_path, params)
|
||||
workspace_mode = "read_only"
|
||||
|
|
@ -687,7 +690,7 @@ def _schedule_task(ctx: ToolContext, internal: Dict[str, Any] | None = None, /,
|
|||
return _publish_scheduling_refusal(
|
||||
ctx, "error", "TOOL_ARG_ERROR", "⚠️ TOOL_ARG_ERROR (schedule_subagent): unsupported argument(s): "
|
||||
f"{bad}. Use the strict schema: subagent_id, objective, expected_output, "
|
||||
"optional role/context/constraints/memory_mode and (for mutative children) "
|
||||
"optional role/context/constraints/memory_mode/workspace_root and (for mutative children) "
|
||||
"write_surface/write_root/protected_paths_grant/external_tool_grants.")
|
||||
internal = dict(internal or {})
|
||||
if set(internal) - _INTERNAL_SCHEDULE_OPTIONS:
|
||||
|
|
|
|||
|
|
@ -71,7 +71,7 @@ def schedule_subagent_properties() -> Dict[str, Any]:
|
|||
"memory, so empty is a blank drive, not a blank context. shared is disabled for live local subagents. "
|
||||
"input_sources=declared independently selects automatic inputs."),
|
||||
},
|
||||
"workspace_root": {"type": "string", "description": "Optional folder, inherited from the parent when omitted. Read-only helpers start exactly here, including Git subdirectories. For self_worktree this selects the Git source copied with its current files into the isolated working tree. An explicit write_root must name the same folder; omit workspace_root for genesis, which provisions an empty project. The folder must already be readable by the parent and grants no read or write authority."},
|
||||
"workspace_root": {"type": "string", "description": "Optional folder, inherited from the parent when omitted. Read-only helpers start exactly here, including Git subdirectories. For self_worktree this selects the Git source copied with its current files into the isolated working tree. For external_workspace an explicit write_root must name the same folder; omit workspace_root for genesis, which provisions an empty project. The folder must already be readable by the parent and grants no read or write authority."},
|
||||
"write_surface": {
|
||||
"type": "string",
|
||||
# No empty-string member: Google Gemini's function-calling validator
|
||||
|
|
@ -84,7 +84,7 @@ def schedule_subagent_properties() -> Dict[str, Any]:
|
|||
"enum": ["read_only", "self_worktree", "external_workspace", "genesis"],
|
||||
"description": "read_only (or omit) = read-only child starting in workspace_root or the inherited folder. A MUTATIVE child uses self_worktree (isolated current-tree Git copy of workspace_root or the inherited source, returning a patch for parent integration), external_workspace (native children write shared files directly), or genesis (standalone project). See tool description for integration. Acting surfaces require mutative subagents enabled (default ON in advanced/pro).",
|
||||
},
|
||||
"write_root": {"type": "string", "description": "For write_surface=external_workspace: the external project directory, with or without Git, never runtime data. An installed skill payload has its own resource address: delegate it directly with delegate_start(subagent_id=..., prompt=..., root='skill_payload', bucket=..., skill_name=...). OMIT write_root to build COOPERATIVELY from scratch — the host mints ONE shared git tree the whole subagent tree writes into together (deeper descendants inherit it), and you verify the combined files with integrate_subagent_patch without reapplying them. For self_worktree, workspace_root selects the source, so write_root cannot select a different folder. Genesis provisions its own empty root."},
|
||||
"write_root": {"type": "string", "description": "For write_surface=external_workspace: the external project directory, with or without Git, never runtime data. An installed skill payload has its own resource address: delegate it directly with delegate_start(subagent_id=..., prompt=..., root='skill_payload', bucket=..., skill_name=...). OMIT write_root to build COOPERATIVELY from scratch — the host mints ONE shared git tree the whole subagent tree writes into together (deeper descendants inherit it), and you verify the combined files with integrate_subagent_patch without reapplying them. Ignored for self_worktree (workspace_root selects its source), read_only and genesis (which provisions its own empty root)."},
|
||||
"directory_strategy": {
|
||||
"type": "string", "enum": ["direct", "copy"],
|
||||
"description": "For an agent_session in an ordinary folder: direct works in the selected folder; copy works in a separate copy of scope_paths and returns changes for application. Choose according to the task and any owner preference. Omit for direct ordinary-folder work. Write-capable children only: a read-only child omits both this and scope_paths (direct with no scope is the same as omitting). Native/API children use shared files directly and do not support copy.",
|
||||
|
|
|
|||
|
|
@ -238,7 +238,7 @@ def _handle_schedule_task(evt: Dict[str, Any], ctx: Any) -> None:
|
|||
"session_id": session_id,
|
||||
"delegation_role": delegation_role,
|
||||
})
|
||||
if workspace_copy:
|
||||
if delegation_role == "subagent" and not acting_reject_detail:
|
||||
task_contract = {**task_contract, "workspace": {
|
||||
**task_contract.get("workspace", {}), "root": workspace_root, "mode": workspace_mode}}
|
||||
live_max_depth = _events().get_max_subagent_depth()
|
||||
|
|
|
|||
|
|
@ -408,12 +408,6 @@ def _resolve_subagent_constraint(
|
|||
readonly = {"mode": LOCAL_READONLY_SUBAGENT_MODE, "allow_enable": False, "allow_review": False}
|
||||
req = requested_constraint if isinstance(requested_constraint, dict) else {}
|
||||
if str(req.get("mode") or "") != ACTING_SUBAGENT_MODE:
|
||||
if workspace_root:
|
||||
from ouroboros.tool_access_reads import readonly_start_folder
|
||||
try:
|
||||
workspace_root = readonly_start_folder(workspace_root)
|
||||
except (OSError, ValueError, RuntimeError) as exc:
|
||||
return readonly, workspace_root, workspace_mode, f"Subagent rejected: {exc}"
|
||||
return readonly, workspace_root, workspace_mode, ""
|
||||
surface = str(req.get("surface") or "").strip().lower()
|
||||
if surface not in VALID_WRITE_SURFACES:
|
||||
|
|
|
|||
|
|
@ -76,13 +76,20 @@ def test_http_policy_uses_actual_operation_and_identity(mode, tmp_path, monkeypa
|
|||
|
||||
@pytest.mark.parametrize("url", [
|
||||
"http://169.254.169.254/latest/meta-data", "http://[fe80::1]/", "http://2852039166/",
|
||||
"http://10.0.0.1/", "file:///outside/workspace/report.html",
|
||||
"http://10.0.0.1/",
|
||||
])
|
||||
def test_cyber_targets_do_not_require_internal_permission(url):
|
||||
assert browser_policy.browser_url_block_reason(url, restricted=False, runtime_mode="cyber_pro") == ""
|
||||
assert browser_policy.browser_url_block_reason(url, restricted=True, runtime_mode="cyber_pro")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["light", "advanced", "pro", "cyber_pro"])
|
||||
def test_local_file_reads_follow_parent_reach_without_a_workspace_fence(mode):
|
||||
for restricted in (False, True):
|
||||
assert browser_policy.browser_url_block_reason(
|
||||
"file:///outside/workspace/report.html", restricted=restricted, runtime_mode=mode) == ""
|
||||
|
||||
|
||||
@pytest.mark.browser
|
||||
@pytest.mark.parametrize("engine", ["chromium", "webkit"])
|
||||
def test_live_evaluate_reads_policy_words_and_cyber_posts_once(control_page, tmp_path, monkeypatch, engine):
|
||||
|
|
|
|||
|
|
@ -250,8 +250,8 @@ class TestChildDriveSkillPayload:
|
|||
assert "launcher-seed" in marker
|
||||
assert "nativeDependency" in dependency
|
||||
assert "node_modules/" in listing and ".seed-origin" in listing
|
||||
assert ".clawhub.json" not in listing
|
||||
assert "BLOCKED" in control
|
||||
assert ".clawhub.json" in listing
|
||||
assert '{"origin":"catalog"}' in control
|
||||
assert "SKILL.md" in search
|
||||
|
||||
write = registry.execute("write_file", {
|
||||
|
|
|
|||
|
|
@ -131,3 +131,23 @@ def test_read_start_mode_alone_cannot_bypass_workspace_write_admission(geometry)
|
|||
result = registry.execute_result("write_file", {"path": "new.txt", "content": "unadmitted"})
|
||||
assert result.status == "blocked" and "WORKSPACE_MODE_BLOCKED" in result.text
|
||||
assert not (selected / "new.txt").exists()
|
||||
|
||||
|
||||
def test_home_search_does_not_replay_ancestry_for_each_file(geometry, monkeypatch):
|
||||
from ouroboros import tool_access_reads
|
||||
|
||||
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
|
||||
home, repo, data, first = geometry
|
||||
root = ToolContext(repo_dir=repo, drive_root=data, task_id="root", workspace_root=first, workspace_mode="external")
|
||||
registry, _ctx = child(geometry, "child", "root", home / "work", capture_parent_workspace(root))
|
||||
documents = home / "documents"
|
||||
documents.mkdir()
|
||||
for index in range(40):
|
||||
(documents / f"input-{index}.txt").write_text(f"LOCAL_INPUT_{index}\n", encoding="utf-8")
|
||||
def unexpected_ancestry(_ctx):
|
||||
pytest.fail("an in-home file asked for the outside-home ancestry policy")
|
||||
monkeypatch.setattr(tool_access_reads, "read_allows_outside_home", unexpected_ancestry)
|
||||
read = registry.execute("read_file", {"root": "user_files", "path": str(documents / "input-0.txt")})
|
||||
assert "LOCAL_INPUT_0" in read, read
|
||||
result = registry.execute("search_code", {"root": "user_files", "path": str(documents), "query": "LOCAL_INPUT_"})
|
||||
assert "Found 40 matches" in result, result
|
||||
|
|
|
|||
|
|
@ -142,7 +142,7 @@ def test_parent_readable_folders_are_admitted_without_mutation_geometry(start_re
|
|||
assert registry._ctx.pending_events[-1]["workspace_root"] == str(folder)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("surface", ["external_workspace", "self_worktree", "genesis"])
|
||||
@pytest.mark.parametrize("surface", ["external_workspace", "genesis"])
|
||||
def test_contradictory_named_start_is_refused_before_enqueue(start_registry, surface):
|
||||
registry, args, home, _repo, _data, _outside = start_registry
|
||||
selected, other = home / "selected", home / "other"
|
||||
|
|
@ -168,3 +168,117 @@ def test_matching_named_start_and_write_root_preserve_existing_external_surface(
|
|||
event = registry._ctx.pending_events[-1]
|
||||
assert event["task_constraint"]["mode"] == "acting_subagent"
|
||||
assert event["task_constraint"]["write_root"] == str(folder)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("surface", ["read_only", "self_worktree"])
|
||||
def test_ignored_write_root_neither_refuses_nor_changes_named_start(start_registry, surface):
|
||||
registry, args, home, _repo, _data, _outside = start_registry
|
||||
folder = home / "selected"
|
||||
folder.mkdir()
|
||||
if surface == "self_worktree":
|
||||
subprocess.run(["git", "init", "--quiet", str(folder)], check=True, capture_output=True)
|
||||
result = registry.execute_result("schedule_subagent", {
|
||||
**args, "workspace_root": str(folder), "write_surface": surface,
|
||||
"write_root": str(home / "ignored-write-target"),
|
||||
})
|
||||
assert result.status == "ok", result.text
|
||||
assert registry._ctx.pending_events[-1]["workspace_root"] == str(folder)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("selection", ["omitted", "absolute", "relative"])
|
||||
@pytest.mark.parametrize("known_address", [False, True])
|
||||
def test_unavailable_project_focus_does_not_block_independent_helper(start_registry, selection, known_address):
|
||||
registry, args, home, repo, data, _outside = start_registry
|
||||
missing = home / "unavailable-project"
|
||||
folder = home / "available-project"
|
||||
folder.mkdir()
|
||||
ctx = registry._ctx
|
||||
ctx.is_direct_chat, ctx.project_id = True, "project-fixture"
|
||||
ctx.task_metadata = {"_project_room_note": "Selected project folder is unavailable"}
|
||||
if known_address:
|
||||
ctx.task_metadata["_project_room_dir"] = str(missing)
|
||||
options = ({"workspace_root": str(folder)} if selection == "absolute" else
|
||||
{"workspace_root": "relative"} if selection == "relative" else {})
|
||||
result = registry.execute_result("schedule_subagent", {**args, **options})
|
||||
if selection == "relative":
|
||||
assert result.code == "TOOL_ARG_ERROR" and "available parent folder" in result.text
|
||||
assert not ctx.pending_events
|
||||
return
|
||||
assert result.status == "ok", result.text
|
||||
event = ctx.pending_events[-1]
|
||||
assert event["parent_workspace"] == {
|
||||
"root": str(missing) if known_address else "", "mode": "", "source": "project_room",
|
||||
"availability": "unavailable", "detail": ctx.task_metadata["_project_room_note"],
|
||||
}
|
||||
assert event.get("workspace_root", "") == (str(folder) if selection == "absolute" else
|
||||
str(missing) if known_address else "")
|
||||
assert ctx.task_metadata["_project_room_note"] == "Selected project folder is unavailable"
|
||||
row = _admit_event(registry, data, repo)
|
||||
child = ToolContext(repo_dir=repo, drive_root=Path(row["drive_root"]), task_id=row["id"],
|
||||
workspace_root=Path(row["workspace_root"]) if row["workspace_root"] else None,
|
||||
workspace_mode=row["workspace_mode"], task_metadata=row["metadata"], project_id=row["project_id"],
|
||||
task_constraint=TaskConstraint(mode="local_readonly_subagent"))
|
||||
assert child.active_repo_dir() != repo
|
||||
assert child.active_repo_dir() == (folder if selection == "absolute" else missing if known_address
|
||||
else Path(row["drive_root"]) / "task_drives" / row["id"])
|
||||
|
||||
|
||||
def _admit_event(registry, data, repo):
|
||||
from tests.test_nested_rights_depth import _fake_ctx
|
||||
from supervisor.events import _handle_schedule_task
|
||||
|
||||
queued = []
|
||||
supervisor = _fake_ctx(data, queued)
|
||||
supervisor.REPO_DIR = repo
|
||||
_handle_schedule_task(registry._ctx.pending_events[-1], supervisor)
|
||||
assert len(queued) == 1
|
||||
return queued[0]
|
||||
|
||||
|
||||
def test_missing_inherited_readonly_folder_keeps_its_address_and_other_reads(start_registry):
|
||||
registry, args, home, repo, data, _outside = start_registry
|
||||
missing = home / "gone-project"
|
||||
registry._ctx.workspace_root, registry._ctx.workspace_mode = missing, "external"
|
||||
result = registry.execute_result("schedule_subagent", args)
|
||||
assert result.status == "ok", result.text
|
||||
row = _admit_event(registry, data, repo)
|
||||
child = ToolContext(repo_dir=repo, drive_root=data, task_id=row["id"], task_metadata=row["metadata"],
|
||||
workspace_root=Path(row["workspace_root"]), workspace_mode=row["workspace_mode"],
|
||||
task_constraint=TaskConstraint(mode="local_readonly_subagent"))
|
||||
registry.set_context(child)
|
||||
assert child.active_repo_dir() == missing and not missing.exists()
|
||||
(repo / "source.txt").write_text("OTHER AUTHORIZED ROOT", encoding="utf-8")
|
||||
assert "OTHER AUTHORIZED ROOT" in registry.execute("read_file", {"root": "system_repo", "path": "source.txt"})
|
||||
assert "OTHER AUTHORIZED ROOT" not in registry.execute("read_file", {"path": "source.txt"})
|
||||
for name, options in (("write_file", {"path": "source.txt", "content": "no"}),
|
||||
("run_command", {"command": "true"})):
|
||||
assert registry.execute_result(name, options).status != "ok"
|
||||
assert not missing.exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("named", [True, False])
|
||||
def test_admitted_external_workspace_is_one_fact_in_contract_result_and_context(start_registry, named):
|
||||
from ouroboros.task_results import load_task_result
|
||||
|
||||
registry, args, home, repo, data, _outside = start_registry
|
||||
parent, selected = home / "parent-project", home / "target-project"
|
||||
parent.mkdir()
|
||||
selected.mkdir()
|
||||
registry._ctx.workspace_root, registry._ctx.workspace_mode = parent, "external"
|
||||
result = registry.execute_result("schedule_subagent", {
|
||||
**args, "write_surface": "external_workspace", "write_root": str(selected),
|
||||
**({"workspace_root": str(selected)} if named else {}),
|
||||
})
|
||||
assert result.status == "ok", result.text
|
||||
row = _admit_event(registry, data, repo)
|
||||
expected = {"root": str(selected), "mode": "external_workspace"}
|
||||
assert {"root": row["workspace_root"], "mode": row["workspace_mode"]} == expected
|
||||
assert row["task_contract"]["workspace"] == expected
|
||||
assert row["metadata"]["task_contract"]["workspace"] == expected
|
||||
assert build_task_contract(row)["workspace"] == expected
|
||||
assert load_task_result(data, row["id"], strict=True)["task_contract"]["workspace"] == expected
|
||||
assert row["task_constraint"]["write_root"] == str(selected)
|
||||
ctx = ToolContext(repo_dir=repo, drive_root=data, workspace_root=Path(row["workspace_root"]),
|
||||
workspace_mode=row["workspace_mode"], task_metadata=row["metadata"],
|
||||
task_constraint=TaskConstraint(mode="acting_subagent", surface="external_workspace"))
|
||||
assert ctx.active_repo_dir() == selected
|
||||
|
|
|
|||
|
|
@ -639,8 +639,10 @@ def test_the_binding_authority_is_its_own_carrier_and_fails_closed():
|
|||
|
||||
|
||||
def _parent(root, metadata, *, task_id="presence-turn-1", ceiling=True):
|
||||
repo = root / "repo"
|
||||
repo.mkdir(parents=True, exist_ok=True)
|
||||
return types.SimpleNamespace(
|
||||
task_depth=0, pending_events=[], drive_root=root, task_id=task_id, task_metadata=metadata,
|
||||
repo_dir=repo, task_depth=0, pending_events=[], drive_root=root, task_id=task_id, task_metadata=metadata,
|
||||
task_contract={"capability_ceiling": presence_ceiling_payload(_ceiling())} if ceiling else {},
|
||||
current_chat_id=4242, is_direct_chat=ceiling, is_workspace_mode=lambda: False,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -71,11 +71,11 @@ def test_inferred_native_selection_preserves_mutation_boundary(tmp_path, operati
|
|||
_resolve(tmp_path, constraint, operation=operation)
|
||||
|
||||
|
||||
def test_inferred_native_selection_preserves_child_selector_boundary(tmp_path):
|
||||
_payload(tmp_path, bucket="native", seeded=True)
|
||||
def test_inferred_native_selection_preserves_acting_parent_read_parity(tmp_path):
|
||||
expected = _payload(tmp_path, bucket="native", seeded=True)
|
||||
constraint = {"mode": "normal", "skill_name": "alpha", "payload_root": "skills/native/alpha"}
|
||||
with pytest.raises(ValueError, match="cannot select skill location=native"):
|
||||
_resolve(tmp_path, constraint, profile="acting_subagent", top_level=False, operation="read")
|
||||
assert _resolve(tmp_path, constraint, profile="acting_subagent", top_level=False,
|
||||
operation="read") == (expected, "native", "alpha")
|
||||
|
||||
|
||||
def test_inferred_native_read_keeps_existing_readonly_authority(tmp_path):
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ def test_api_all_fields_form_queues_without_changing_write_surface(registry, acc
|
|||
"objective": "Inspect the assigned source.", "expected_output": "Findings.",
|
||||
"role": "", "context": "", "constraints": "", "memory_mode": "forked",
|
||||
"input_sources": "shared",
|
||||
"write_surface": surface, "write_root": "", "directory_strategy": "direct",
|
||||
"write_surface": surface, "write_root": "", "workspace_root": "", "directory_strategy": "direct",
|
||||
"scope_paths": [], "protected_paths_grant": False, "external_tool_grants": [],
|
||||
"allowed_origins": [], "delegation_intent": "", "may_mutate": False,
|
||||
"may_fan_out": True, "max_children": 0, "requested_depth": 0,
|
||||
|
|
|
|||
|
|
@ -35,7 +35,10 @@ def test_schedule_task_live_emits_strict_contract_and_requested_status(tmp_path,
|
|||
|
||||
_configure_test_subagent(monkeypatch)
|
||||
event_queue = _FakeEventQueue(status_root=tmp_path)
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir(exist_ok=True)
|
||||
ctx = SimpleNamespace(
|
||||
repo_dir=repo,
|
||||
task_depth=0,
|
||||
pending_events=[],
|
||||
event_queue=event_queue,
|
||||
|
|
@ -96,7 +99,10 @@ def test_schedule_task_falls_back_to_pending_events_when_live_queue_unavailable(
|
|||
from ouroboros.tools.control import _schedule_task
|
||||
|
||||
_configure_test_subagent(monkeypatch)
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir(exist_ok=True)
|
||||
ctx = SimpleNamespace(
|
||||
repo_dir=repo,
|
||||
task_depth=0,
|
||||
pending_events=[],
|
||||
event_queue=_FakeEventQueue(fail=True),
|
||||
|
|
@ -323,7 +329,10 @@ def test_schedule_task_memory_modes_prepare_declared_drive_shape(tmp_path, monke
|
|||
(parent_memory / "knowledge" / "pattern.md").write_text("stable pattern", encoding="utf-8")
|
||||
|
||||
event_queue = _FakeEventQueue()
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir(exist_ok=True)
|
||||
ctx = SimpleNamespace(
|
||||
repo_dir=repo,
|
||||
task_depth=0,
|
||||
pending_events=[],
|
||||
event_queue=event_queue,
|
||||
|
|
@ -382,7 +391,10 @@ def test_configured_session_child_materializes_initial_and_steered_attachments(t
|
|||
attachment_manifest=steered_manifest,
|
||||
)
|
||||
event_queue = _FakeEventQueue()
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir(exist_ok=True)
|
||||
ctx = SimpleNamespace(
|
||||
repo_dir=repo,
|
||||
task_depth=0, pending_events=[], event_queue=event_queue,
|
||||
drive_root=tmp_path, task_id="parent-attachments",
|
||||
task_contract={"attachment_manifest": [dict(row) for row in parent_manifest]},
|
||||
|
|
@ -422,7 +434,10 @@ def test_schedule_task_rejects_legacy_description_schema(tmp_path, monkeypatch):
|
|||
from ouroboros.tools.control import _schedule_task
|
||||
|
||||
_configure_test_subagent(monkeypatch)
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir(exist_ok=True)
|
||||
ctx = SimpleNamespace(
|
||||
repo_dir=repo,
|
||||
task_depth=0,
|
||||
pending_events=[],
|
||||
event_queue=None,
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ def test_ordinary_top_level_presets_share_one_exact_principal_matrix():
|
|||
def test_shared_top_level_principal_does_not_widen_specialized_profiles():
|
||||
assert "shell" not in _POLICY["local_readonly_subagent"]["skill_payload"]
|
||||
assert "skill_repair" not in _POLICY
|
||||
assert "skill_payload" not in _POLICY["acting_subagent"]
|
||||
assert _POLICY["acting_subagent"]["skill_payload"] == {"read", "list", "search"}
|
||||
for profile in ("local_readonly_subagent", "acting_subagent"):
|
||||
assert {"read", "list", "search"} <= _POLICY[profile]["runtime_data"]
|
||||
assert not {"write", "edit"} & _POLICY[profile]["runtime_data"]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue