dongarandClaude Opus 5
53ed11f9c4
A frontier model receives a scoped sub-task, works autonomously with full tool access, and hands back; the local loop reads the handback note and the file state, never the reasoning. - Branch at delegation start, and abort discards it. That required committing each turn's work onto the branch: without commits, deleting a branch discards nothing, so "abort discards the work" was a comforting phrase rather than a fact. The caller's own in-flight changes are committed to their branch first, so an abort cannot destroy work nobody delegated - Hard caps on turns and wall clock; a breach is a defined failure return, and its result is not usable, so the caller gates rather than resuming on a half-edited tree - Every event is checkpointed as it arrives, so a crash mid-delegation leaves the branch at a known commit and a truthful transcript - Handback is validated locally: blank fields or a non-boolean success flag are refused and the delegation continues rather than accepting a useless note Two failure modes found by running it against a real model: - A success claim with nothing changed was accepted. The handback plus file state is the entire interface back, so a note describing work that does not exist is worse than no note. Success now requires a successful mutating tool call; reporting failure still does not, since "this is impossible" is a legitimate finding - The delegate invented the absolute path /workspace, wrote there, ran its tests there, and handed back truthfully — with the repo empty and git seeing nothing. Writes outside the workspace are now refused with a message naming the real root, and both prompts state where the workspace is instead of leaving the model to guess. This is not sandboxing: work outside the repo is invisible to git, so the milestone would publish an empty PR Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
159 lines
5.5 KiB
Python
159 lines
5.5 KiB
Python
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nisse_dev.tools import EXECUTOR_TOOLS, MAX_OUTPUT_CHARS, Workspace
|
|
|
|
|
|
@pytest.fixture
|
|
def ws(tmp_path: Path) -> Workspace:
|
|
return Workspace(tmp_path / "work", timeout_s=10)
|
|
|
|
|
|
def test_manifest_is_small_and_flat() -> None:
|
|
# A15: fewer tools, flatter arguments for the local executor.
|
|
names = [t["function"]["name"] for t in EXECUTOR_TOOLS]
|
|
assert names == ["read_file", "write_file", "list_dir", "run_command"]
|
|
for tool in EXECUTOR_TOOLS:
|
|
props = tool["function"]["parameters"]["properties"]
|
|
assert all(p["type"] == "string" for p in props.values())
|
|
|
|
|
|
def test_write_then_read_round_trip(ws: Workspace) -> None:
|
|
assert ws.write_file("src/app.py", "print('hi')\n").success
|
|
outcome = ws.read_file("src/app.py")
|
|
assert outcome.success
|
|
assert outcome.result == "print('hi')\n"
|
|
|
|
|
|
def test_relative_paths_land_in_the_workspace(ws: Workspace) -> None:
|
|
ws.write_file("notes.txt", "x")
|
|
assert (ws.root / "notes.txt").exists()
|
|
|
|
|
|
def test_read_missing_file_is_a_failed_result_not_an_exception(
|
|
ws: Workspace,
|
|
) -> None:
|
|
outcome = ws.read_file("nope.txt")
|
|
assert not outcome.success
|
|
assert "could not read" in outcome.result
|
|
|
|
|
|
def test_list_dir(ws: Workspace) -> None:
|
|
ws.write_file("a.txt", "a")
|
|
ws.write_file("sub/b.txt", "b")
|
|
outcome = ws.list_dir(".")
|
|
assert outcome.success
|
|
assert outcome.result.splitlines() == ["a.txt", "sub/"]
|
|
|
|
|
|
def test_list_empty_dir(ws: Workspace) -> None:
|
|
assert ws.list_dir(".").result == "(empty)"
|
|
|
|
|
|
def test_run_command_success_reports_exit_code_and_stdout(ws: Workspace) -> None:
|
|
outcome = ws.run_command("echo hello")
|
|
assert outcome.success
|
|
assert "exit code: 0" in outcome.result
|
|
assert "hello" in outcome.result
|
|
|
|
|
|
def test_nonzero_exit_is_a_result_not_an_error(ws: Workspace) -> None:
|
|
outcome = ws.run_command("exit 3")
|
|
assert not outcome.success # feeds the A12 failure count
|
|
assert "exit code: 3" in outcome.result
|
|
|
|
|
|
def test_command_runs_in_the_workspace(ws: Workspace) -> None:
|
|
ws.write_file("marker.txt", "here")
|
|
assert "marker.txt" in ws.run_command("ls").result
|
|
|
|
|
|
def test_command_timeout_is_reported(tmp_path: Path) -> None:
|
|
ws = Workspace(tmp_path / "work", timeout_s=1)
|
|
outcome = ws.run_command("sleep 30")
|
|
assert not outcome.success
|
|
assert "timed out" in outcome.result
|
|
|
|
|
|
def _is_running(pid: int) -> bool:
|
|
"""os.kill(pid, 0) is not enough: a killed orphan stays a zombie until
|
|
something reaps it, and no init does that inside a CI container. Read the
|
|
process state instead."""
|
|
try:
|
|
stat = Path(f"/proc/{pid}/stat").read_text()
|
|
except FileNotFoundError:
|
|
return False
|
|
# State is the first field after the parenthesised comm.
|
|
return stat.rsplit(")", 1)[1].split()[0] != "Z"
|
|
|
|
|
|
def test_timeout_kills_the_whole_process_group(tmp_path: Path) -> None:
|
|
ws = Workspace(tmp_path / "work", timeout_s=1)
|
|
pidfile = ws.root / "child.pid"
|
|
# A backgrounded grandchild must not outlive the timeout.
|
|
ws.run_command(f"(sleep 30 & echo $! > {pidfile}); sleep 30")
|
|
assert pidfile.exists()
|
|
pid = int(pidfile.read_text().strip())
|
|
assert _is_running(os.getpid()), "liveness check is broken, not the kill"
|
|
# SIGKILL to a process group is asynchronous, and the grandchild is an
|
|
# orphan by now, so its exit is not synchronised with our return. Wait
|
|
# for death rather than assuming it is instantaneous.
|
|
deadline = time.monotonic() + 5
|
|
while _is_running(pid) and time.monotonic() < deadline:
|
|
time.sleep(0.05)
|
|
assert not _is_running(pid)
|
|
|
|
|
|
def test_agent_credentials_are_not_inherited(
|
|
ws: Workspace, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("NISSE_LLM_TOKEN", "sk-secret")
|
|
monkeypatch.setenv("PATH", os.environ["PATH"])
|
|
outcome = ws.run_command("echo [${NISSE_LLM_TOKEN}]")
|
|
assert "sk-secret" not in outcome.result
|
|
assert "[]" in outcome.result
|
|
|
|
|
|
def test_stdin_is_closed_so_prompts_fail_fast(ws: Workspace) -> None:
|
|
outcome = ws.run_command("read line && echo got")
|
|
assert not outcome.success
|
|
assert "got" not in outcome.result
|
|
|
|
|
|
def test_long_output_is_truncated(ws: Workspace) -> None:
|
|
outcome = ws.run_command("printf 'x%.0s' $(seq 1 20000)")
|
|
assert len(outcome.result) < MAX_OUTPUT_CHARS + 200
|
|
assert "truncated" in outcome.result
|
|
|
|
|
|
def test_dispatch_unknown_tool_and_missing_argument(ws: Workspace) -> None:
|
|
assert not ws.dispatch("fly", {}).success
|
|
missing = ws.dispatch("read_file", {})
|
|
assert not missing.success
|
|
assert "missing argument" in missing.result
|
|
|
|
|
|
def test_write_outside_the_workspace_is_refused(ws: Workspace, tmp_path: Path) -> None:
|
|
# Not sandboxing: a file written outside the repo is invisible to git, so
|
|
# the work would vanish while the model truthfully reported success.
|
|
# Models really do invent absolute paths like /workspace.
|
|
stray = tmp_path / "elsewhere" / "slugify.py"
|
|
outcome = ws.write_file(str(stray), "x")
|
|
assert not outcome.success
|
|
assert "outside the workspace" in outcome.result
|
|
assert str(ws.root) in outcome.result # tells it where to write instead
|
|
assert not stray.exists()
|
|
|
|
|
|
def test_escaping_via_parent_traversal_is_refused(ws: Workspace) -> None:
|
|
outcome = ws.write_file("../escaped.txt", "x")
|
|
assert not outcome.success
|
|
assert not (ws.root.parent / "escaped.txt").exists()
|
|
|
|
|
|
def test_describe_names_the_root(ws: Workspace) -> None:
|
|
assert str(ws.root) in ws.describe()
|