dongarandClaude Opus 5
0fd0b2e4a5
- tools.py: four flat tools (read_file, write_file, list_dir, run_command) per A15's reduced manifest. Deliberately unsandboxed: the VM is the trust boundary (B3) and tool use is auto-approved inside an approved unit of work (A4), so path confinement would add failure modes without adding safety. What is enforced is operational — timeout with process-group kill so a backgrounded server cannot outlive its command, output truncation so a build log cannot eat the context window, closed stdin so prompts fail fast, and a scrubbed environment so project code never sees the agent's credentials - executor.py: renders the log, calls the executor role with the reduced manifest, dispatches tool calls, and checkpoints every result (A13). Escalation is observable, never self-assessed (A12): consecutive command failures and a hard turn cap. Failed reads do not count — looking around is not failing - Until delegation exists (step 7), escalation gates to the human with the failures attached - control.py: start milestone now runs the executor; a stopped milestone gates to AWAIT_INPUT Two bugs found by running it live against real models: - A turn-ending gate tool (submit_plan) left a tool_call with no tool_result, which providers reject outright, so every planning turn after the first failed with a 400. The planner now records the gate as the call's outcome, which is also the truthful transcript - The renderer now emits a placeholder for any tool_call lacking a result, so a crash between call and result (A3) cannot poison the transcript permanently Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
import os
|
|
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 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())
|
|
with pytest.raises(ProcessLookupError):
|
|
os.kill(pid, 0)
|
|
|
|
|
|
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
|