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>
106 lines
3.8 KiB
Python
106 lines
3.8 KiB
Python
import json
|
|
|
|
from nisse_dev.canonical import (
|
|
AssistantText,
|
|
Compaction,
|
|
GateOpened,
|
|
StoredEvent,
|
|
ToolCall,
|
|
ToolResult,
|
|
UserMessage,
|
|
)
|
|
from nisse_dev.render import OpenAIChatRenderer, effective_events
|
|
|
|
|
|
def stored(events: list) -> list[StoredEvent]:
|
|
return [StoredEvent(id=i + 1, ts="t", event=e) for i, e in enumerate(events)]
|
|
|
|
|
|
def test_renders_mixed_transcript_in_order() -> None:
|
|
events = stored(
|
|
[
|
|
UserMessage(text="run the tests", source="m:1"),
|
|
ToolCall(call_id="call-9", tool="run_tests", arguments={"path": "x"}),
|
|
ToolResult(call_id="call-9", result="1 passed", success=True),
|
|
AssistantText(text="all green", model="local/qwen"),
|
|
]
|
|
)
|
|
body = OpenAIChatRenderer().render(events, tools=None, model="local/qwen")
|
|
assert body["model"] == "local/qwen"
|
|
roles = [m["role"] for m in body["messages"]]
|
|
assert roles == ["user", "assistant", "tool", "assistant"]
|
|
call = body["messages"][1]["tool_calls"][0]
|
|
assert call["id"] == "call-9"
|
|
assert call["function"]["name"] == "run_tests"
|
|
assert json.loads(call["function"]["arguments"]) == {"path": "x"}
|
|
assert body["messages"][2]["tool_call_id"] == "call-9"
|
|
assert "tools" not in body
|
|
|
|
|
|
def test_system_prompt_and_tools_pass_through() -> None:
|
|
tools = [{"type": "function", "function": {"name": "run_tests"}}]
|
|
body = OpenAIChatRenderer().render(
|
|
stored([UserMessage(text="hi", source="m:1")]),
|
|
tools=tools,
|
|
model="local/qwen",
|
|
system="you are nisse-dev",
|
|
)
|
|
assert body["messages"][0] == {"role": "system", "content": "you are nisse-dev"}
|
|
assert body["tools"] == tools
|
|
|
|
|
|
def test_compaction_replaces_its_range_only() -> None:
|
|
events = stored(
|
|
[
|
|
UserMessage(text="old-1", source="m:1"),
|
|
AssistantText(text="old-2", model="q"),
|
|
UserMessage(text="kept", source="m:3"),
|
|
]
|
|
)
|
|
events.append(
|
|
StoredEvent(
|
|
id=4,
|
|
ts="t",
|
|
event=Compaction(summary="the old stuff", model="q", replaces=[1, 2]),
|
|
)
|
|
)
|
|
remaining = effective_events(events)
|
|
assert [e.id for e in remaining] == [3, 4]
|
|
body = OpenAIChatRenderer().render(events, tools=None, model="q")
|
|
contents = [m["content"] for m in body["messages"]]
|
|
assert contents == ["kept", "[summary of earlier conversation]\nthe old stuff"]
|
|
|
|
|
|
def test_tool_call_without_a_result_gets_a_placeholder() -> None:
|
|
# Providers reject a dangling tool_use; this happens after a crash (A3)
|
|
# or with turn-ending gate tools like submit_plan.
|
|
events = stored(
|
|
[
|
|
UserMessage(text="go", source="m:1"),
|
|
ToolCall(call_id="call-x", tool="submit_plan", arguments={}),
|
|
]
|
|
)
|
|
body = OpenAIChatRenderer().render(events, tools=None, model="q")
|
|
assert [m["role"] for m in body["messages"]] == ["user", "assistant", "tool"]
|
|
assert body["messages"][2]["tool_call_id"] == "call-x"
|
|
assert body["messages"][2]["content"] == "[no result recorded]"
|
|
|
|
|
|
def test_answered_tool_call_gets_no_placeholder() -> None:
|
|
events = stored(
|
|
[
|
|
ToolCall(call_id="call-x", tool="run_command", arguments={}),
|
|
ToolResult(call_id="call-x", result="ok", success=True),
|
|
]
|
|
)
|
|
body = OpenAIChatRenderer().render(events, tools=None, model="q")
|
|
assert [m["role"] for m in body["messages"]] == ["assistant", "tool"]
|
|
assert body["messages"][1]["content"] == "ok"
|
|
|
|
|
|
def test_gate_event_renders_as_context() -> None:
|
|
events = stored([GateOpened(gate="arch", reason="plan submitted (rev-1)")])
|
|
body = OpenAIChatRenderer().render(events, tools=None, model="q")
|
|
assert body["messages"][0]["role"] == "user"
|
|
assert "arch" in body["messages"][0]["content"]
|