diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index 1382ee5..f950a28 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/nisse-dev/src/nisse_dev/control.py @@ -59,6 +59,14 @@ class AskHuman: question: str +class MilestoneExecutor(Protocol): + """What the orchestrator needs from an executor loop; the concrete one + lives in nisse_dev.executor (imported there, not here, to keep the + dependency pointing one way).""" + + def run(self, milestone: int) -> Any: ... + + class PlanningAgent(Protocol): def plan( self, @@ -91,7 +99,9 @@ def _fresh_data() -> dict[str, Any]: return { "revs": {}, "gate": None, - "question": None, + # What an operator reply in AWAIT_INPUT should resume. Every path into + # AWAIT_INPUT must set this, or the reply has nowhere to go. + "pending": None, "milestone": 1, "task_id": None, "roles": None, # A11 mapping, pinned at task start @@ -106,11 +116,13 @@ class Orchestrator: agent: PlanningAgent, operators: Iterable[str], roles: RoleMap | None = None, + executor: "MilestoneExecutor | None" = None, ) -> None: self._store = store self._transport = transport self._agent = agent self._roles = roles + self._executor = executor self._operators = set(operators) self._data = _fresh_data() self._project: str | None = None @@ -128,11 +140,14 @@ class Orchestrator: self._store.oplog.task_id = self._data.get("task_id") gate = self._data.get("gate") if report.action == "await_input": - self._transition(AWAIT_INPUT, cause="boot: unresolved intents") - self._transport.ask_question( - "restart found unresolved work: " - + "; ".join(report.notes) - + " — how should I proceed?" + self._await_input( + cause="boot: unresolved intents", + prompt=( + "restart found unresolved work: " + + "; ".join(report.notes) + + " — how should I proceed?" + ), + pending={"kind": "boot"}, ) elif state.state == IDLE or ( state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF) and gate @@ -147,9 +162,10 @@ class Orchestrator: else: # Interrupted mid-turn (planning, executing, unanswered question): # never blind-resume (A3.6). - self._transition(AWAIT_INPUT, cause=f"boot: restarted in {state.state}") - self._transport.ask_question( - f"restarted mid-{state.state}; continue or re-plan?" + self._await_input( + cause=f"boot: restarted in {state.state}", + prompt=f"restarted mid-{state.state}; continue or re-plan?", + pending={"kind": "boot"}, ) def _handle(self, msg: InboundMessage) -> None: @@ -207,17 +223,18 @@ class Orchestrator: gate_name, self._project, feedback=feedback, answer=answer ) except PlanningError as exc: - self._transition(AWAIT_INPUT, cause=f"planning failed: {exc}") - self._transport.ask_question( - f"planning for {gate_name} failed ({exc}); retry or re-plan?" + self._await_input( + cause=f"planning failed: {exc}", + prompt=f"planning for {gate_name} failed ({exc}); how should I proceed?", + pending={"kind": "planning_failed", "gate": gate_name}, ) return if isinstance(action, AskHuman): - self._data["question"] = {"text": action.question, "gate": gate_name} - self._transition( - AWAIT_INPUT, cause=f"agent question during {gate_name} planning" + self._await_input( + cause=f"agent question during {gate_name} planning", + prompt=action.question, + pending={"kind": "question", "gate": gate_name}, ) - self._transport.ask_question(action.question) return rev = self._data["revs"].get(gate_name, 0) + 1 self._data["revs"][gate_name] = rev @@ -259,7 +276,7 @@ class Orchestrator: self._transition( EXECUTING, cause=f"operator command: start milestone {milestone}" ) - self._transport.send_message(f"[stub] executing milestone {milestone}") + self._execute_milestone(milestone) else: self._transport.send_message( f"plan approved; say 'start milestone {milestone}' to begin" @@ -313,18 +330,59 @@ class Orchestrator: f"milestone plan approved; say 'start milestone {milestone}' to begin" ) + def _execute_milestone(self, milestone: int) -> None: + if self._executor is None: + self._transport.send_message(f"[stub] executing milestone {milestone}") + return + self._transport.send_message(f"executing milestone {milestone}") + outcome = self._executor.run(milestone) + if outcome.status == "completed": + # PR_OPEN and the review loop arrive with the Gitea layer (B2); + # until then a finished milestone simply reports and waits. + self._transport.send_message(f"milestone {milestone} done: {outcome.note}") + return + self._await_input( + cause=f"milestone {milestone} {outcome.status}", + prompt=f"milestone {milestone} stopped ({outcome.status}): {outcome.note}", + pending={"kind": "milestone_stopped", "milestone": milestone}, + ) + + def _await_input(self, *, cause: str, prompt: str, pending: dict[str, Any]) -> None: + self._data["pending"] = pending + self._transition(AWAIT_INPUT, cause=cause) + self._transport.ask_question(prompt) + def _handle_answer(self, text: str) -> None: - question = self._data.get("question") - if question is None: + pending = self._data.get("pending") + if pending is None: + # Reached only if state was forced externally; say what works + # rather than dead-ending the operator. self._transport.send_message( - "awaiting operator resolution; no pending question to answer" + "waiting for direction, but I have no pending item to resume; " + "try 'status', or 'use ' to start over" ) return - self._data["question"] = None - gate_name = question["gate"] - back = PLANNING_ARCH if gate_name == "arch" else MILESTONE_PLANNING - self._transition(back, cause="operator answered question") - self._run_planning_turn(gate_name, answer=text) + self._data["pending"] = None + match pending.get("kind"): + case "question" | "planning_failed": + gate_name = pending["gate"] + back = PLANNING_ARCH if gate_name == "arch" else MILESTONE_PLANNING + self._transition(back, cause="operator answered; resuming planning") + self._run_planning_turn(gate_name, answer=text) + case "milestone_stopped": + milestone = pending["milestone"] + self._transition( + AWAIT_MILESTONE_SIGNOFF, + cause="operator answered; plan still approved", + ) + self._transport.send_message( + f"noted. the milestone {milestone} plan is still approved — " + f"say 'start milestone {milestone}' to run it again" + ) + case _: + self._transport.send_message( + "noted. say 'status' for where things stand" + ) def _transition(self, new_state: str, *, cause: str) -> None: state = self._store.get_state() diff --git a/nisse-dev/src/nisse_dev/executor.py b/nisse-dev/src/nisse_dev/executor.py new file mode 100644 index 0000000..28c78ab --- /dev/null +++ b/nisse-dev/src/nisse_dev/executor.py @@ -0,0 +1,107 @@ +"""Local executor loop (spec A12 escalation triggers, A15 accommodations). + +Runs the approved plan with the executor role — the cheap local model doing +mechanical work. Every tool call and result is checkpointed to the canonical +log as it happens (A13), so a crash mid-milestone leaves a resumable record. + +Escalation is decided from outside the model (A12): consecutive command +failures and hard caps on turns, never the model's own claim that it is stuck +or finished-enough. Until delegation exists (step 7), an escalation gates to +the human with the failures attached rather than engaging a frontier model. +""" + +from dataclasses import dataclass +from typing import Literal + +from nisse_dev.canonical import ToolCall, ToolResult +from nisse_dev.llm import LLMClient, LLMError, RoleMap +from nisse_dev.state import StateStore +from nisse_dev.tools import EXECUTOR_TOOLS, Workspace + +EXECUTOR_SYSTEM = """You are nisse-dev's executor, doing mechanical work. + +Follow the approved plan above exactly. It was written to be executed +verbatim; do not redesign it, expand its scope, or skip steps. + +Work by calling tools. Verify your work by running the project's tests. +When every step of the plan is done and the tests pass, reply with a short +plain-text summary and no tool call — that ends the milestone.""" + +DEFAULT_MAX_TURNS = 24 +DEFAULT_FAILURE_LIMIT = 2 + + +@dataclass(frozen=True) +class ExecutorOutcome: + status: Literal["completed", "capped", "escalate"] + note: str + + +class LLMExecutor: + def __init__( + self, + client: LLMClient, + store: StateStore, + roles: RoleMap, + workspace: Workspace, + *, + max_turns: int = DEFAULT_MAX_TURNS, + failure_limit: int = DEFAULT_FAILURE_LIMIT, + ) -> None: + self._client = client + self._store = store + self._roles = roles + self._workspace = workspace + self._max_turns = max_turns + self._failure_limit = failure_limit + + def run(self, milestone: int) -> ExecutorOutcome: + model = self._roles.resolve("executor") + system = f"{EXECUTOR_SYSTEM}\n\nYou are executing milestone {milestone}." + consecutive_failures = 0 + + for _turn in range(self._max_turns): + try: + produced = self._client.complete( + self._store.read_events(), + tools=EXECUTOR_TOOLS, + model=model, + system=system, + ) + except LLMError as exc: + return ExecutorOutcome("escalate", f"executor call failed: {exc}") + + calls: list[ToolCall] = [] + for event in produced: + self._store.append_event(event) + if isinstance(event, ToolCall): + calls.append(event) + + if not calls: + return ExecutorOutcome("completed", "executor reported the plan done") + + for call in calls: + outcome = self._workspace.dispatch(call.tool, call.arguments) + self._store.append_event( + ToolResult( + call_id=call.call_id, + result=outcome.result, + success=outcome.success, + ) + ) + # Only commands count toward escalation: a failed read is the + # model looking around, a failed command is work not working. + if call.tool == "run_command": + consecutive_failures = ( + 0 if outcome.success else consecutive_failures + 1 + ) + if consecutive_failures >= self._failure_limit: + return ExecutorOutcome( + "escalate", + f"{consecutive_failures} consecutive command failures; " + f"last: {outcome.result[:500]}", + ) + + return ExecutorOutcome( + "capped", f"hit the {self._max_turns}-turn cap for milestone {milestone}" + ) diff --git a/nisse-dev/src/nisse_dev/planner.py b/nisse-dev/src/nisse_dev/planner.py index 653d68f..be599b0 100644 --- a/nisse-dev/src/nisse_dev/planner.py +++ b/nisse-dev/src/nisse_dev/planner.py @@ -11,7 +11,7 @@ crash mid-planning leaves a truthful transcript. from typing import Any -from nisse_dev.canonical import AssistantText, ToolCall +from nisse_dev.canonical import AssistantText, ToolCall, ToolResult from nisse_dev.control import AskHuman, PlanningError, SubmitPlan from nisse_dev.llm import LLMClient, LLMError, RoleMap from nisse_dev.state import StateStore @@ -142,6 +142,21 @@ class LLMPlanningAgent: continue if isinstance(event, ToolCall): action = self._interpret(event) + if action is not None: + # The gate is this call's outcome. Recording it keeps the + # transcript renderable: providers reject a tool call with + # no result, and the next planning turn replays this log. + self._store.append_event( + ToolResult( + call_id=event.call_id, + result=( + "plan posted for approval; turn ended" + if isinstance(action, SubmitPlan) + else "question posted; awaiting the operator" + ), + success=True, + ) + ) return action def _interpret(self, call: ToolCall) -> SubmitPlan | AskHuman | None: diff --git a/nisse-dev/src/nisse_dev/render.py b/nisse-dev/src/nisse_dev/render.py index 21db51b..63d4b2f 100644 --- a/nisse-dev/src/nisse_dev/render.py +++ b/nisse-dev/src/nisse_dev/render.py @@ -57,8 +57,24 @@ class OpenAIChatRenderer: messages: list[dict[str, Any]] = [] if system is not None: messages.append({"role": "system", "content": system}) - for stored in effective_events(events): + kept = effective_events(events) + answered = {e.event.call_id for e in kept if isinstance(e.event, ToolResult)} + for stored in kept: messages.append(self._message_for(stored.event)) + # Providers reject a tool call with no result following it. One + # can legitimately be missing: a crash between the call and its + # result (A3), or a turn-ending gate tool. Render a placeholder so + # the transcript stays replayable rather than poisoned. + if isinstance(stored.event, ToolCall) and ( + stored.event.call_id not in answered + ): + messages.append( + { + "role": "tool", + "tool_call_id": stored.event.call_id, + "content": "[no result recorded]", + } + ) body: dict[str, Any] = {"model": model, "messages": messages} if tools: body["tools"] = tools diff --git a/nisse-dev/src/nisse_dev/tools.py b/nisse-dev/src/nisse_dev/tools.py new file mode 100644 index 0000000..b873a9c --- /dev/null +++ b/nisse-dev/src/nisse_dev/tools.py @@ -0,0 +1,204 @@ +"""Executor tools (spec A15: reduced schema, flat arguments). + +Deliberately unsandboxed. The VM is the trust boundary (B3) and tool use is +auto-approved inside an approved unit of work (A4), so per-command +confinement would add failure modes without adding safety. The workspace is +where work happens — the default for relative paths and the command cwd — not +a wall. Protecting agent state from the agent is a mount-layout concern +(A2/B3), not this module's job. + +What *is* enforced here is operational: commands cannot hang the loop, cannot +consume the context window with megabytes of build log, and do not inherit +the agent's own credentials. +""" + +import os +import shutil +import signal +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +DEFAULT_TIMEOUT_S = 120 +MAX_OUTPUT_CHARS = 4000 + +EXECUTOR_TOOLS: list[dict[str, Any]] = [ + { + "type": "function", + "function": { + "name": "read_file", + "description": "Read a UTF-8 text file.", + "parameters": { + "type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "write_file", + "description": "Write a UTF-8 text file, creating parent directories.", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "content": {"type": "string"}, + }, + "required": ["path", "content"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "list_dir", + "description": "List the entries of a directory.", + "parameters": { + "type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "run_command", + "description": ( + "Run a shell command in the workspace. Returns its exit code " + "and output; a non-zero exit is a result, not an error." + ), + "parameters": { + "type": "object", + "properties": {"command": {"type": "string"}}, + "required": ["command"], + }, + }, + }, +] + + +def _truncate(text: str) -> str: + if len(text) <= MAX_OUTPUT_CHARS: + return text + dropped = len(text) - MAX_OUTPUT_CHARS + return f"{text[:MAX_OUTPUT_CHARS]}\n[... {dropped} characters truncated]" + + +def _child_env() -> dict[str, str]: + """The project's build and test code is arbitrary; it has no business + reading the agent's LiteLLM, Matrix, or Gitea credentials.""" + return {k: v for k, v in os.environ.items() if not k.startswith("NISSE_")} + + +@dataclass(frozen=True) +class ToolOutcome: + result: str + success: bool + + +class Workspace: + def __init__(self, root: Path, *, timeout_s: int = DEFAULT_TIMEOUT_S) -> None: + self.root = Path(root) + self.root.mkdir(parents=True, exist_ok=True) + self._timeout_s = timeout_s + + def _resolve(self, path: str) -> Path: + candidate = Path(path) + return candidate if candidate.is_absolute() else self.root / candidate + + def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome: + try: + match tool: + case "read_file": + return self.read_file(str(arguments["path"])) + case "write_file": + return self.write_file( + str(arguments["path"]), str(arguments["content"]) + ) + case "list_dir": + return self.list_dir(str(arguments.get("path", "."))) + case "run_command": + return self.run_command(str(arguments["command"])) + case _: + return ToolOutcome(f"no such tool: {tool}", success=False) + except KeyError as exc: + return ToolOutcome(f"missing argument {exc} for {tool}", success=False) + + def read_file(self, path: str) -> ToolOutcome: + target = self._resolve(path) + try: + return ToolOutcome(_truncate(target.read_text()), success=True) + except (OSError, UnicodeDecodeError) as exc: + return ToolOutcome(f"could not read {path}: {exc}", success=False) + + def write_file(self, path: str, content: str) -> ToolOutcome: + target = self._resolve(path) + try: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content) + except OSError as exc: + return ToolOutcome(f"could not write {path}: {exc}", success=False) + return ToolOutcome(f"wrote {len(content)} characters to {path}", success=True) + + def list_dir(self, path: str) -> ToolOutcome: + target = self._resolve(path) + try: + entries = sorted( + f"{p.name}/" if p.is_dir() else p.name for p in target.iterdir() + ) + except OSError as exc: + return ToolOutcome(f"could not list {path}: {exc}", success=False) + return ToolOutcome(_truncate("\n".join(entries) or "(empty)"), success=True) + + def run_command(self, command: str) -> ToolOutcome: + try: + # Popen rather than run(): on timeout, run() kills only the direct + # child, and TimeoutExpired carries no pid, so a backgrounded + # grandchild (a dev server, say) would survive. Holding the pid + # lets us kill the whole session. + process = subprocess.Popen( + command, + shell=True, + cwd=self.root, + env=_child_env(), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + stdin=subprocess.DEVNULL, + text=True, + start_new_session=True, + ) + except OSError as exc: + return ToolOutcome(f"could not run command: {exc}", success=False) + try: + stdout, stderr = process.communicate(timeout=self._timeout_s) + except subprocess.TimeoutExpired: + self._kill_group(process) + return ToolOutcome( + f"command timed out after {self._timeout_s}s: {command}", + success=False, + ) + body = _truncate( + f"exit code: {process.returncode}\nstdout:\n{stdout}\nstderr:\n{stderr}" + ) + return ToolOutcome(body, success=process.returncode == 0) + + @staticmethod + def _kill_group(process: "subprocess.Popen[str]") -> None: + try: + os.killpg(os.getpgid(process.pid), signal.SIGKILL) + except (ProcessLookupError, PermissionError): + process.kill() + # Drain the pipes of the now-dead group so no zombie is left behind. + try: + process.communicate(timeout=5) + except subprocess.TimeoutExpired: + pass + + def reset(self) -> None: + """Discard workspace contents; used by tests, not by the agent.""" + shutil.rmtree(self.root, ignore_errors=True) + self.root.mkdir(parents=True, exist_ok=True) diff --git a/nisse-dev/tests/test_executor.py b/nisse-dev/tests/test_executor.py new file mode 100644 index 0000000..70b43f4 --- /dev/null +++ b/nisse-dev/tests/test_executor.py @@ -0,0 +1,173 @@ +import io +from pathlib import Path +from typing import Any + +import pytest + +from nisse_dev.canonical import AssistantText, CanonicalEvent, StoredEvent, ToolCall +from nisse_dev.executor import LLMExecutor +from nisse_dev.llm import LLMError, RoleMap +from nisse_dev.oplog import OpLog +from nisse_dev.state import StateStore +from nisse_dev.tools import Workspace + +ROLES = RoleMap( + executor="llama3.2", + planner="claude-haiku-4-5", + delegate="claude-haiku-4-5", + consultant="claude-haiku-4-5", +) + + +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="done", model=model)] + return self._turns.pop(0) + + +class FailingClient: + def complete(self, *args: Any, **kwargs: Any) -> list[CanonicalEvent]: + raise LLMError("proxy exploded") + + +def call(tool: str, call_id: str = "c1", **arguments: Any) -> ToolCall: + return ToolCall(call_id=call_id, tool=tool, arguments=arguments) + + +@pytest.fixture +def store(tmp_path: Path) -> StateStore: + return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + + +@pytest.fixture +def ws(tmp_path: Path) -> Workspace: + return Workspace(tmp_path / "work", timeout_s=10) + + +def build(client: Any, store: StateStore, ws: Workspace, **kw: Any) -> LLMExecutor: + return LLMExecutor(client, store, ROLES, ws, **kw) + + +def test_tool_calls_are_executed_and_results_logged( + store: StateStore, ws: Workspace +) -> None: + client = ScriptedClient( + [ + [call("write_file", path="app.py", content="print(1)\n")], + [AssistantText(text="wrote the file", model="llama3.2")], + ] + ) + outcome = build(client, store, ws).run(milestone=1) + + assert outcome.status == "completed" + assert (ws.root / "app.py").read_text() == "print(1)\n" + kinds = [type(e.event).__name__ for e in store.read_events()] + assert kinds == ["ToolCall", "ToolResult", "AssistantText"] + + +def test_uses_executor_role_and_reduced_manifest( + store: StateStore, ws: Workspace +) -> None: + client = ScriptedClient([]) + build(client, store, ws).run(milestone=2) + assert client.calls[0]["model"] == "llama3.2" # never the planner + names = [t["function"]["name"] for t in client.calls[0]["tools"]] + assert names == ["read_file", "write_file", "list_dir", "run_command"] + assert "milestone 2" in client.calls[0]["system"] + + +def test_no_tool_call_ends_the_milestone(store: StateStore, ws: Workspace) -> None: + client = ScriptedClient([[AssistantText(text="all steps done", model="m")]]) + assert build(client, store, ws).run(milestone=1).status == "completed" + + +def test_two_consecutive_command_failures_escalate( + store: StateStore, ws: Workspace +) -> None: + client = ScriptedClient( + [ + [call("run_command", command="exit 1")], + [call("run_command", command="exit 1")], + [call("run_command", command="exit 1")], + ] + ) + outcome = build(client, store, ws, failure_limit=2).run(milestone=1) + assert outcome.status == "escalate" + assert "2 consecutive command failures" in outcome.note + assert len(client.calls) == 2 # stopped, did not keep trying + + +def test_a_success_resets_the_failure_count(store: StateStore, ws: Workspace) -> None: + client = ScriptedClient( + [ + [call("run_command", command="exit 1")], + [call("run_command", command="true")], + [call("run_command", command="exit 1")], + [AssistantText(text="done", model="m")], + ] + ) + assert build(client, store, ws, failure_limit=2).run(milestone=1).status == ( + "completed" + ) + + +def test_failed_reads_do_not_count_toward_escalation( + store: StateStore, ws: Workspace +) -> None: + client = ScriptedClient( + [ + [call("read_file", path="missing-a.txt")], + [call("read_file", path="missing-b.txt")], + [AssistantText(text="looked around, done", model="m")], + ] + ) + assert build(client, store, ws, failure_limit=2).run(milestone=1).status == ( + "completed" + ) + + +def test_turn_cap_is_enforced(store: StateStore, ws: Workspace) -> None: + forever = [[call("list_dir", path=".")] for _ in range(50)] + client = ScriptedClient(forever) + outcome = build(client, store, ws, max_turns=3).run(milestone=1) + assert outcome.status == "capped" + assert "3-turn cap" in outcome.note + assert len(client.calls) == 3 + + +def test_llm_error_escalates_rather_than_crashing( + store: StateStore, ws: Workspace +) -> None: + outcome = build(FailingClient(), store, ws).run(milestone=1) + assert outcome.status == "escalate" + assert "proxy exploded" in outcome.note + + +def test_multiple_calls_in_one_turn_all_run(store: StateStore, ws: Workspace) -> None: + client = ScriptedClient( + [ + [ + call("write_file", call_id="a", path="one.txt", content="1"), + call("write_file", call_id="b", path="two.txt", content="2"), + ], + [AssistantText(text="done", model="m")], + ] + ) + build(client, store, ws).run(milestone=1) + assert (ws.root / "one.txt").exists() and (ws.root / "two.txt").exists() + results = [ + e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult" + ] + assert {r.call_id for r in results} == {"a", "b"} diff --git a/nisse-dev/tests/test_planner.py b/nisse-dev/tests/test_planner.py index d30e58f..9dcce14 100644 --- a/nisse-dev/tests/test_planner.py +++ b/nisse-dev/tests/test_planner.py @@ -12,7 +12,14 @@ from nisse_dev.canonical import ( ToolCall, UserMessage, ) -from nisse_dev.control import AskHuman, Orchestrator, PlanningError, SubmitPlan +from nisse_dev.control import ( + AWAIT_ARCH_SIGNOFF, + AWAIT_MILESTONE_SIGNOFF, + AskHuman, + Orchestrator, + PlanningError, + SubmitPlan, +) from nisse_dev.llm import RoleMap from nisse_dev.oplog import OpLog from nisse_dev.planner import PLANNER_TOOLS, LLMPlanningAgent @@ -108,7 +115,18 @@ def test_model_output_is_appended_to_the_log(store: StateStore) -> None: ) LLMPlanningAgent(client, store, ROLES).plan("arch", "demo") kinds = [type(e.event).__name__ for e in store.read_events()] - assert kinds == ["AssistantText", "ToolCall"] + assert kinds == ["AssistantText", "ToolCall", "ToolResult"] + + +def test_gate_tool_call_is_answered_in_the_log(store: StateStore) -> None: + # submit_plan ends the turn and has no natural result, but a tool call + # left dangling makes the next render invalid for the provider. + client = ScriptedClient([[plan_call("the plan")]]) + LLMPlanningAgent(client, store, ROLES).plan("arch", "demo") + events = [e.event for e in store.read_events()] + assert [type(e).__name__ for e in events] == ["ToolCall", "ToolResult"] + assert events[1].call_id == events[0].call_id + assert "posted for approval" in events[1].result def test_prose_only_turn_retries_once_then_raises(store: StateStore) -> None: @@ -170,7 +188,13 @@ def test_use_project_produces_llm_plan_and_transcript(tmp_path: Path) -> None: events = [e.event for e in store.read_events()] kinds = [type(e).__name__ for e in events] - assert kinds == ["UserMessage", "ToolCall", "PlanSubmitted", "GateOpened"] + assert kinds == [ + "UserMessage", + "ToolCall", + "ToolResult", + "PlanSubmitted", + "GateOpened", + ] submitted = next(e for e in events if isinstance(e, PlanSubmitted)) assert submitted.model == "anthropic/claude-opus-5" # pinned planner @@ -195,6 +219,53 @@ def test_planning_failure_gates_to_await_input(tmp_path: Path) -> None: assert transport.outbound[-1].kind == "question" +def test_operator_can_recover_from_a_planning_failure(tmp_path: Path) -> None: + # Regression: AWAIT_INPUT reached by any path other than ask_human used to + # record no pending item, so every operator reply was refused with "no + # pending question" and the agent was stuck for good. + prose = [AssistantText(text="prose", model="m")] + client = ScriptedClient([prose, prose, [plan_call("recovered plan")]]) + _, store, transport = build(tmp_path, client) + transport.receive("don", "use demo") + assert store.get_state().state == "AWAIT_INPUT" + + transport.receive("don", "try again, keep it short") + + assert store.get_state().state == AWAIT_ARCH_SIGNOFF + assert "recovered plan" in transport.outbound[-1].text + + +def test_operator_can_recover_after_a_stopped_milestone(tmp_path: Path) -> None: + class StoppedExecutor: + def run(self, milestone: int) -> Any: + from nisse_dev.executor import ExecutorOutcome + + return ExecutorOutcome("escalate", "tests failed twice") + + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + transport = InMemoryTransport() + client = ScriptedClient([[plan_call("p1")], [plan_call("p2")]]) + orch = Orchestrator( + store, + transport, + LLMPlanningAgent(client, store, ROLES), + operators={"don"}, + roles=ROLES, + executor=StoppedExecutor(), + ) + orch.start() + transport.receive("don", "use demo") + transport.receive("don", "approve") + transport.receive("don", "approve") + transport.receive("don", "start milestone 1") + assert store.get_state().state == "AWAIT_INPUT" + + transport.receive("don", "ok, I fixed the dependency") + + assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF + assert "start milestone 1" in transport.outbound[-1].text + + def test_rejection_feedback_reaches_the_planner_via_the_log( tmp_path: Path, ) -> None: diff --git a/nisse-dev/tests/test_render.py b/nisse-dev/tests/test_render.py index 9cf7654..39b40a1 100644 --- a/nisse-dev/tests/test_render.py +++ b/nisse-dev/tests/test_render.py @@ -71,6 +71,33 @@ def test_compaction_replaces_its_range_only() -> None: 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") diff --git a/nisse-dev/tests/test_tools.py b/nisse-dev/tests/test_tools.py new file mode 100644 index 0000000..5e66c4f --- /dev/null +++ b/nisse-dev/tests/test_tools.py @@ -0,0 +1,129 @@ +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 _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" + 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 diff --git a/tests/integration/test_milestone_live.py b/tests/integration/test_milestone_live.py new file mode 100644 index 0000000..8a8ea62 --- /dev/null +++ b/tests/integration/test_milestone_live.py @@ -0,0 +1,107 @@ +"""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 ( + AWAIT_ARCH_SIGNOFF, + AWAIT_INPUT, + AWAIT_MILESTONE_SIGNOFF, + 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 _drive_to_execution( + store: StateStore, transport: InMemoryTransport, max_steps: int = 10 +) -> None: + """Play the operator. A scripted approve/approve/start sequence is not + robust against a live model: ask_human is a legitimate move at any gate, + and a question shifts the state machine under a fixed script. So respond + to whatever the agent actually did.""" + for _ in range(max_steps): + state = store.get_state() + assert state is not None + if state.state == EXECUTING: + return + gate = state.data.get("gate") or {} + if state.state == AWAIT_INPUT: + transport.receive("don", "Use your best judgement and proceed.") + elif state.state == AWAIT_MILESTONE_SIGNOFF and gate.get("approved"): + transport.receive("don", "start milestone 1") + elif state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF): + transport.receive("don", "approve") + else: + raise AssertionError(f"unexpected state {state.state}") + raise AssertionError( + f"never reached execution; last message: {transport.outbound[-1]}" + ) + + +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, !', and verify it by running python.", + ) + _drive_to_execution(store, transport) + + 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