Merge pull request 'Discover a model's vendor from LiteLLM instead of its name' (#35) from alias-discovery into main

Reviewed-on: dongar/nisse#35
Reviewed-by: Don Garrett <dgarrett@acm.org>

.env.example

@@ -34,3 +34,26 @@ NISSE_GITEA_TEST_REPO=owner/sandbox-repo
 # Gitea logins allowed to drive the agent via PR comments (A4 per-channel
 # allowlist; a Matrix MXID and a Gitea login are different names).
 NISSE_GITEA_OPERATORS=your-gitea-login
+
+# --- Running the agent (the demo, not the tests) ---
+# The tests use NISSE_GITEA_TEST_REPO; a running agent uses these.
+#
+# Layout. NISSE_WORKSPACE is the *clone the agent works in*, not the directory
+# the agent runs from — they must not be the same place:
+#
+#   ~/nisse-demo-workspace/     deployment root, not a git repo
+#   ├── .env                    this file
+#   ├── state.db                the A2 state store
+#   └── project/                NISSE_WORKSPACE, the clone
+#
+# The agent commits its workspace with `git add -A` and pushes it, so anything
+# inside the clone reaches the project repository. A .env or state store placed
+# there would publish credentials as a side effect of the agent working
+# correctly; startup refuses to proceed if either is inside the workspace.
+NISSE_STATE_DB=~/nisse-demo-workspace/state.db
+NISSE_WORKSPACE=~/nisse-demo-workspace/project
+NISSE_PROJECT=demo
+NISSE_GITEA_REPO=owner/demo-repo
+# NISSE_BASE_BRANCH=main
+# Path to this file, if not ./.env
+# NISSE_ENV=~/nisse-demo-workspace/.env

nisse-dev/src/nisse_dev/agent.py

@@ -0,0 +1,205 @@
+"""Composition root: the one place that knows how the whole agent fits together.
+
+Every component below has been unit-tested in isolation for some time; nothing
+had ever assembled them. That gap hid two live bugs — the delegation harness
+and the Gitea comment poller were both fully built and reachable from nothing
+but their own tests — which is the argument for having a single root rather
+than constructing pieces ad hoc.
+"""
+
+import threading
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from nisse_dev.approvals import PlanBoard
+from nisse_dev.catalog import ModelCatalog, select_client
+from nisse_dev.config import AgentConfig
+from nisse_dev.control import Orchestrator
+from nisse_dev.delegation import Delegation
+from nisse_dev.executor import LLMExecutor
+from nisse_dev.expert import ExpertDesk
+from nisse_dev.git import GitRepo, authenticated_url
+from nisse_dev.gitea import Comment, GiteaClient
+from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient
+from nisse_dev.matrix import MatrixTransport
+from nisse_dev.oplog import OpLog
+from nisse_dev.planner import LLMPlanningAgent
+from nisse_dev.poller import CommentPoller
+from nisse_dev.publish import MilestonePublisher
+from nisse_dev.state import StateStore
+from nisse_dev.tools import Workspace
+
+TOOL_DRIVING_ROLES = ("planner", "executor", "delegate")
+
+
+def discover_models(config: AgentConfig, oplog: OpLog) -> ModelCatalog:
+    """Resolve every configured role against LiteLLM before anything runs.
+
+    Refusing to boot is the whole point: a role pointed at a model this key
+    cannot see is a misconfiguration, and the cheapest place to learn that is
+    here rather than ten minutes into a milestone. An unreachable proxy is
+    reported separately -- it is an outage, not a mistake, and telling them
+    apart is what lets an operator know whether to edit config or wait.
+    """
+    catalog = ModelCatalog.discover(config.llm_base_url, config.llm_token)
+    catalog.validate(config.roles.pinned())
+    for role in TOOL_DRIVING_ROLES:
+        facts = catalog.facts(config.roles.resolve(role))
+        if facts.supports_tools is False:
+            # A warning, not a refusal: models LiteLLM declares tool-less are
+            # observed driving tool loops anyway, and A15 expects weak local
+            # models to be accommodated rather than excluded.
+            oplog.emit(
+                "model_tools_unsupported",
+                cause="startup",
+                role=role,
+                model=facts.alias,
+                backing=facts.backing,
+            )
+    return catalog
+
+
+@dataclass
+class Agent:
+    config: AgentConfig
+    store: StateStore
+    transport: MatrixTransport
+    orchestrator: Orchestrator
+    gitea: GiteaClient
+    account: str
+    _stopped: threading.Event
+
+    def start(self) -> None:
+        """Boot in the order A3 requires: reconcile against external truth
+        before accepting a single operator command."""
+        self.orchestrator.start()
+        self.transport.start()
+        state = self.store.get_state()
+        where = state.state if state else "IDLE"
+        self.transport.send_message(
+            f"nisse-dev up as {self.account} — project {self.config.project}, "
+            f"state {where}, planner {self.config.roles.planner}, "
+            f"executor {self.config.roles.executor}"
+        )
+
+    def wait(self) -> None:
+        self._stopped.wait()
+
+    def stop(self) -> None:
+        """Shut down outermost-first: stop accepting work, then stop talking,
+        then close the state store last so nothing writes to a closed file."""
+        self.orchestrator.stop_watching()
+        self.transport.close()
+        self.gitea.close()
+        self.store.close()
+        self._stopped.set()
+
+
+def prepare_workspace(config: AgentConfig, account: str) -> GitRepo:
+    """Ensure the workspace is a clone of the project repo.
+
+    Cloning here rather than on `use <project>` is the single-project
+    shortcut: resolving a project name to a repo is issue #27's work.
+    """
+    repo = GitRepo(config.workspace)
+    remote = authenticated_url(
+        config.gitea_url, config.gitea_repo, account, config.gitea_token
+    )
+    if not (config.workspace / ".git").exists():
+        config.workspace.parent.mkdir(parents=True, exist_ok=True)
+        GitRepo(config.workspace.parent).clone(remote, config.workspace)
+    repo.configure_identity(account, f"{account}@users.noreply.local")
+    return repo
+
+
+def build_agent(
+    config: AgentConfig, *, notify: Callable[[str], None] | None = None
+) -> Agent:
+    oplog = OpLog("nisse-dev")
+    store = StateStore(config.state_db, oplog=oplog)
+    # Before the clone, before the token checks: a typo'd role name should
+    # cost a second, not a repository.
+    catalog = discover_models(config, oplog)
+
+    gitea = GiteaClient(config.gitea_url, config.gitea_token)
+    # Whoami rather than another config variable: it names the identity every
+    # Gitea permission is evaluated against, and proves the token works before
+    # anything depends on it.
+    account = gitea.whoami()
+
+    git_repo = prepare_workspace(config, account)
+    workspace = Workspace(config.workspace)
+
+    normalized = LiteLLMClient(config.llm_base_url, config.llm_token)
+    # Delegations prefer the vendor-native route; fidelity is cheap
+    # elsewhere (A5). Which route a given model gets is decided by
+    # select_client, not by where the caller happens to sit.
+    passthrough = AnthropicPassthroughClient(config.llm_base_url, config.llm_token)
+
+    transport = MatrixTransport(config.matrix)
+    announce = notify or transport.send_message
+
+    expert = ExpertDesk(normalized, store, config.roles)
+    delegate_model = config.roles.resolve("delegate")
+    delegation = Delegation(
+        select_client(delegate_model, normalized, passthrough, catalog),
+        store,
+        workspace,
+        git_repo,
+        model=delegate_model,
+        backing_model=catalog.facts(delegate_model).backing,
+    )
+    executor = LLMExecutor(
+        normalized,
+        store,
+        config.roles,
+        workspace,
+        expert=expert,
+        delegate=delegation,
+        notify=announce,
+    )
+    publisher = MilestonePublisher(
+        store,
+        gitea,
+        config.gitea_repo,
+        git_repo,
+        base_url=config.gitea_url,
+        account=account,
+        token=config.gitea_token,
+        base_branch=config.base_branch,
+    )
+
+    def poller_factory(on_comment: Callable[[Comment], None]) -> CommentPoller:
+        return CommentPoller(
+            gitea,
+            config.gitea_repo,
+            on_comment,
+            oplog=oplog,
+            # Its own comments are not new input; without this the agent
+            # answers itself.
+            ignore_authors={account},
+        )
+
+    orchestrator = Orchestrator(
+        store,
+        transport,
+        LLMPlanningAgent(normalized, store, config.roles),
+        operators=config.operators,
+        roles=config.roles,
+        executor=executor,
+        publisher=publisher,
+        gitea_operators=config.gitea_operators,
+        board=PlanBoard(gitea, config.gitea_repo),
+        delegate=delegation,
+        poller_factory=poller_factory,
+        catalog=catalog,
+    )
+    return Agent(
+        config=config,
+        store=store,
+        transport=transport,
+        orchestrator=orchestrator,
+        gitea=gitea,
+        account=account,
+        _stopped=threading.Event(),
+    )

nisse-dev/src/nisse_dev/canonical.py

@@ -56,8 +56,12 @@ class GateOpened:
 class DelegationStarted:
     brief: str
     role: str
-    model: str
+    model: str  # the alias asked for, e.g. "remote-architecture"
     caps: dict[str, Any]  # turns / tokens / wall-clock limits
+    # What the alias resolved to at task start. An alias is free to change
+    # models between tasks, so eval history (A16) keyed on the alias alone
+    # would average results across whatever it happened to point at.
+    backing_model: str = ""
 
 
 @dataclass(frozen=True)

nisse-dev/src/nisse_dev/catalog.py

@@ -0,0 +1,172 @@
+"""Runtime model discovery (spec A5, A11, A13).
+
+LiteLLM is the VM's single LLM destination, which makes it the authority on
+what any given model *is*. Asking it at startup keeps vendor knowledge in one
+place instead of duplicating it as name patterns in agent code -- and that
+stops being a tidiness argument the moment a role alias no longer names its
+vendor. An alias like ``remote-architecture`` is deliberately free to point at
+Anthropic today and something else tomorrow; nothing in the name can tell us
+which request shape the model behind it expects.
+
+Discovery is per key, not global: ``/model/info`` returns only the models the
+presented token is allowed to use. So the same check that catches a typo'd
+alias also catches a role pointed at a model this key cannot reach.
+"""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from nisse_dev.llm import LLMClient, LLMError
+
+# LiteLLM's provider name for models we have a vendor-native adapter for.
+ANTHROPIC_PROVIDER = "anthropic"
+
+
+class CatalogUnavailable(LLMError):
+    """The catalog could not be fetched: the proxy is down, unreachable, or
+    refused the token. An outage, not a mistake in the configuration."""
+
+
+class UnknownModel(LLMError):
+    """A model was named that this key's catalog does not contain: a typo, a
+    stale alias, or a key without access. Always a misconfiguration."""
+
+
+@dataclass(frozen=True)
+class ModelFacts:
+    alias: str  # what we ask for, e.g. "remote-architecture"
+    backing: str  # what LiteLLM calls, e.g. "anthropic/claude-opus-5"
+    provider: str  # "anthropic", "openai", "ollama", ...
+    supports_tools: bool | None  # None when LiteLLM does not say
+
+    @property
+    def native_anthropic(self) -> bool:
+        return self.provider == ANTHROPIC_PROVIDER
+
+
+class ModelCatalog:
+    """What one LiteLLM key can see, as of the moment it was asked."""
+
+    def __init__(self, facts: Mapping[str, ModelFacts]) -> None:
+        self._facts = dict(facts)
+
+    @classmethod
+    def from_payload(cls, payload: Mapping[str, Any]) -> "ModelCatalog":
+        facts: dict[str, ModelFacts] = {}
+        for entry in payload.get("data") or []:
+            alias = entry.get("model_name")
+            if not alias:
+                continue
+            params = entry.get("litellm_params") or {}
+            info = entry.get("model_info") or {}
+            backing = str(params.get("model") or "")
+            # litellm_provider is the direct answer; the backing model's
+            # "vendor/model" prefix says the same thing and is present even
+            # when the provider field is not.
+            provider = info.get("litellm_provider") or backing.partition("/")[0]
+            facts[alias] = ModelFacts(
+                alias=alias,
+                backing=backing,
+                provider=str(provider),
+                supports_tools=info.get("supports_function_calling"),
+            )
+        return cls(facts)
+
+    @classmethod
+    def discover(
+        cls,
+        base_url: str,
+        token: str,
+        *,
+        timeout_s: float = 30.0,
+        transport: httpx.BaseTransport | None = None,
+    ) -> "ModelCatalog":
+        try:
+            with httpx.Client(
+                base_url=base_url.rstrip("/"),
+                headers={"Authorization": f"Bearer {token}"},
+                timeout=timeout_s,
+                transport=transport,
+            ) as http:
+                response = http.get("/model/info")
+        except httpx.HTTPError as exc:
+            raise CatalogUnavailable(
+                f"could not reach the LiteLLM proxy at {base_url} to discover "
+                f"models: {exc}"
+            ) from exc
+        if response.status_code != 200:
+            raise CatalogUnavailable(
+                f"the LiteLLM proxy refused a model listing with "
+                f"{response.status_code}: {response.text[:300]}"
+            )
+        return cls.from_payload(response.json())
+
+    def aliases(self) -> list[str]:
+        return sorted(self._facts)
+
+    def facts(self, alias: str) -> ModelFacts:
+        try:
+            return self._facts[alias]
+        except KeyError:
+            raise UnknownModel(
+                f"{alias!r} is not a model this LiteLLM key can see; "
+                f"available: {', '.join(self.aliases()) or '(none)'}"
+            ) from None
+
+    def validate(self, roles: Mapping[str, str]) -> None:
+        """Check a whole role mapping at once, before anything uses it.
+
+        Reported together rather than one per boot: an operator fixing a
+        config wants the whole list, not to rediscover the next typo on the
+        next start (the same reason ``AgentConfig.from_env`` batches its
+        missing variables).
+        """
+        unknown = {
+            role: alias for role, alias in roles.items() if alias not in self._facts
+        }
+        if unknown:
+            named = ", ".join(
+                f"{role}={alias!r}" for role, alias in sorted(unknown.items())
+            )
+            raise UnknownModel(
+                f"role(s) point at models this LiteLLM key cannot see: {named}; "
+                f"available: {', '.join(self.aliases()) or '(none)'}"
+            )
+
+    def pinned(self, roles: Mapping[str, str]) -> dict[str, dict[str, Any]]:
+        """The resolved mapping to record at task start (A11, A16).
+
+        Recording only the alias would let an alias that silently changes
+        models muddle the eval history behind it: "remote-architecture: 80%
+        good" averaged over however many models it pointed at is not a fact
+        about anything.
+        """
+        return {
+            role: {
+                "alias": alias,
+                "backing": self.facts(alias).backing,
+                "provider": self.facts(alias).provider,
+            }
+            for role, alias in roles.items()
+        }
+
+
+def select_client(
+    model: str,
+    normalized: LLMClient,
+    passthrough: LLMClient,
+    catalog: ModelCatalog,
+) -> LLMClient:
+    """Choose the route from what the model *is*, not from what it is called.
+
+    A13 spells out the rule: use the vendor-native adapter where one exists,
+    and otherwise provisionally accept LiteLLM's normalization for that
+    vendor -- validating it against the A16 eval set before trusting it with
+    delegations. There is deliberately no name-based fallback here: a second
+    heuristic could disagree with the proxy, and an unknown model raises
+    rather than quietly taking the normalized route.
+    """
+    return passthrough if catalog.facts(model).native_anthropic else normalized

nisse-dev/src/nisse_dev/config.py

@@ -0,0 +1,123 @@
+"""Agent configuration (spec A5: from env/.env, nothing hardcoded).
+
+Every value is collected and validated **before** anything is built, and a
+single error names every missing variable at once. The alternative — failing
+at first use — surfaces a missing token twenty minutes into a milestone, on
+the call that mattered.
+"""
+
+import os
+from collections.abc import Mapping
+from dataclasses import dataclass
+from pathlib import Path
+
+from nisse_dev.llm import RoleMap
+from nisse_dev.matrix import MatrixConfig
+
+REQUIRED = (
+    "NISSE_STATE_DB",
+    "NISSE_WORKSPACE",
+    "NISSE_PROJECT",
+    "NISSE_LLM_BASE_URL",
+    "NISSE_LLM_TOKEN",
+    "NISSE_ROLE_EXECUTOR",
+    "NISSE_ROLE_PLANNER",
+    "NISSE_ROLE_DELEGATE",
+    "NISSE_ROLE_CONSULTANT",
+    "NISSE_MATRIX_HOMESERVER",
+    "NISSE_MATRIX_USER_ID",
+    "NISSE_MATRIX_TOKEN",
+    "NISSE_MATRIX_ROOM_ID",
+    "NISSE_MATRIX_OPERATORS",
+    "NISSE_GITEA_URL",
+    "NISSE_GITEA_TOKEN",
+    "NISSE_GITEA_REPO",
+    "NISSE_GITEA_OPERATORS",
+)
+
+
+class ConfigError(RuntimeError):
+    pass
+
+
+def _people(raw: str) -> frozenset[str]:
+    """Operator allowlists are comma-separated; blanks are dropped so a
+    trailing comma cannot admit an empty identity."""
+    return frozenset(part.strip() for part in raw.split(",") if part.strip())
+
+
+@dataclass(frozen=True)
+class AgentConfig:
+    state_db: Path
+    workspace: Path
+    project: str
+    llm_base_url: str
+    llm_token: str
+    roles: RoleMap
+    matrix: MatrixConfig
+    operators: frozenset[str]
+    gitea_url: str
+    gitea_token: str
+    gitea_repo: str
+    gitea_operators: frozenset[str]
+    base_branch: str = "main"
+
+    @classmethod
+    def from_env(cls, env: Mapping[str, str] = os.environ) -> "AgentConfig":
+        # Whitespace-only counts as missing: a token of spaces would pass a
+        # truthiness check and fail at the first request instead.
+        missing = [name for name in REQUIRED if not (env.get(name) or "").strip()]
+        if missing:
+            raise ConfigError(
+                "cannot start: these are unset or empty — " + ", ".join(missing)
+            )
+        config = cls(
+            state_db=Path(env["NISSE_STATE_DB"]).expanduser(),
+            workspace=Path(env["NISSE_WORKSPACE"]).expanduser(),
+            project=env["NISSE_PROJECT"],
+            llm_base_url=env["NISSE_LLM_BASE_URL"],
+            llm_token=env["NISSE_LLM_TOKEN"],
+            roles=RoleMap.from_env(env),
+            matrix=MatrixConfig.from_env(env),
+            operators=_people(env["NISSE_MATRIX_OPERATORS"]),
+            gitea_url=env["NISSE_GITEA_URL"],
+            gitea_token=env["NISSE_GITEA_TOKEN"],
+            gitea_repo=env["NISSE_GITEA_REPO"],
+            gitea_operators=_people(env["NISSE_GITEA_OPERATORS"]),
+            base_branch=env.get("NISSE_BASE_BRANCH", "main"),
+        )
+        config._refuse_secrets_inside_the_clone(env)
+        return config
+
+    def _refuse_secrets_inside_the_clone(self, env: Mapping[str, str]) -> None:
+        """The workspace is a git clone the agent commits with ``git add -A``
+        and pushes. Anything inside it reaches the project repository, so a
+        state store or an env file placed there would publish credentials as a
+        side effect of the agent working correctly — silently, and once.
+        """
+        workspace = self.workspace.expanduser().resolve()
+        offenders = []
+        for label, path in (
+            ("NISSE_STATE_DB", self.state_db),
+            ("NISSE_ENV", Path(env.get("NISSE_ENV", ".env"))),
+        ):
+            resolved = path.expanduser().resolve()
+            if resolved == workspace or workspace in resolved.parents:
+                offenders.append(f"{label} ({resolved})")
+        if offenders:
+            raise ConfigError(
+                "refusing to start: these sit inside the workspace clone at "
+                f"{workspace} and would be committed and pushed to the project "
+                "repository — " + ", ".join(offenders)
+            )
+
+    def redacted(self) -> dict[str, str]:
+        """What the agent may safely announce about how it is configured."""
+        return {
+            "project": self.project,
+            "workspace": str(self.workspace),
+            "repo": self.gitea_repo,
+            "planner": self.roles.planner,
+            "executor": self.roles.executor,
+            "delegate": self.roles.delegate,
+        }

nisse-dev/src/nisse_dev/control.py

@@ -143,6 +143,10 @@ def _fresh_data() -> dict[str, Any]:
         "milestone": 1,
         "task_id": None,
         "roles": None,  # A11 mapping, pinned at task start
+        # What each pinned alias resolved to, discovered from LiteLLM. An
+        # alias is free to point somewhere else tomorrow, so the transcript
+        # has to say what actually answered (A11, A16).
+        "models": None,
         "pr": None,  # the open PR for the current milestone, once published
         # Pointer to the Gitea plan record (Part H): the state store holds
         # only this, so gate state can be re-derived from Gitea after a
@@ -163,11 +167,15 @@ class Orchestrator:
         publisher: "MilestonePublisher | None" = None,
         gitea_operators: Iterable[str] = (),
         board: "PlanBoard | None" = None,
+        delegate: Any = None,
+        poller_factory: Any = None,
+        catalog: Any = None,
     ) -> None:
         self._store = store
         self._transport = transport
         self._agent = agent
         self._roles = roles
+        self._catalog = catalog
         self._executor = executor
         self._publisher = publisher
         self._operators = set(operators)
@@ -176,6 +184,12 @@ class Orchestrator:
         # the other.
         self._gitea_operators = set(gitea_operators)
         self._board = board
+        self._delegate = delegate
+        # Builds a poller bound to a comment handler; the orchestrator
+        # owns *when* to watch, which is a state-machine concern, while
+        # the factory owns how to build one.
+        self._poller_factory = poller_factory
+        self._pollers: dict[str, Any] = {}
         # Transport and the comment poller deliver on separate threads;
         # state transitions must not interleave.
         self._lock = threading.RLock()
@@ -184,6 +198,11 @@ class Orchestrator:
         transport.on_message(self._handle)
 
     def start(self, reconcilers: Mapping[str, Reconciler] | None = None) -> None:
+        # A3 step 5 runs before the generic sweep: an interrupted delegation is
+        # resumed or discarded by inspecting the branch and the log, and only a
+        # genuinely unclean cut is left for the sweep below to gate on.
+        if self._delegate is not None:
+            self._delegate.recover(self._transport.send_message)
         # The publisher owns the external-truth query for its own intents, so
         # a configured publisher always contributes its reconciler (A3).
         registered: dict[str, Reconciler] = dict(reconcilers or {})
@@ -274,7 +293,17 @@ class Orchestrator:
         # 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}
+        resolved = (
+            self._catalog.pinned(pinned)
+            if self._catalog is not None and pinned is not None
+            else None
+        )
+        self._data = {
+            **_fresh_data(),
+            "task_id": task_id,
+            "roles": pinned,
+            "models": resolved,
+        }
         self._project = project
         self._store.oplog.task_id = task_id
         if self._board is not None:
@@ -458,6 +487,9 @@ class Orchestrator:
             # durability of the record, not the ability to approve.
             return
         board_state["comment_id"] = post.comment_id
+        # Approvals can arrive on the issue as well as over transport (A4),
+        # so watch it for as long as the gate stands.
+        self._watch("plan", board_state["issue"], self.handle_plan_comment)
 
     def handle_plan_comment(self, comment: Any) -> None:
         """A comment on the project issue may satisfy an open plan gate (A4).
@@ -579,6 +611,8 @@ class Orchestrator:
             "url": result.pull_request.url,
             "milestone": milestone,
         }
+        self._unwatch("plan")  # the gate is behind us; the PR is the thread now
+        self._watch("review", result.pull_request.number, self.handle_review_comment)
         self._transition(PR_OPEN, cause=f"milestone {milestone} published")
         self._transport.send_message(
             f"{result.note} — review it, then say 'next milestone' or "
@@ -587,12 +621,14 @@ class Orchestrator:
 
     def _handle_pr_open(self, text: str) -> None:
         if text == "next milestone":
+            self._unwatch("review")
             self._data["milestone"] = int(self._data.get("milestone", 1)) + 1
             self._data["gate"] = None
             self._data["pr"] = None
             self._transition(MILESTONE_PLANNING, cause="operator: next milestone")
             self._run_planning_turn(f"milestone-{self._data['milestone']}")
         elif text == "project complete":
+            self.stop_watching()
             self._data = {**_fresh_data()}
             self._project = None
             self._store.oplog.task_id = None
@@ -607,6 +643,27 @@ class Orchestrator:
                 "'project complete'"
             )
 
+    def _watch(self, kind: str, index: int, handler: Any) -> None:
+        """Start watching one Gitea thread, replacing any previous watch of
+        the same kind. Priming first, so attaching does not replay comments
+        already addressed."""
+        if self._poller_factory is None:
+            return
+        self._unwatch(kind)
+        poller = self._poller_factory(handler)
+        poller.prime(index)
+        poller.start(index)
+        self._pollers[kind] = poller
+
+    def _unwatch(self, kind: str) -> None:
+        poller = self._pollers.pop(kind, None)
+        if poller is not None:
+            poller.stop()
+
+    def stop_watching(self) -> None:
+        for kind in list(self._pollers):
+            self._unwatch(kind)
+
     def _await_input(self, *, cause: str, prompt: str, pending: dict[str, Any]) -> None:
         self._data["pending"] = pending
         self._transition(AWAIT_INPUT, cause=cause)

nisse-dev/src/nisse_dev/delegation.py

@@ -36,8 +36,9 @@ from nisse_dev.canonical import (
     ToolResult,
 )
 from nisse_dev.git import GitError, GitRepo
-from nisse_dev.llm import LLMClient, LLMError
+from nisse_dev.llm import LLMClient, LLMError, UpstreamUnavailable
-from nisse_dev.state import StateStore
+from nisse_dev.reconcile import Reconciler
+from nisse_dev.state import Intent, StateStore
 from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, Workspace
 
 DELEGATION_INTENT = "delegation"
@@ -128,7 +129,7 @@ class Handback:
 
 @dataclass
 class DelegationResult:
-    status: Literal["handback", "capped", "failed"]
+    status: Literal["handback", "capped", "failed", "interrupted"]
     note: str
     handback: Handback | None = None
     branch: str | None = None
@@ -136,6 +137,17 @@ class DelegationResult:
     tokens: int = 0
     duration_s: float = 0.0
 
+    @property
+    def interrupted(self) -> bool:
+        """Stopped by something external rather than by the delegate.
+
+        The work on the branch may be perfectly good, so it is preserved and
+        the intent left pending: the next boot treats it exactly like a crash
+        mid-delegation (A3) and resumes if the cut was clean. Discarding here
+        would throw away work because the budget ran out.
+        """
+        return self.status == "interrupted"
+
     @property
     def usable(self) -> bool:
         """Only a successful handback may be built upon. A cap breach or a
@@ -153,6 +165,7 @@ class Delegation:
         git_repo: GitRepo,
         *,
         model: str,
+        backing_model: str = "",
         caps: Caps | None = None,
         now: Callable[[], float] = time.monotonic,
     ) -> None:
@@ -161,6 +174,7 @@ class Delegation:
         self._workspace = workspace
         self._git = git_repo
         self._model = model
+        self._backing_model = backing_model
         self._caps = caps or Caps()
         self._now = now
 
@@ -169,6 +183,19 @@ class Delegation:
     ) -> DelegationResult:
         started_at = self._now()
         base_branch = self._git.current_branch()
+        # Recorded before the branch exists, so a crash anywhere after this
+        # leaves a durable statement of what was in flight (A2/A3).
+        intent_id = self._store.begin_intent(
+            DELEGATION_INTENT,
+            {
+                "branch": branch,
+                "base": base_branch,
+                "brief": brief,
+                "role": role,
+                "caps": self._caps.as_payload(),
+            },
+            cause=f"delegating to {self._model}",
+        )
         try:
             # Commit whatever the caller had in flight onto their own branch
             # first. Otherwise it would ride along onto the delegation branch
@@ -176,6 +203,7 @@ class Delegation:
             self._git.commit_all("Checkpoint before delegating")
             self._git.checkout_new_branch(branch)
         except GitError as exc:
+            self._store.abort_intent(intent_id, cause=f"could not branch: {exc}")
             return DelegationResult("failed", f"could not branch for delegation: {exc}")
 
         self._store.append_event(
@@ -184,6 +212,7 @@ class Delegation:
                 role=role,
                 model=self._model,
                 caps=self._caps.as_payload(),
+                backing_model=self._backing_model,
             )
         )
         result = self._loop(brief, branch, started_at)
@@ -197,12 +226,86 @@ class Delegation:
                 exit_status=result.status,
             )
         )
+        if result.interrupted:
+            # Leave the branch and the intent alone: this delegation is not
+            # finished, it was cut off, and A3's resume rule is exactly the
+            # machinery for picking it back up.
+            return result
         if not result.usable:
             # Abort discards the branch: the caller must never resume on top
             # of a partially-edited tree (A13).
             self._abandon(branch, base_branch)
+        # Settled either way: an aborted delegation is a *finished* one, and
+        # leaving the intent pending would make the next boot think work was
+        # still in flight.
+        self._store.commit_intent(intent_id, cause=f"delegation {result.status}")
         return result
 
+    def resume(self, intent: Intent) -> DelegationResult:
+        """Continue a delegation the process died in the middle of (A3 step 5).
+
+        The transcript is already in the canonical log, so the loop re-renders
+        it and carries on. No new branch and no second ``delegation_started``:
+        this is the same delegation, which is why its vendor-opaque reasoning
+        may still be replayed (A10 defines a session as the delegation, not the
+        process lifetime).
+        """
+        payload = intent.payload
+        branch = str(payload["branch"])
+        started_at = self._now()
+        try:
+            self._git.checkout_existing_branch(branch)
+        except GitError as exc:
+            self._store.commit_intent(
+                intent.id, cause=f"delegation branch unusable: {exc}"
+            )
+            return DelegationResult("failed", f"could not resume on {branch}: {exc}")
+
+        result = self._loop(str(payload.get("brief", "")), branch, started_at)
+        result.duration_s = self._now() - started_at
+        self._store.append_event(
+            DelegationCompleted(
+                handback=result.handback.render() if result.handback else result.note,
+                artifacts=result.artifacts,
+                tokens=result.tokens,
+                duration_s=result.duration_s,
+                exit_status=result.status,
+            )
+        )
+        if not result.usable:
+            self._abandon(branch, str(payload.get("base", "main")))
+        self._store.commit_intent(intent.id, cause=f"delegation {result.status}")
+        return result
+
+    def recover(self, notify: Callable[[str], None]) -> DelegationResult | None:
+        """Sweep pending delegation intents at boot (A3 step 5).
+
+        Settles what is already finished, discards what never started, resumes
+        a clean cut — announcing it with a *notice*, not a question, because a
+        clean cut needs no decision from the operator — and leaves an unclean
+        one pending so the generic boot sweep gates on it.
+        """
+        for intent in self._store.pending_intents():
+            if intent.kind != DELEGATION_INTENT:
+                continue
+            verdict = delegation_verdict(self._store, self._git, intent)
+            branch = intent.payload.get("branch")
+            if verdict == "settled":
+                self._store.commit_intent(
+                    intent.id, cause="delegation had already completed"
+                )
+            elif verdict == "absent":
+                self._store.abort_intent(
+                    intent.id, cause=f"delegation branch {branch} never created"
+                )
+            elif verdict == "resume":
+                notify(
+                    f"resuming the delegation interrupted on {branch}; "
+                    "the transcript and branch are intact"
+                )
+                return self.resume(intent)
+        return None
+
     def _loop(self, brief: str, branch: str, started_at: float) -> DelegationResult:
         system = (
             f"{SYSTEM_PROMPT}\n\n{self._workspace.describe()}"
@@ -210,6 +313,7 @@ class Delegation:
         )
         touched: set[str] = set()
         did_work = False
+        spent = 0
 
         for turn in range(self._caps.turns):
             elapsed = self._now() - started_at
@@ -220,6 +324,7 @@ class Delegation:
                     f"after {turn} turns",
                     branch=branch,
                     artifacts=sorted(touched),
+                    tokens=spent,
                 )
             try:
                 produced = self._client.complete(
@@ -228,9 +333,35 @@ class Delegation:
                     model=self._model,
                     system=system,
                 )
+            except UpstreamUnavailable as exc:
+                return DelegationResult(
+                    "interrupted",
+                    f"stopped by something upstream, not by the work: {exc}",
+                    branch=branch,
+                    artifacts=sorted(touched),
+                    tokens=spent,
+                )
             except LLMError as exc:
                 return DelegationResult(
-                    "failed", f"delegate call failed: {exc}", branch=branch
+                    "failed",
+                    f"delegate call failed: {exc}",
+                    branch=branch,
+                    tokens=spent,
+                )
+            # Metered from what the provider reported, not estimated. The cap
+            # is a scope limit (A13), distinct from the money cap A5 keeps on
+            # the LiteLLM key — but an unbounded delegation can still exhaust
+            # a daily budget and take external models away until it resets.
+            usage = getattr(self._client, "last_usage", {}) or {}
+            spent += int(usage.get("input", 0)) + int(usage.get("output", 0))
+            if spent >= self._caps.tokens:
+                return DelegationResult(
+                    "capped",
+                    f"token cap of {self._caps.tokens} reached ({spent} spent) "
+                    f"after {turn + 1} turns",
+                    branch=branch,
+                    artifacts=sorted(touched),
+                    tokens=spent,
                 )
 
             calls: list[ToolCall] = []
@@ -252,6 +383,7 @@ class Delegation:
                     "delegate produced neither a tool call nor text",
                     branch=branch,
                     artifacts=sorted(touched),
+                    tokens=spent,
                 )
 
             for call in calls:
@@ -301,6 +433,7 @@ class Delegation:
                         handback=handback if handback.succeeded else None,
                         branch=branch,
                         artifacts=sorted(touched),
+                        tokens=spent,
                     )
 
                 outcome = self._workspace.dispatch(call.tool, call.arguments)
@@ -327,6 +460,7 @@ class Delegation:
             f"turn cap of {self._caps.turns} reached without a handback",
             branch=branch,
             artifacts=sorted(touched),
+            tokens=spent,
         )
 
     def _commit_turn(self, turn: int) -> None:
@@ -361,3 +495,64 @@ def _parse_handback(arguments: dict[str, Any]) -> Handback | None:
         next=str(arguments["next"]).strip(),
         succeeded=succeeded,
     )
+
+
+def delegation_reconciler(store: StateStore, git_repo: GitRepo) -> Reconciler:
+    """A3 step 5: decide whether an interrupted delegation may be resumed.
+
+    The judgement is an observable check, never model self-assessment — the
+    same reasoning as A12. Returns:
+
+    - ``True``  the delegation is settled or cleanly resumable
+    - ``False`` it never really started, so there is nothing to resume
+    - ``None``  the cut is unclean, so boot gates to the operator
+
+    "The branch sits at the expected commit" is approximated by the branch
+    existing and carrying the delegation's commits: per-turn shas are not
+    recorded in the log today, so a branch rewritten out from under us would
+    not be detected. Stated rather than glossed, because the check is weaker
+    than A3's wording implies.
+    """
+
+    def check(intent: Intent) -> bool | None:
+        verdict = delegation_verdict(store, git_repo, intent)
+        if verdict in ("resume", "settled"):
+            return True
+        return False if verdict == "absent" else None
+
+    return check
+
+
+def delegation_verdict(
+    store: StateStore, git_repo: GitRepo, intent: Intent
+) -> Literal["resume", "settled", "absent", "gate"]:
+    """The A3 step 5 decision, separated from the Reconciler contract because
+    boot must *act* on it — resuming is not the same as judging settled."""
+    branch = intent.payload.get("branch")
+    if not branch:
+        return "gate"
+    if _completed_after_start(store):
+        return "settled"  # it finished; the intent merely outlived its commit
+    if not git_repo.branch_exists(str(branch)):
+        return "absent"  # the crash landed before any work existed
+    events = store.read_events()
+    if not events:
+        return "gate"
+    last = events[-1].event
+    if isinstance(last, ToolCall):
+        # A call with no recorded result: its effect is unknown and cannot be
+        # verified by inspection, so gate (A3).
+        return "gate"
+    if not isinstance(last, AssistantText | ToolResult):
+        return "gate"
+    return "resume"
+
+
+def _completed_after_start(store: StateStore) -> bool:
+    """True when the last delegation event in the log is a completion."""
+    for stored in reversed(store.read_events()):
+        if isinstance(stored.event, DelegationCompleted):
+            return True
+        if isinstance(stored.event, DelegationStarted):
+            return False
+    return False

nisse-dev/src/nisse_dev/evals.py

@@ -0,0 +1,88 @@
+"""Delegation and consult evaluation capture (spec A16).
+
+Each ``delegation_started`` / ``delegation_completed`` pair is already a clean
+input-output record; this pairs them up and attaches a verdict, so "best model
+this week" can be decided by re-running a captured set rather than by benchmark
+chatter. Without it, swapping the frontier tier becomes something to avoid,
+because the effect is unmeasurable.
+
+Verdicts live outside the canonical log: the log records what happened, a
+verdict is a later judgement about it, and judgements get revised.
+"""
+
+import json
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from nisse_dev.canonical import DelegationCompleted, DelegationStarted
+from nisse_dev.state import StateStore
+
+UNMARKED = "unmarked"
+
+
+@dataclass(frozen=True)
+class EvalRecord:
+    event_id: int  # of the completion, which is what a verdict attaches to
+    role: str
+    model: str
+    brief: str
+    caps: dict[str, Any]
+    handback: str
+    artifacts: list[str]
+    tokens: int
+    duration_s: float
+    exit_status: str
+    verdict: str = UNMARKED
+    note: str = ""
+
+
+def collect(store: StateStore, after: int = 0) -> list[EvalRecord]:
+    """Pair each start with the completion that follows it.
+
+    A start with no completion is skipped rather than reported half-formed:
+    it means the process died mid-delegation, which A3 handles, and an
+    unfinished record would pollute the eval set.
+    """
+    verdicts = store.outcomes()
+    records: list[EvalRecord] = []
+    pending: tuple[int, DelegationStarted] | None = None
+    for stored in store.read_events(after=after):
+        if isinstance(stored.event, DelegationStarted):
+            pending = (stored.id, stored.event)
+        elif isinstance(stored.event, DelegationCompleted) and pending is not None:
+            _, started = pending
+            verdict, note = verdicts.get(stored.id, (UNMARKED, ""))
+            records.append(
+                EvalRecord(
+                    event_id=stored.id,
+                    role=started.role,
+                    model=started.model,
+                    brief=started.brief,
+                    caps=started.caps,
+                    handback=stored.event.handback,
+                    artifacts=stored.event.artifacts,
+                    tokens=stored.event.tokens,
+                    duration_s=stored.event.duration_s,
+                    exit_status=stored.event.exit_status,
+                    verdict=verdict,
+                    note=note,
+                )
+            )
+            pending = None
+    return records
+
+
+def to_jsonl(records: list[EvalRecord]) -> str:
+    """JSON lines, so a captured set can be diffed and re-run per record."""
+    return "\n".join(json.dumps(asdict(r)) for r in records)
+
+
+def summarise(records: list[EvalRecord]) -> dict[str, Any]:
+    by_model: dict[str, dict[str, int]] = {}
+    for record in records:
+        bucket = by_model.setdefault(
+            record.model, {"total": 0, "good": 0, "bad": 0, UNMARKED: 0}
+        )
+        bucket["total"] += 1
+        bucket[record.verdict if record.verdict in ("good", "bad") else UNMARKED] += 1
+    return by_model

nisse-dev/src/nisse_dev/executor.py

@@ -10,13 +10,16 @@ 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 collections.abc import Callable
 from dataclasses import dataclass
-from typing import Literal
+from typing import Any, Literal
 
 from nisse_dev.canonical import ToolCall, ToolResult
+from nisse_dev.delegation import Delegation
+from nisse_dev.expert import ASK_EXPERT_TOOL, BudgetExhausted, ExpertDesk
 from nisse_dev.llm import LLMClient, LLMError, RoleMap
 from nisse_dev.state import StateStore
-from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, Workspace
+from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, ToolOutcome, Workspace
 
 EXECUTOR_SYSTEM = """You are nisse-dev's executor, doing mechanical work.
 
@@ -56,6 +59,9 @@ class LLMExecutor:
         *,
         max_turns: int = DEFAULT_MAX_TURNS,
         failure_limit: int = DEFAULT_FAILURE_LIMIT,
+        expert: ExpertDesk | None = None,
+        delegate: Delegation | None = None,
+        notify: Callable[[str], None] | None = None,
     ) -> None:
         self._client = client
         self._store = store
@@ -63,6 +69,76 @@ class LLMExecutor:
         self._workspace = workspace
         self._max_turns = max_turns
         self._failure_limit = failure_limit
+        self._expert = expert
+        self._delegate = delegate
+        # The loop runs synchronously inside the transport callback, so
+        # without this the agent is silent for the whole milestone and a
+        # working run looks exactly like a hung one.
+        self._notify = notify
+
+    def _report(self, turn: int, calls: list[ToolCall]) -> None:
+        if self._notify is None:
+            return
+        summary = ", ".join(
+            f"{c.tool}({c.arguments.get('path') or c.arguments.get('command', '')[:40]})"
+            if c.arguments
+            else c.tool
+            for c in calls
+        )
+        self._notify(f"turn {turn + 1}: {summary}")
+
+    def _escalate(
+        self, milestone: int, failures: int, last_failure: str
+    ) -> ExecutorOutcome | None:
+        """Hand the sub-task to a stronger model, per A12.
+
+        Returns None when the delegation handed back usable work and the local
+        loop should carry on; an ExecutorOutcome when the caller must stop.
+        """
+        note = f"{failures} consecutive command failures; last: {last_failure[:500]}"
+        if self._delegate is None:
+            return ExecutorOutcome("escalate", note)
+
+        # A12 says the failures go in the brief: the delegate must not have to
+        # rediscover what the local model already learned the hard way.
+        brief = (
+            f"Milestone {milestone} is stuck. The local executor tried and its "
+            f"commands failed {failures} times in a row.\n\n"
+            f"Most recent failure:\n{last_failure[:2000]}\n\n"
+            "The approved plan and the full transcript are in the conversation "
+            "above. Finish this sub-task, staying inside the approved plan."
+        )
+        result = self._delegate.run(
+            brief=brief, branch=f"delegation/milestone-{milestone}-{failures}"
+        )
+        if result.usable:
+            return None
+        # A13: a cap breach or failure leaves a half-edited tree, so the local
+        # loop must not resume on top of it.
+        return ExecutorOutcome(
+            "escalate", f"{note}; delegation {result.status}: {result.note}"
+        )
+
+    def _tools(self) -> list[dict[str, Any]]:
+        # A12: ask_expert exists in addition to the observable escalation
+        # triggers, never as a replacement for them.
+        if self._expert is None:
+            return EXECUTOR_TOOLS
+        return [*EXECUTOR_TOOLS, ASK_EXPERT_TOOL]
+
+    def _consult(self, arguments: dict[str, Any]) -> ToolOutcome:
+        assert self._expert is not None
+        question = str(arguments.get("question", "")).strip()
+        if not question:
+            return ToolOutcome("ask_expert needs a question", success=False)
+        try:
+            return ToolOutcome(self._expert.ask(question).render(), success=True)
+        except BudgetExhausted as exc:
+            # Not an error the model should route around: it is the signal
+            # that it is circling and should get on with the work or fail.
+            return ToolOutcome(str(exc), success=False)
+        except LLMError as exc:
+            return ToolOutcome(f"the consultant did not answer: {exc}", success=False)
 
     def run(self, milestone: int) -> ExecutorOutcome:
         model = self._roles.resolve("executor")
@@ -74,11 +150,11 @@ class LLMExecutor:
         did_work = False
         challenged = False
 
-        for _turn in range(self._max_turns):
+        for turn in range(self._max_turns):
             try:
                 produced = self._client.complete(
                     self._store.read_events(),
-                    tools=EXECUTOR_TOOLS,
+                    tools=self._tools(),
                     model=model,
                     system=system,
                 )
@@ -107,9 +183,13 @@ class LLMExecutor:
                     "executor claimed the milestone was done without making any change",
                 )
 
+            self._report(turn, calls)
             for call in calls:
                 if call.tool in MUTATING_TOOLS:
                     did_work = True
+                if call.tool == "ask_expert":
+                    outcome = self._consult(call.arguments)
+                else:
                     outcome = self._workspace.dispatch(call.tool, call.arguments)
                 self._store.append_event(
                     ToolResult(
@@ -125,11 +205,18 @@ class LLMExecutor:
                         0 if outcome.success else consecutive_failures + 1
                     )
                     if consecutive_failures >= self._failure_limit:
-                        return ExecutorOutcome(
+                        # A12: this is the trigger delegation exists for.
-                            "escalate",
+                        # Escalating to the human is the fallback when there
-                            f"{consecutive_failures} consecutive command failures; "
+                        # is no delegate, not the intended path.
-                            f"last: {outcome.result[:500]}",
+                        escalation = self._escalate(
+                            milestone,
+                            consecutive_failures,
+                            outcome.result,
                         )
+                        if escalation is not None:
+                            return escalation
+                        consecutive_failures = 0
+                        did_work = True
 
         return ExecutorOutcome(
             "capped", f"hit the {self._max_turns}-turn cap for milestone {milestone}"

nisse-dev/src/nisse_dev/expert.py

@@ -0,0 +1,214 @@
+"""`ask_expert` (spec A14).
+
+The one case delegation does not cover: a question needing no repo access —
+"which of these two approaches is better", "is this API deprecated". A single
+stateless call, a text answer, no session state, no diff application.
+
+Two constraints from A14 shape the implementation:
+
+**Structured output is contracted at the prompt level, not the API level.**
+Vendors differ on JSON mode and schema enforcement, so the answer is parsed and
+validated against a local schema, with one retry. This runs over LiteLLM's
+*normalized* route, which A13 reserves for calls where fidelity is cheap.
+
+**A per-task budget**, so a confused executor cannot spend twenty consults
+circling. The count comes from the canonical log rather than a counter in
+memory: it survives a restart, and it is observable from outside the model,
+which is the same reasoning A12 applies to escalation triggers.
+
+Consults are recorded as ``delegation_started`` / ``delegation_completed`` with
+role ``consultant``. That is deliberate — A16 wants a "consult/delegation eval
+set", and reusing the composite events gets consult evaluation for free rather
+than inventing a parallel record.
+"""
+
+import json
+import re
+from dataclasses import dataclass
+
+from nisse_dev.canonical import (
+    DelegationCompleted,
+    DelegationStarted,
+    StoredEvent,
+    UserMessage,
+)
+from nisse_dev.llm import LLMClient, LLMError, RoleMap
+from nisse_dev.state import StateStore
+
+CONSULTANT_ROLE = "consultant"
+DEFAULT_BUDGET = 3
+
+SYSTEM_PROMPT = """You are a consultant answering one self-contained question.
+
+You have no access to the caller's repository, so answer from the question
+alone. If the question cannot be answered without seeing the code, say so
+plainly in `answer` rather than guessing.
+
+Reply with a single JSON object and nothing else:
+
+{"answer": "...", "confidence": "high" | "medium" | "low", "caveats": "..."}
+
+Keep `answer` direct and specific. Put what would change your mind, or what you
+are assuming, in `caveats`."""
+
+_RETRY_NUDGE = (
+    "\n\nYour previous reply was not a single valid JSON object with the keys "
+    "answer, confidence and caveats. Reply with only that object."
+)
+_CONFIDENCES = {"high", "medium", "low"}
+
+
+ASK_EXPERT_TOOL: dict = {
+    "type": "function",
+    "function": {
+        "name": "ask_expert",
+        "description": (
+            "Ask a stronger model one self-contained question that needs no "
+            "access to this repository — a design trade-off, or how a library "
+            "behaves. It cannot see your files. Use sparingly; the budget is "
+            "small."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {"question": {"type": "string"}},
+            "required": ["question"],
+        },
+    },
+}
+
+
+class BudgetExhausted(RuntimeError):
+    """Raised rather than silently answering: a caller that has burned its
+    consult budget is circling, and should escalate or ask the operator."""
+
+
+@dataclass(frozen=True)
+class ExpertAnswer:
+    answer: str
+    confidence: str
+    caveats: str
+
+    def render(self) -> str:
+        body = f"{self.answer}\n\n(confidence: {self.confidence})"
+        return f"{body}\n\nCaveats: {self.caveats}" if self.caveats else body
+
+
+class ExpertDesk:
+    def __init__(
+        self,
+        client: LLMClient,
+        store: StateStore,
+        roles: RoleMap,
+        *,
+        budget: int = DEFAULT_BUDGET,
+    ) -> None:
+        self._client = client
+        self._store = store
+        self._roles = roles
+        self._budget = budget
+
+    def consults_used(self, since_event_id: int = 0) -> int:
+        return sum(
+            1
+            for e in self._store.read_events(after=since_event_id)
+            if isinstance(e.event, DelegationStarted)
+            and e.event.role == CONSULTANT_ROLE
+        )
+
+    def ask(self, question: str, *, since_event_id: int = 0) -> ExpertAnswer:
+        used = self.consults_used(since_event_id)
+        if used >= self._budget:
+            raise BudgetExhausted(
+                f"consult budget of {self._budget} is spent for this task "
+                f"({used} used); escalate or ask the operator instead"
+            )
+        model = self._roles.resolve(CONSULTANT_ROLE)
+        self._store.append_event(
+            DelegationStarted(
+                brief=question,
+                role=CONSULTANT_ROLE,
+                model=model,
+                caps={"consults": self._budget, "used": used},
+            )
+        )
+        answer, note = self._answer(question, model)
+        self._store.append_event(
+            DelegationCompleted(
+                handback=answer.render() if answer else note,
+                artifacts=[],
+                tokens=0,
+                duration_s=0.0,
+                exit_status="answered" if answer else "failed",
+            )
+        )
+        if answer is None:
+            raise LLMError(f"ask_expert produced no usable answer: {note}")
+        return answer
+
+    def _answer(self, question: str, model: str) -> tuple[ExpertAnswer | None, str]:
+        system = SYSTEM_PROMPT
+        last = "no reply"
+        for attempt in (1, 2):
+            try:
+                # Stateless by design (A14): the question travels as a
+                # synthesised message that is never appended to the log, so a
+                # consult neither depends on nor leaks repo context. It cannot
+                # ride in the system prompt alone — the Messages API requires
+                # at least one non-system message.
+                produced = self._client.complete(
+                    [
+                        StoredEvent(
+                            id=0,
+                            ts="",
+                            event=UserMessage(text=question, source="ask_expert"),
+                        )
+                    ],
+                    tools=None,
+                    model=model,
+                    system=system,
+                )
+            except LLMError as exc:
+                return None, str(exc)
+            text = "\n".join(
+                getattr(e, "text", "") for e in produced if hasattr(e, "text")
+            ).strip()
+            parsed = parse_answer(text)
+            if parsed is not None:
+                return parsed, "ok"
+            last = text[:300] or "empty reply"
+            if attempt == 1:
+                system = SYSTEM_PROMPT + _RETRY_NUDGE
+        return None, f"unparseable reply: {last}"
+
+
+def parse_answer(text: str) -> ExpertAnswer | None:
+    """Validated locally because the prompt-level contract is all we have.
+
+    Tolerates a fenced code block, since models wrap JSON in one habitually,
+    but not missing or malformed fields — those are a failed turn.
+    """
+    candidate = text.strip()
+    fenced = re.search(r"```(?:json)?\s*(.*?)```", candidate, re.DOTALL)
+    if fenced:
+        candidate = fenced.group(1).strip()
+    start, end = candidate.find("{"), candidate.rfind("}")
+    if start == -1 or end <= start:
+        return None
+    try:
+        data = json.loads(candidate[start : end + 1])
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(data, dict):
+        return None
+    answer = data.get("answer")
+    confidence = data.get("confidence")
+    if not isinstance(answer, str) or not answer.strip():
+        return None
+    if not isinstance(confidence, str) or confidence.lower() not in _CONFIDENCES:
+        return None
+    caveats = data.get("caveats")
+    return ExpertAnswer(
+        answer=answer.strip(),
+        confidence=confidence.lower(),
+        caveats=caveats.strip() if isinstance(caveats, str) else "",
+    )

nisse-dev/src/nisse_dev/git.py

@@ -59,6 +59,10 @@ class GitRepo:
         self.root.mkdir(parents=True, exist_ok=True)
         self._run("init", "-q", "-b", default_branch)
 
+    def clone(self, remote_url: str, dest: Path, *, token: str | None = None) -> None:
+        self.root.mkdir(parents=True, exist_ok=True)
+        self._run("clone", "-q", remote_url, str(dest), secret=token)
+
     def configure_identity(self, name: str, email: str) -> None:
         self._run("config", "user.name", name)
         self._run("config", "user.email", email)
@@ -66,6 +70,16 @@ class GitRepo:
     def checkout_new_branch(self, name: str) -> None:
         self._run("checkout", "-q", "-b", name)
 
+    def checkout_existing_branch(self, name: str) -> None:
+        self._run("checkout", "-q", name)
+
+    def branch_exists(self, name: str) -> bool:
+        try:
+            self._run("rev-parse", "--verify", "--quiet", f"refs/heads/{name}")
+        except GitError:
+            return False
+        return True
+
     def commit_all(self, message: str) -> str | None:
         """Returns the new commit sha, or None when there was nothing to
         commit — an empty milestone is a fact to report, not an error."""

nisse-dev/src/nisse_dev/llm.py

@@ -41,6 +41,60 @@ class LLMError(RuntimeError):
 RATE_LIMIT_STATUS = 429
 RATE_LIMIT_RETRY_S = 20.0
 
+# Generous by default. Output is billed by what is produced, not by the
+# ceiling, and a ceiling too low does not fail loudly — it truncates a tool
+# call mid-argument and hands back something that parses. Measured: Opus
+# writing an architecture plan routinely spends 3,000-4,000 output tokens and
+# was observed at 4,064 against a 4,096 limit.
+DEFAULT_MAX_OUTPUT_TOKENS = 16384
+
+
+def max_output_tokens(env: Mapping[str, str] = os.environ) -> int:
+    raw = env.get("NISSE_MAX_OUTPUT_TOKENS")
+    return int(raw) if raw else DEFAULT_MAX_OUTPUT_TOKENS
+
+
+class UpstreamUnavailable(LLMError):
+    """The proxy or provider could not be reached, or refused for reasons
+    unrelated to what was asked: a network blip, a timeout, a 5xx, a rate
+    limit that did not clear, an exhausted budget.
+
+    Worth distinguishing because the *work* is not at fault. A delegation
+    stopped by one of these has a branch that may be perfectly good, so it is
+    preserved and resumed rather than discarded — unlike a delegate that made
+    a mess, which is cleaned up.
+    """
+
+
+class BudgetExhausted(UpstreamUnavailable):
+    """The proxy refused because a spend limit is gone.
+
+    Distinct from the rest of its family because it will not clear on its
+    own within the hour: an exhausted daily budget takes external models away
+    until it resets, so retrying wastes time and the operator needs telling.
+    A5 keeps the budget itself at the proxy; recognising its refusal is ours.
+    """
+
+
+def _looks_like_budget_refusal(text: str) -> bool:
+    # Pattern-matched rather than contracted: LiteLLM's wording is not a
+    # documented interface, so this errs toward missing one rather than
+    # misreporting an unrelated failure as a budget problem.
+    lowered = text.lower()
+    return "budget" in lowered and (
+        "exceed" in lowered or "exhaust" in lowered or "limit" in lowered
+    )
+
+
+class TruncatedResponse(LLMError):
+    """The model was cut off mid-answer.
+
+    Raised rather than returned, because the damage is silent: a truncated
+    tool call arrives with required arguments simply absent, and a truncated
+    write_file would write half a file and report success. A cut-off turn is
+    a failed turn, and every caller already retries or gates on LLMError.
+    """
+
 
 class LLMClient(Protocol):
     def complete(
@@ -92,6 +146,7 @@ class LiteLLMClient:
         timeout_s: float = 120.0,
         transport: httpx.BaseTransport | None = None,
         sleep: Any = time.sleep,
+        max_tokens: int | None = None,
     ) -> None:
         self._http = httpx.Client(
             base_url=base_url.rstrip("/"),
@@ -101,6 +156,11 @@ class LiteLLMClient:
         )
         self._renderer = OpenAIChatRenderer()
         self._sleep = sleep
+        self._max_tokens = max_tokens or max_output_tokens()
+        # Set after each call so a caller can meter itself (A13 caps);
+        # a side channel rather than a return value, so the LLMClient
+        # protocol stays the same for every implementation.
+        self.last_usage: dict[str, int] = {}
 
     @classmethod
     def from_env(cls, env: Mapping[str, str] = os.environ) -> "LiteLLMClient":
@@ -118,10 +178,15 @@ class LiteLLMClient:
         system: str | None = None,
     ) -> list[CanonicalEvent]:
         body = self._renderer.render(events, tools, model, system=system)
+        # A ceiling on this route too: without one a runaway turn is bounded
+        # only by the provider's default, and the same truncation check has
+        # nothing to check against.
+        body["max_tokens"] = self._max_tokens
         response = _post_with_rate_limit_retry(
             self._http, "/chat/completions", body, self._sleep
         )
         if response.status_code != 200:
+            _raise_for_upstream(response)
             raise LLMError(
                 f"LLM request returned {response.status_code}: {response.text[:500]}"
             )
@@ -131,10 +196,21 @@ class LiteLLMClient:
         self, data: dict[str, Any], requested_model: str
     ) -> list[CanonicalEvent]:
         try:
-            message = data["choices"][0]["message"]
+            choice = data["choices"][0]
+            message = choice["message"]
         except (KeyError, IndexError) as exc:
             raise LLMError(f"malformed completion response: {exc}") from exc
+        if choice.get("finish_reason") == "length":
+            raise TruncatedResponse(
+                f"{requested_model} hit its output limit mid-answer; "
+                "raise NISSE_MAX_OUTPUT_TOKENS"
+            )
         produced_by = data.get("model") or requested_model
+        usage = data.get("usage") or {}
+        self.last_usage = {
+            "input": int(usage.get("prompt_tokens") or 0),
+            "output": int(usage.get("completion_tokens") or 0),
+        }
         out: list[CanonicalEvent] = []
         if message.get("content"):
             out.append(AssistantText(text=message["content"], model=produced_by))
@@ -184,7 +260,7 @@ class AnthropicPassthroughClient:
         token: str,
         *,
         timeout_s: float = 300.0,
-        max_tokens: int = 4096,
+        max_tokens: int | None = None,
         transport: httpx.BaseTransport | None = None,
         sleep: Any = time.sleep,
     ) -> None:
@@ -199,8 +275,9 @@ class AnthropicPassthroughClient:
             transport=transport,
         )
         self._renderer = AnthropicMessagesRenderer()
-        self._max_tokens = max_tokens
+        self._max_tokens = max_tokens or max_output_tokens()
         self._sleep = sleep
+        self.last_usage: dict[str, int] = {}
 
     @classmethod
     def from_env(
@@ -226,6 +303,7 @@ class AnthropicPassthroughClient:
             self._http, "/v1/messages", body, self._sleep
         )
         if response.status_code != 200:
+            _raise_for_upstream(response)
             raise LLMError(
                 f"passthrough returned {response.status_code}: {response.text[:500]}"
             )
@@ -234,10 +312,20 @@ class AnthropicPassthroughClient:
     def _parse(
         self, data: dict[str, Any], requested_model: str
     ) -> list[CanonicalEvent]:
+        if data.get("stop_reason") == "max_tokens":
+            raise TruncatedResponse(
+                f"{requested_model} hit the output limit of {self._max_tokens} "
+                f"tokens mid-answer; raise NISSE_MAX_OUTPUT_TOKENS"
+            )
         content = data.get("content")
         if not isinstance(content, list):
             raise LLMError(f"malformed passthrough response: {str(data)[:300]}")
         produced_by = data.get("model") or requested_model
+        usage = data.get("usage") or {}
+        self.last_usage = {
+            "input": int(usage.get("input_tokens") or 0),
+            "output": int(usage.get("output_tokens") or 0),
+        }
         out: list[CanonicalEvent] = []
         for block in content:
             kind = block.get("type")
@@ -271,8 +359,30 @@ def _post_with_rate_limit_retry(
         try:
             response = http.post(path, json=body)
         except httpx.HTTPError as exc:
-            raise LLMError(f"LLM request failed: {exc}") from exc
+            # Could not reach the proxy at all: nothing about the request was
+            # wrong, so callers holding work in progress should keep it.
+            raise UpstreamUnavailable(f"LLM request failed: {exc}") from exc
         if response.status_code != RATE_LIMIT_STATUS or attempt == 2:
             return response
         sleep(RATE_LIMIT_RETRY_S)
     raise AssertionError("unreachable")
+
+
+# Model families we have a native renderer adapter for. A13: delegations take
+# the vendor's passthrough route because a normalising middlebox is lossy
+# exactly where a long-running delegation lives.
+def _raise_for_upstream(response: httpx.Response) -> None:
+    """Classify a non-200 as upstream trouble where it plainly is.
+
+    A 4xx that is not a rate limit means the request itself was wrong, and
+    preserving work on the strength of one would hide a real bug — so the
+    benefit of the doubt stops at the server's own door.
+    """
+    if _looks_like_budget_refusal(response.text):
+        raise BudgetExhausted(
+            f"the proxy refused on budget grounds: {response.text[:300]}"
+        )
+    if response.status_code >= 500 or response.status_code == RATE_LIMIT_STATUS:
+        raise UpstreamUnavailable(
+            f"upstream returned {response.status_code}: {response.text[:300]}"
+        )

nisse-dev/src/nisse_dev/main.py

@@ -1,11 +1,46 @@
-"""Placeholder entry point for the dev agent."""
+"""Entry point for the dev agent.
 
-from nisse_core.greeting import greeting
+Loads configuration, builds the agent, and stays up until told to stop.
+Shutdown is deliberate rather than abrupt: A2's durability guarantees assume
+the process may die at any moment, but a clean stop still beats relying on
+them.
+"""
+
+import os
+import signal
+import sys
+from pathlib import Path
+from types import FrameType
+
+from nisse_dev.agent import build_agent
+from nisse_dev.config import AgentConfig, ConfigError
+from nisse_dev.envfile import load_env_file
 
 
-def main() -> None:
+def main() -> int:
-    print(greeting("nisse-dev"))
+    load_env_file(Path(os.environ.get("NISSE_ENV", ".env")).expanduser())
+    try:
+        config = AgentConfig.from_env()
+    except ConfigError as exc:
+        # Fail here rather than twenty minutes into a milestone on the one
+        # call that needed the missing value.
+        print(f"nisse-dev: {exc}", file=sys.stderr)
+        return 2
+
+    agent = build_agent(config)
+
+    def shutdown(_signum: int, _frame: FrameType | None) -> None:
+        print("nisse-dev: stopping", file=sys.stderr)
+        agent.stop()
+
+    signal.signal(signal.SIGINT, shutdown)
+    signal.signal(signal.SIGTERM, shutdown)
+
+    agent.start()
+    print(f"nisse-dev: ready as {agent.account} on {config.project}", file=sys.stderr)
+    agent.wait()
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

nisse-dev/src/nisse_dev/planner.py

@@ -13,7 +13,7 @@ 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 LLMClient, LLMError, RoleMap
+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
@@ -125,6 +125,11 @@ class LLMPlanningAgent:
         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)
 

nisse-dev/src/nisse_dev/state.py

@@ -27,7 +27,7 @@ from nisse_dev.canonical import (
 )
 from nisse_dev.oplog import OpLog
 
-CURRENT_SCHEMA_VERSION = 2
+CURRENT_SCHEMA_VERSION = 3
 
 Migration = Callable[[sqlite3.Connection], None]
 
@@ -41,12 +41,29 @@ CREATE TABLE canonical_events (
 """
 
 
+_EVAL_OUTCOMES_DDL = """
+CREATE TABLE eval_outcomes (
+    event_id INTEGER PRIMARY KEY,
+    verdict TEXT NOT NULL,
+    note TEXT NOT NULL DEFAULT '',
+    marked_at TEXT NOT NULL
+)
+"""
+
+
 def _migrate_v1_add_canonical_events(conn: sqlite3.Connection) -> None:
     conn.execute(_CANONICAL_EVENTS_DDL)
 
 
+def _migrate_v2_add_eval_outcomes(conn: sqlite3.Connection) -> None:
+    conn.execute(_EVAL_OUTCOMES_DDL)
+
+
 # Maps version N to the migration that takes the schema from N to N+1.
-MIGRATIONS: Mapping[int, Migration] = {1: _migrate_v1_add_canonical_events}
+MIGRATIONS: Mapping[int, Migration] = {
+    1: _migrate_v1_add_canonical_events,
+    2: _migrate_v2_add_eval_outcomes,
+}
 
 
 class StateSchemaError(RuntimeError):
@@ -93,6 +110,7 @@ _SCHEMA = [
     )
     """,
     _CANONICAL_EVENTS_DDL,
+    _EVAL_OUTCOMES_DDL,
 ]
 
 
@@ -294,6 +312,27 @@ class StateStore:
             for row in rows
         ]
 
+    def mark_outcome(self, event_id: int, verdict: str, note: str = "") -> None:
+        """Record whether a delegation or consult was good (A16).
+
+        Kept out of the canonical log on purpose: the log is an append-only
+        record of what happened, and a verdict is a later judgement about it
+        that may be revised.
+        """
+        self._conn.execute(
+            "INSERT INTO eval_outcomes (event_id, verdict, note, marked_at) "
+            "VALUES (?, ?, ?, ?) ON CONFLICT(event_id) DO UPDATE SET "
+            "verdict = excluded.verdict, note = excluded.note, "
+            "marked_at = excluded.marked_at",
+            (event_id, verdict, note, _now()),
+        )
+
+    def outcomes(self) -> dict[int, tuple[str, str]]:
+        rows = self._conn.execute(
+            "SELECT event_id, verdict, note FROM eval_outcomes"
+        ).fetchall()
+        return {r["event_id"]: (r["verdict"], r["note"]) for r in rows}
+
     def close(self) -> None:
         self._conn.close()
 

nisse-dev/tests/test_agent_build.py

@@ -0,0 +1,289 @@
+"""The composition root. These tests exist because two components — the
+delegation harness and the comment poller — were fully built, fully tested,
+and reachable from nothing but their own tests. Unit tests that construct a
+component directly cannot catch that; only assembly can."""
+
+import sqlite3
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pytest
+from test_config import COMPLETE
+
+from nisse_dev.agent import build_agent, discover_models, prepare_workspace
+from nisse_dev.catalog import ModelCatalog, UnknownModel
+from nisse_dev.config import AgentConfig
+from nisse_dev.git import GitRepo
+from nisse_dev.oplog import OpLog
+
+# What the test roles resolve to, as LiteLLM would report them.
+TEST_CATALOG = ModelCatalog.from_payload(
+    {
+        "data": [
+            {
+                "model_name": alias,
+                "litellm_params": {"model": f"anthropic/{alias}"},
+                "model_info": {
+                    "litellm_provider": "anthropic",
+                    "supports_function_calling": True,
+                },
+            }
+            for alias in ("claude-opus-5", "claude-haiku-4-5")
+        ]
+    }
+)
+
+
+def config_for(tmp_path: Path, **overrides: str) -> AgentConfig:
+    env = {
+        **COMPLETE,
+        "NISSE_STATE_DB": str(tmp_path / "state.db"),
+        "NISSE_WORKSPACE": str(tmp_path / "work"),
+        **overrides,
+    }
+    return AgentConfig.from_env(env)
+
+
+@pytest.fixture
+def origin(tmp_path: Path) -> Path:
+    """A bare repo standing in for the project on the forge."""
+    remote = tmp_path / "origin.git"
+    GitRepo(tmp_path)._run("init", "-q", "--bare", "-b", "main", str(remote))
+    seed = GitRepo(tmp_path / "seed")
+    seed.init()
+    seed.configure_identity("seed", "seed@test")
+    (seed.root / "README.md").write_text("project\n")
+    seed.commit_all("seed")
+    seed.push(str(remote), "main")
+    return remote
+
+
+def test_workspace_is_cloned_when_absent(tmp_path: Path, origin: Path) -> None:
+    config = config_for(tmp_path, NISSE_GITEA_URL=f"file://{origin}")
+    # authenticated_url would mangle a file:// remote, so clone directly.
+    GitRepo(tmp_path).clone(str(origin), config.workspace)
+    repo = prepare_workspace(config, "nisse-dev-testing")
+
+    assert (config.workspace / "README.md").exists()
+    assert repo.current_branch() == "main"
+
+
+def test_workspace_is_reused_when_already_cloned(tmp_path: Path, origin: Path) -> None:
+    config = config_for(tmp_path)
+    GitRepo(tmp_path).clone(str(origin), config.workspace)
+    (config.workspace / "local-work.txt").write_text("in progress")
+
+    prepare_workspace(config, "nisse-dev-testing")
+
+    # Re-cloning would silently destroy uncommitted work.
+    assert (config.workspace / "local-work.txt").exists()
+
+
+class FakeTransport:
+    def __init__(self) -> None:
+        self.sent: list[str] = []
+        self.started = False
+        self.closed = False
+
+    def send_message(self, text: str) -> None:
+        self.sent.append(text)
+
+    def send_approval_request(self, text: str) -> None:
+        self.sent.append(text)
+
+    def ask_question(self, text: str) -> None:
+        self.sent.append(text)
+
+    def on_message(self, callback: Any) -> None:
+        self.callback = callback
+
+    def start(self) -> None:
+        self.started = True
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def build_offline(tmp_path: Path, origin: Path, monkeypatch: Any) -> Any:
+    """Assemble the real graph with only the network boundaries faked."""
+    config = config_for(tmp_path)
+    GitRepo(tmp_path).clone(str(origin), config.workspace)
+
+    monkeypatch.setattr(
+        "nisse_dev.agent.GiteaClient.whoami", lambda self: "nisse-dev-testing"
+    )
+    monkeypatch.setattr(
+        "nisse_dev.agent.ModelCatalog.discover",
+        classmethod(lambda cls, *a, **kw: TEST_CATALOG),
+    )
+    monkeypatch.setattr("nisse_dev.agent.MatrixTransport", lambda cfg: FakeTransport())
+    monkeypatch.setattr(
+        "nisse_dev.agent.authenticated_url",
+        lambda base, repo, user, token: str(origin),
+    )
+    return build_agent(config)
+
+
+def test_the_whole_graph_assembles(tmp_path: Path, origin: Path, monkeypatch) -> None:
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    orchestrator = agent.orchestrator
+
+    # Every component that was previously unreachable is now connected.
+    assert orchestrator._executor is not None
+    assert orchestrator._publisher is not None
+    assert orchestrator._board is not None
+    assert orchestrator._delegate is not None
+    assert orchestrator._poller_factory is not None
+    assert orchestrator._executor._delegate is not None  # A12 escalation path
+    assert orchestrator._executor._expert is not None  # A14 consult path
+    agent.stop()
+
+
+def test_start_announces_where_it_resumed(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    agent.start()
+
+    assert agent.transport.started
+    readiness = agent.transport.sent[-1]
+    assert "nisse-dev up as nisse-dev-testing" in readiness
+    assert "greet" in readiness  # the project
+    assert "claude-opus-5" in readiness  # the pinned planner
+    agent.stop()
+
+
+def test_stop_closes_everything_and_releases_the_wait(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    agent.start()
+    agent.stop()
+
+    assert agent.transport.closed
+    agent.wait()  # returns immediately rather than hanging
+    with pytest.raises(sqlite3.ProgrammingError):
+        agent.store.get_state()  # the store really is closed
+
+
+def test_poller_factory_ignores_the_agents_own_comments(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    # Without this the agent answers its own PR replies in a loop.
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    poller = agent.orchestrator._poller_factory(lambda c: None)
+    assert "nisse-dev-testing" in poller._ignore
+    agent.stop()
+
+
+def test_delegation_uses_the_passthrough_route(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    # A13: delegations run vendor-native; the normalized route is for the
+    # places where fidelity is cheap.
+    from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient
+
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    delegation = agent.orchestrator._delegate
+    assert isinstance(delegation._client, AnthropicPassthroughClient)
+    assert isinstance(agent.orchestrator._executor._client, LiteLLMClient)
+    agent.stop()
+
+
+def test_httpx_clients_are_not_shared_between_routes(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    normalized = agent.orchestrator._executor._client._http
+    passthrough = agent.orchestrator._delegate._client._http
+    assert isinstance(normalized, httpx.Client)
+    assert normalized is not passthrough  # different base URLs and headers
+    agent.stop()
+
+
+# --- startup model discovery (A5/A11) ------------------------------------
+
+
+def test_a_typod_role_refuses_to_boot(tmp_path: Path, monkeypatch) -> None:
+    # Misconfiguration should cost a second at startup, not be discovered ten
+    # minutes into a milestone.
+    config = config_for(tmp_path, NISSE_ROLE_PLANNER="clause-opus-5")
+    monkeypatch.setattr(
+        "nisse_dev.agent.ModelCatalog.discover",
+        classmethod(lambda cls, *a, **kw: TEST_CATALOG),
+    )
+    with pytest.raises(UnknownModel, match="planner='clause-opus-5'"):
+        discover_models(config, OpLog("test"))
+
+
+def test_a_role_this_key_cannot_reach_refuses_to_boot(
+    tmp_path: Path, monkeypatch
+) -> None:
+    # /model/info is scoped per key, so the same check catches a role pointed
+    # at a real model that this token has no access to.
+    config = config_for(tmp_path, NISSE_ROLE_EXECUTOR="local-fast")
+    monkeypatch.setattr(
+        "nisse_dev.agent.ModelCatalog.discover",
+        classmethod(lambda cls, *a, **kw: TEST_CATALOG),
+    )
+    with pytest.raises(UnknownModel, match="executor='local-fast'"):
+        discover_models(config, OpLog("test"))
+
+
+def test_a_tool_less_model_warns_but_still_boots(tmp_path: Path, monkeypatch) -> None:
+    # A15 expects weak local models to be accommodated, and models LiteLLM
+    # declares tool-less are observed driving tool loops anyway.
+    import io
+
+    tool_less = ModelCatalog.from_payload(
+        {
+            "data": [
+                {
+                    "model_name": alias,
+                    "litellm_params": {"model": "openai/qwen3-8b-awq"},
+                    "model_info": {
+                        "litellm_provider": "openai",
+                        "supports_function_calling": False,
+                    },
+                }
+                for alias in ("claude-opus-5", "claude-haiku-4-5")
+            ]
+        }
+    )
+    monkeypatch.setattr(
+        "nisse_dev.agent.ModelCatalog.discover",
+        classmethod(lambda cls, *a, **kw: tool_less),
+    )
+    stream = io.StringIO()
+    discover_models(config_for(tmp_path), OpLog("test", stream))
+
+    logged = stream.getvalue()
+    assert "model_tools_unsupported" in logged
+    assert '"role": "executor"' in logged
+    assert '"role": "consultant"' not in logged  # ask_expert drives no tools
+
+
+def test_the_resolved_model_is_pinned_at_task_start(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    agent.orchestrator._begin_project("greet")
+
+    pinned = agent.orchestrator._data["models"]
+    assert pinned["planner"] == {
+        "alias": "claude-opus-5",
+        "backing": "anthropic/claude-opus-5",
+        "provider": "anthropic",
+    }
+    agent.stop()
+
+
+def test_delegation_records_what_the_alias_resolved_to(
+    tmp_path: Path, origin: Path, monkeypatch
+) -> None:
+    # A16 eval history keyed on an alias alone would average across whatever
+    # that alias pointed at over time.
+    agent = build_offline(tmp_path, origin, monkeypatch)
+    assert agent.orchestrator._delegate._backing_model == "anthropic/claude-opus-5"
+    agent.stop()

nisse-dev/tests/test_canonical.py

@@ -16,7 +16,7 @@ from nisse_dev.canonical import (
     from_payload,
     mint_call_id,
 )
-from nisse_dev.state import StateStore
+from nisse_dev.state import CURRENT_SCHEMA_VERSION, StateStore
 
 ALL_EVENTS = [
     UserMessage(text="use demo", source="matrix:$abc"),
@@ -116,9 +116,12 @@ def test_default_migration_upgrades_a_real_v1_store(tmp_path: Path) -> None:
     conn.close()
 
     with StateStore(db) as store:
-        assert store.schema_version == 2
+        assert store.schema_version == CURRENT_SCHEMA_VERSION
         state = store.get_state()
         assert state is not None
         assert state.state == "IDLE"  # v1 data intact
         store.append_event(UserMessage(text="hello", source="m:1"))
         assert len(store.read_events()) == 1
+        # Two migrations were traversed, so both later tables must exist.
+        store.mark_outcome(1, "good", "arrived via v1 -> v3")
+        assert store.outcomes()[1] == ("good", "arrived via v1 -> v3")

nisse-dev/tests/test_catalog.py

@@ -0,0 +1,202 @@
+"""Runtime model discovery (spec A5/A11/A13).
+
+The point of these tests is that nothing here reads the model's *name*. A
+semantic alias like "remote-architecture" is free to point anywhere, so the
+route has to come from what LiteLLM says the model is.
+"""
+
+import httpx
+import pytest
+
+from nisse_dev.catalog import (
+    CatalogUnavailable,
+    ModelCatalog,
+    UnknownModel,
+    select_client,
+)
+
+
+def entry(alias: str, backing: str, provider: str | None, tools: bool | None = None):
+    """One /model/info row, shaped as the live proxy returns it."""
+    info: dict = {"supports_function_calling": tools}
+    if provider is not None:
+        info["litellm_provider"] = provider
+    return {
+        "model_name": alias,
+        "litellm_params": {"model": backing, "api_base": "http://proxy.test/v1"},
+        "model_info": info,
+    }
+
+
+def catalog(*entries) -> ModelCatalog:
+    return ModelCatalog.from_payload({"data": list(entries)})
+
+
+# --- parsing -------------------------------------------------------------
+
+
+def test_facts_are_read_from_the_payload() -> None:
+    cat = catalog(
+        entry("claude-haiku-4-5", "anthropic/claude-haiku-4-5", "anthropic", True)
+    )
+    facts = cat.facts("claude-haiku-4-5")
+    assert facts.backing == "anthropic/claude-haiku-4-5"
+    assert facts.provider == "anthropic"
+    assert facts.supports_tools is True
+    assert facts.native_anthropic
+
+
+def test_provider_falls_back_to_the_backing_prefix() -> None:
+    # litellm_provider is absent on some deployments; the backing model's
+    # own "vendor/model" prefix says the same thing.
+    cat = catalog(entry("mystery", "anthropic/claude-opus-5", None))
+    assert cat.facts("mystery").provider == "anthropic"
+
+
+def test_unnamed_entries_are_skipped() -> None:
+    cat = ModelCatalog.from_payload({"data": [{"litellm_params": {"model": "x"}}]})
+    assert cat.aliases() == []
+
+
+def test_missing_tool_support_is_unknown_not_false() -> None:
+    # LiteLLM reports null for models it has no capability data on. That is
+    # not the same as declaring the model tool-less, and the startup warning
+    # depends on the difference.
+    cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai", None))
+    assert cat.facts("local-fast").supports_tools is None
+
+
+# --- route selection by provider, never by name --------------------------
+
+
+def test_a_semantic_alias_routes_by_its_provider() -> None:
+    # The whole feature: nothing in "remote-architecture" says Anthropic.
+    cat = catalog(entry("remote-architecture", "anthropic/claude-opus-5", "anthropic"))
+    normalized, passthrough = object(), object()
+    chosen = select_client("remote-architecture", normalized, passthrough, cat)
+    assert chosen is passthrough
+
+
+def test_the_same_alias_reroutes_when_it_is_repointed() -> None:
+    # A13: provisionally accept LiteLLM's normalization for a vendor with no
+    # native adapter, rather than sending it Anthropic-shaped bodies.
+    cat = catalog(entry("remote-architecture", "openai/gpt-6-astra", "openai"))
+    normalized, passthrough = object(), object()
+    assert (
+        select_client("remote-architecture", normalized, passthrough, cat) is normalized
+    )
+
+
+def test_a_name_that_merely_mentions_a_vendor_is_not_that_vendor() -> None:
+    cat = catalog(entry("claude-clone-9", "openai/some-clone", "openai"))
+    normalized, passthrough = object(), object()
+    assert select_client("claude-clone-9", normalized, passthrough, cat) is normalized
+
+
+def test_local_models_take_the_normalized_route() -> None:
+    cat = catalog(
+        entry("local-coder", "openai/qwen3-14b", "openai"),
+        entry("local-fast", "openai/qwen3-8b-awq", "openai"),
+    )
+    normalized, passthrough = object(), object()
+    for alias in ("local-coder", "local-fast"):
+        assert select_client(alias, normalized, passthrough, cat) is normalized
+
+
+# --- misconfiguration is loud --------------------------------------------
+
+
+def test_an_unknown_model_raises_rather_than_guessing_a_route() -> None:
+    cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai"))
+    with pytest.raises(UnknownModel, match="local-fsat"):
+        select_client("local-fsat", object(), object(), cat)
+
+
+def test_the_error_lists_what_is_available() -> None:
+    cat = catalog(
+        entry("local-fast", "openai/qwen3-8b-awq", "openai"),
+        entry("claude-opus-5", "anthropic/claude-opus-5", "anthropic"),
+    )
+    with pytest.raises(UnknownModel, match="claude-opus-5, local-fast"):
+        cat.facts("typo")
+
+
+def test_validate_names_every_bad_role_at_once() -> None:
+    # One boot should surface the whole list, not make the operator
+    # rediscover the next typo on the next start.
+    cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai"))
+    with pytest.raises(UnknownModel) as caught:
+        cat.validate(
+            {
+                "executor": "local-fast",
+                "planner": "clause-opus-5",
+                "delegate": "remote-architecture",
+                "consultant": "local-fast",
+            }
+        )
+    message = str(caught.value)
+    assert "planner='clause-opus-5'" in message
+    assert "delegate='remote-architecture'" in message
+    assert "executor" not in message  # the good ones are not noise
+
+
+def test_validate_passes_a_whole_good_mapping() -> None:
+    cat = catalog(
+        entry("local-fast", "openai/qwen3-8b-awq", "openai"),
+        entry("claude-opus-5", "anthropic/claude-opus-5", "anthropic"),
+    )
+    cat.validate({"executor": "local-fast", "planner": "claude-opus-5"})
+
+
+# --- an outage is not a misconfiguration ---------------------------------
+
+
+def test_an_unreachable_proxy_is_reported_as_an_outage() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("no route to host")
+
+    with pytest.raises(CatalogUnavailable, match="could not reach"):
+        ModelCatalog.discover(
+            "http://litellm.test", "sk-test", transport=httpx.MockTransport(handler)
+        )
+
+
+def test_a_refused_listing_is_reported_as_an_outage() -> None:
+    cat = httpx.MockTransport(lambda request: httpx.Response(401, text="bad key"))
+    with pytest.raises(CatalogUnavailable, match="401"):
+        ModelCatalog.discover("http://litellm.test", "sk-test", transport=cat)
+
+
+def test_discovery_asks_the_proxy_with_the_token() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        seen["auth"] = request.headers.get("Authorization")
+        return httpx.Response(
+            200, json={"data": [entry("local-fast", "openai/qwen3-8b-awq", "openai")]}
+        )
+
+    cat = ModelCatalog.discover(
+        "http://litellm.test/", "sk-test", transport=httpx.MockTransport(handler)
+    )
+    assert seen["url"] == "http://litellm.test/model/info"
+    assert seen["auth"] == "Bearer sk-test"
+    assert cat.aliases() == ["local-fast"]
+
+
+# --- what gets pinned ----------------------------------------------------
+
+
+def test_pinning_records_what_the_alias_resolved_to() -> None:
+    # An alias may point somewhere else tomorrow; eval history (A16) keyed on
+    # the alias alone would average across whatever it pointed at.
+    cat = catalog(entry("remote-architecture", "anthropic/claude-opus-5", "anthropic"))
+    pinned = cat.pinned({"planner": "remote-architecture"})
+    assert pinned == {
+        "planner": {
+            "alias": "remote-architecture",
+            "backing": "anthropic/claude-opus-5",
+            "provider": "anthropic",
+        }
+    }

nisse-dev/tests/test_config.py

@@ -0,0 +1,132 @@
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.config import AgentConfig, ConfigError
+
+COMPLETE = {
+    "NISSE_STATE_DB": "~/demo/state.db",
+    "NISSE_WORKSPACE": "~/demo/work",
+    "NISSE_PROJECT": "greet",
+    "NISSE_LLM_BASE_URL": "https://litellm.test",
+    "NISSE_LLM_TOKEN": "sk-x",
+    "NISSE_ROLE_EXECUTOR": "claude-haiku-4-5",
+    "NISSE_ROLE_PLANNER": "claude-opus-5",
+    "NISSE_ROLE_DELEGATE": "claude-opus-5",
+    "NISSE_ROLE_CONSULTANT": "claude-haiku-4-5",
+    "NISSE_MATRIX_HOMESERVER": "https://matrix.test",
+    "NISSE_MATRIX_USER_ID": "@nisse-dev-testing:test",
+    "NISSE_MATRIX_TOKEN": "syt_x",
+    "NISSE_MATRIX_ROOM_ID": "!room:test",
+    "NISSE_MATRIX_OPERATORS": "@don:test",
+    "NISSE_GITEA_URL": "https://gitea.test",
+    "NISSE_GITEA_TOKEN": "gt",
+    "NISSE_GITEA_REPO": "don/demo",
+    "NISSE_GITEA_OPERATORS": "don",
+}
+
+
+def test_complete_environment_builds() -> None:
+    config = AgentConfig.from_env(COMPLETE)
+    assert config.project == "greet"
+    assert config.roles.planner == "claude-opus-5"
+    assert config.matrix.room_id == "!room:test"
+    assert config.gitea_repo == "don/demo"
+    assert config.base_branch == "main"
+
+
+def test_paths_expand_the_home_directory() -> None:
+    config = AgentConfig.from_env(COMPLETE)
+    assert config.workspace.is_absolute()
+    assert "~" not in str(config.state_db)
+
+
+def test_every_missing_variable_is_reported_at_once() -> None:
+    # Failing at first use would surface a missing token twenty minutes into
+    # a milestone, on the call that mattered.
+    partial = {k: v for k, v in COMPLETE.items() if k != "NISSE_MATRIX_TOKEN"}
+    partial.pop("NISSE_GITEA_REPO")
+    with pytest.raises(ConfigError) as excinfo:
+        AgentConfig.from_env(partial)
+    message = str(excinfo.value)
+    assert "NISSE_MATRIX_TOKEN" in message
+    assert "NISSE_GITEA_REPO" in message
+
+
+def test_empty_is_treated_as_missing() -> None:
+    with pytest.raises(ConfigError, match="NISSE_GITEA_TOKEN"):
+        AgentConfig.from_env({**COMPLETE, "NISSE_GITEA_TOKEN": "   "})
+
+
+def test_operator_lists_are_split_and_trimmed() -> None:
+    config = AgentConfig.from_env(
+        {
+            **COMPLETE,
+            "NISSE_MATRIX_OPERATORS": " @don:test , @other:test ,",
+            "NISSE_GITEA_OPERATORS": "don,",
+        }
+    )
+    assert config.operators == {"@don:test", "@other:test"}
+    # A trailing comma must not admit an empty identity into an allowlist.
+    assert "" not in config.operators
+    assert config.gitea_operators == {"don"}
+
+
+def test_base_branch_is_overridable() -> None:
+    config = AgentConfig.from_env({**COMPLETE, "NISSE_BASE_BRANCH": "trunk"})
+    assert config.base_branch == "trunk"
+
+
+def test_redacted_summary_carries_no_secrets() -> None:
+    summary = AgentConfig.from_env(COMPLETE).redacted()
+    flat = " ".join(summary.values())
+    for secret in ("sk-x", "syt_x", "gt"):
+        assert secret not in flat
+    assert summary["project"] == "greet"
+
+
+def test_state_db_and_workspace_are_separate(tmp_path: Path) -> None:
+    # A2: build artefacts must not share a volume with agent state.
+    config = AgentConfig.from_env(COMPLETE)
+    assert config.state_db.parent != config.workspace
+
+
+def test_state_db_inside_the_clone_is_refused(tmp_path: Path) -> None:
+    # git add -A would commit it, and the push would publish the agent's
+    # own state alongside the project.
+    work = tmp_path / "project"
+    with pytest.raises(ConfigError, match="inside the workspace"):
+        AgentConfig.from_env(
+            {
+                **COMPLETE,
+                "NISSE_WORKSPACE": str(work),
+                "NISSE_STATE_DB": str(work / "state.db"),
+            }
+        )
+
+
+def test_env_file_inside_the_clone_is_refused(tmp_path: Path) -> None:
+    # The severe case: the agent would commit and push its own credentials
+    # as a side effect of working correctly.
+    work = tmp_path / "project"
+    with pytest.raises(ConfigError, match="NISSE_ENV"):
+        AgentConfig.from_env(
+            {
+                **COMPLETE,
+                "NISSE_WORKSPACE": str(work),
+                "NISSE_STATE_DB": str(tmp_path / "state.db"),
+                "NISSE_ENV": str(work / ".env"),
+            }
+        )
+
+
+def test_sibling_paths_are_fine(tmp_path: Path) -> None:
+    config = AgentConfig.from_env(
+        {
+            **COMPLETE,
+            "NISSE_WORKSPACE": str(tmp_path / "project"),
+            "NISSE_STATE_DB": str(tmp_path / "state.db"),
+            "NISSE_ENV": str(tmp_path / ".env"),
+        }
+    )
+    assert config.workspace.name == "project"

nisse-dev/tests/test_delegation.py

@@ -11,14 +11,17 @@ from nisse_dev.canonical import (
     DelegationStarted,
     StoredEvent,
     ToolCall,
+    ToolResult,
 )
 from nisse_dev.delegation import (
     DELEGATE_TOOLS,
+    DELEGATION_INTENT,
     Caps,
     Delegation,
+    delegation_reconciler,
 )
 from nisse_dev.git import GitRepo
-from nisse_dev.llm import LLMError
+from nisse_dev.llm import BudgetExhausted, LLMError, UpstreamUnavailable
 from nisse_dev.oplog import OpLog
 from nisse_dev.state import StateStore
 from nisse_dev.tools import Workspace
@@ -302,3 +305,345 @@ def test_failure_may_be_reported_without_doing_work(
     assert result.status == "failed"
     assert not result.usable
     assert len(client.calls) == 1  # believed immediately, not challenged
+
+
+# --- A3 step 5: the resume rule -----------------------------------------
+
+
+def pending_delegation_intent(store: StateStore, branch: str = "d") -> Any:
+    store.begin_intent(
+        DELEGATION_INTENT,
+        {"branch": branch, "base": "main", "brief": "finish it"},
+        cause="test",
+    )
+    return store.pending_intents()[-1]
+
+
+def test_intent_is_recorded_and_settled_by_a_completed_delegation(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [[call("write_file", path="a.py", content="x")], [hand_back()]]
+    )
+    build(client, store, ws, repo).run(brief="b", branch="d")
+    # Nothing left in flight: an aborted delegation is a finished one too.
+    assert store.pending_intents() == []
+
+
+def test_aborted_delegation_also_settles_its_intent(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient([[hand_back(succeeded=False, changed="cannot")]])
+    build(client, store, ws, repo).run(brief="b", branch="d")
+    assert store.pending_intents() == []
+
+
+def test_clean_cut_after_a_tool_result_is_resumable(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    repo.checkout_new_branch("d")
+    (repo.root / "partial.py").write_text("x")
+    repo.commit_all("partial work")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(ToolCall(call_id="c", tool="write_file", arguments={}))
+    store.append_event(ToolResult(call_id="c", result="wrote", success=True))
+    intent = pending_delegation_intent(store)
+
+    assert delegation_reconciler(store, repo)(intent) is True
+
+
+def test_dangling_tool_call_gates(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # The effect of the call is unknown and not verifiable by inspection.
+    repo.checkout_new_branch("d")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(ToolCall(call_id="c", tool="run_command", arguments={}))
+    intent = pending_delegation_intent(store)
+
+    assert delegation_reconciler(store, repo)(intent) is None
+
+
+def test_missing_branch_means_nothing_started(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(ToolResult(call_id="c", result="ok", success=True))
+    intent = pending_delegation_intent(store, branch="never-created")
+
+    assert delegation_reconciler(store, repo)(intent) is False
+
+
+def test_a_finished_delegation_settles_rather_than_resumes(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    repo.checkout_new_branch("d")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(
+        DelegationCompleted(
+            handback="done",
+            artifacts=[],
+            tokens=0,
+            duration_s=1.0,
+            exit_status="handback",
+        )
+    )
+    intent = pending_delegation_intent(store)
+
+    assert delegation_reconciler(store, repo)(intent) is True
+
+
+def test_resume_continues_the_same_delegation(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # Interrupt after one turn, then resume: no second delegation_started,
+    # because A10 defines a session as the delegation, not the process.
+    repo.checkout_new_branch("d")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(ToolResult(call_id="c", result="wrote a.py", success=True))
+    intent = pending_delegation_intent(store)
+
+    client = ScriptedClient(
+        [[call("write_file", path="b.py", content="x")], [hand_back()]]
+    )
+    result = build(client, store, ws, repo).resume(intent)
+
+    assert result.usable
+    starts = [e for e in store.read_events() if isinstance(e.event, DelegationStarted)]
+    assert len(starts) == 1
+    assert store.pending_intents() == []
+    assert (repo.root / "b.py").exists()
+
+
+def test_resume_onto_a_missing_branch_fails_cleanly(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    intent = pending_delegation_intent(store, branch="gone")
+    result = build(ScriptedClient([]), store, ws, repo).resume(intent)
+
+    assert result.status == "failed"
+    assert not result.usable
+    assert store.pending_intents() == []  # not left dangling for the next boot
+
+
+# --- boot recovery sweep ------------------------------------------------
+
+
+def test_recover_resumes_a_clean_cut_with_a_notice(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    repo.checkout_new_branch("d")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(ToolResult(call_id="c", result="wrote a.py", success=True))
+    pending_delegation_intent(store)
+    notices: list[str] = []
+
+    client = ScriptedClient(
+        [[call("write_file", path="b.py", content="x")], [hand_back()]]
+    )
+    result = build(client, store, ws, repo).recover(notices.append)
+
+    assert result is not None and result.usable
+    # A3: a notice, not a question — a clean cut needs no operator decision.
+    assert notices and "resuming the delegation" in notices[0]
+
+
+def test_recover_settles_a_finished_delegation_without_resuming(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    repo.checkout_new_branch("d")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(
+        DelegationCompleted(
+            handback="done",
+            artifacts=[],
+            tokens=0,
+            duration_s=1.0,
+            exit_status="handback",
+        )
+    )
+    pending_delegation_intent(store)
+    client = ScriptedClient([])
+
+    assert build(client, store, ws, repo).recover(lambda _m: None) is None
+    assert store.pending_intents() == []
+    assert client.calls == []  # the model was never asked
+
+
+def test_recover_discards_an_intent_whose_branch_never_existed(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    store.append_event(ToolResult(call_id="c", result="ok", success=True))
+    pending_delegation_intent(store, branch="never-made")
+
+    assert build(ScriptedClient([]), store, ws, repo).recover(lambda _m: None) is None
+    assert store.pending_intents() == []
+
+
+def test_recover_leaves_an_unclean_cut_for_the_boot_gate(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    repo.checkout_new_branch("d")
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(ToolCall(call_id="c", tool="run_command", arguments={}))
+    pending_delegation_intent(store)
+
+    assert build(ScriptedClient([]), store, ws, repo).recover(lambda _m: None) is None
+    # Still pending, so reconcile_on_boot gates rather than blind-resuming.
+    assert [i.kind for i in store.pending_intents()] == [DELEGATION_INTENT]
+
+
+# --- the token cap, which A13 calls hard and was decoration -------------
+
+
+class MeteredClient(ScriptedClient):
+    """A client that reports usage, as the real ones do."""
+
+    def __init__(self, turns: list, per_call: int = 1000) -> None:
+        super().__init__(turns)
+        self._per_call = per_call
+        self.last_usage: dict[str, int] = {}
+
+    def complete(self, *args: Any, **kwargs: Any):
+        produced = super().complete(*args, **kwargs)
+        self.last_usage = {"input": self._per_call, "output": self._per_call}
+        return produced
+
+
+def test_token_cap_stops_a_delegation(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # An unbounded delegation can exhaust a daily budget and take external
+    # models away until it resets, so the cap has to be real.
+    client = MeteredClient(
+        [[call("list_dir", path=".")] for _ in range(20)], per_call=600
+    )
+    result = build(client, store, ws, repo, caps=Caps(turns=20, tokens=2000)).run(
+        brief="b", branch="d"
+    )
+
+    assert result.status == "capped"
+    assert "token cap of 2000" in result.note
+    # 600 in + 600 out per call: stops on the second turn, not the twentieth.
+    assert len(client.calls) == 2
+
+
+def test_tokens_spent_are_reported_not_left_at_zero(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = MeteredClient(
+        [[call("write_file", path="a.py", content="x")], [hand_back()]], per_call=500
+    )
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.usable
+    assert result.tokens == 2000  # two calls, 1000 each
+
+    completed = next(
+        e.event for e in store.read_events() if isinstance(e.event, DelegationCompleted)
+    )
+    # A16 evaluates cost against outcome, which needs a real number.
+    assert completed.tokens == 2000
+
+
+def test_a_client_without_usage_still_works(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # Scripted and third-party clients need not report usage; the cap simply
+    # does not bind rather than crashing the delegation.
+    client = ScriptedClient(
+        [[call("write_file", path="a.py", content="x")], [hand_back()]]
+    )
+    assert build(client, store, ws, repo).run(brief="b", branch="d").usable
+
+
+def test_budget_exhaustion_preserves_the_work_for_later(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    """An outage is not a bad delegate.
+
+    The branch may hold perfectly good work, so it is kept and the intent
+    left pending — the next boot treats it exactly like a crash mid-delegation
+    and resumes if the cut was clean. Discarding would throw away work
+    because the budget ran out.
+    """
+
+    class BrokeClient(ScriptedClient):
+        def complete(self, *args: Any, **kwargs: Any):
+            if self.calls:
+                raise BudgetExhausted("daily budget exhausted")
+            return super().complete(*args, **kwargs)
+
+    client = BrokeClient([[call("write_file", path="progress.py", content="real")]])
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.status == "interrupted"
+    assert result.interrupted and not result.usable
+    # The work survives, on its branch.
+    assert repo.current_branch() == "d"
+    assert (repo.root / "progress.py").exists()
+    # And the intent stays pending, which is what A3's resume rule keys on.
+    assert [i.kind for i in store.pending_intents()] == [DELEGATION_INTENT]
+
+
+def test_an_ordinary_failure_still_discards(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # Only external interruptions are preserved; a delegate that made a mess
+    # is still cleaned up (A13).
+    client = ScriptedClient([[hand_back(succeeded=False, changed="cannot")]])
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.status == "failed"
+    assert not result.interrupted
+    assert repo.current_branch() == "main"
+    assert store.pending_intents() == []
+
+
+def test_a_network_outage_also_preserves_the_work(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # Same reasoning as an exhausted budget: the branch may be fine, and the
+    # outage says nothing about the quality of the work on it.
+    class FlakyClient(ScriptedClient):
+        def complete(self, *args: Any, **kwargs: Any):
+            if self.calls:
+                raise UpstreamUnavailable("LLM request failed: connection refused")
+            return super().complete(*args, **kwargs)
+
+    client = FlakyClient([[call("write_file", path="progress.py", content="real")]])
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.status == "interrupted"
+    assert (repo.root / "progress.py").exists()
+    assert [i.kind for i in store.pending_intents()] == [DELEGATION_INTENT]
+
+
+def test_a_bad_request_is_not_given_the_benefit_of_the_doubt(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # A 4xx that is not a rate limit means we sent something wrong; treating
+    # that as an outage would preserve branches and hide a real bug.
+    class BadRequestClient(ScriptedClient):
+        def complete(self, *args: Any, **kwargs: Any):
+            raise LLMError("LLM request returned 400: malformed tool schema")
+
+    result = build(BadRequestClient([]), store, ws, repo).run(brief="b", branch="d")
+    assert result.status == "failed"
+    assert not result.interrupted

nisse-dev/tests/test_dev_main.py

@@ -1,10 +1,26 @@
+from pathlib import Path
+
 import pytest
 
+from nisse_dev.config import REQUIRED
 from nisse_dev.main import main
 
 
-def test_main_prints_greeting(capsys: pytest.CaptureFixture[str]) -> None:
+def test_main_refuses_to_start_without_configuration(
-    main()
+    monkeypatch: pytest.MonkeyPatch,
-    assert (
+    capsys: pytest.CaptureFixture[str],
-        capsys.readouterr().out == "Hello from nisse-dev, the house spirit is awake.\n"
+    tmp_path: Path,
-    )
+) -> None:
+    # Clear the real variables rather than swapping os.environ: from_env binds
+    # the environment object as a default argument, so a replacement would be
+    # ignored. And point NISSE_ENV at a file that does not exist, because the
+    # working directory's .env holds live credentials.
+    for name in REQUIRED:
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("NISSE_ENV", str(tmp_path / "absent.env"))
+
+    assert main() == 2
+
+    err = capsys.readouterr().err
+    assert "cannot start" in err
+    assert "NISSE_LLM_TOKEN" in err

nisse-dev/tests/test_evals.py

@@ -0,0 +1,129 @@
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    DelegationCompleted,
+    DelegationStarted,
+)
+from nisse_dev.evals import UNMARKED, collect, summarise, to_jsonl
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> StateStore:
+    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+
+
+def record_pair(
+    store: StateStore,
+    *,
+    brief: str = "add greet",
+    model: str = "claude-haiku-4-5",
+    role: str = "delegate",
+    status: str = "handback",
+) -> int:
+    store.append_event(
+        DelegationStarted(brief=brief, role=role, model=model, caps={"turns": 5})
+    )
+    store.append_event(
+        DelegationCompleted(
+            handback="what changed: greet.py",
+            artifacts=["greet.py"],
+            tokens=1200,
+            duration_s=9.5,
+            exit_status=status,
+        )
+    )
+    return store.read_events()[-1].id
+
+
+def test_pairs_a_start_with_its_completion(store: StateStore) -> None:
+    record_pair(store)
+    [record] = collect(store)
+    assert record.brief == "add greet"
+    assert record.model == "claude-haiku-4-5"
+    assert record.exit_status == "handback"
+    assert record.artifacts == ["greet.py"]
+    assert record.verdict == UNMARKED
+
+
+def test_unpaired_start_is_skipped(store: StateStore) -> None:
+    # A crash mid-delegation leaves a start with no completion; a half-formed
+    # record would pollute the set A16 exists to make trustworthy.
+    store.append_event(
+        DelegationStarted(brief="died", role="delegate", model="m", caps={})
+    )
+    assert collect(store) == []
+
+
+def test_unrelated_events_between_the_pair_are_ignored(store: StateStore) -> None:
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(AssistantText(text="working", model="m"))
+    store.append_event(
+        DelegationCompleted(
+            handback="done",
+            artifacts=[],
+            tokens=1,
+            duration_s=1.0,
+            exit_status="handback",
+        )
+    )
+    assert len(collect(store)) == 1
+
+
+def test_verdicts_attach_to_the_completion(store: StateStore) -> None:
+    event_id = record_pair(store)
+    store.mark_outcome(event_id, "good", "clean handback, tests passed")
+    [record] = collect(store)
+    assert record.verdict == "good"
+    assert "tests passed" in record.note
+
+
+def test_a_verdict_can_be_revised(store: StateStore) -> None:
+    # Judgements get revised; the log they describe does not.
+    event_id = record_pair(store)
+    store.mark_outcome(event_id, "good")
+    store.mark_outcome(event_id, "bad", "regressed later")
+    [record] = collect(store)
+    assert record.verdict == "bad"
+    assert record.note == "regressed later"
+
+
+def test_consults_and_delegations_share_the_set(store: StateStore) -> None:
+    record_pair(store, role="delegate")
+    record_pair(store, role="consultant", brief="which approach?")
+    roles = [r.role for r in collect(store)]
+    assert roles == ["delegate", "consultant"]
+
+
+def test_collect_can_start_after_a_marker(store: StateStore) -> None:
+    record_pair(store, brief="old")
+    marker = store.read_events()[-1].id
+    record_pair(store, brief="new")
+    assert [r.brief for r in collect(store, after=marker)] == ["new"]
+
+
+def test_jsonl_round_trips(store: StateStore) -> None:
+    record_pair(store)
+    line = to_jsonl(collect(store))
+    assert json.loads(line)["model"] == "claude-haiku-4-5"
+
+
+def test_summary_groups_by_model(store: StateStore) -> None:
+    good = record_pair(store, model="haiku")
+    record_pair(store, model="haiku", status="capped")
+    record_pair(store, model="opus")
+    store.mark_outcome(good, "good")
+
+    summary = summarise(collect(store))
+    assert summary["haiku"]["total"] == 2
+    assert summary["haiku"]["good"] == 1
+    assert summary["haiku"][UNMARKED] == 1
+    assert summary["opus"]["total"] == 1

nisse-dev/tests/test_executor.py

@@ -223,3 +223,95 @@ def test_multiple_calls_in_one_turn_all_run(store: StateStore, ws: Workspace) ->
         e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
     ]
     assert {r.call_id for r in results} == {"a", "b"}
+
+
+class StubDesk:
+    """Stands in for ExpertDesk without a model behind it."""
+
+    def __init__(self, answer: str = "use sqlite", error: Exception | None = None):
+        self.asked: list[str] = []
+        self._answer = answer
+        self._error = error
+
+    def ask(self, question: str, **kw: Any) -> Any:
+        self.asked.append(question)
+        if self._error is not None:
+            raise self._error
+
+        class _A:
+            def render(self_inner) -> str:
+                return self._answer
+
+        return _A()
+
+
+def test_ask_expert_is_offered_only_when_a_desk_exists(
+    store: StateStore, ws: Workspace
+) -> None:
+    client = ScriptedClient([])
+    build(client, store, ws).run(milestone=1)
+    assert all(t["function"]["name"] != "ask_expert" for t in client.calls[0]["tools"])
+
+    client2 = ScriptedClient([])
+    build(client2, store, ws, expert=StubDesk()).run(milestone=1)
+    names = [t["function"]["name"] for t in client2.calls[0]["tools"]]
+    assert names[-1] == "ask_expert"
+    # A12: in addition to the observable triggers, not instead of them.
+    assert "run_command" in names
+
+
+def test_ask_expert_answer_comes_back_as_a_tool_result(
+    store: StateStore, ws: Workspace
+) -> None:
+    desk = StubDesk(answer="prefer sqlite here")
+    client = ScriptedClient(
+        [
+            [call("ask_expert", question="sqlite or postgres?")],
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="done", model="m")],
+        ]
+    )
+    build(client, store, ws, expert=desk).run(milestone=1)
+
+    assert desk.asked == ["sqlite or postgres?"]
+    results = [
+        e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
+    ]
+    assert "prefer sqlite here" in results[0].result
+
+
+def test_consulting_is_not_doing_the_work(store: StateStore, ws: Workspace) -> None:
+    # Asking is not changing anything, so a completion claim backed only by
+    # consults is still challenged.
+    client = ScriptedClient(
+        [
+            [call("ask_expert", question="what should I do?")],
+            [AssistantText(text="I think it is done", model="m")],
+            [AssistantText(text="really done", model="m")],
+        ]
+    )
+    outcome = build(client, store, ws, expert=StubDesk()).run(milestone=1)
+    assert outcome.status == "escalate"
+
+
+def test_exhausted_budget_is_reported_to_the_model_not_raised(
+    store: StateStore, ws: Workspace
+) -> None:
+    from nisse_dev.expert import BudgetExhausted
+
+    desk = StubDesk(error=BudgetExhausted("consult budget of 3 is spent"))
+    client = ScriptedClient(
+        [
+            [call("ask_expert", question="again?")],
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="done", model="m")],
+        ]
+    )
+    outcome = build(client, store, ws, expert=desk).run(milestone=1)
+
+    assert outcome.status == "completed"  # it got on with the work
+    results = [
+        e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
+    ]
+    assert not results[0].success
+    assert "budget" in results[0].result

nisse-dev/tests/test_expert.py

@@ -0,0 +1,186 @@
+import io
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    CanonicalEvent,
+    DelegationCompleted,
+    DelegationStarted,
+    StoredEvent,
+)
+from nisse_dev.expert import (
+    DEFAULT_BUDGET,
+    BudgetExhausted,
+    ExpertAnswer,
+    ExpertDesk,
+    parse_answer,
+)
+from nisse_dev.llm import LLMError, RoleMap
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+
+ROLES = RoleMap(
+    executor="llama3.2",
+    planner="haiku",
+    delegate="haiku",
+    consultant="claude-haiku-4-5",
+)
+
+GOOD = '{"answer": "Use SQLite; it is enough here.", "confidence": "high", "caveats": "Assumes one writer."}'
+
+
+class ScriptedClient:
+    def __init__(self, replies: list[str]) -> None:
+        self._replies = replies
+        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._replies:
+            return [AssistantText(text="nothing more", model=model)]
+        return [AssistantText(text=self._replies.pop(0), model=model)]
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> StateStore:
+    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+
+
+def desk(client: Any, store: StateStore, **kw: Any) -> ExpertDesk:
+    return ExpertDesk(client, store, ROLES, **kw)
+
+
+# --- parsing (the contract is prompt-level, so parsing is the guard) ------
+
+
+def test_parses_a_clean_object() -> None:
+    parsed = parse_answer(GOOD)
+    assert parsed == ExpertAnswer(
+        answer="Use SQLite; it is enough here.",
+        confidence="high",
+        caveats="Assumes one writer.",
+    )
+
+
+def test_parses_a_fenced_block_because_models_wrap_json() -> None:
+    assert parse_answer(f"Here you go:\n```json\n{GOOD}\n```") is not None
+
+
+def test_missing_or_bad_fields_are_a_failed_turn() -> None:
+    assert parse_answer('{"answer": "", "confidence": "high"}') is None
+    assert parse_answer('{"answer": "x", "confidence": "certain"}') is None
+    assert parse_answer('{"confidence": "high"}') is None
+    assert parse_answer("no json at all") is None
+    assert parse_answer('["not", "an", "object"]') is None
+
+
+def test_absent_caveats_is_tolerated() -> None:
+    parsed = parse_answer('{"answer": "yes", "confidence": "low"}')
+    assert parsed is not None
+    assert parsed.caveats == ""
+
+
+# --- the desk ------------------------------------------------------------
+
+
+def test_ask_returns_a_validated_answer(store: StateStore) -> None:
+    client = ScriptedClient([GOOD])
+    answer = desk(client, store).ask("SQLite or Postgres?")
+    assert answer.confidence == "high"
+    assert "SQLite" in answer.render()
+
+
+def test_consult_is_stateless_and_uses_the_consultant_role(
+    store: StateStore,
+) -> None:
+    store.append_event(AssistantText(text="secret repo context", model="m"))
+    client = ScriptedClient([GOOD])
+    desk(client, store).ask("Is this API deprecated?")
+
+    call = client.calls[0]
+    # A14: no repo access, no session state — exactly the question, and none
+    # of the log the caller was working from.
+    assert [e.event.text for e in call["events"]] == ["Is this API deprecated?"]
+    assert all("secret repo context" not in e.event.text for e in call["events"])
+    assert call["tools"] is None
+    assert call["model"] == "claude-haiku-4-5"
+
+
+def test_unparseable_reply_is_retried_once_then_raises(store: StateStore) -> None:
+    client = ScriptedClient(["waffle", "more waffle"])
+    with pytest.raises(LLMError, match="no usable answer"):
+        desk(client, store).ask("q")
+    assert len(client.calls) == 2
+    assert "not a single valid JSON object" in client.calls[1]["system"]
+
+
+def test_retry_succeeds_after_a_bad_first_reply(store: StateStore) -> None:
+    client = ScriptedClient(["waffle", GOOD])
+    assert desk(client, store).ask("q").confidence == "high"
+
+
+def test_consults_are_recorded_as_composite_events(store: StateStore) -> None:
+    # A16 wants a consult/delegation eval set; reusing the pair gets consult
+    # evaluation for free.
+    desk(ScriptedClient([GOOD]), store).ask("q")
+    events = [e.event for e in store.read_events()]
+    started = next(e for e in events if isinstance(e, DelegationStarted))
+    completed = next(e for e in events if isinstance(e, DelegationCompleted))
+    assert started.role == "consultant"
+    assert started.brief == "q"
+    assert completed.exit_status == "answered"
+
+
+def test_a_failed_consult_is_recorded_too(store: StateStore) -> None:
+    with pytest.raises(LLMError):
+        desk(ScriptedClient(["waffle", "waffle"]), store).ask("q")
+    completed = [
+        e.event for e in store.read_events() if isinstance(e.event, DelegationCompleted)
+    ]
+    assert completed[0].exit_status == "failed"
+
+
+def test_budget_is_enforced_across_calls(store: StateStore) -> None:
+    client = ScriptedClient([GOOD] * 10)
+    d = desk(client, store, budget=2)
+    d.ask("one")
+    d.ask("two")
+    with pytest.raises(BudgetExhausted, match="budget of 2"):
+        d.ask("three")
+    assert len(client.calls) == 2  # the third never reached the model
+
+
+def test_budget_survives_a_restart(tmp_path: Path) -> None:
+    # Counted from the canonical log, not from memory, so a crash cannot
+    # hand a circling executor a fresh allowance.
+    db = tmp_path / "state.db"
+    with StateStore(db, oplog=OpLog("d", io.StringIO())) as store:
+        desk(ScriptedClient([GOOD]), store, budget=1).ask("one")
+    with (
+        StateStore(db, oplog=OpLog("d", io.StringIO())) as store,
+        pytest.raises(BudgetExhausted),
+    ):
+        desk(ScriptedClient([GOOD]), store, budget=1).ask("two")
+
+
+def test_budget_counts_only_after_the_task_marker(store: StateStore) -> None:
+    d = desk(ScriptedClient([GOOD] * 5), store, budget=1)
+    d.ask("earlier task")
+    marker = store.read_events()[-1].id
+    # A new task starts its allowance afresh from its own marker.
+    assert d.ask("new task", since_event_id=marker).confidence == "high"
+
+
+def test_default_budget_is_small() -> None:
+    assert DEFAULT_BUDGET <= 5  # A14: do not let this become a second subsystem

nisse-dev/tests/test_llm.py

@@ -11,7 +11,18 @@ from nisse_dev.canonical import (
     UserMessage,
 )
 from nisse_dev.envfile import load_env_file
-from nisse_dev.llm import RATE_LIMIT_RETRY_S, LiteLLMClient, LLMError, RoleMap
+from nisse_dev.llm import (
+    DEFAULT_MAX_OUTPUT_TOKENS,
+    RATE_LIMIT_RETRY_S,
+    AnthropicPassthroughClient,
+    BudgetExhausted,
+    LiteLLMClient,
+    LLMError,
+    RoleMap,
+    TruncatedResponse,
+    UpstreamUnavailable,
+    max_output_tokens,
+)
 
 EVENTS = [StoredEvent(id=1, ts="t", event=UserMessage(text="hi", source="m:1"))]
 
@@ -203,3 +214,142 @@ def test_load_env_file_missing_is_noop(tmp_path: Path) -> None:
     env: dict[str, str] = {}
     load_env_file(tmp_path / "nope.env", env)
     assert env == {}
+
+
+# --- truncation (silent until measured against Opus) ---------------------
+
+
+def test_openai_route_raises_on_truncation() -> None:
+    # finish_reason "length" means the answer was cut off; parsing it as
+    # complete yields a tool call with required arguments simply absent.
+    truncated = httpx.Response(
+        200,
+        json={
+            "model": "gpt-test-1",
+            "choices": [
+                {
+                    "finish_reason": "length",
+                    "message": {
+                        "content": None,
+                        "tool_calls": [
+                            {
+                                "id": "c",
+                                "type": "function",
+                                "function": {
+                                    "name": "submit_plan",
+                                    "arguments": '{"type": "arch"}',
+                                },
+                            }
+                        ],
+                    },
+                }
+            ],
+        },
+    )
+    client = make_client(lambda request: truncated)
+    with pytest.raises(TruncatedResponse, match="output limit"):
+        client.complete(EVENTS, tools=None, model="gpt-6-astra")
+
+
+def test_anthropic_route_raises_on_truncation() -> None:
+    # Observed for real: Opus spent 4,064 output tokens against a 4,096
+    # ceiling, and a longer plan came back as submit_plan with no content.
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "model": "claude-opus-5",
+                "stop_reason": "max_tokens",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "t",
+                        "name": "submit_plan",
+                        "input": {"type": "arch"},
+                    }
+                ],
+            },
+        )
+
+    client = AnthropicPassthroughClient(
+        "https://litellm.test",
+        "tok",
+        transport=httpx.MockTransport(handler),
+        sleep=lambda _s: None,
+    )
+    with pytest.raises(TruncatedResponse, match="output limit"):
+        client.complete(EVENTS, tools=None, model="claude-opus-5")
+
+
+def test_a_complete_answer_is_not_mistaken_for_truncation() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "model": "claude-opus-5",
+                "stop_reason": "tool_use",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "t",
+                        "name": "submit_plan",
+                        "input": {"type": "arch", "content": "a real plan"},
+                    }
+                ],
+            },
+        )
+
+    client = AnthropicPassthroughClient(
+        "https://litellm.test", "tok", transport=httpx.MockTransport(handler)
+    )
+    [call] = client.complete(EVENTS, tools=None, model="claude-opus-5")
+    assert call.arguments["content"] == "a real plan"
+
+
+def test_output_ceiling_is_configurable_and_generous_by_default() -> None:
+    # Output is billed by what is produced, so a low ceiling buys nothing and
+    # costs silent truncation.
+    assert DEFAULT_MAX_OUTPUT_TOKENS >= 8192
+    assert max_output_tokens({}) == DEFAULT_MAX_OUTPUT_TOKENS
+    assert max_output_tokens({"NISSE_MAX_OUTPUT_TOKENS": "2048"}) == 2048
+
+
+def test_truncation_is_an_llm_error_so_callers_already_handle_it() -> None:
+    # The planner retries, the executor and delegation gate. None of them
+    # need to learn a new exception type.
+    assert issubclass(TruncatedResponse, LLMError)
+
+
+def test_a_5xx_is_upstream_trouble() -> None:
+    client = make_client(lambda request: httpx.Response(503, text="bad gateway"))
+    with pytest.raises(UpstreamUnavailable, match="503"):
+        client.complete(EVENTS, tools=None, model="local-fast")
+
+
+def test_a_connection_failure_is_upstream_trouble() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("refused")
+
+    with pytest.raises(UpstreamUnavailable, match="request failed"):
+        make_client(handler).complete(EVENTS, tools=None, model="local-fast")
+
+
+def test_a_persistent_rate_limit_is_upstream_trouble() -> None:
+    # It cleared for neither attempt, so the caller should keep its work
+    # rather than discard it.
+    client = make_client(lambda request: httpx.Response(429, text="slow down"))
+    with pytest.raises(UpstreamUnavailable):
+        client.complete(EVENTS, tools=None, model="local-fast")
+
+
+def test_a_bad_request_is_not_upstream_trouble() -> None:
+    client = make_client(lambda request: httpx.Response(400, text="bad schema"))
+    with pytest.raises(LLMError) as excinfo:
+        client.complete(EVENTS, tools=None, model="local-fast")
+    assert not isinstance(excinfo.value, UpstreamUnavailable)
+
+
+def test_budget_refusal_is_a_kind_of_upstream_trouble() -> None:
+    # So one catch covers the family, but it can still be told apart where
+    # the difference matters — it will not clear within the hour.
+    assert issubclass(BudgetExhausted, UpstreamUnavailable)

nisse-dev/tests/test_planner.py

@@ -20,7 +20,7 @@ from nisse_dev.control import (
     PlanningError,
     SubmitPlan,
 )
-from nisse_dev.llm import RoleMap
+from nisse_dev.llm import BudgetExhausted, RoleMap
 from nisse_dev.oplog import OpLog
 from nisse_dev.planner import PLANNER_TOOLS, LLMPlanningAgent
 from nisse_dev.state import StateStore
@@ -279,3 +279,21 @@ def test_rejection_feedback_reaches_the_planner_via_the_log(
         isinstance(e, UserMessage) and "too vague" in e.text for e in replan_context
     )
     assert "v2" in transport.outbound[-1].text
+
+
+def test_budget_exhaustion_is_not_retried(store: StateStore) -> None:
+    """A budget will not reappear within a turn, so a retry buys nothing and
+    delays telling the operator."""
+
+    class BrokeClient:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        def complete(self, *args: Any, **kwargs: Any):
+            self.calls += 1
+            raise BudgetExhausted("daily budget exhausted")
+
+    client = BrokeClient()
+    with pytest.raises(BudgetExhausted):
+        LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
+    assert client.calls == 1  # not two

tests/integration/test_entry_points.py

@@ -1,5 +1,6 @@
 """Integration tests: run the installed console scripts as real subprocesses."""
 
+import os
 import subprocess
 
 import pytest
@@ -7,10 +8,35 @@ import pytest
 pytestmark = pytest.mark.integration
 
 
-@pytest.mark.parametrize("agent", ["nisse-dev", "nisse-ops"])
+def test_ops_entry_point_runs_and_greets() -> None:
-def test_entry_point_runs_and_greets(agent: str) -> None:
     result = subprocess.run(
-        [agent], capture_output=True, text=True, timeout=30, check=False
+        ["nisse-ops"], capture_output=True, text=True, timeout=30, check=False
     )
     assert result.returncode == 0
-    assert result.stdout == f"Hello from {agent}, the house spirit is awake.\n"
+    assert result.stdout == "Hello from nisse-ops, the house spirit is awake.\n"
+
+
+def test_dev_entry_point_refuses_to_start_unconfigured() -> None:
+    """Missing configuration must fail at startup, naming everything absent.
+
+    Run with a deliberately empty environment and a .env path that does not
+    exist: the real .env in the working directory holds live credentials, and
+    a test must never be one cwd away from starting a real agent.
+    """
+    result = subprocess.run(
+        ["nisse-dev"],
+        capture_output=True,
+        text=True,
+        timeout=30,
+        check=False,
+        env={
+            "PATH": os.environ["PATH"],
+            "HOME": os.environ["HOME"],
+            "NISSE_ENV": "/nonexistent/.env",
+        },
+    )
+    assert result.returncode == 2
+    assert "cannot start" in result.stderr
+    # All of them at once, not the first one it happened to reach.
+    for expected in ("NISSE_STATE_DB", "NISSE_MATRIX_TOKEN", "NISSE_GITEA_REPO"):
+        assert expected in result.stderr

tests/integration/test_expert_live.py

@@ -0,0 +1,60 @@
+"""A real consult over the normalized route (A14).
+
+The point is the prompt-level contract: a live model must produce something
+that parses against our local schema, since no API-level enforcement exists.
+"""
+
+import io
+import os
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.envfile import load_env_file
+from nisse_dev.evals import collect
+from nisse_dev.expert import BudgetExhausted, ExpertDesk
+from nisse_dev.llm import LiteLLMClient, RoleMap
+from nisse_dev.oplog import OpLog
+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 live frontier route configured ({SENTINEL} unset)",
+    ),
+]
+
+
+def desk(tmp_path: Path, budget: int = 2) -> tuple[ExpertDesk, StateStore]:
+    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()))
+    return ExpertDesk(LiteLLMClient.from_env(), store, roles, budget=budget), store
+
+
+def test_live_consult_parses_against_the_local_schema(tmp_path: Path) -> None:
+    expert, store = desk(tmp_path)
+
+    answer = expert.ask(
+        "For a single-process agent storing a few thousand rows of task state, "
+        "is SQLite sufficient, or is PostgreSQL warranted? Answer briefly."
+    )
+
+    assert answer.confidence in {"high", "medium", "low"}
+    assert len(answer.answer) > 20
+    # The consult is captured for A16 without any extra bookkeeping.
+    [record] = collect(store)
+    assert record.role == "consultant"
+    assert record.exit_status == "answered"
+
+
+def test_live_budget_stops_a_circling_caller(tmp_path: Path) -> None:
+    expert, _ = desk(tmp_path, budget=1)
+    expert.ask("Name one advantage of WAL mode in SQLite. Be brief.")
+    with pytest.raises(BudgetExhausted):
+        expert.ask("And another?")

tests/integration/test_model_capability.py

@@ -0,0 +1,171 @@
+"""Capability matrix across models (spec Part H, A15).
+
+Part H says the thresholds that depend on local capability — A12's failure
+count, A14's consult budget, A15's manifest size — must be tuned "at
+build-order step 5 on the real hardware". This is the measurement that makes
+that possible, rather than reasoning about model strength from reputation.
+
+What it measures is the thing that actually breaks: whether a model can hold
+the tool-call contract our loops depend on. llama3.2 could not — it returned
+`{` as an entire plan — and no amount of prompt work fixed it.
+
+Opt in with NISSE_CAPABILITY_MODELS, comma separated. Skipped otherwise,
+because it spends real tokens across several models.
+"""
+
+import os
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    StoredEvent,
+    ToolCall,
+    ToolResult,
+    UserMessage,
+)
+from nisse_dev.catalog import ModelCatalog, select_client
+from nisse_dev.envfile import load_env_file
+from nisse_dev.expert import parse_answer
+from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient, LLMError
+from nisse_dev.planner import PLANNER_TOOLS
+from nisse_dev.tools import EXECUTOR_TOOLS
+
+load_env_file(Path(__file__).resolve().parents[2] / ".env")
+
+SENTINEL = "NISSE_CAPABILITY_MODELS"
+MODELS = [m.strip() for m in os.environ.get(SENTINEL, "").split(",") if m.strip()]
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not MODELS, reason=f"set {SENTINEL} to a comma-separated model list"
+    ),
+]
+
+
+def clients() -> tuple[LiteLLMClient, AnthropicPassthroughClient]:
+    return LiteLLMClient.from_env(), AnthropicPassthroughClient.from_env()
+
+
+@pytest.fixture(scope="module")
+def catalog() -> ModelCatalog:
+    """The live catalog this key can see. /model/info is scoped per key, so
+    the probe list and the key have to agree."""
+    return ModelCatalog.discover(
+        os.environ["NISSE_LLM_BASE_URL"], os.environ["NISSE_LLM_TOKEN"]
+    )
+
+
+def user(text: str) -> list[StoredEvent]:
+    return [StoredEvent(id=1, ts="t", event=UserMessage(text=text, source="probe"))]
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_model_answers_at_all(model: str, catalog: ModelCatalog) -> None:
+    normalized, passthrough = clients()
+    client = select_client(model, normalized, passthrough, catalog)
+    produced = client.complete(
+        user("Reply with exactly one word: pong"), None, model, system="Be terse."
+    )
+    texts = [e.text for e in produced if isinstance(e, AssistantText)]
+    assert texts, f"{model} produced no text: {produced!r}"
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_model_holds_the_planner_contract(model: str, catalog: ModelCatalog) -> None:
+    """A4's gating is structural at the tool layer: a model that answers in
+    prose cannot be gated, because there is no plan to approve."""
+    normalized, passthrough = clients()
+    client = select_client(model, normalized, passthrough, catalog)
+    produced = client.complete(
+        user("Draft an architecture plan for a command-line slug generator."),
+        PLANNER_TOOLS,
+        model,
+        system=(
+            "You must answer by calling exactly one tool: submit_plan or "
+            "ask_human. Plain prose is not accepted."
+        ),
+    )
+    calls = [e for e in produced if isinstance(e, ToolCall)]
+    assert calls, f"{model} answered in prose instead of calling a tool"
+    assert calls[0].tool in {"submit_plan", "ask_human"}
+    if calls[0].tool == "submit_plan":
+        content = calls[0].arguments.get("content", "")
+        # llama3.2 emitted "{" here — structurally a tool call, substantively
+        # nothing. Length is a crude proxy, but it separates the two.
+        assert len(str(content)) > 80, (
+            f"{model} submitted a degenerate plan: {content!r}"
+        )
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_model_round_trips_a_tool_result(model: str, catalog: ModelCatalog) -> None:
+    """The executor loop feeds results back; a model that cannot consume its
+    own tool result cannot run a milestone."""
+    normalized, passthrough = clients()
+    client = select_client(model, normalized, passthrough, catalog)
+    events = user("List the files in the current directory using your tools.")
+    produced = client.complete(events, EXECUTOR_TOOLS, model, system="Use a tool.")
+    calls = [e for e in produced if isinstance(e, ToolCall)]
+    assert calls, f"{model} made no tool call"
+
+    next_id = 2
+    for event in produced:
+        events.append(StoredEvent(id=next_id, ts="t", event=event))
+        next_id += 1
+    events.append(
+        StoredEvent(
+            id=next_id,
+            ts="t",
+            event=ToolResult(
+                call_id=calls[0].call_id, result="app.py\ntests/", success=True
+            ),
+        )
+    )
+    follow_up = client.complete(events, EXECUTOR_TOOLS, model, system="Use a tool.")
+    assert follow_up, f"{model} produced nothing after its tool result"
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_model_holds_the_prompt_level_json_contract(
+    model: str, catalog: ModelCatalog
+) -> None:
+    """A14 contracts structured output at the prompt level because vendors
+    differ on JSON enforcement. This measures whether that holds per model."""
+    normalized, passthrough = clients()
+    client = select_client(model, normalized, passthrough, catalog)
+    produced = client.complete(
+        user("Is SQLite sufficient for a single-process agent's task state?"),
+        None,
+        model,
+        system=(
+            "Reply with a single JSON object and nothing else: "
+            '{"answer": "...", "confidence": "high"|"medium"|"low", '
+            '"caveats": "..."}'
+        ),
+    )
+    text = "\n".join(e.text for e in produced if isinstance(e, AssistantText))
+    assert parse_answer(text) is not None, (
+        f"{model} did not produce parseable JSON: {text[:200]!r}"
+    )
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_route_selection_matches_the_model(model: str, catalog: ModelCatalog) -> None:
+    """A non-Anthropic model must not be sent Anthropic-shaped bodies. This
+    was hardcoded until gpt-6-astra made it observable, and name-sniffed until
+    semantic aliases made the name stop meaning anything."""
+    normalized, passthrough = clients()
+    facts = catalog.facts(model)
+    chosen = select_client(model, normalized, passthrough, catalog)
+    if facts.provider == "anthropic":
+        assert chosen is passthrough
+    else:
+        assert chosen is normalized
+    # And the chosen route must actually work for this model.
+    try:
+        chosen.complete(user("say pong"), None, model, system="Be terse.")
+    except LLMError as exc:
+        pytest.fail(f"{model} failed on its selected route: {exc}")