diff --git a/agentcli/agent/registry.py b/agentcli/agent/registry.py index 556da60..1ca1a03 100644 --- a/agentcli/agent/registry.py +++ b/agentcli/agent/registry.py @@ -302,9 +302,7 @@ async def execute( validate(payload, self.specs[agent_type].schema) await self.context.policy.aauthorize(agent_type, payload) agent = factory() - result: SubAgentResult = await asyncio.wait_for( - agent.run(task), self.context.policy.timeout_seconds - ) + result: SubAgentResult = await agent.run(task) from .evidence import annotate annotate( diff --git a/agentcli/subagents/base.py b/agentcli/subagents/base.py index a2a05a7..5c0fc35 100644 --- a/agentcli/subagents/base.py +++ b/agentcli/subagents/base.py @@ -464,7 +464,7 @@ async def guarded(self: Any, task: SubAgentTask) -> SubAgentResult: "Helper operation 'filesystem_tool' is unsupported on this foundation runtime" ) - result = await asyncio.wait_for(handler(self, task), policy.timeout_seconds) + result = await handler(self, task) return bound_result(result, policy.max_output_bytes) except ( OSError, diff --git a/agentcli/subagents/file_ops.py b/agentcli/subagents/file_ops.py index 9e9fc98..a4f6af2 100644 --- a/agentcli/subagents/file_ops.py +++ b/agentcli/subagents/file_ops.py @@ -5,12 +5,15 @@ from __future__ import annotations +import asyncio import os -import shutil +import tempfile +from dataclasses import replace from pathlib import Path from typing import TYPE_CHECKING, Any from ..files import is_hidden_or_system +from ..policy import tool_policy from .base import SubAgent, SubAgentResult, SubAgentTask, SubAgentType if TYPE_CHECKING: @@ -30,15 +33,20 @@ def __init__( message_bus: MessageBus | None = None, ) -> None: super().__init__(SubAgentType.FILE_OPS, config, message_bus) - self.working_dir = Path(self.config.get("working_dir", os.getcwd())).resolve() + self.working_dir = self.policy.root + from ..agent.checkpoints import CheckpointManager + + self.journal: CheckpointManager = self.config.get("journal") or CheckpointManager( + policy=tool_policy(self.config) + ) self.allow_outside = bool(self.config.get("allow_outside_working_dir", False)) - # Phase 10: Read-only by default unless allow_write or read_only=False is explicitly configured + # Read-only by default unless allow_write or read_only=False is explicitly configured self.read_only = bool( self.config.get("read_only", not self.config.get("allow_write", False)) ) def _resolve_path(self, path: str, operation: str = "") -> Path: - """Resolve and validate a file path. + """Resolve and validate a file path under the effective policy. Args: path: The path to resolve (can be relative or absolute). @@ -50,39 +58,209 @@ def _resolve_path(self, path: str, operation: str = "") -> Path: Raises: ValueError: If path is outside working directory and not allowed. """ - raw_path = Path(path) - is_explicit_abs = raw_path.is_absolute() - if not is_explicit_abs: - path_obj = (self.working_dir / raw_path).resolve() - else: - path_obj = raw_path.resolve() - - # Check if path is within working directory - if not self.allow_outside: - # Allow read-only operations on explicit absolute paths - if is_explicit_abs and operation in ("read", "list", "stat", "exists"): - return path_obj + is_mutation = operation in {"write", "create", "append", "delete", "mkdir"} + pol = tool_policy(self.config) + return pol.path(path, write=is_mutation) + + def _atomic_write(self, path: Path, data: bytes) -> None: + """Safely write data to a file atomically via tempfile replacement.""" + policy = tool_policy(self.config) + if len(data) > min(policy.max_file_bytes, 1048576): + raise ValueError("Mutation exceeds restorable file byte limit") + fd, name = tempfile.mkstemp(prefix=".agentcli-write-", dir=path.parent) + temporary = Path(name) + try: + with os.fdopen(fd, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + if path.exists(): + try: + os.chmod(temporary, path.stat().st_mode) + except OSError: + pass + policy.path(path, write=True) + os.replace(temporary, path) + finally: + if temporary.exists(): + try: + os.chmod(temporary, 0o600) + except OSError: + pass + try: + temporary.unlink() + except OSError: + pass + async def run(self, task: SubAgentTask) -> SubAgentResult: + """Parent dispatch for file operations. + + In worker mode: executes _run_local directly. + In parent mode: captures pre-state, dispatches to helper, acquires post-state, + and preserves cancellation cleanly. + """ + from ..bounds import bound_result + from ..io_boundary import filesystem_request, in_worker, query, unpack_snapshot + from ..policy import PolicyDenied, active_policy + + if in_worker: + return await self._run_local(task) + + policy = tool_policy(self.config) + is_governed = active_policy.get() is not None + + payload = task.payload + if not isinstance(payload, dict): + res = SubAgentResult( + task_id=task.id, + agent_type=self.agent_type, + success=False, + error="Payload must be a dictionary", + ) + return res if is_governed else bound_result(res, policy.max_output_bytes) + + op_raw = payload.get("operation") + if not isinstance(op_raw, str) or not op_raw.strip(): + res = SubAgentResult( + task_id=task.id, + agent_type=self.agent_type, + success=False, + error="No operation specified", + ) + return res if is_governed else bound_result(res, policy.max_output_bytes) + operation = op_raw.strip().lower() + + path_raw = next( + ( + payload.get(k) + for k in ("path", "file_path", "filename", "file", "target") + if payload.get(k) is not None + ), + None, + ) + if path_raw is None or not isinstance(path_raw, str) or not path_raw.strip(): + res = SubAgentResult( + task_id=task.id, + agent_type=self.agent_type, + success=False, + error="No path specified", + ) + return res if is_governed else bound_result(res, policy.max_output_bytes) + path = path_raw + + checkpoint_id: str | None = None + + try: + if operation in {"write", "create", "append", "delete"} and not self.read_only: + if operation in {"write", "create", "append"}: + content = payload.get("content", "") + encoding = payload.get("encoding", "utf-8") + if not isinstance(content, str) or not isinstance(encoding, str): + raise ValueError("Content and encoding must be strings") + data = content.encode(encoding) + limit = min(policy.max_file_bytes, 1048576) + if len(data) > limit: + raise ValueError("Mutation exceeds restorable file byte limit") + + # Capture pre-state via helper + snapshot = unpack_snapshot( + await query(filesystem_request("capture", policy, path=path), policy) + ) + if operation == "append": + content = payload.get("content", "") + encoding = payload.get("encoding", "utf-8") + data = content.encode(encoding) if isinstance(content, str) else b"" + limit = min(policy.max_file_bytes, 1048576) + if len(snapshot.data or b"") + len(data) > limit: + raise ValueError("Mutation exceeds restorable file byte limit") + + checkpoint_id = self.journal.create_checkpoint(f"{operation}: {snapshot.rel_path}") + self.journal.record_snapshot(snapshot) + + # Primary mutation query with cancellation preservation + primary_exc: BaseException | None = None try: - path_obj.relative_to(self.working_dir) - except ValueError: - raise ValueError( - f"Path '{path}' is outside working directory '{self.working_dir}' " - "and cross-directory access is not enabled" + req = filesystem_request( + "file_ops", + policy, + payload=payload, + task_id=task.id, + read_only=self.read_only, + ) + h_res = await query(req, policy) + result = SubAgentResult( + task_id=task.id, + agent_type=task.agent_type, + success=h_res.get("success", False), + output=h_res.get("output"), + error=h_res.get("error"), ) + except BaseException as exc: + primary_exc = exc + raise + finally: + if checkpoint_id: + cleanup_budget = min(5.0, policy.timeout_seconds) + post_policy = replace(policy, timeout_seconds=cleanup_budget) + checkpoint = self.journal._get_checkpoint(checkpoint_id) + if checkpoint is not None: + for snap_path, snapshot in list(checkpoint.files.items()): + try: + fp_req = filesystem_request( + "fingerprint", post_policy, path=snap_path + ) + snapshot.expected = await query(fp_req, post_policy) + snapshot.sealed = True + from ..agent.checkpoints import unchanged - return path_obj + if unchanged(snapshot): + checkpoint.files.pop(snap_path) + except BaseException as cleanup_exc: + if primary_exc is not None and isinstance( + primary_exc, asyncio.CancelledError + ): + raise primary_exc from cleanup_exc + raise + self.journal.discard_empty_checkpoint(checkpoint_id) - async def run(self, task: SubAgentTask) -> SubAgentResult: - """Execute a file operation task. + if not is_governed: + return bound_result(result, policy.max_output_bytes) + return result + + except ( + PolicyDenied, + ValueError, + TypeError, + OSError, + TimeoutError, + NotImplementedError, + ) as exc: + if is_governed: + raise + err_msg = str(exc) + if isinstance(exc, PolicyDenied): + if "outside the granted workspace" in err_msg: + err_msg = f"Operation failed: Path is outside working directory ({exc})" + elif "read-only mode" in err_msg: + err_msg = f"Operation '{operation}' is not permitted in read-only mode ({exc})" + res = SubAgentResult( + task_id=task.id, + agent_type=self.agent_type, + success=False, + error=err_msg, + ) + return bound_result(res, policy.max_output_bytes) + + async def _run_local(self, task: SubAgentTask) -> SubAgentResult: + """Execute a file operation task locally within active policy context. Expected payload: - - operation: "read" | "write" | "delete" | "list" | "mkdir" + - operation: "read" | "write" | "delete" | "list" | "mkdir" | "exists" | "stat" | "append" | "create" - path: target file/directory path - - content: content to write (for write operation) + - content: content to write (for write/append/create) - encoding: text encoding (default: utf-8) """ - payload = task.payload + payload = task.payload or {} operation = payload.get("operation", "").lower() path = ( payload.get("path") @@ -120,6 +298,8 @@ async def run(self, task: SubAgentTask) -> SubAgentResult: error=f"Operation '{operation}' is not permitted in read-only mode. Use --allow-write to enable mutations.", ) + policy = tool_policy(self.config) + if operation == "read": if not resolved_path.exists(): return SubAgentResult( @@ -129,7 +309,13 @@ async def run(self, task: SubAgentTask) -> SubAgentResult: error=f"File not found: {path}", ) encoding = payload.get("encoding", "utf-8") - content = resolved_path.read_text(encoding=encoding) + if not isinstance(encoding, str): + raise ValueError("Encoding must be a string") + with resolved_path.open("rb") as stream: + data = stream.read(policy.max_file_bytes + 1) + if len(data) > policy.max_file_bytes: + raise ValueError("File read exceeds byte limit") + content = data.decode(encoding, errors="replace") return SubAgentResult( task_id=task.id, agent_type=self.agent_type, @@ -140,31 +326,47 @@ async def run(self, task: SubAgentTask) -> SubAgentResult: elif operation in ("write", "create"): content = payload.get("content", "") encoding = payload.get("encoding", "utf-8") + if not isinstance(content, str) or not isinstance(encoding, str): + raise ValueError("Content and encoding must be strings") + prepared_data = content.encode(encoding) + limit = min(policy.max_file_bytes, 1048576) + if len(prepared_data) > limit: + raise ValueError("Mutation exceeds restorable file byte limit") resolved_path.parent.mkdir(parents=True, exist_ok=True) - resolved_path.write_text(content, encoding=encoding, newline="") + self._atomic_write(resolved_path, prepared_data) return SubAgentResult( task_id=task.id, agent_type=self.agent_type, success=True, output={ "path": str(resolved_path), - "bytes_written": len(content.encode(encoding)), + "bytes_written": len(prepared_data), }, ) elif operation == "append": content = payload.get("content", "") encoding = payload.get("encoding", "utf-8") + if not isinstance(content, str) or not isinstance(encoding, str): + raise ValueError("Content and encoding must be strings") + prepared_data = content.encode(encoding) + appended_bytes = len(prepared_data) + limit = min(policy.max_file_bytes, 1048576) + if resolved_path.exists(): + with resolved_path.open("rb") as stream: + original = stream.read(limit + 1) + prepared_data = original + prepared_data + if len(prepared_data) > limit: + raise ValueError("Mutation exceeds restorable file byte limit") resolved_path.parent.mkdir(parents=True, exist_ok=True) - with resolved_path.open("a", encoding=encoding, newline="") as f: - f.write(content) + self._atomic_write(resolved_path, prepared_data) return SubAgentResult( task_id=task.id, agent_type=self.agent_type, success=True, output={ "path": str(resolved_path), - "bytes_appended": len(content.encode(encoding)), + "bytes_appended": appended_bytes, }, ) @@ -177,9 +379,10 @@ async def run(self, task: SubAgentTask) -> SubAgentResult: error=f"Path not found: {path}", ) if resolved_path.is_dir(): - shutil.rmtree(resolved_path) - else: - resolved_path.unlink() + raise ValueError( + "Directory deletion is unsupported; file undo cannot restore directories" + ) + resolved_path.unlink() return SubAgentResult( task_id=task.id, agent_type=self.agent_type, @@ -202,17 +405,28 @@ async def run(self, task: SubAgentTask) -> SubAgentResult: success=False, error=f"Not a directory: {path}", ) + from ..bounds import walk_entries + include_hidden = bool(payload.get("include_hidden", payload.get("all", False))) - items = [] - for item in resolved_path.iterdir(): + items: list[dict[str, Any]] = [] + # The shared walker counts native acquisitions before filtering, + # rejects protected/link entries before metadata access and closes + # scandir on exhaustion/error. An exhausted budget fails closed. + for item, _, directory in walk_entries( + resolved_path, policy, max_entries=4096, max_depth=1 + ): if not include_hidden and is_hidden_or_system(item): continue + try: + size = item.stat().st_size if item.is_file() else None + except OSError: + size = None items.append( { "name": item.name, "path": str(item), - "is_dir": item.is_dir(), - "size": item.stat().st_size if item.is_file() else None, + "is_dir": directory, + "size": size, } ) return SubAgentResult( @@ -235,7 +449,12 @@ async def run(self, task: SubAgentTask) -> SubAgentResult: exists = resolved_path.exists() is_file = resolved_path.is_file() if exists else False is_dir = resolved_path.is_dir() if exists else False - size = resolved_path.stat().st_size if is_file else None + size = None + if is_file: + try: + size = resolved_path.stat().st_size + except OSError: + size = None return SubAgentResult( task_id=task.id, agent_type=self.agent_type, diff --git a/tests/test_helper_request_contracts.py b/tests/test_helper_request_contracts.py index 6b1e336..7ab7c4d 100644 --- a/tests/test_helper_request_contracts.py +++ b/tests/test_helper_request_contracts.py @@ -15,6 +15,7 @@ import socket import sys from pathlib import Path +from typing import Any from unittest.mock import patch import pytest @@ -334,18 +335,42 @@ def test_io_worker_truthful_unsupported_failures( ): _io_worker.filesystem(req_unknown) - # file_ops operation: on this frozen foundation runtime base, FileOpsAgent lacks _run_local + # file_ops interface check: rejects FileOpsAgent implementations lacking _run_local + class MissingRunLocalAgent: + def __init__(self, config: dict[str, Any] | None = None) -> None: + self.config = config or {} + + from agentcli.subagents import file_ops as file_ops_module + + dummy_file = tmp_path / "dummy.txt" + dummy_file.write_text("dummy content", encoding="utf-8") req_fileops = io_boundary.filesystem_request( "file_ops", policy, - task_id="t-fileops", + task_id="t-fileops-missing", read_only=True, - payload={"operation": "read", "path": str(tmp_path / "dummy.txt")}, + payload={"operation": "read", "path": str(dummy_file)}, ) - with pytest.raises( - NotImplementedError, match="FileOps in-worker local execution is unavailable" - ): - _io_worker.filesystem(req_fileops) + with monkeypatch.context() as m: + m.setattr(file_ops_module, "FileOpsAgent", MissingRunLocalAgent) + with pytest.raises( + NotImplementedError, match="FileOps in-worker local execution is unavailable" + ): + _io_worker.filesystem(req_fileops) + + # file_ops positive execution: real FileOpsAgent with _run_local executes successfully + req_positive = io_boundary.filesystem_request( + "file_ops", + policy, + task_id="t-fileops-pos", + read_only=True, + payload={"operation": "read", "path": str(dummy_file)}, + ) + pos_res = _io_worker.filesystem(req_positive) + assert isinstance(pos_res, dict) + assert pos_res["success"] is True + assert pos_res["output"]["content"] == "dummy content" + assert pos_res["error"] is None # ============================================================================ diff --git a/tests/test_i01_i02_regressions.py b/tests/test_i01_i02_regressions.py new file mode 100644 index 0000000..ae686c3 --- /dev/null +++ b/tests/test_i01_i02_regressions.py @@ -0,0 +1,1706 @@ +"""I01/I02 Permanent Regression Tests. + +These tests reproduce the original I01/I02 failures and verify the fixes. +Adapted for issue-76-corrections worktree: +- Explicit child worker bootstrap to ensure child subprocesses import this checkout. +- Explicit RunPolicy injection into FileOpsAgent configs. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +import tempfile +from pathlib import Path +from typing import Any, Self + +import pytest + +import agentcli +from agentcli.bounds import bound_result +from agentcli.policy import RunPolicy +from agentcli.subagents.base import SubAgent, SubAgentResult, SubAgentTask, SubAgentType +from agentcli.subagents.file_ops import FileOpsAgent + +# Path to the current worktree root for child subprocess bootstrap +WORKTREE_ROOT = Path(agentcli.__file__).resolve().parent.parent +CHILD_BOOTSTRAP = f"import sys; sys.path.insert(0, {str(WORKTREE_ROOT)!r}); " + + +@pytest.fixture(autouse=True) +def ensure_child_worktree_bootstrap(monkeypatch: pytest.MonkeyPatch) -> None: + """Ensure that default io_boundary.worker_command bootstraps sys.path to this worktree.""" + from agentcli import io_boundary + + def bootstrapped_worker_command() -> list[str]: + return [ + sys.executable, + "-I", + "-c", + CHILD_BOOTSTRAP + + "import runpy; runpy.run_module('agentcli._io_worker', run_name='__main__')", + ] + + monkeypatch.setattr(io_boundary, "worker_command", bootstrapped_worker_command) + + +class TestI01CancellationPreservation: + """I01: FileOps cancellation swallowed by post-state fingerprint timeout. + + When a FileOps write is cancelled after the atomic replacement commits, + the `finally` block's fingerprint acquisition (5s cleanup budget) can + timeout before the outer tool deadline (10s). The original cancellation + must be preserved, not replaced by a generic timeout error. + """ + + @pytest.mark.asyncio + async def test_cancellation_preserved_across_fingerprint_timeout(self, tmp_path: Path) -> None: + """Reproduce I01: cancellation preserved across fingerprint timeout. + + This test uses a custom worker that adds a 10s delay in fingerprint, + while the outer policy has 10s timeout and cleanup has 5s budget. + The cancellation should propagate, not be replaced by timeout error. + """ + marker = tmp_path / "fingerprint.pid" + + # Custom worker that adds delay in fingerprint - signals after atomic commit + header = ( + CHILD_BOOTSTRAP + + "import os,time,json\nfrom pathlib import Path\n" + + "marker=Path(" + + repr(str(marker)) + + ")\n" + + "original_fingerprint=None\n" + + "def slow_fingerprint(path):\n" + + " marker.write_text(json.dumps({'pid': os.getpid(), 'time': time.time()}))\n" + + " time.sleep(10)\n" + + " return original_fingerprint(path)\n" + + "import agentcli.agent.checkpoints as checkpoints_mod\n" + + "original_fingerprint = checkpoints_mod.fingerprint\n" + + "checkpoints_mod.fingerprint = slow_fingerprint\n" + ) + worker_cmd = ( + header + "import runpy\nrunpy.run_module('agentcli._io_worker',run_name='__main__')\n" + ) + + from agentcli import io_boundary + + original_worker_cmd = io_boundary.worker_command + io_boundary.worker_command = lambda: [sys.executable, "-I", "-c", worker_cmd] + + try: + with tempfile.TemporaryDirectory(dir=tmp_path) as test_dir_str: + test_dir = Path(test_dir_str) + test_file = test_dir / "test.txt" + test_file.write_text("original content", encoding="utf-8") + + # Policy with 10s outer timeout, 5s cleanup budget + policy = RunPolicy( + root=test_dir, + allow_write=True, + timeout_seconds=10.0, + max_output_bytes=65536, + ) + + # Apply the policy to the agent via config + agent = FileOpsAgent( + config={ + "policy": policy, + "working_dir": str(test_dir), + "allow_write": True, + "read_only": False, + "command_timeout": 10.0, # Explicit 10s timeout + } + ) + + task = SubAgentTask( + id="test-cancel-1", + agent_type=SubAgentType.FILE_OPS, + payload={ + "operation": "write", + "path": str(test_file), + "content": "new content " + "x" * 100000, + "encoding": "utf-8", + }, + ) + + task_future = asyncio.create_task(agent.run(task)) + + # Wait for fingerprint to start (it runs in finally block after write) + for _ in range(100): + if marker.exists(): + break + await asyncio.sleep(0.05) + assert marker.exists(), "Fingerprint never started" + + # Verify the atomic write committed + assert test_file.read_text(encoding="utf-8") == "new content " + "x" * 100000, ( + "Atomic commit not observed" + ) + + # Cancel the task - fingerprint is running and will take 10s + # The outer policy timeout is 10s, cleanup is 5s + # The cancellation should propagate, not be replaced by timeout error + task_future.cancel() + + try: + await task_future + except asyncio.CancelledError: + # This is what should happen - cancellation propagates as exception + return + # If we get here, cancellation was swallowed + raise AssertionError("Cancellation was swallowed - expected CancelledError") + + finally: + io_boundary.worker_command = original_worker_cmd + + +def check_alive(pid: int) -> bool: + """Query process liveness natively; fail closed on query errors other than 87.""" + if sys.platform == "win32": + import ctypes + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel.OpenProcess.restype = wintypes.HANDLE + kernel.GetExitCodeProcess.argtypes = [ + wintypes.HANDLE, + ctypes.POINTER(wintypes.DWORD), + ] + kernel.GetExitCodeProcess.restype = wintypes.BOOL + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + + handle = kernel.OpenProcess(0x1000, False, pid) + if not handle: + last_err = ctypes.get_last_error() + if last_err == 87: # ERROR_INVALID_PARAMETER (process does not exist) + return False + raise ctypes.WinError(last_err) + try: + code = wintypes.DWORD() + assert kernel.GetExitCodeProcess(handle, ctypes.byref(code)) + return code.value == 259 # STILL_ACTIVE + finally: + kernel.CloseHandle(handle) + else: + import os + + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + + +class TestI01CancellationControls: + """I01: Additional cancellation control tests.""" + + @pytest.mark.asyncio + async def test_repeated_cancellation_same_task(self, tmp_path: Path) -> None: + """Cancelling the same pending request twice preserves CancelledError.""" + target = tmp_path / "target.bin" + target.write_bytes(b"original\x00\xff\r\n") + committed = tmp_path / "committed.pid" + sealing = tmp_path / "sealing.pid" + + worker = ( + CHILD_BOOTSTRAP + + """import os,time +from pathlib import Path +from agentcli import _io_worker +original_execute=_io_worker.execute +original_replace=os.replace +def execute(request): + if request.get('operation')=='fingerprint': + Path(SEAL_MARKER).write_text(str(os.getpid())) + time.sleep(20) + return original_execute(request) +def replace(source,destination,*args,**kwargs): + value=original_replace(source,destination,*args,**kwargs) + if Path(destination).name=='target.bin': + Path(COMMIT_MARKER).write_text(str(os.getpid())) + time.sleep(20) + return value +_io_worker.execute=execute +os.replace=replace +_io_worker.main() +""".replace("SEAL_MARKER", repr(str(sealing))).replace("COMMIT_MARKER", repr(str(committed))) + ) + + from agentcli import io_boundary + from agentcli.agent.registry import ToolRegistry + from agentcli.runtime import RunContext + + original_command = io_boundary.worker_command + io_boundary.worker_command = lambda: [sys.executable, "-I", "-c", worker] + context = RunContext( + policy=RunPolicy(root=tmp_path, allow_write=True, timeout_seconds=10.0) + ) + pending = None + commit_pid = None + cleanup_pid = None + import time + + try: + registry = ToolRegistry(context=context) + pending = asyncio.create_task( + registry.execute( + "file_ops", + {"operation": "write", "path": "target.bin", "content": "changed"}, + ) + ) + + # 1. Observe atomic commit and capture mutation PID + ready_deadline = time.monotonic() + 8 + while time.monotonic() < ready_deadline: + if committed.exists() and committed.read_text(encoding="utf-8").isdigit(): + break + if pending.done(): + raise AssertionError("mutation finished before controlled cancellation") + await asyncio.sleep(0.01) + assert committed.exists() and committed.read_text(encoding="utf-8").isdigit(), ( + "atomic replacement never entered" + ) + commit_pid = int(committed.read_text(encoding="utf-8")) + assert target.read_bytes() == b"changed" + + # 2. Assert pending request and request first cancellation + assert not pending.done(), "request finished before first cancellation" + first_cancel_requested = pending.cancel() + assert first_cancel_requested, "first cancellation must be requested" + + # 3. Await fresh DISTINCT parent-cleanup marker (bounded wait) + cleanup_ready_deadline = time.monotonic() + 4 + while time.monotonic() < cleanup_ready_deadline: + if sealing.exists() and sealing.read_text(encoding="utf-8").isdigit(): + break + assert not pending.done(), "request finished before controlled cleanup readiness" + await asyncio.sleep(0.01) + assert sealing.exists() and sealing.read_text(encoding="utf-8").isdigit(), ( + "cleanup helper never entered" + ) + + # 4. Capture cleanup PID; assert SAME original request remains pending + cleanup_pid = int(sealing.read_text(encoding="utf-8")) + assert not pending.done(), "second cancellation must target pending request" + + # 5. Request second cancellation and assert request accepted + second_cancel_requested = pending.cancel() + assert second_cancel_requested, "second cancellation must be requested" + + # 6. Require propagated asyncio.CancelledError on final await + done, _ = await asyncio.wait({pending}, timeout=8) + assert pending in done, "post-state cleanup exceeded reviewer watchdog" + try: + result = await pending + raise AssertionError(f"Expected CancelledError, got result: {result}") + except asyncio.CancelledError: + pass + + # 7. Assert retained snapshot.sealed is False + assert len(context.journal._checkpoints) >= 1, "Checkpoint must exist" + for cp in context.journal._checkpoints: + for snap in cp.files.values(): + assert snap.sealed is False, "snapshot must not be sealed" + + # 8. Verify rollback refuses unknown post-state, reverts no files and leaves committed bytes unchanged + rollback_result = context.journal.rollback() + assert rollback_result["success"] is False + assert rollback_result.get("reverted_files") == [] + error_msg = rollback_result.get("error", "").lower() + assert ( + "unsealed" in error_msg + or "missing expected post-state" in error_msg + or "conflict" in error_msg + ), f"Rollback should fail for unsealed checkpoint: {rollback_result}" + assert target.read_bytes() == b"changed", "committed bytes must remain unchanged" + + # 9. Explicitly verify BOTH captured helper PIDs are no longer live + await context.aclose() + assert not check_alive(commit_pid), f"Commit helper {commit_pid} still alive" + assert not check_alive(cleanup_pid), f"Cleanup helper {cleanup_pid} still alive" + finally: + if pending is not None and not pending.done(): + pending.cancel() + if pending is not None: + await asyncio.gather(pending, return_exceptions=True) + await context.aclose() + io_boundary.worker_command = original_command + + def test_windows_liveness_query_negative_control(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Negative control: error 5 (Access Denied) fails closed, error 87 succeeds as absent.""" + if sys.platform != "win32": + pytest.skip("Windows-only negative control for Win32 OpenProcess query semantics") + + import ctypes + + class StubFunction: + def __call__(self, *args: Any) -> int: + return 0 + + class StubKernel: + OpenProcess = StubFunction() + GetExitCodeProcess = StubFunction() + CloseHandle = StubFunction() + + monkeypatch.setattr(ctypes, "WinDLL", lambda *args, **kwargs: StubKernel()) + + # Error 87 (ERROR_INVALID_PARAMETER) correctly signals non-existent process + monkeypatch.setattr(ctypes, "get_last_error", lambda: 87) + assert check_alive(12345) is False + + # Error 5 (ERROR_ACCESS_DENIED) must fail closed (raise WinError/PermissionError), never return False + monkeypatch.setattr(ctypes, "get_last_error", lambda: 5) + with pytest.raises(PermissionError): + check_alive(12345) + + # Unknown error 1 (ERROR_INVALID_FUNCTION) must also fail closed + monkeypatch.setattr(ctypes, "get_last_error", lambda: 1) + with pytest.raises(OSError): + check_alive(12345) + + @pytest.mark.asyncio + async def test_non_cancelled_sealing_failure_returns_error(self, tmp_path: Path) -> None: + """Non-cancelled fingerprint failure returns error, not cancelled.""" + marker = tmp_path / "fingerprint.pid" + + # Worker that fails fingerprint with IOError + header = ( + CHILD_BOOTSTRAP + + "import os,time,json\nfrom pathlib import Path\n" + + "marker=Path(" + + repr(str(marker)) + + ")\n" + + "original_fingerprint=None\n" + + "def failing_fingerprint(path):\n" + + " marker.write_text(json.dumps({'pid': os.getpid(), 'time': time.time()}))\n" + + " raise IOError('Simulated fingerprint failure')\n" + + "import agentcli.agent.checkpoints as checkpoints_mod\n" + + "original_fingerprint = checkpoints_mod.fingerprint\n" + + "checkpoints_mod.fingerprint = failing_fingerprint\n" + ) + worker_cmd = ( + header + "import runpy\nrunpy.run_module('agentcli._io_worker',run_name='__main__')\n" + ) + + from agentcli import io_boundary + + original_worker_cmd = io_boundary.worker_command + io_boundary.worker_command = lambda: [sys.executable, "-I", "-c", worker_cmd] + + try: + with tempfile.TemporaryDirectory(dir=tmp_path) as test_dir_str: + test_dir = Path(test_dir_str) + test_file = test_dir / "test.txt" + test_file.write_text("original content", encoding="utf-8") + + policy = RunPolicy( + root=test_dir, + allow_write=True, + timeout_seconds=10.0, + max_output_bytes=65536, + ) + + agent = FileOpsAgent( + config={ + "policy": policy, + "working_dir": str(test_dir), + "allow_write": True, + "read_only": False, + "command_timeout": 10.0, + } + ) + + task = SubAgentTask( + id="test-cancel-3", + agent_type=SubAgentType.FILE_OPS, + payload={ + "operation": "write", + "path": str(test_file), + "content": "new content", + "encoding": "utf-8", + }, + ) + + result = await agent.run(task) + + # Should fail with fingerprint error, not cancellation + assert result.success is False + assert ( + "fingerprint" in (result.error or "").lower() + or "ioerror" in (result.error or "").lower() + ), f"Expected fingerprint error, got: {result.error}" + + finally: + io_boundary.worker_command = original_worker_cmd + + +class TestI02OutputCapEnforcement: + """I02: Tiny output caps exceeded by truncation envelope.""" + + @pytest.mark.asyncio + async def test_fileops_read_with_tiny_cap_returns_error(self, tmp_path: Path) -> None: + """Test that FileOps read with 16-byte cap returns error, not oversized envelope.""" + with tempfile.TemporaryDirectory(dir=tmp_path) as test_dir_str: + test_dir = Path(test_dir_str) + test_file = test_dir / "test.txt" + test_file.write_text( + "This is a test file with some content that exceeds 16 bytes", encoding="utf-8" + ) + + # Explicit policy with 16-byte cap + policy = RunPolicy( + root=test_dir, + allow_write=True, + timeout_seconds=10.0, + max_output_bytes=16, + ) + + agent = FileOpsAgent( + config={ + "policy": policy, + "working_dir": str(test_dir), + "allow_write": True, + "read_only": False, + "max_output_bytes": 16, + } + ) + + task = SubAgentTask( + id="test-read-1", + agent_type=SubAgentType.FILE_OPS, + payload={ + "operation": "read", + "path": str(test_file), + "encoding": "utf-8", + }, + ) + + result = await agent.run(task) + + # Should fail with envelope error, not return oversized envelope + assert result.success is False + assert result.output is None + assert "envelope" in (result.error or "").lower() + + @pytest.mark.asyncio + async def test_bound_result_16_byte_cap_rejects_non_control_output(self) -> None: + """Test bound_result directly with non-control output and 16-byte cap.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={ + "content": "This is a test file with some content that exceeds 16 bytes", + "path": "/test.txt", + }, + ) + + bounded = bound_result(result, 16) + + # Should fail with envelope error + assert bounded.success is False + assert bounded.output is None + assert "envelope" in (bounded.error or "").lower() + + @pytest.mark.asyncio + async def test_bound_result_allows_output_under_cap(self) -> None: + """Test that output under the cap is returned unchanged (no envelope).""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={ + "content": "tiny", + "path": "/test.txt", + }, + ) + + bounded = bound_result(result, 64) + + # Small output fits without truncation - returned as-is + assert bounded.success is True + assert bounded.output is not None + assert bounded.output.get("truncated") is not True + + +class TestI02OutputCapRepresentability: + """I02: Envelope size representability at various caps.""" + + @pytest.mark.asyncio + async def test_envelope_created_when_output_exceeds_cap(self) -> None: + """Output exceeding cap gets wrapped in envelope with truncated flag.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "x" * 50, "path": "/test.txt"}, + ) + + # 86 bytes raw, cap at 63 -> should truncate + bounded = bound_result(result, 63) + assert bounded.success is True + assert bounded.output is not None + assert bounded.output.get("truncated") is True + assert bounded.output.get("evidence_available") is False + + @pytest.mark.asyncio + async def test_output_under_cap_returned_unchanged(self) -> None: + """Output under cap is returned unchanged without envelope.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "small", "path": "/test.txt"}, + ) + + # 56 bytes raw, cap at 63 -> no envelope + bounded = bound_result(result, 63) + assert bounded.success is True + assert bounded.output is not None + assert bounded.output.get("truncated") is not True + + @pytest.mark.asyncio + async def test_envelope_boundary_at_63_bytes(self) -> None: + """Envelope overhead is 63 bytes; minimal cap for any envelope is 63.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "x" * 50, "path": "/test.txt"}, + ) + + # 86 bytes raw, cap 63 -> envelope of 63 bytes (empty content) + bounded = bound_result(result, 63) + assert bounded.success is True + assert bounded.output.get("content") == "" + assert bounded.output.get("truncated") is True + assert bounded.output.get("evidence_available") is False + + @pytest.mark.asyncio + async def test_envelope_at_62_bytes_rejected(self) -> None: + """62 bytes is below envelope overhead, should reject.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "x" * 50, "path": "/test.txt"}, + ) + + bounded = bound_result(result, 62) + assert bounded.success is False + assert "envelope" in (bounded.error or "").lower() + + @pytest.mark.asyncio + async def test_envelope_at_64_bytes_allows_content(self) -> None: + """64 bytes allows envelope with 1 byte content.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "x" * 50, "path": "/test.txt"}, + ) + + bounded = bound_result(result, 64) + # At 64 bytes, budget is (64-63)//6 = 0, so content is still empty + assert bounded.success is True + assert len(bounded.output.get("content", "")) == 0 + + # At 69 bytes, budget becomes 1 - need fresh result since bound_result modifies in place + result2 = SubAgentResult( + task_id="test-2", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "x" * 50, "path": "/test.txt"}, + ) + bounded = bound_result(result2, 69) + assert bounded.success is True + assert len(bounded.output.get("content", "")) >= 1 + + @pytest.mark.asyncio + async def test_unicode_content_byte_count_respected(self) -> None: + """Unicode content byte count properly respected in cap.""" + # Each emoji is 4 bytes in UTF-8 + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "\U0001f389" * 5, "path": "/test.txt"}, + ) + + # 5 emojis * 4 = 20 bytes content + JSON overhead ~56 bytes = 76 bytes + # Should fit at 80 bytes + bounded = bound_result(result, 80) + assert bounded.success is True + + # Verify exact byte count + output_json = json.dumps(bounded.output, default=str).encode("utf-8") + assert len(output_json) <= 80 + + # Should fail at 60 bytes + result2 = SubAgentResult( + task_id="test-2", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": "\U0001f389" * 5, "path": "/test.txt"}, + ) + bounded = bound_result(result2, 60) + assert bounded.success is False + assert "envelope" in (bounded.error or "").lower() + + @pytest.mark.asyncio + async def test_special_characters_escaped_in_json(self) -> None: + """Special characters (quotes, backslashes, newlines) properly escaped in JSON.""" + result = SubAgentResult( + task_id="test-1", + agent_type=SubAgentType.FILE_OPS, + success=True, + output={"content": 'quote"backslash\\newline\n' * 20, "path": "/test.txt"}, + ) + + # Should handle escapes properly - cap at 80 to force truncation + bounded = bound_result(result, 80) + assert bounded.success is True + assert "truncated" in bounded.output + + # Verify actual byte count + output_json = json.dumps(bounded.output, default=str).encode("utf-8") + assert len(output_json) <= 80 + + +class TestI03PermanentReviewCoverage: + """R100 permanent review coverage: + + - ToolRegistry successful file write, retained sealed snapshot and byte-exact rollback. + - Rollback conflict refusal. + - Read/append byte caps with unchanged bytes on rejection. + - Denied outside/read-only/protected paths. + - Standalone default output cap and failed-result compatibility. + - Protected listing plus ordinary sibling. + - Actual helper test state restoration with real MonkeyPatch. + """ + + @pytest.mark.asyncio + async def test_tool_registry_write_seal_and_rollback(self, tmp_path: Path) -> None: + """Cover ToolRegistry successful write, retained sealed snapshot and byte-exact rollback.""" + from agentcli.agent.registry import ToolRegistry + from agentcli.runtime import RunContext + + target = tmp_path / "journal_target.txt" + # Non-UTF8 bytes plus CRLF: only byte-exact undo can restore this. + original_bytes = b"initial \xff\xfe\r\nstate\n" + target.write_bytes(original_bytes) + + context = RunContext( + policy=RunPolicy(root=tmp_path, allow_write=True, timeout_seconds=10.0) + ) + try: + registry = ToolRegistry(context=context) + res = await registry.execute( + "file_ops", + { + "operation": "write", + "path": "journal_target.txt", + "content": "updated content\n", + }, + ) + assert res.success is True + assert target.read_bytes() == b"updated content\n" + + # Checkpoint retained and sealed + assert len(context.journal._checkpoints) >= 1 + last_cp = context.journal._checkpoints[-1] + resolved_key = str(target.resolve()) + assert resolved_key in last_cp.files + snap = last_cp.files[resolved_key] + assert snap.sealed is True + assert snap.expected is not None + + # Rollback to exact initial bytes + rb = context.journal.rollback() + assert rb["success"] is True + assert any("journal_target.txt" in f for f in rb["reverted_files"]) + assert target.read_bytes() == original_bytes + finally: + await context.aclose() + + @pytest.mark.asyncio + async def test_tool_registry_rollback_conflict_refusal(self, tmp_path: Path) -> None: + """Cover rollback conflict refusal when external modification occurs.""" + from agentcli.agent.registry import ToolRegistry + from agentcli.runtime import RunContext + + target = tmp_path / "conflict_target.txt" + target.write_bytes(b"initial state\n") + + context = RunContext( + policy=RunPolicy(root=tmp_path, allow_write=True, timeout_seconds=10.0) + ) + try: + registry = ToolRegistry(context=context) + res = await registry.execute( + "file_ops", + {"operation": "write", "path": "conflict_target.txt", "content": "agent write\n"}, + ) + assert res.success is True + assert target.read_bytes() == b"agent write\n" + + # External modification alters file content and fingerprint + target.write_bytes(b"tampered by external process\n") + + # Rollback refuses to overwrite conflicted file + rb = context.journal.rollback() + assert rb["success"] is False + assert "conflict" in rb.get("error", "").lower() + assert target.read_bytes() == b"tampered by external process\n" + finally: + await context.aclose() + + @pytest.mark.asyncio + async def test_read_and_append_byte_caps_with_unchanged_bytes(self, tmp_path: Path) -> None: + """Read and append byte caps reject oversized payloads and preserve original bytes.""" + target = tmp_path / "capped_file.txt" + initial_bytes = b"initial payload\n" + target.write_bytes(initial_bytes) + + # Policy with max_file_bytes = 30 + policy = RunPolicy(root=tmp_path, allow_write=True, max_file_bytes=30) + agent = FileOpsAgent( + config={"policy": policy, "working_dir": str(tmp_path), "allow_write": True} + ) + + # New content alone fits the cap, but existing + new exceeds the total budget + oversized_append = "y" * 20 + res_append = await agent.run( + SubAgentTask( + id="append-cap", + agent_type=SubAgentType.FILE_OPS, + payload={ + "operation": "append", + "path": "capped_file.txt", + "content": oversized_append, + }, + ) + ) + assert res_append.success is False + assert "byte limit" in (res_append.error or "").lower() + assert target.read_bytes() == initial_bytes + + # Write large file to test read cap + large_target = tmp_path / "large_read.txt" + large_target.write_bytes(b"y" * 100) + res_read = await agent.run( + SubAgentTask( + id="read-cap", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "read", "path": "large_read.txt"}, + ) + ) + assert res_read.success is False + assert "byte limit" in (res_read.error or "").lower() + assert large_target.read_bytes() == b"y" * 100 + + @pytest.mark.asyncio + async def test_denied_outside_and_readonly_paths(self, tmp_path: Path) -> None: + """Denied outside root and read-only mutations return failed SubAgentResult.""" + agent = FileOpsAgent(config={"working_dir": str(tmp_path)}) + + # Outside root + res_outside = await agent.run( + SubAgentTask( + id="outside-root", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "read", "path": "../outside.txt"}, + ) + ) + assert res_outside.success is False + assert "outside working directory" in (res_outside.error or "") + + # Read-only mutation + res_ro = await agent.run( + SubAgentTask( + id="ro-mutation", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "write", "path": "test.txt", "content": "forbidden"}, + ) + ) + assert res_ro.success is False + assert "read-only mode" in (res_ro.error or "") + + # Protected credential file read is denied and never returns secret content + secret = tmp_path / ".env.local" + secret.write_bytes(b"SECRET_TOKEN=do-not-leak") + res_protected = await agent.run( + SubAgentTask( + id="protected-read", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "read", "path": ".env.local"}, + ) + ) + assert res_protected.success is False + assert "Credential/instruction metadata" in (res_protected.error or "") + # No secret content may leak through output or the error message + assert "do-not-leak" not in json.dumps(res_protected.output, default=str) + assert "do-not-leak" not in (res_protected.error or "") + + @pytest.mark.asyncio + async def test_standalone_default_cap_and_compatibility(self, tmp_path: Path) -> None: + """Standalone unmanaged calls enforce default 65536 byte cap.""" + (tmp_path / "large_standalone.txt").write_text("x" * 70000, encoding="utf-8") + agent = FileOpsAgent(config={"working_dir": str(tmp_path)}) + task = SubAgentTask( + id="standalone-cap", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "read", "path": "large_standalone.txt"}, + ) + result = await agent.run(task) + assert result.success is True + size = len(json.dumps(result.output, ensure_ascii=False).encode("utf-8")) + assert size <= 65536 + assert (result.output or {}).get("truncated") is True + + @pytest.mark.asyncio + async def test_protected_listing_and_ordinary_siblings(self, tmp_path: Path) -> None: + """Protected credential/instruction files omitted, ordinary siblings retained.""" + (tmp_path / "normal_file.txt").write_text("hello", encoding="utf-8") + (tmp_path / ".env.production").write_text("SECRET=1", encoding="utf-8") + (tmp_path / ".git").mkdir() + (tmp_path / ".git" / "config").write_text("repo", encoding="utf-8") + + policy = RunPolicy(root=tmp_path) + agent = FileOpsAgent(config={"policy": policy, "working_dir": str(tmp_path)}) + + listing = await agent.run( + SubAgentTask( + id="list-protected", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "list", "path": ".", "include_hidden": True}, + ) + ) + assert listing.success is True + names = [item["name"] for item in (listing.output or {}).get("items", [])] + assert "normal_file.txt" in names + assert ".env.production" not in names + assert ".git" not in names + + def test_actual_helper_test_state_restoration(self, tmp_path: Path) -> None: + """Actual installed test_io_worker_truthful_unsupported_failures restores in_worker state.""" + import importlib.util + + from agentcli import io_boundary + + spec = importlib.util.spec_from_file_location( + "installed_helper_tests", + WORKTREE_ROOT / "tests/test_helper_request_contracts.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + before = io_boundary.in_worker + patches = pytest.MonkeyPatch() + try: + module.test_io_worker_truthful_unsupported_failures(tmp_path, patches) + finally: + patches.undo() + + assert io_boundary.in_worker == before + + +class _CountedNativeScandir: + """Deterministic ``os.scandir`` seam that repeats one real DirEntry. + + Counts every native acquisition attempt, including entries the walker later + filters, so the 4096-entry budget can be exercised without disk fixtures. + """ + + def __init__(self, entry: os.DirEntry[str], available: int) -> None: + self._entry = entry + self._available = available + self.acquired = 0 + self.closed = False + + def __iter__(self) -> _CountedNativeScandir: + return self + + def __next__(self) -> os.DirEntry[str]: + if self.acquired >= self._available: + raise StopIteration + self.acquired += 1 + return self._entry + + def __enter__(self) -> Self: + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + def close(self) -> None: + self.closed = True + + +class TestI04BoundedNativeListing: + """R101: helper-local FileOps listing acquisitions are bounded. + + The shared walker counts native acquisitions before filtering, so protected + or otherwise filtered entries must still exhaust the 4096-entry budget and + fail closed instead of returning a successful, silently incomplete listing. + """ + + # 4096-entry budget plus one exhaustion lookahead acquisition. + ACQUISITION_CAP = 4097 + + @staticmethod + def _install_counted_scandir( + monkeypatch: pytest.MonkeyPatch, target: Path, entry_name: str, available: int + ) -> _CountedNativeScandir: + """Repeat one real DirEntry for ``target`` while counting acquisitions.""" + original_scandir = os.scandir + with original_scandir(target) as real_entries: + entry = next(item for item in real_entries if item.name == entry_name) + seam = _CountedNativeScandir(entry, available) + + def counted_scandir(path: Any, *args: Any, **kwargs: Any) -> Any: + if Path(path) == target: + return seam + return original_scandir(path, *args, **kwargs) + + monkeypatch.setattr(os, "scandir", counted_scandir) + return seam + + async def _run_seeded_listing( + self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, entry_name: str + ) -> tuple[SubAgentResult, _CountedNativeScandir]: + from agentcli import io_boundary + + listing_dir = tmp_path / "listing" + listing_dir.mkdir() + (listing_dir / entry_name).write_text("payload", encoding="utf-8") + seam = self._install_counted_scandir(monkeypatch, listing_dir, entry_name, 20000) + # Helper-local execution: exercise the native walker without a child process. + monkeypatch.setattr(io_boundary, "in_worker", True) + policy = RunPolicy(root=tmp_path) + agent = FileOpsAgent(config={"policy": policy, "working_dir": str(tmp_path)}) + result = await agent.run( + SubAgentTask( + id="native-list", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "list", "path": "listing", "include_hidden": True}, + ) + ) + return result, seam + + @pytest.mark.asyncio + async def test_filtered_protected_entries_exhaust_budget( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """All acquisitions filtered as protected still consume the entry budget.""" + hidden = ".env.secret" # policy denies this entry, yet it must be counted + result, seam = await self._run_seeded_listing(monkeypatch, tmp_path, hidden) + + assert seam.acquired > 0 # counted before filtering, not by emitted rows + assert seam.acquired <= self.ACQUISITION_CAP + assert seam.closed is True + + assert result.success is False # no false complete success + assert "entry limit" in (result.error or "").lower() + assert result.output is None + + @pytest.mark.asyncio + async def test_ordinary_entries_exhaust_budget( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Ordinary entries beyond the budget fail closed with the same diagnostic.""" + result, seam = await self._run_seeded_listing(monkeypatch, tmp_path, "ordinary.txt") + + assert seam.acquired > 0 + assert seam.acquired <= self.ACQUISITION_CAP + assert seam.closed is True + + assert result.success is False + assert "entry limit" in (result.error or "").lower() + assert result.output is None + + @pytest.mark.asyncio + async def test_small_native_listing_stays_correct( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Small ordinary and protected listings keep their existing output shape.""" + from agentcli import io_boundary + + listing_dir = tmp_path / "listing" + listing_dir.mkdir() + (listing_dir / "ordinary_file.txt").write_text("visible", encoding="utf-8") + (listing_dir / ".env.production").write_text("SECRET=1", encoding="utf-8") + (listing_dir / ".git").mkdir() + monkeypatch.setattr(io_boundary, "in_worker", True) + + policy = RunPolicy(root=tmp_path) + agent = FileOpsAgent(config={"policy": policy, "working_dir": str(tmp_path)}) + result = await agent.run( + SubAgentTask( + id="native-small-list", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "list", "path": "listing", "include_hidden": True}, + ) + ) + assert result.success is True + names = [item["name"] for item in (result.output or {}).get("items", [])] + assert "ordinary_file.txt" in names + assert ".env.production" not in names + assert ".git" not in names + + +class _CooperativeOwnedHandler: + """Public duck-typed handler whose cancellation cleanup stays owned. + + The handler never completes on its own. Its ``finally`` finalizer records + that cleanup began and then blocks on an explicit release event, so the + public request must remain pending until the test releases it. A repeat + cancellation is delivered to that same finalizer instead of a detached + execution path, which makes request/handler ownership directly observable. + """ + + def __init__(self) -> None: + self.started = asyncio.Event() + self.finalizer_entered = asyncio.Event() + self.finalizer_finished = asyncio.Event() + self.release = asyncio.Event() + self.owner_task: asyncio.Task[Any] | None = None + self.finalizer_entries = 0 + self.repeat_cancellations_delivered = 0 + + async def run(self, task: SubAgentTask) -> SubAgentResult: + self.owner_task = asyncio.current_task() + self.started.set() + try: + while True: + await asyncio.sleep(1) + finally: + self.finalizer_entries += 1 + self.finalizer_entered.set() + # Remain owned: the request must not settle until this same finalizer + # is explicitly released, even across repeat cancellation. + while not self.release.is_set(): + try: + await self.release.wait() + except asyncio.CancelledError: + self.repeat_cancellations_delivered += 1 + self.finalizer_finished.set() + return SubAgentResult(task_id=task.id, agent_type=task.agent_type, success=True, output={}) + + +class _GuardedSlowAgent(SubAgent): + """Cooperative guarded handler that always exceeds its policy deadline.""" + + async def run(self, task: SubAgentTask) -> SubAgentResult: + await asyncio.sleep(10) + return SubAgentResult( + task_id=task.id, agent_type=task.agent_type, success=True, output={"ok": True} + ) + + +class _GuardedFastAgent(SubAgent): + """Cooperative guarded handler that completes well inside its deadline.""" + + async def run(self, task: SubAgentTask) -> SubAgentResult: + return SubAgentResult( + task_id=task.id, agent_type=task.agent_type, success=True, output={"ok": True} + ) + + +class TestI05CancellationOwnershipAndDeadlines: + """R102: the public request owns handler teardown across repeat cancellation. + + These regressions run on native CPython (including 3.11) without a child + helper: they assert public request/handler ownership and the unchanged + enclosing deadline contexts, not task identity or source shape alone. + """ + + @pytest.mark.asyncio + async def test_public_request_owns_handler_until_finalizer_released( + self, tmp_path: Path + ) -> None: + """Repeat cancellation stays inside one owned finalizer until released.""" + from agentcli.agent.registry import ToolRegistry + from agentcli.runtime import RunContext + + base_tasks = set(asyncio.all_tasks()) + context = RunContext(policy=RunPolicy(root=tmp_path, timeout_seconds=30.0)) + handler = _CooperativeOwnedHandler() + request: asyncio.Task[Any] | None = None + try: + registry = ToolRegistry(context=context) + registry.register( + "owned_contract", + lambda: handler, + schema={"type": "object", "properties": {}, "additionalProperties": False}, + effect="observation", + ) + request = asyncio.create_task(registry.execute("owned_contract", {})) + await asyncio.wait_for(handler.started.wait(), timeout=5.0) + assert not request.done(), "request settled before any cancellation" + + # First cancellation must enter the owned finalizer. + assert request.cancel() is True + await asyncio.wait_for(handler.finalizer_entered.wait(), timeout=5.0) + assert handler.finalizer_entries == 1 + # The request stays pending while cleanup is unfinished, and it is the + # task that actually owns the handler (no detached execution path). + assert not request.done(), "request returned before cleanup settled" + assert handler.owner_task is request, "handler ran outside the public request" + + # Second cancellation must reach that same (single) finalizer. + assert request.cancel() is True + watch = asyncio.get_running_loop().time() + 5.0 + while ( + handler.repeat_cancellations_delivered < 1 + and asyncio.get_running_loop().time() < watch + ): + await asyncio.sleep(0.01) + assert handler.repeat_cancellations_delivered >= 1 + assert handler.finalizer_entries == 1 + assert not request.done(), "request returned while finalizer stayed blocked" + + # Explicit release is the only thing that lets finalization settle. + handler.release.set() + done, _ = await asyncio.wait({request}, timeout=5.0) + assert request in done, "finalization did not settle after release" + with pytest.raises(asyncio.CancelledError): + await request + assert handler.finalizer_finished.is_set() + assert handler.owner_task.done() + finally: + handler.release.set() + if request is not None: + if not request.done(): + request.cancel() + await asyncio.gather(request, return_exceptions=True) + await context.aclose() + leaked = set(asyncio.all_tasks()) - base_tasks + assert leaked == set(), f"owned tasks leaked after return: {leaked}" + + @pytest.mark.asyncio + async def test_guarded_subagent_deadline_returns_truthful_failure(self, tmp_path: Path) -> None: + """The unchanged enclosing timeout still fails a slow guarded handler.""" + policy = RunPolicy(root=tmp_path, timeout_seconds=0.2) + agent = _GuardedSlowAgent(agent_type=SubAgentType.CODE_ANALYZER, config={"policy": policy}) + task = SubAgentTask(id="deadline-slow", agent_type=SubAgentType.CODE_ANALYZER, payload={}) + with policy.activate(): + result = await asyncio.wait_for(agent.run(task), timeout=5.0) + assert result.success is False + assert "timed out" in (result.error or "").lower() + + @pytest.mark.asyncio + async def test_guarded_subagent_within_deadline_succeeds(self, tmp_path: Path) -> None: + """The unchanged enclosing timeout does not fail a handler inside its deadline.""" + policy = RunPolicy(root=tmp_path, timeout_seconds=5.0) + agent = _GuardedFastAgent(agent_type=SubAgentType.CODE_ANALYZER, config={"policy": policy}) + task = SubAgentTask(id="deadline-fast", agent_type=SubAgentType.CODE_ANALYZER, payload={}) + with policy.activate(): + result = await asyncio.wait_for(agent.run(task), timeout=5.0) + assert result.success is True + assert (result.output or {}).get("ok") is True + + @pytest.mark.asyncio + async def test_public_registry_deadline_bounds_cooperative_handler( + self, tmp_path: Path + ) -> None: + """The registry's enclosing timeout still bounds a non-cancelled handler.""" + from agentcli.agent.registry import ToolRegistry + from agentcli.runtime import RunContext + + async def slow_handler() -> dict[str, bool]: + await asyncio.sleep(10) + return {"ok": True} + + context = RunContext(policy=RunPolicy(root=tmp_path, timeout_seconds=0.2)) + try: + registry = ToolRegistry(context=context) + registry.register_callable( + "slow_contract", + slow_handler, + description="slow contract", + schema={"type": "object", "properties": {}, "additionalProperties": False}, + ) + result = await asyncio.wait_for(registry.execute("slow_contract", {}), timeout=5.0) + finally: + await context.aclose() + assert result.success is False + assert isinstance(result.error, str) + + +class TestI06NativeFileOpsContracts: + """R103: helper-local native FileOps contract tests. + + These run ``FileOpsAgent._run_local``/``_atomic_write`` in-process against real + temporary files with a real active ``RunPolicy`` and ``io_boundary.in_worker`` + scoped by monkeypatch. They are helper-local contract tests, not end-to-end + claims; the real public subprocess tests above remain authoritative. + """ + + @staticmethod + async def _dispatch( + root: Path, + monkeypatch: pytest.MonkeyPatch, + policy: RunPolicy, + task: SubAgentTask, + *, + read_only: bool = False, + ) -> SubAgentResult: + from agentcli import io_boundary + + monkeypatch.setattr(io_boundary, "in_worker", True) + agent = FileOpsAgent( + config={ + "policy": policy, + "working_dir": str(root), + "allow_write": True, + "read_only": read_only, + } + ) + with policy.activate(): + return await agent.run(task) + + @staticmethod + def _task(operation: str, **payload: Any) -> SubAgentTask: + return SubAgentTask( + id=f"native-{operation}", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": operation, **payload}, + ) + + @pytest.mark.asyncio + async def test_atomic_write_create_bytes_and_temp_cleanup( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Write/create absent and existing files atomically, preserving bytes and mode.""" + import stat + + policy = RunPolicy(root=tmp_path, allow_write=True, max_file_bytes=4096) + text = "h\u00e9llo\x00world" + expected = text.encode("utf-8") + + created = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="created.bin", content=text), + ) + assert created.success is True + assert (created.output or {}).get("bytes_written") == len(expected) + assert (tmp_path / "created.bin").read_bytes() == expected + + nested = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("create", path="nested/dir/created2.txt", content="new"), + ) + assert nested.success is True + assert (tmp_path / "nested" / "dir" / "created2.txt").read_bytes() == b"new" + + existing = tmp_path / "existing.txt" + existing.write_bytes(b"original") + original_mode = stat.S_IMODE(existing.stat().st_mode) if os.name != "nt" else None + replaced = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="existing.txt", content="replaced"), + ) + assert replaced.success is True + assert existing.read_bytes() == b"replaced" + if original_mode is not None: + assert stat.S_IMODE(existing.stat().st_mode) == original_mode + + assert not list(tmp_path.rglob(".agentcli-write-*")), "staging temp files remained" + + @pytest.mark.asyncio + async def test_append_preserves_bytes_and_reports_count( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Append combines binary-safe original bytes with encoded new data.""" + policy = RunPolicy(root=tmp_path, allow_write=True, max_file_bytes=4096) + target = tmp_path / "append.bin" + original = b"\x00\xffAB\r\n" + target.write_bytes(original) + added = "\u00e9" + + appended = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("append", path="append.bin", content=added), + ) + assert appended.success is True + assert (appended.output or {}).get("bytes_appended") == len(added.encode("utf-8")) + assert target.read_bytes() == original + added.encode("utf-8") + + fresh = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("append", path="fresh.txt", content="seed"), + ) + assert fresh.success is True + assert (tmp_path / "fresh.txt").read_bytes() == b"seed" + + @pytest.mark.asyncio + async def test_oversized_mutation_rejected_without_change( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Oversized append/write is rejected and leaves original bytes intact.""" + policy = RunPolicy(root=tmp_path, allow_write=True, max_file_bytes=8) + target = tmp_path / "small.bin" + target.write_bytes(b"123456") + + oversized_append = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("append", path="small.bin", content="789012"), + ) + assert oversized_append.success is False + assert "byte limit" in (oversized_append.error or "").lower() + assert target.read_bytes() == b"123456" + + oversized_write = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="small.bin", content="x" * 20), + ) + assert oversized_write.success is False + assert "byte limit" in (oversized_write.error or "").lower() + assert target.read_bytes() == b"123456" + + @pytest.mark.asyncio + async def test_delete_success_missing_and_directory_rejection( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """File deletion succeeds, missing paths fail, directories are refused.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + victim = tmp_path / "victim.txt" + victim.write_bytes(b"bye") + + deleted = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("delete", path="victim.txt") + ) + assert deleted.success is True + assert (deleted.output or {}).get("deleted") is True + assert not victim.exists() + + missing = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("delete", path="victim.txt") + ) + assert missing.success is False + assert "not found" in (missing.error or "").lower() + + directory = tmp_path / "keepdir" + directory.mkdir() + (directory / "inner.txt").write_bytes(b"inner") + rejected = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("delete", path="keepdir") + ) + assert rejected.success is False + assert "directory deletion is unsupported" in (rejected.error or "").lower() + assert directory.is_dir() + assert (directory / "inner.txt").read_bytes() == b"inner" + + @pytest.mark.asyncio + async def test_mkdir_and_exists_stat_field_accuracy( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """mkdir, then exists/stat report accurate file/directory/size fields.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + + before = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("stat", path="data/child.txt") + ) + assert before.success is True + assert (before.output or {}).get("exists") is False + assert (before.output or {}).get("is_file") is False + assert (before.output or {}).get("is_dir") is False + assert (before.output or {}).get("size") is None + + made = await self._dispatch(tmp_path, monkeypatch, policy, self._task("mkdir", path="data")) + assert made.success is True + assert (made.output or {}).get("created") is True + assert (tmp_path / "data").is_dir() + + payload = b"payload!" + (tmp_path / "data" / "child.txt").write_bytes(payload) + after = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("exists", path="data/child.txt") + ) + assert after.success is True + assert (after.output or {}).get("exists") is True + assert (after.output or {}).get("is_file") is True + assert (after.output or {}).get("is_dir") is False + assert (after.output or {}).get("size") == len(payload) + + directory_stat = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("stat", path="data") + ) + assert directory_stat.success is True + assert (directory_stat.output or {}).get("is_dir") is True + assert (directory_stat.output or {}).get("is_file") is False + + @pytest.mark.asyncio + async def test_read_success_missing_oversized_and_encoding( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Read handles success, missing files, oversized files and encodings.""" + policy = RunPolicy(root=tmp_path, allow_write=True, max_file_bytes=64) + text = "Gr\u00fc\u00dfe \u00e9 text" + (tmp_path / "read.txt").write_bytes(text.encode("utf-8")) + + ok = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("read", path="read.txt") + ) + assert ok.success is True + assert (ok.output or {}).get("content") == text + + missing = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("read", path="absent.txt") + ) + assert missing.success is False + assert "file not found" in (missing.error or "").lower() + + (tmp_path / "big.bin").write_bytes(b"z" * 200) + oversized = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("read", path="big.bin") + ) + assert oversized.success is False + assert "byte limit" in (oversized.error or "").lower() + + utf16 = "wide" + (tmp_path / "utf16.txt").write_bytes(utf16.encode("utf-16")) + encoded = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("read", path="utf16.txt", encoding="utf-16"), + ) + assert encoded.success is True + assert (encoded.output or {}).get("content") == utf16 + + bad_encoding = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("read", path="read.txt", encoding=123), + ) + assert bad_encoding.success is False + assert "encoding must be a string" in (bad_encoding.error or "").lower() + + @pytest.mark.asyncio + async def test_read_directory_reports_os_error( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Reading a directory surfaces the native OS error as a failed result.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + (tmp_path / "adir").mkdir() + result = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("read", path="adir") + ) + assert result.success is False + assert "operation failed" in (result.error or "").lower() + + @pytest.mark.asyncio + async def test_replacement_failure_preserves_original_and_removes_temp( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A controlled replacement OS error keeps original bytes and cleans staging.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + target = tmp_path / "replace.txt" + target.write_bytes(b"original bytes") + + real_replace = os.replace + failing = {"active": True} + + def guarded_replace(src: Any, dst: Any) -> None: + if failing["active"]: + raise OSError("simulated replacement failure") + real_replace(src, dst) + + monkeypatch.setattr(os, "replace", guarded_replace) + failed = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="replace.txt", content="updated"), + ) + assert failed.success is False + assert (failed.error or "").lower().startswith("operation failed") + assert target.read_bytes() == b"original bytes" + assert not list(tmp_path.rglob(".agentcli-write-*")) + + # The scoped seam restores real behaviour for later operations. + failing["active"] = False + succeeded = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="replace.txt", content="updated"), + ) + assert succeeded.success is True + assert target.read_bytes() == b"updated" + + @pytest.mark.asyncio + async def test_invalid_content_and_encoding_types( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Non-string content/encoding fail through real public error handling.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + bad_content = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="bad.txt", content=123), + ) + assert bad_content.success is False + assert "must be strings" in (bad_content.error or "").lower() + + bad_encoding = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("append", path="bad.txt", content="ok", encoding=5), + ) + assert bad_encoding.success is False + assert "must be a string" in (bad_encoding.error or "").lower() + + @pytest.mark.asyncio + async def test_read_only_mutation_rejected( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A read-only agent refuses mutations without creating the target.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + result = await self._dispatch( + tmp_path, + monkeypatch, + policy, + self._task("write", path="blocked.txt", content="nope"), + read_only=True, + ) + assert result.success is False + assert "read-only mode" in (result.error or "").lower() + assert not (tmp_path / "blocked.txt").exists() + + @pytest.mark.asyncio + async def test_standalone_missing_operation_path_and_unknown( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Missing operation/path and unknown operations use public error handling.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + no_op = await self._dispatch( + tmp_path, + monkeypatch, + policy, + SubAgentTask( + id="native-no-op", + agent_type=SubAgentType.FILE_OPS, + payload={"path": "x.txt"}, + ), + ) + assert no_op.success is False + assert "no operation specified" in (no_op.error or "").lower() + + no_path = await self._dispatch( + tmp_path, + monkeypatch, + policy, + SubAgentTask( + id="native-no-path", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "read"}, + ), + ) + assert no_path.success is False + assert "no path specified" in (no_path.error or "").lower() + + unknown = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("frobnicate", path="x.txt") + ) + assert unknown.success is False + assert "unknown operation" in (unknown.error or "").lower() + + @pytest.mark.asyncio + async def test_list_hidden_filter_and_target_errors( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Listing filters hidden entries by default and reports target errors.""" + policy = RunPolicy(root=tmp_path, allow_write=True) + listing = tmp_path / "listing" + listing.mkdir() + (listing / "visible.txt").write_text("v", encoding="utf-8") + (listing / ".hidden_note.txt").write_text("h", encoding="utf-8") + + filtered = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("list", path="listing") + ) + assert filtered.success is True + names = [item["name"] for item in (filtered.output or {}).get("items", [])] + assert "visible.txt" in names + assert ".hidden_note.txt" not in names + + missing = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("list", path="absent_dir") + ) + assert missing.success is False + assert "path not found" in (missing.error or "").lower() + + not_dir = await self._dispatch( + tmp_path, monkeypatch, policy, self._task("list", path="listing/visible.txt") + ) + assert not_dir.success is False + assert "not a directory" in (not_dir.error or "").lower() + + @pytest.mark.asyncio + async def test_public_parent_dispatch_malformed_payloads(self, tmp_path: Path) -> None: + """Standalone parent dispatch reports real malformed-operation errors.""" + agent = FileOpsAgent( + config={"working_dir": str(tmp_path), "allow_write": True, "max_file_bytes": 8} + ) + no_operation = await agent.run( + SubAgentTask(id="parent-no-op", agent_type=SubAgentType.FILE_OPS, payload={}) + ) + assert no_operation.success is False + assert "no operation specified" in (no_operation.error or "").lower() + + no_path = await agent.run( + SubAgentTask( + id="parent-no-path", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "read"}, + ) + ) + assert no_path.success is False + assert "no path specified" in (no_path.error or "").lower() + + bad_content = await agent.run( + SubAgentTask( + id="parent-bad-content", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "write", "path": "x.txt", "content": 123}, + ) + ) + assert bad_content.success is False + assert "must be strings" in (bad_content.error or "").lower() + + oversized = await agent.run( + SubAgentTask( + id="parent-oversized", + agent_type=SubAgentType.FILE_OPS, + payload={"operation": "write", "path": "x.txt", "content": "x" * 20}, + ) + ) + assert oversized.success is False + assert "byte limit" in (oversized.error or "").lower() + assert not (tmp_path / "x.txt").exists() diff --git a/tests/test_shell_prototype_empirical.py b/tests/test_shell_prototype_empirical.py index e94acc9..98d451e 100644 --- a/tests/test_shell_prototype_empirical.py +++ b/tests/test_shell_prototype_empirical.py @@ -649,7 +649,10 @@ async def test_p03_cancellation_and_grandchild_teardown(tmp_path: Path) -> None: grandchild_script = tmp_path / "grandchild_runner.py" grandchild_script.write_text( f"import os, time, pathlib\n" - f"pathlib.Path(r'{grandchild_pid_file}').write_text(str(os.getpid()))\n" + f"_pid_target = pathlib.Path(r'{grandchild_pid_file}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getpid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target)\n" f"time.sleep(60)\n", encoding="utf-8", ) @@ -658,8 +661,14 @@ async def test_p03_cancellation_and_grandchild_teardown(tmp_path: Path) -> None: child_script = tmp_path / "child_runner.py" child_script.write_text( f"import os, subprocess, sys, time, pathlib\n" - f"pathlib.Path(r'{child_pid_file}').write_text(str(os.getpid()))\n" - f"pathlib.Path(r'{helper_pid_file}').write_text(str(os.getppid()))\n" + f"_pid_target = pathlib.Path(r'{child_pid_file}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getpid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target)\n" + f"_pid_target = pathlib.Path(r'{helper_pid_file}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getppid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target)\n" f"proc = subprocess.Popen([sys.executable, str(pathlib.Path(r'{grandchild_script}'))])\n" f"proc.wait()\n", encoding="utf-8", @@ -772,7 +781,10 @@ async def test_p03_normal_command_exit_retaining_descendant(tmp_path: Path) -> N grandchild_script = tmp_path / "gc_retain.py" grandchild_script.write_text( f"import os, time, pathlib\n" - f"pathlib.Path(r'{grandchild_pid_file}').write_text(str(os.getpid()))\n" + f"_pid_target = pathlib.Path(r'{grandchild_pid_file}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getpid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target)\n" f"time.sleep(60)\n", encoding="utf-8", ) @@ -847,7 +859,10 @@ async def test_p03_abrupt_client_server_shutdown_with_execution_pending(tmp_path child_cmd = ( f'{sys.executable} -c "' f"import os, time, pathlib; " - f"pathlib.Path(r'{pid_file}').write_text(str(os.getpid())); " + f"_pid_target = pathlib.Path(r'{pid_file}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getpid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target); " f'time.sleep(60)"' ) @@ -985,7 +1000,10 @@ async def run_governed_lifecycle_case(mutate_teardown: bool) -> None: gc_script = sub_dir / "gc.py" gc_script.write_text( f"import os, time, pathlib\n" - f"pathlib.Path(r'{marker_gc}').write_text(str(os.getpid()))\n" + f"_pid_target = pathlib.Path(r'{marker_gc}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getpid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target)\n" f"time.sleep(45)\n" ) @@ -993,7 +1011,10 @@ async def run_governed_lifecycle_case(mutate_teardown: bool) -> None: child_script.write_text( f"import os, subprocess, sys, pathlib, json, time\n" f"p = subprocess.Popen([sys.executable, r'{gc_script}'])\n" - f"pathlib.Path(r'{marker_pids}').write_text(json.dumps([os.getppid(), os.getpid(), p.pid]))\n" + f"_pid_target = pathlib.Path(r'{marker_pids}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(json.dumps([os.getppid(), os.getpid(), p.pid]), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target)\n" f"time.sleep(45)\n" ) @@ -1120,7 +1141,10 @@ async def test_empirical_child_command_teardown_and_pids(tmp_path: Path) -> None child_cmd = ( f'{sys.executable} -c "' f"import os, time, pathlib; " - f"pathlib.Path(r'{pid_file}').write_text(str(os.getpid())); " + f"_pid_target = pathlib.Path(r'{pid_file}'); " + f"_pid_staging = _pid_target.with_suffix('.tmp'); " + f"_pid_staging.write_text(str(os.getpid()), encoding='utf-8'); " + f"os.replace(_pid_staging, _pid_target); " f'time.sleep(60)"' )