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>
305 lines
9.6 KiB
Python
305 lines
9.6 KiB
Python
import io
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from nisse_dev.canonical import (
|
|
AssistantText,
|
|
CanonicalEvent,
|
|
DelegationCompleted,
|
|
DelegationStarted,
|
|
StoredEvent,
|
|
ToolCall,
|
|
)
|
|
from nisse_dev.delegation import (
|
|
DELEGATE_TOOLS,
|
|
Caps,
|
|
Delegation,
|
|
)
|
|
from nisse_dev.git import GitRepo
|
|
from nisse_dev.llm import LLMError
|
|
from nisse_dev.oplog import OpLog
|
|
from nisse_dev.state import StateStore
|
|
from nisse_dev.tools import Workspace
|
|
|
|
|
|
class ScriptedClient:
|
|
def __init__(self, turns: list[list[CanonicalEvent]]) -> None:
|
|
self._turns = turns
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
def complete(
|
|
self,
|
|
events: list[StoredEvent],
|
|
tools: list[dict[str, Any]] | None,
|
|
model: str,
|
|
system: str | None = None,
|
|
) -> list[CanonicalEvent]:
|
|
self.calls.append({"tools": tools, "model": model, "system": system})
|
|
if not self._turns:
|
|
return [AssistantText(text="still thinking", model=model)]
|
|
return self._turns.pop(0)
|
|
|
|
|
|
class FailingClient:
|
|
def complete(self, *args: Any, **kwargs: Any) -> list[CanonicalEvent]:
|
|
raise LLMError("frontier unavailable")
|
|
|
|
|
|
def call(tool: str, cid: str = "c1", **arguments: Any) -> ToolCall:
|
|
return ToolCall(call_id=cid, tool=tool, arguments=arguments)
|
|
|
|
|
|
def hand_back(succeeded: bool = True, cid: str = "hb", **overrides: Any) -> ToolCall:
|
|
args = {
|
|
"changed": "added greet.py",
|
|
"now_true": "greet() returns a greeting and its test passes",
|
|
"next": "open the PR",
|
|
"succeeded": succeeded,
|
|
}
|
|
args.update(overrides)
|
|
return ToolCall(call_id=cid, tool="hand_back", arguments=args)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Path) -> StateStore:
|
|
return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(tmp_path: Path) -> GitRepo:
|
|
r = GitRepo(tmp_path / "work")
|
|
r.init()
|
|
r.configure_identity("bot", "bot@example.test")
|
|
(r.root / "seed.txt").write_text("seed")
|
|
r.commit_all("seed")
|
|
return r
|
|
|
|
|
|
@pytest.fixture
|
|
def ws(repo: GitRepo) -> Workspace:
|
|
return Workspace(repo.root, timeout_s=10)
|
|
|
|
|
|
def build(client: Any, store: StateStore, ws: Workspace, repo: GitRepo, **kw: Any):
|
|
return Delegation(client, store, ws, repo, model="claude-haiku-4-5", **kw)
|
|
|
|
|
|
def test_successful_delegation_returns_a_structured_handback(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient(
|
|
[
|
|
[call("write_file", path="greet.py", content="def greet(): ...")],
|
|
[hand_back()],
|
|
]
|
|
)
|
|
result = build(client, store, ws, repo).run(brief="add greet", branch="d1")
|
|
|
|
assert result.usable
|
|
assert result.handback is not None
|
|
assert "greet.py" in result.handback.changed
|
|
assert "what to do next" in result.handback.render().lower()
|
|
assert result.artifacts == ["greet.py"]
|
|
|
|
|
|
def test_delegation_works_on_its_own_branch(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient(
|
|
[[call("write_file", path="a.py", content="x")], [hand_back()]]
|
|
)
|
|
build(client, store, ws, repo).run(brief="b", branch="deleg-1")
|
|
assert repo.current_branch() == "deleg-1"
|
|
|
|
|
|
def test_abort_discards_the_branch_and_its_work(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
# A13: abort must leave nothing to resume on top of.
|
|
client = ScriptedClient(
|
|
[
|
|
[call("write_file", path="half-done.py", content="oops")],
|
|
[hand_back(succeeded=False, changed="could not finish")],
|
|
]
|
|
)
|
|
result = build(client, store, ws, repo).run(brief="b", branch="deleg-2")
|
|
|
|
assert not result.usable
|
|
assert repo.current_branch() == "main"
|
|
assert not (repo.root / "half-done.py").exists()
|
|
|
|
|
|
def test_turn_cap_is_hard(store: StateStore, ws: Workspace, repo: GitRepo) -> None:
|
|
forever = [[call("list_dir", path=".")] for _ in range(20)]
|
|
client = ScriptedClient(forever)
|
|
result = build(client, store, ws, repo, caps=Caps(turns=3)).run(
|
|
brief="b", branch="d"
|
|
)
|
|
|
|
assert result.status == "capped"
|
|
assert "turn cap of 3" in result.note
|
|
assert len(client.calls) == 3
|
|
|
|
|
|
def test_wall_clock_cap_is_hard(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
ticks = iter([0.0, 1.0, 99.0, 99.0, 99.0])
|
|
client = ScriptedClient([[call("list_dir", path=".")] for _ in range(10)])
|
|
result = build(
|
|
client,
|
|
store,
|
|
ws,
|
|
repo,
|
|
caps=Caps(turns=10, wall_clock_s=60.0),
|
|
now=lambda: next(ticks),
|
|
).run(brief="b", branch="d")
|
|
|
|
assert result.status == "capped"
|
|
assert "wall-clock cap" in result.note
|
|
|
|
|
|
def test_cap_breach_is_not_usable_and_discards_work(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient([[call("write_file", path="partial.py", content="x")]] * 5)
|
|
result = build(client, store, ws, repo, caps=Caps(turns=2)).run(
|
|
brief="b", branch="d"
|
|
)
|
|
|
|
assert not result.usable # the caller must gate, not resume (A13)
|
|
assert not (repo.root / "partial.py").exists()
|
|
|
|
|
|
def test_events_are_checkpointed_as_they_arrive(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient(
|
|
[[call("write_file", path="a.py", content="x")], [hand_back()]]
|
|
)
|
|
build(client, store, ws, repo).run(brief="add a", branch="d")
|
|
|
|
kinds = [type(e.event).__name__ for e in store.read_events()]
|
|
assert kinds == [
|
|
"DelegationStarted",
|
|
"ToolCall",
|
|
"ToolResult",
|
|
"ToolCall", # hand_back
|
|
"ToolResult",
|
|
"DelegationCompleted",
|
|
]
|
|
|
|
|
|
def test_composite_events_record_the_brief_caps_and_outcome(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient(
|
|
[[call("write_file", path="a.py", content="x")], [hand_back()]]
|
|
)
|
|
caps = Caps(turns=7, tokens=1234, wall_clock_s=42.0)
|
|
build(client, store, ws, repo, caps=caps).run(brief="the brief", branch="d")
|
|
|
|
events = [e.event for e in store.read_events()]
|
|
started = next(e for e in events if isinstance(e, DelegationStarted))
|
|
completed = next(e for e in events if isinstance(e, DelegationCompleted))
|
|
assert started.brief == "the brief"
|
|
assert started.model == "claude-haiku-4-5"
|
|
assert started.caps == {"turns": 7, "tokens": 1234, "wall_clock_s": 42.0}
|
|
# A16: this pair is the eval record, so it must carry the outcome.
|
|
assert completed.exit_status == "handback"
|
|
assert "greet()" in completed.handback
|
|
|
|
|
|
def test_hand_back_tool_is_offered_alongside_the_executor_tools() -> None:
|
|
names = [t["function"]["name"] for t in DELEGATE_TOOLS]
|
|
assert names[-1] == "hand_back"
|
|
assert "run_command" in names # full tool access, not a consult (A13)
|
|
|
|
|
|
def test_malformed_handback_is_rejected_and_the_delegation_continues(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient(
|
|
[
|
|
[hand_back(cid="bad", changed=" ")], # blank field
|
|
[call("write_file", path="a.py", content="x")],
|
|
[hand_back(cid="good")],
|
|
]
|
|
)
|
|
result = build(client, store, ws, repo).run(brief="b", branch="d")
|
|
|
|
assert result.usable # recovered rather than accepting a useless note
|
|
assert len(client.calls) == 3
|
|
|
|
|
|
def test_non_boolean_success_flag_is_rejected(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
client = ScriptedClient(
|
|
[
|
|
[call("write_file", path="a.py", content="x")],
|
|
[hand_back(succeeded="yes")], # not a bool: unparseable
|
|
[hand_back(cid="ok")],
|
|
]
|
|
)
|
|
result = build(client, store, ws, repo).run(brief="b", branch="d")
|
|
assert result.usable
|
|
assert len(client.calls) == 3
|
|
|
|
|
|
def test_client_failure_is_a_defined_return_not_a_crash(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
result = build(FailingClient(), store, ws, repo).run(brief="b", branch="d")
|
|
assert result.status == "failed"
|
|
assert "frontier unavailable" in result.note
|
|
assert not result.usable
|
|
|
|
|
|
def test_a_delegate_that_never_hands_back_is_capped_not_believed(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
# Prose forever: the delegate never reports, so its tree state is unknown.
|
|
client = ScriptedClient([])
|
|
result = build(client, store, ws, repo, caps=Caps(turns=3)).run(
|
|
brief="b", branch="d"
|
|
)
|
|
assert result.status == "capped"
|
|
assert not result.usable
|
|
|
|
|
|
def test_success_claimed_without_any_change_is_refused(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
# The handback plus file state is the whole interface back, so a note
|
|
# describing work that was never done is worse than no note.
|
|
client = ScriptedClient(
|
|
[
|
|
[hand_back(cid="early")],
|
|
[call("write_file", path="real.py", content="x")],
|
|
[hand_back(cid="earned")],
|
|
]
|
|
)
|
|
result = build(client, store, ws, repo).run(brief="b", branch="d")
|
|
|
|
assert result.usable
|
|
assert len(client.calls) == 3 # the first claim was sent back
|
|
assert (repo.root / "real.py").exists()
|
|
|
|
|
|
def test_failure_may_be_reported_without_doing_work(
|
|
store: StateStore, ws: Workspace, repo: GitRepo
|
|
) -> None:
|
|
# "This sub-task is impossible" is a legitimate finding that requires no
|
|
# changes; only a claim of success needs evidence.
|
|
client = ScriptedClient(
|
|
[[hand_back(succeeded=False, changed="nothing: the API does not exist")]]
|
|
)
|
|
result = build(client, store, ws, repo).run(brief="b", branch="d")
|
|
|
|
assert result.status == "failed"
|
|
assert not result.usable
|
|
assert len(client.calls) == 1 # believed immediately, not challenged
|