Add LLM-backed planning agent (A4/A11/A13) — step 5a

- planner.py: LLMPlanningAgent offers exactly two tools, submit_plan
  and ask_human, so gating stays structural at the tool layer; no
  proceed tool exists. System prompt requires explicit, low-inference
  plans because a weaker model executes them (A13)
- Model output is appended to the canonical log as it arrives, before
  being interpreted, so a crash mid-planning leaves a truthful
  transcript
- A bad turn (prose instead of a tool call, malformed arguments, or a
  failed call) retries once, then raises PlanningError; the
  orchestrator gates to AWAIT_INPUT rather than inventing a plan
- Orchestrator: appends every operator message, plan_submitted, and
  gate_opened to the transcript, so re-planning sees rejection
  feedback as conversation rather than out-of-band parameters; pins
  the A11 role mapping at task start
- llm.py: non-JSON or non-object tool arguments now raise LLMError
  instead of escaping as JSONDecodeError — observed from llama3.2,
  which emits truncated argument strings
- Live planning test is opt-in on NISSE_LIVE_PLANNER_MODEL: planning
  escalates unconditionally (A12) and the local model does not hold
  the tool-call contract, so it awaits a frontier route

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.env.example

@@ -11,6 +11,12 @@ NISSE_ROLE_PLANNER=anthropic/claude-opus-5
 NISSE_ROLE_DELEGATE=anthropic/claude-opus-5
 NISSE_ROLE_CONSULTANT=anthropic/claude-opus-5
 
+# Opt-in for the live planning test: set to a frontier model the virtual key
+# can reach. Planning escalates unconditionally (A12) and small local models
+# do not hold the tool-call contract reliably, so this stays unset until a
+# frontier route exists.
+# NISSE_LIVE_PLANNER_MODEL=anthropic/claude-opus-5
+
 # --- Matrix transport (A1), used by nisse-dev's MatrixTransport ---
 NISSE_MATRIX_HOMESERVER=https://matrix.example
 NISSE_MATRIX_USER_ID=@nisse-dev:example

nisse-dev/src/nisse_dev/control.py

@@ -13,7 +13,9 @@ from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 from typing import Any, Protocol
 
+from nisse_dev.canonical import GateOpened, PlanSubmitted, UserMessage
 from nisse_dev.gates import Approved, Gate, Ignored, PlanRevision, Rejected
+from nisse_dev.llm import RoleMap
 from nisse_dev.reconcile import Reconciler, reconcile_on_boot
 from nisse_dev.state import StateStore
 from nisse_dev.transport import InboundMessage, Transport
@@ -41,6 +43,12 @@ class TransitionError(RuntimeError):
     pass
 
 
+class PlanningError(RuntimeError):
+    """A planning turn produced nothing usable. Part of the PlanningAgent
+    contract: the orchestrator gates to AWAIT_INPUT rather than inventing a
+    plan or retrying forever."""
+
+
 @dataclass(frozen=True)
 class SubmitPlan:
     content: str
@@ -86,6 +94,7 @@ def _fresh_data() -> dict[str, Any]:
         "question": None,
         "milestone": 1,
         "task_id": None,
+        "roles": None,  # A11 mapping, pinned at task start
     }
 
 
@@ -96,10 +105,12 @@ class Orchestrator:
         transport: Transport,
         agent: PlanningAgent,
         operators: Iterable[str],
+        roles: RoleMap | None = None,
     ) -> None:
         self._store = store
         self._transport = transport
         self._agent = agent
+        self._roles = roles
         self._operators = set(operators)
         self._data = _fresh_data()
         self._project: str | None = None
@@ -144,6 +155,12 @@ class Orchestrator:
     def _handle(self, msg: InboundMessage) -> None:
         if msg.sender not in self._operators:
             return
+        # Every operator message enters the canonical transcript, so planning
+        # turns see feedback and answers as conversation rather than as
+        # out-of-band parameters (A10).
+        self._store.append_event(
+            UserMessage(text=msg.text.strip(), source=f"transport:{msg.sender}")
+        )
         state = self._store.get_state()
         current = state.state if state else IDLE
         text = msg.text.strip()
@@ -172,7 +189,10 @@ class Orchestrator:
 
     def _begin_project(self, project: str) -> None:
         task_id = f"{project}-{uuid.uuid4().hex[:8]}"
-        self._data = {**_fresh_data(), "task_id": task_id}
+        # Pin the role mapping now (A11): a mid-task config change must not
+        # split one task across two planners.
+        pinned = self._roles.pinned() if self._roles is not None else None
+        self._data = {**_fresh_data(), "task_id": task_id, "roles": pinned}
         self._project = project
         self._store.oplog.task_id = task_id
         self._transition(PLANNING_ARCH, cause=f"operator command: use {project}")
@@ -182,9 +202,16 @@ class Orchestrator:
         self, gate_name: str, feedback: str | None = None, answer: str | None = None
     ) -> None:
         assert self._project is not None
-        action = self._agent.plan(
+        try:
-            gate_name, self._project, feedback=feedback, answer=answer
+            action = self._agent.plan(
-        )
+                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?"
+            )
+            return
         if isinstance(action, AskHuman):
             self._data["question"] = {"text": action.question, "gate": gate_name}
             self._transition(
@@ -203,6 +230,13 @@ class Orchestrator:
         await_state = (
             AWAIT_ARCH_SIGNOFF if gate_name == "arch" else AWAIT_MILESTONE_SIGNOFF
         )
+        roles = self._data.get("roles") or {}
+        self._store.append_event(
+            PlanSubmitted(plan=action.content, model=roles.get("planner", "stub"))
+        )
+        self._store.append_event(
+            GateOpened(gate=gate_name, reason=f"plan submitted (rev-{rev})")
+        )
         self._transition(await_state, cause=f"plan submitted ({gate_name} rev-{rev})")
         self._transport.send_approval_request(
             f"{action.content}\n\nreply 'approve' to approve rev-{rev}; "

nisse-dev/src/nisse_dev/llm.py

@@ -131,11 +131,26 @@ class LiteLLMClient:
         # gets a locally-minted canonical ID; the provider's is dropped, since
         # every future request is rendered wholly from the canonical log.
         for call in message.get("tool_calls") or []:
+            raw = call["function"]["arguments"] or "{}"
+            # Weak models emit truncated or non-JSON argument strings; that is
+            # a bad model turn, not a client crash (A15).
+            try:
+                arguments = json.loads(raw)
+            except json.JSONDecodeError as exc:
+                raise LLMError(
+                    f"tool call {call['function']['name']!r} had non-JSON "
+                    f"arguments: {raw[:200]!r}"
+                ) from exc
+            if not isinstance(arguments, dict):
+                raise LLMError(
+                    f"tool call {call['function']['name']!r} arguments were "
+                    f"{type(arguments).__name__}, expected object"
+                )
             out.append(
                 ToolCall(
                     call_id=mint_call_id(),
                     tool=call["function"]["name"],
-                    arguments=json.loads(call["function"]["arguments"] or "{}"),
+                    arguments=arguments,
                 )
             )
         return out

nisse-dev/src/nisse_dev/planner.py

@@ -0,0 +1,156 @@
+"""LLM-backed planning agent (spec A4 gating, A11 roles, A13 plan quality).
+
+Gating is structural at the tool layer: the planner is offered exactly two
+tools — ``submit_plan``, which posts a plan for approval and ends the turn,
+and ``ask_human``, which pauses for an answer. No "proceed" tool exists, so
+the model has no path to advance itself past a gate.
+
+Model output is appended to the canonical log as it arrives (A2/A10), so a
+crash mid-planning leaves a truthful transcript.
+"""
+
+from typing import Any
+
+from nisse_dev.canonical import AssistantText, ToolCall
+from nisse_dev.control import AskHuman, PlanningError, SubmitPlan
+from nisse_dev.llm import LLMClient, LLMError, RoleMap
+from nisse_dev.state import StateStore
+
+# Plans may be executed by a materially weaker model, which cannot
+# reconstruct intent (A13).
+SYSTEM_PROMPT = """You are nisse-dev, a software engineering agent.
+
+You are planning, not implementing. Produce plans that are explicit and
+low-inference: a materially weaker model will execute them and cannot
+reconstruct your intent. Name files, commands, and acceptance criteria
+concretely; never write "handle the edge cases" or "wire it up".
+
+You must answer by calling exactly one tool:
+- submit_plan: your plan is posted for human approval and your turn ends.
+- ask_human: ask one question when a decision is genuinely the operator's.
+
+You cannot approve your own plan or start work; a human gate follows every
+plan you submit."""
+
+PLANNER_TOOLS: list[dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "submit_plan",
+            "description": "Post a plan for human approval. Ends your turn.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "type": {
+                        "type": "string",
+                        "description": "Which plan this is, e.g. 'arch' or 'milestone-1'.",
+                    },
+                    "content": {
+                        "type": "string",
+                        "description": "The plan itself, explicit enough to execute verbatim.",
+                    },
+                },
+                "required": ["type", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "ask_human",
+            "description": "Ask the operator one question, then wait.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "question": {"type": "string"},
+                },
+                "required": ["question"],
+            },
+        },
+    },
+]
+
+_NUDGE = (
+    "You did not call a tool. Call submit_plan or ask_human now; "
+    "plain prose is not accepted."
+)
+
+
+class LLMPlanningAgent:
+    def __init__(
+        self,
+        client: LLMClient,
+        store: StateStore,
+        roles: RoleMap,
+        *,
+        system_prompt: str = SYSTEM_PROMPT,
+    ) -> None:
+        self._client = client
+        self._store = store
+        self._roles = roles
+        self._system_prompt = system_prompt
+
+    def plan(
+        self,
+        gate: str,
+        project: str,
+        feedback: str | None = None,
+        answer: str | None = None,
+    ) -> SubmitPlan | AskHuman:
+        """Operator feedback and answers are already durable in the canonical
+        log (the orchestrator appends every inbound message), so they are not
+        re-injected here; the log is the single source of context."""
+        model = self._roles.resolve("planner")
+        instruction = (
+            f"{self._system_prompt}\n\n"
+            f"Current task: draft the {gate} plan for project {project!r}."
+        )
+        action, first_error = self._try_turn(model, instruction)
+        if action is not None:
+            return action
+        action, second_error = self._try_turn(model, f"{instruction}\n\n{_NUDGE}")
+        if action is None:
+            detail = second_error or first_error
+            raise PlanningError(
+                f"planner ({model}) produced no submit_plan or ask_human call "
+                f"after a retry{f': {detail}' if detail else ''}"
+            )
+        return action
+
+    def _try_turn(
+        self, model: str, system: str
+    ) -> tuple[SubmitPlan | AskHuman | None, str | None]:
+        """A malformed or failed call is a bad turn, not a crash: the caller
+        retries once and then gates to the human (A12 — judgment about model
+        output is observable, never the model's own claim)."""
+        try:
+            return self._turn(model, system), None
+        except LLMError as exc:
+            return None, str(exc)
+
+    def _turn(self, model: str, system: str) -> SubmitPlan | AskHuman | None:
+        events = self._store.read_events()
+        produced = self._client.complete(
+            events, tools=PLANNER_TOOLS, model=model, system=system
+        )
+        action: SubmitPlan | AskHuman | None = None
+        for event in produced:
+            # Append before interpreting: the transcript must be truthful even
+            # if the call below rejects the arguments.
+            self._store.append_event(event)
+            if isinstance(event, AssistantText) or action is not None:
+                continue
+            if isinstance(event, ToolCall):
+                action = self._interpret(event)
+        return action
+
+    def _interpret(self, call: ToolCall) -> SubmitPlan | AskHuman | None:
+        if call.tool == "submit_plan":
+            content = call.arguments.get("content")
+            if isinstance(content, str) and content.strip():
+                return SubmitPlan(content.strip())
+        elif call.tool == "ask_human":
+            question = call.arguments.get("question")
+            if isinstance(question, str) and question.strip():
+                return AskHuman(question.strip())
+        return None

nisse-dev/tests/test_llm.py

@@ -76,6 +76,38 @@ def test_tool_calls_get_locally_minted_canonical_ids() -> None:
     assert event.call_id.startswith("call-")
 
 
+def test_truncated_tool_arguments_raise_llm_error() -> None:
+    message = {
+        "content": None,
+        "tool_calls": [
+            {
+                "id": "c",
+                "type": "function",
+                "function": {"name": "submit_plan", "arguments": '{"content": "'},
+            }
+        ],
+    }
+    client = make_client(lambda request: completion(message))
+    with pytest.raises(LLMError, match="non-JSON arguments"):
+        client.complete(EVENTS, tools=None, model="local/qwen")
+
+
+def test_non_object_tool_arguments_raise_llm_error() -> None:
+    message = {
+        "content": None,
+        "tool_calls": [
+            {
+                "id": "c",
+                "type": "function",
+                "function": {"name": "submit_plan", "arguments": '"just a string"'},
+            }
+        ],
+    }
+    client = make_client(lambda request: completion(message))
+    with pytest.raises(LLMError, match="expected object"):
+        client.complete(EVENTS, tools=None, model="local/qwen")
+
+
 def test_http_error_status_raises_llm_error() -> None:
     client = make_client(lambda request: httpx.Response(429, text="rate limited"))
     with pytest.raises(LLMError, match="429"):

nisse-dev/tests/test_planner.py

@@ -0,0 +1,210 @@
+import io
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    CanonicalEvent,
+    PlanSubmitted,
+    StoredEvent,
+    ToolCall,
+    UserMessage,
+)
+from nisse_dev.control import AskHuman, Orchestrator, PlanningError, SubmitPlan
+from nisse_dev.llm import RoleMap
+from nisse_dev.oplog import OpLog
+from nisse_dev.planner import PLANNER_TOOLS, LLMPlanningAgent
+from nisse_dev.state import StateStore
+from nisse_dev.transport import InMemoryTransport
+
+ROLES = RoleMap(
+    executor="local/qwen",
+    planner="anthropic/claude-opus-5",
+    delegate="anthropic/claude-opus-5",
+    consultant="anthropic/claude-opus-5",
+)
+
+
+class ScriptedClient:
+    """Returns canned canonical events per call, recording what it was asked."""
+
+    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(
+            {"events": events, "tools": tools, "model": model, "system": system}
+        )
+        if not self._turns:
+            return []
+        return self._turns.pop(0)
+
+
+def plan_call(content: str, gate: str = "arch") -> ToolCall:
+    return ToolCall(
+        call_id="call-1",
+        tool="submit_plan",
+        arguments={"type": gate, "content": content},
+    )
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> StateStore:
+    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+
+
+def test_submit_plan_tool_call_becomes_submit_plan(store: StateStore) -> None:
+    client = ScriptedClient([[plan_call("step 1: create src/app.py")]])
+    agent = LLMPlanningAgent(client, store, ROLES)
+    action = agent.plan("arch", "demo")
+    assert action == SubmitPlan("step 1: create src/app.py")
+
+
+def test_ask_human_tool_call_becomes_ask_human(store: StateStore) -> None:
+    client = ScriptedClient(
+        [
+            [
+                ToolCall(
+                    call_id="call-1",
+                    tool="ask_human",
+                    arguments={"question": "postgres or sqlite?"},
+                )
+            ]
+        ]
+    )
+    action = LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    assert action == AskHuman("postgres or sqlite?")
+
+
+def test_planner_role_and_tools_are_used(store: StateStore) -> None:
+    client = ScriptedClient([[plan_call("plan")]])
+    LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    call = client.calls[0]
+    assert call["model"] == "anthropic/claude-opus-5"  # never the executor
+    assert [t["function"]["name"] for t in call["tools"]] == [
+        "submit_plan",
+        "ask_human",
+    ]
+    assert "demo" in call["system"] and "arch" in call["system"]
+
+
+def test_no_proceed_tool_is_offered() -> None:
+    names = {tool["function"]["name"] for tool in PLANNER_TOOLS}
+    assert names == {"submit_plan", "ask_human"}  # no self-advancement path
+
+
+def test_model_output_is_appended_to_the_log(store: StateStore) -> None:
+    client = ScriptedClient(
+        [[AssistantText(text="thinking out loud", model="m"), plan_call("the plan")]]
+    )
+    LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    kinds = [type(e.event).__name__ for e in store.read_events()]
+    assert kinds == ["AssistantText", "ToolCall"]
+
+
+def test_prose_only_turn_retries_once_then_raises(store: StateStore) -> None:
+    client = ScriptedClient(
+        [
+            [AssistantText(text="here's what I think", model="m")],
+            [AssistantText(text="still prose", model="m")],
+        ]
+    )
+    with pytest.raises(PlanningError, match="no submit_plan or ask_human"):
+        LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    assert len(client.calls) == 2
+    assert "did not call a tool" in client.calls[1]["system"]
+
+
+def test_retry_succeeds_after_prose(store: StateStore) -> None:
+    client = ScriptedClient(
+        [[AssistantText(text="prose", model="m")], [plan_call("the real plan")]]
+    )
+    action = LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    assert action == SubmitPlan("the real plan")
+
+
+def test_malformed_tool_arguments_are_rejected(store: StateStore) -> None:
+    blank = ToolCall(call_id="c", tool="submit_plan", arguments={"content": "   "})
+    client = ScriptedClient([[blank], [blank]])
+    with pytest.raises(PlanningError):
+        LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+
+
+def test_planner_sees_prior_transcript(store: StateStore) -> None:
+    store.append_event(UserMessage(text="use demo", source="transport:@don"))
+    client = ScriptedClient([[plan_call("plan")]])
+    LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    seen = [e.event for e in client.calls[0]["events"]]
+    assert UserMessage(text="use demo", source="transport:@don") in seen
+
+
+# --- orchestrator integration (still no network) -------------------------
+
+
+def build(tmp_path: Path, client: ScriptedClient) -> tuple[Orchestrator, StateStore]:
+    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+    transport = InMemoryTransport()
+    agent = LLMPlanningAgent(client, store, ROLES)
+    orch = Orchestrator(store, transport, agent, operators={"don"}, roles=ROLES)
+    orch.start()
+    return orch, store, transport
+
+
+def test_use_project_produces_llm_plan_and_transcript(tmp_path: Path) -> None:
+    client = ScriptedClient([[plan_call("1. create app.py\n2. add a test")]])
+    _, store, transport = build(tmp_path, client)
+    transport.receive("don", "use demo")
+
+    request = transport.outbound[-1]
+    assert request.kind == "approval_request"
+    assert "create app.py" in request.text
+
+    events = [e.event for e in store.read_events()]
+    kinds = [type(e).__name__ for e in events]
+    assert kinds == ["UserMessage", "ToolCall", "PlanSubmitted", "GateOpened"]
+    submitted = next(e for e in events if isinstance(e, PlanSubmitted))
+    assert submitted.model == "anthropic/claude-opus-5"  # pinned planner
+
+
+def test_roles_pinned_at_task_start(tmp_path: Path) -> None:
+    client = ScriptedClient([[plan_call("plan")]])
+    _, store, transport = build(tmp_path, client)
+    transport.receive("don", "use demo")
+    state = store.get_state()
+    assert state is not None
+    assert state.data["roles"]["planner"] == "anthropic/claude-opus-5"
+
+
+def test_planning_failure_gates_to_await_input(tmp_path: Path) -> None:
+    prose = [AssistantText(text="prose", model="m")]
+    client = ScriptedClient([prose, prose])
+    _, store, transport = build(tmp_path, client)
+    transport.receive("don", "use demo")
+    state = store.get_state()
+    assert state is not None
+    assert state.state == "AWAIT_INPUT"
+    assert transport.outbound[-1].kind == "question"
+
+
+def test_rejection_feedback_reaches_the_planner_via_the_log(
+    tmp_path: Path,
+) -> None:
+    client = ScriptedClient([[plan_call("v1")], [plan_call("v2")]])
+    _, _store, transport = build(tmp_path, client)
+    transport.receive("don", "use demo")
+    transport.receive("don", "too vague, name the files")
+
+    replan_context = [e.event for e in client.calls[1]["events"]]
+    assert any(
+        isinstance(e, UserMessage) and "too vague" in e.text for e in replan_context
+    )
+    assert "v2" in transport.outbound[-1].text

tests/integration/test_planning_live.py

@@ -0,0 +1,64 @@
+"""Live planning turn: a real model must answer with a submit_plan or
+ask_human tool call, not prose.
+
+Deliberately opt-in via NISSE_LIVE_PLANNER_MODEL rather than the usual
+NISSE_LLM_BASE_URL sentinel. Planning escalates to a frontier model
+unconditionally (A12), and the small local model does not hold the tool-call
+contract reliably — measured, see the PR — so this test needs a frontier
+route that does not exist yet. Set the variable to the model to plan with
+once one is configured; until then this skips on purpose, not by accident.
+"""
+
+import io
+import os
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.canonical import UserMessage
+from nisse_dev.control import AskHuman, SubmitPlan
+from nisse_dev.envfile import load_env_file
+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
+
+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 frontier planner route configured ({SENTINEL} unset)",
+    ),
+]
+
+
+def test_live_planning_turn_returns_a_tool_action(tmp_path: Path) -> None:
+    assert os.environ.get("NISSE_LLM_TOKEN"), (
+        f"{SENTINEL} is set, so the live planning path is expected to work, "
+        "but NISSE_LLM_TOKEN is missing"
+    )
+    model = os.environ[SENTINEL]
+    roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
+    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+    store.append_event(
+        UserMessage(
+            text=(
+                "Project 'hello-cli': a Python script that prints a greeting. "
+                "Draft the architecture plan."
+            ),
+            source="live-test",
+        )
+    )
+    agent = LLMPlanningAgent(LiteLLMClient.from_env(), store, roles)
+
+    action = agent.plan("arch", "hello-cli")
+
+    assert isinstance(action, SubmitPlan | AskHuman)
+    text = action.content if isinstance(action, SubmitPlan) else action.question
+    assert text.strip(), "tool call carried empty content"
+    # The turn must be durable regardless of what the model said (A2/A10).
+    assert len(store.read_events()) > 1