nisse/nisse-dev/src/nisse_dev/planner.py

"""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, ToolResult
from nisse_dev.control import AskHuman, PlanningError, SubmitPlan
from nisse_dev.llm import BudgetExhausted, 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 BudgetExhausted:
            # Retrying cannot help: the budget will not reappear within a
            # turn. Surface it now so the operator is told rather than the
            # loop spending a second call to fail identically.
            raise
        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)
                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:
        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