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>
78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
"""End-to-end milestone against real models: plan with the planner role,
|
|
approve through the gate, then let the executor role actually create and test
|
|
a file on disk.
|
|
|
|
Gated on NISSE_LIVE_PLANNER_MODEL like the planning test — it needs a model
|
|
that holds the tool-call contract on both ends.
|
|
"""
|
|
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nisse_dev.control import EXECUTING, Orchestrator
|
|
from nisse_dev.envfile import load_env_file
|
|
from nisse_dev.executor import LLMExecutor
|
|
from nisse_dev.llm import LiteLLMClient, RoleMap
|
|
from nisse_dev.oplog import OpLog
|
|
from nisse_dev.planner import LLMPlanningAgent
|
|
from nisse_dev.state import StateStore
|
|
from nisse_dev.tools import Workspace
|
|
from nisse_dev.transport import InMemoryTransport
|
|
|
|
load_env_file(Path(__file__).resolve().parents[2] / ".env")
|
|
|
|
SENTINEL = "NISSE_LIVE_PLANNER_MODEL"
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.skipif(
|
|
not os.environ.get(SENTINEL),
|
|
reason=f"no live planner route configured ({SENTINEL} unset)",
|
|
),
|
|
]
|
|
|
|
|
|
def test_plan_approve_execute_produces_a_real_file(tmp_path: Path) -> None:
|
|
assert os.environ.get("NISSE_LLM_TOKEN"), (
|
|
f"{SENTINEL} is set, so the live path is expected to work, "
|
|
"but NISSE_LLM_TOKEN is missing"
|
|
)
|
|
# Both roles use the capable model here: the point is the whole gated
|
|
# loop over a real proxy, not local-model competence (measured separately).
|
|
model = os.environ[SENTINEL]
|
|
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
|
|
client = LiteLLMClient.from_env()
|
|
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
|
|
workspace = Workspace(tmp_path / "work", timeout_s=60)
|
|
transport = InMemoryTransport()
|
|
orchestrator = Orchestrator(
|
|
store,
|
|
transport,
|
|
LLMPlanningAgent(client, store, roles),
|
|
operators={"don"},
|
|
roles=roles,
|
|
executor=LLMExecutor(client, store, roles, workspace, max_turns=10),
|
|
)
|
|
orchestrator.start()
|
|
|
|
transport.receive(
|
|
"don",
|
|
"use greet. Milestone 1: create greet.py in the workspace root defining "
|
|
"greet(name) returning 'Hello, <name>!', and verify it by running python.",
|
|
)
|
|
transport.receive("don", "approve") # architecture gate
|
|
transport.receive("don", "approve") # milestone gate
|
|
transport.receive("don", "start milestone 1")
|
|
|
|
state = store.get_state()
|
|
assert state is not None
|
|
assert state.state in (EXECUTING, "AWAIT_INPUT")
|
|
|
|
created = list(workspace.root.rglob("*.py"))
|
|
assert created, f"executor created no Python file; said: {transport.outbound[-1]}"
|
|
combined = "\n".join(p.read_text() for p in created)
|
|
assert "def greet" in combined
|