Add the end-to-end test, run after each milestone (M1.6)

tests/integration/test_end_to_end.py drives the whole assembled agent
(real models, the Gitea sandbox, a real build host) through a milestone
with a scripted operator, and checks what bring-up run 3 checked by hand:
a PR holding only the work, pushed with the push credential; no
credential in the clone; the planner's view carrying facts, not the
executor's turns. It cleans up after itself on the sandbox and the build
host.

scripts/e2e.sh runs it on the bring-up build VM (or --local). It has its
own marker, e2e, kept out of check.sh and CI since it spends tokens.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

.gitea/workflows/ci.yml

@@ -40,7 +40,7 @@ jobs:
         run: uv run ruff format --check .
 
       - name: Unit tests
-        run: uv run pytest -m "not integration"
+        run: uv run pytest -m "not integration and not e2e"
 
       - name: Push key
         # The sandbox's deploy key, where nisse-test.yaml expects it.
@@ -59,4 +59,4 @@ jobs:
           NISSE_LLM_TOKEN: ${{ secrets.NISSE_LLM_TOKEN }}
           NISSE_MATRIX_TOKEN: ${{ secrets.NISSE_MATRIX_TOKEN }}
           NISSE_GITEA_TOKEN: ${{ secrets.NISSE_GITEA_TOKEN }}
-        run: uv run pytest -m integration -rs
+        run: uv run pytest -m "integration and not e2e" -rs

nisse-dev/src/nisse_dev/transport.py

@@ -55,6 +55,12 @@ class InMemoryTransport:
     def on_message(self, callback: Callable[[InboundMessage], None]) -> None:
         self._callback = callback
 
+    def start(self) -> None:
+        """Nothing to connect to: the transport's lifecycle, for the agent."""
+
+    def close(self) -> None:
+        """Nothing to disconnect from."""
+
     def receive(self, sender: str, text: str, original: str | None = None) -> None:
         assert self._callback is not None, "no message handler registered"
         self._callback(InboundMessage(sender, text, original))

pyproject.toml

@@ -17,6 +17,7 @@ testpaths = [
 addopts = "--strict-markers"
 markers = [
     "integration: end-to-end tests that exercise installed entry points or external services",
+    "e2e: the whole agent against real services, run after each milestone (scripts/e2e.sh)",
 ]
 
 [tool.ruff]

scripts/check.sh

@@ -6,5 +6,5 @@ cd "$(dirname "$0")/.."
 uv sync --all-packages
 uv run ruff check .
 uv run ruff format --check .
-uv run pytest -m "not integration"
+uv run pytest -m "not integration and not e2e"
-uv run pytest -m integration
+uv run pytest -m "integration and not e2e"

scripts/e2e.sh

@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+# The end-to-end test (Development Plan M1.6): the whole agent against real
+# services, driven by a scripted operator. Run after each milestone; it
+# spends model tokens and takes minutes, so it is not part of check.sh or CI.
+#
+#   scripts/e2e.sh            on the bring-up build VM (configured below)
+#   scripts/e2e.sh --local    on a throwaway local sshd instead
+#
+# Uses tests/integration/nisse-test.yaml and its secrets (.env at the repo
+# root), like the integration tests.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+
+if [ "${1:-}" = "--local" ]; then
+    shift
+else
+    export NISSE_E2E_BUILD_HOST=${NISSE_E2E_BUILD_HOST:-nisse-dev.garrett-hall.com}
+    export NISSE_E2E_BUILD_KEY=${NISSE_E2E_BUILD_KEY:-/root/nisse-e2e/ssh/id_ed25519}
+fi
+exec uv run pytest -m e2e -rs tests/integration/test_end_to_end.py "$@"

tests/integration/test_end_to_end.py

@@ -0,0 +1,202 @@
+"""The end-to-end test (Development Plan M1.6): the whole agent, assembled as
+it runs in production, against real models, the Gitea sandbox and a real
+build host, driven by a scripted operator.
+
+It automates the checks of bring-up run 3:
+- a milestone reaches a PR on the sandbox that contains only the work, no
+  build artifacts (#80), pushed with the push credential;
+- the clone on the build host holds no credential (#89);
+- after the milestone, the planner's view carries the orchestrator's facts
+  and none of the executor's turns (#78).
+
+Only the Matrix transport is replaced, by the in-memory one: the operator's
+Matrix credentials are not something a test should hold. Run it with
+scripts/e2e.sh after each milestone; it spends model tokens and takes minutes.
+"""
+
+import contextlib
+import os
+import shlex
+import uuid
+from collections.abc import Iterator
+from dataclasses import replace
+from fnmatch import fnmatch
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_dev.agent import Agent, build_agent
+from nisse_dev.canonical import UserMessage
+from nisse_dev.config import AgentConfig, BuildHost
+from nisse_dev.control import (
+    AWAIT_ARCH_SIGNOFF,
+    AWAIT_INPUT,
+    AWAIT_MILESTONE_SIGNOFF,
+    PR_OPEN,
+)
+from nisse_dev.execution import SSHHost
+from nisse_dev.gitea import GiteaError
+from nisse_dev.transport import InMemoryTransport
+
+pytestmark = [pytest.mark.integration, pytest.mark.e2e]
+
+OPERATOR = "@dongar:dongar.net"  # an allowlisted operator in nisse-test.yaml
+SETTLE_S = 1200  # a milestone, with real models, inside one message
+TASK = (
+    "Add a function `count_vowels(text)` to the sandbox package that returns "
+    "how many of a, e, i, o, u (either case) the text contains, with tests."
+)
+
+
+def build_host(ssh_host: SSHHost) -> BuildHost:
+    """The operator's build VM when scripts/e2e.sh names one, otherwise the
+    throwaway local sshd."""
+    address = os.environ.get("NISSE_E2E_BUILD_HOST")
+    if address:
+        key = Path(os.environ["NISSE_E2E_BUILD_KEY"])
+        return BuildHost(address=address, key=key)
+    return BuildHost(
+        address=ssh_host.address,
+        key=ssh_host._key,
+        user=ssh_host.user,
+        port=ssh_host.port,
+    )
+
+
+@pytest.fixture
+def agent(
+    live_config: AgentConfig,
+    ssh_host: SSHHost,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> Iterator[tuple[Agent, InMemoryTransport]]:
+    run = uuid.uuid4().hex[:8]
+    config = replace(
+        live_config,
+        state_db=tmp_path / "state.db",
+        workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
+        build_host=build_host(ssh_host),
+    )
+    transport = InMemoryTransport()
+    monkeypatch.setattr("nisse_dev.agent.MatrixTransport", lambda cfg: transport)
+    agent = None
+    try:
+        agent = build_agent(config)
+        agent.start()
+        yield agent, transport
+    finally:
+        if agent is not None:
+            agent.stop()
+        # The run's clone on the build host goes too, even if startup failed.
+        host = SSHHost(
+            config.build_host.address,
+            key=config.build_host.key,
+            user=config.build_host.user,
+            port=config.build_host.port,
+            state_dir=tmp_path / "ssh",
+        )
+        host.execute(f"rm -rf /tmp/nisse-e2e-{run}")
+
+
+class Operator:
+    """Says things to the agent and waits for it to finish with each."""
+
+    def __init__(self, agent: Agent, transport: InMemoryTransport) -> None:
+        self.agent = agent
+        self.transport = transport
+
+    def say(self, text: str) -> None:
+        self.transport.receive(OPERATOR, text)
+        # Messages are handled in order on the agent's worker, so a no-op
+        # behind this one returns when it, and the work it started, is done.
+        self.agent.worker.call(lambda: None, timeout=SETTLE_S)
+
+    def state(self) -> Any:
+        return self.agent.worker.call(self.agent.store.get_state)
+
+    def drive_to_pr(self, max_steps: int = 12) -> Any:
+        """Answer whatever the agent actually does: a live model may ask a
+        question at any gate, so a fixed script would break."""
+        for _ in range(max_steps):
+            state = self.state()
+            gate = state.data.get("gate") or {}
+            if state.state == PR_OPEN:
+                return state
+            if state.state == AWAIT_INPUT:
+                self.say("Use your best judgement and proceed.")
+            elif state.state == AWAIT_MILESTONE_SIGNOFF and gate.get("approved"):
+                self.say("start milestone 1")
+            elif state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF):
+                self.say("approve")
+            else:
+                raise AssertionError(f"stuck in {state.state}: {self.last()}")
+        raise AssertionError(f"no PR after {max_steps} steps: {self.last()}")
+
+    def last(self) -> str:
+        return self.transport.outbound[-1].text if self.transport.outbound else ""
+
+
+# The test's own idea of an artifact, independent of the code's list, so
+# emptying that list can't make this pass.
+def is_artifact(path: str) -> bool:
+    parts = path.split("/")
+    return (
+        "__pycache__" in parts
+        or ".pytest_cache" in parts
+        or fnmatch(parts[-1], "*.py[co]")
+    )
+
+
+def test_a_milestone_end_to_end(agent: tuple[Agent, InMemoryTransport]) -> None:
+    running, transport = agent
+    operator = Operator(running, transport)
+    gitea, repo = running.gitea, running.config.gitea_repo
+    pr: dict[str, Any] = {}
+    try:
+        # The task first, before any project: it must be kept, not lost.
+        operator.say(TASK)
+        operator.say(f"use {running.config.project}")
+        state = operator.drive_to_pr()
+        pr = state.data["pr"]
+        task = state.data["task_id"]
+
+        # --- the PR holds the work and only the work (D4, #80) ---
+        files = gitea._request("GET", f"/repos/{repo}/pulls/{pr['number']}/files")
+        changed = [f["filename"] for f in files]
+        assert changed, "the PR is empty"
+        assert not [p for p in changed if is_artifact(p)], changed
+        found = gitea._request("GET", f"/repos/{repo}/pulls/{pr['number']}")
+        assert found["head"]["ref"] == f"{task}-milestone-1"
+
+        # --- no credential in the clone on the build host (#89) ---
+        host = SSHHost(
+            running.config.build_host.address,
+            key=running.config.build_host.key,
+            user=running.config.build_host.user,
+            port=running.config.build_host.port,
+            state_dir=running.config.state_db.parent / "ssh",
+        )
+        config = host.execute(
+            f"cat {shlex.quote(str(running.config.workspace / '.git' / 'config'))}"
+        ).stdout
+        assert "PRIVATE KEY" not in config
+        assert running.config.gitea_token not in config
+
+        # --- the planner sees facts, not the executor's turns (#78) ---
+        operator.say("next milestone")
+        store = running.store
+        view = running.worker.call(lambda: store.transcript(task, "planner").view())
+        said = [e.event.text for e in view if isinstance(e.event, UserMessage)]
+        assert any("[orchestrator] milestone 1 finished" in t for t in said), said
+        assert any("[orchestrator] milestone 1 published" in t for t in said), said
+        assert not [e for e in view if e.actor == "executor"], "executor turns leaked"
+    finally:
+        # Leave the sandbox as it was: closes the PR and the plan issue.
+        operator.say("abandon")
+        if pr:
+            branch = gitea._request("GET", f"/repos/{repo}/pulls/{pr['number']}")[
+                "head"
+            ]["ref"]
+            with contextlib.suppress(GiteaError):  # cleanup is best effort
+                gitea._request("DELETE", f"/repos/{repo}/branches/{branch}")