dev/nisse
2
1 Fork 0
Code Issues 23 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

M2.8: switch over to the new loop
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 2m16s
Details
CI / e2e (pull_request) Successful in 10m51s
Details

Browse Source 
nisse-dev now runs only the conversational loop. The gated state machine
it was built alongside is gone:

- Removed control.py, planner.py, executor.py, approvals.py, gates.py
  (command_form moves to loop/reader.py), publish.py and poller.py, the
  old assembly in agent.py (its shared helpers move to loop/build.py),
  the store's state-machine view (get_state, transition), their tests,
  the old end-to-end and live planning/milestone/gate tests, and
  specs/orchestrator.qnt with its scenarios.
- Roles are driver, helper, expert and classifier; driver is required.
  loop:, roles.planner, roles.executor and tuning.executor are refused
  with what to use instead; tuning.driver takes the turn cap and failure
  limit.
- The benchmark and the capability matrix measure the driver: the
  loop's own system prompt and tools, the task's plan as the agreed
  restatement. Task plans end with finish_milestone, so earlier runs
  show as (old); stored runs' executor tuning reads as the driver's.
- A delegation on the loop records what its helper's alias resolved to
  (L9), as the old path's did.
- The dump shows a work item's revision and what it waits for.
- Behaviour the old tests covered and the loop shares stays covered:
  driver runs that fail on a model call, stop mid-turn, reset the
  failure count; go-aheads ignoring case and punctuation; a request too
  big to fit is never sent; the assembly tests, now on build_loop_agent.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
This commit is contained in:
dongarandClaude Opus 5.5 committed 2026-10-08 03:12:58 +00:00
1 parent 09232dcbdd
commit 6ae2375194
89 files changed
+951 -8592
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/integration/conftest.py
+2 -2
View File
@@ -33,8 +33,8 @@ def live_config() -> AgentConfig:
@pytest.fixture
def live_model(live_config: AgentConfig) -> str:
"""The model the live tests plan, execute and delegate with."""
return live_config.roles.planner
"""The model the live tests drive and delegate with."""
return live_config.roles.driver
@pytest.fixture
tests/integration/e2e_support.py
+18 -1
View File
@@ -10,6 +10,8 @@ from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from nisse_dev.config import BuildHost
from nisse_dev.execution import SSHHost
from nisse_dev.gitea import GiteaClient, GiteaError
from nisse_dev.llm import RoleMap
@@ -23,7 +25,7 @@ E2E_MODEL = "qwen3.8-27b-nothink"
def e2e_roles() -> RoleMap:
model = os.environ.get("NISSE_E2E_MODEL", "").strip() or E2E_MODEL
return RoleMap(executor=model, planner=model, helper=model, expert=model)
return RoleMap(driver=model, helper=model, expert=model)
def task(run: str) -> str:
@@ -177,3 +179,18 @@ def fresh_repo(
finally:
if not os.environ.get("NISSE_E2E_KEEP_REPOS"):
delete_repo(operator, repo.name)
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,
)
tests/integration/nisse-test.yaml
+1 -2
View File
@@ -36,8 +36,7 @@ llm:
# not local-model competence (measured separately by test_model_capability),
# so every role uses a model that holds the tool-call contract.
roles:
executor: claude-haiku-4-5
planner: claude-haiku-4-5
driver: claude-haiku-4-5
helper: claude-haiku-4-5
expert: claude-haiku-4-5
tests/integration/test_benchmark_live.py
+3 -3
View File
@@ -1,16 +1,16 @@
"""One live benchmark run: the simplest task, through the agent's own
client wiring, with the test config's executor."""
client wiring, with the test config's driver."""
from pathlib import Path
import pytest
from nisse_dev.agent import make_llm_clients
from nisse_dev.bench.runner import run_task
from nisse_dev.bench.tasks import load_tasks
from nisse_dev.catalog import ModelCatalog
from nisse_dev.config import AgentConfig
from nisse_dev.execution import SSHHost
from nisse_dev.loop.build import make_llm_clients
pytestmark = pytest.mark.integration
@@ -20,7 +20,7 @@ def test_the_simplest_task_passes_live(
) -> None:
catalog = ModelCatalog.discover(live_config.llm_base_url, live_config.llm_token)
normalized, _ = make_llm_clients(live_config, catalog)
facts = catalog.facts(live_config.roles.executor)
facts = catalog.facts(live_config.roles.driver)
[task] = load_tasks(["write-function"])
record = run_task(
tests/integration/test_crash_durability.py
+6 -5
View File
@@ -19,7 +19,8 @@ from nisse_dev.canonical import UserMessage
from nisse_dev.state import StateStore
store = StateStore(sys.argv[1])
store.transition("EXECUTING", cause="test", project="demo", data={"milestone": 2})
store.open_work_item("demo-1", title="t", scope="demo", position="RUNNING", data={"milestone": 2}, cause="test")
store.set_active("demo-1", cause="test")
store.begin_intent("push_branch", {"branch": "milestone-2"}, cause="test")
store.transcript("t", "operator").append(UserMessage(text="start milestone 2", source="m:1"))
os._exit(1) # die without closing anything, as a power cut would
@@ -38,10 +39,10 @@ def test_state_survives_process_death(tmp_path: Path) -> None:
assert result.returncode == 1, result.stderr
with StateStore(db) as store:
state = store.get_state()
assert state is not None
assert state.state == "EXECUTING"
assert state.data == {"milestone": 2}
item = store.active_item()
assert item is not None
assert item.position == "RUNNING"
assert item.data == {"milestone": 2}
report = reconcile_on_boot(store, {"push_branch": lambda intent: None})
assert report.action == "await_input"
tests/integration/test_end_to_end.py
-214
View File
@@ -1,214 +0,0 @@
"""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 e2e_support import e2e_roles, fresh_repo, operator_gitea
from e2e_support import task as make_task
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 GiteaClient, 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
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, str]]:
run = uuid.uuid4().hex[:8]
stack = contextlib.ExitStack()
operator = operator_gitea(live_config.gitea_url)
account = GiteaClient(live_config.gitea_url, live_config.gitea_token).whoami()
fresh = stack.enter_context(
fresh_repo(operator, run, tmp_path / "keys", agent=account)
)
config = replace(
live_config.with_project(
repo=fresh.name,
base_branch="main",
push_key=fresh.push_key,
workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
build_host=build_host(ssh_host),
),
roles=e2e_roles(),
state_db=tmp_path / "state.db",
)
transport = InMemoryTransport()
monkeypatch.setattr("nisse_dev.agent.MatrixTransport", lambda cfg: transport)
agent = None
try:
agent = build_agent(config)
agent.start()
yield agent, transport, run
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}")
stack.close() # the run's repository
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, str],
) -> None:
running, transport, run = 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(make_task(run))
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}")
tests/integration/test_end_to_end_loop.py
+33 -25
View File
@@ -22,7 +22,14 @@ from pathlib import Path
from typing import Any
import pytest
from e2e_support import dependent_task, e2e_roles, fresh_repo, operator_gitea, task
from e2e_support import (
build_host,
dependent_task,
e2e_roles,
fresh_repo,
operator_gitea,
task,
)
from nisse_dev.config import AgentConfig, LoopTuning
from nisse_dev.execution import SSHHost
@@ -54,8 +61,6 @@ def running(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> Iterator[tuple[RunningLoop, InMemoryTransport, str]]:
from test_end_to_end import build_host
run = uuid.uuid4().hex[:8]
stack = contextlib.ExitStack()
# Kept outside pytest's temp directories, which pytest prunes, so a
@@ -66,15 +71,18 @@ def running(
operator = operator_gitea(live_config.gitea_url)
account = GiteaClient(live_config.gitea_url, live_config.gitea_token).whoami()
fresh = stack.enter_context(fresh_repo(operator, run, keep / "keys", agent=account))
[project] = live_config.projects
project = replace(
project,
repo=fresh.name,
base_branch="main",
push_key=fresh.push_key,
workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
build_host=build_host(ssh_host),
)
config = replace(
live_config.with_project(
repo=fresh.name,
base_branch="main",
push_key=fresh.push_key,
workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
build_host=build_host(ssh_host),
),
loop="new",
live_config,
projects=(project,),
roles=e2e_roles(),
state_db=keep / "state.db",
# Short, so a bug's grace period runs out within the test.
@@ -97,10 +105,10 @@ def running(
if agent is not None:
agent.stop()
host = SSHHost(
config.build_host.address,
key=config.build_host.key,
user=config.build_host.user,
port=config.build_host.port,
project.build_host.address,
key=project.build_host.key,
user=project.build_host.user,
port=project.build_host.port,
state_dir=tmp_path / "ssh",
)
host.execute(f"rm -rf /tmp/nisse-e2e-{run}")
@@ -197,7 +205,7 @@ class Operator:
def merge_and_close(self, issue: int) -> None:
"""Approve and merge the item's PR, and see the agent close the item."""
gitea, repo = self.agent.gitea, self.agent.config.gitea_repo
gitea, repo = self.agent.gitea, self.agent.config.projects[0].repo
as_operator = operator_gitea(self.agent.config.gitea_url)
pr = loop_of(self.item(issue)).pr
with contextlib.suppress(GiteaError): # approval may not be required
@@ -222,7 +230,7 @@ class Operator:
def tidy(self, issues: list[int], prs: list[int]) -> None:
"""Leave the sandbox tidy: whatever is still open gets closed."""
gitea, repo = self.agent.gitea, self.agent.config.gitea_repo
gitea, repo = self.agent.gitea, self.agent.config.projects[0].repo
as_operator = operator_gitea(self.agent.config.gitea_url)
for issue in issues:
item = self.item(issue)
@@ -245,7 +253,7 @@ def test_the_slice_end_to_end(
) -> None:
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
gitea, repo = agent.gitea, agent.config.projects[0].repo
as_operator = operator_gitea(agent.config.gitea_url)
issue = None
pr = 0
@@ -297,7 +305,7 @@ def test_an_issue_blocked_on_another_waits_for_it(
closes; then B starts from a workspace that has A's work."""
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
gitea, repo = agent.gitea, agent.config.projects[0].repo
as_operator = operator_gitea(agent.config.gitea_url)
owner, name = repo.split("/")
issues: list[int] = []
@@ -360,7 +368,7 @@ def test_a_bug_goes_ahead_after_its_grace_period(
the grace period runs out, with no go from the operator."""
agent, transport, run = running
operator = Operator(agent, transport)
repo, base = agent.config.gitea_repo, agent.config.base_branch
repo, base = agent.config.projects[0].repo, agent.config.projects[0].base_branch
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
@@ -412,7 +420,7 @@ def test_a_plain_yes_is_a_go_ahead(
and the agent says so before acting."""
agent, transport, run = running
operator = Operator(agent, transport)
repo = agent.config.gitea_repo
repo = agent.config.projects[0].repo
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
@@ -454,7 +462,7 @@ def test_a_two_milestone_feature(
it; the item closes after both merge."""
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
gitea, repo = agent.gitea, agent.config.projects[0].repo
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
@@ -510,7 +518,7 @@ def test_a_conversation_becomes_filed_work(
is proposed, agreed, filed as a record issue, and started as agreed."""
agent, transport, run = running
operator = Operator(agent, transport)
gitea, repo = agent.gitea, agent.config.gitea_repo
gitea, repo = agent.gitea, agent.config.projects[0].repo
issues: list[int] = []
prs: list[int] = []
@@ -602,7 +610,7 @@ def test_a_conversation_becomes_a_new_project(
with fresh_repo(
as_operator, f"{run}-new", keep / "keys", agent=agent.account, seed=False
) as repo:
host = agent.config.build_host
host = agent.config.projects[0].build_host
filled = (
entry.group(1)
.replace(f"<owner>/{name}", repo.name)
@@ -646,7 +654,7 @@ def test_a_question_leaves_the_restatement_up(
nothing: the same revision stays up, and a go binds to it."""
agent, transport, run = running
operator = Operator(agent, transport)
repo = agent.config.gitea_repo
repo = agent.config.projects[0].repo
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
try:
tests/integration/test_expert_live.py
+1 -1
View File
@@ -22,7 +22,7 @@ pytestmark = pytest.mark.integration
def desk(tmp_path: Path, litellm: LiteLLMClient, live_model: str):
def make(budget: int = 2) -> tuple[ExpertDesk, StateStore]:
model = live_model
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
roles = RoleMap(driver=model, helper=model, expert=model)
store = StateStore(
tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())
)
tests/integration/test_gate_restart.py
-65
View File
@@ -1,65 +0,0 @@
"""Integration: a restart mid-gate resumes the open gate with its revision
history intact; unresolved intents gate the whole boot to AWAIT_INPUT
(architecture L8 and L3 together)."""
import io
from pathlib import Path
import pytest
from nisse_dev.control import (
AWAIT_ARCH_SIGNOFF,
AWAIT_INPUT,
AWAIT_MILESTONE_SIGNOFF,
Orchestrator,
StubAgent,
)
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport
pytestmark = pytest.mark.integration
def boot(db: Path) -> tuple[Orchestrator, StateStore, InMemoryTransport]:
store = StateStore(db, oplog=OpLog("nisse-dev", io.StringIO()))
transport = InMemoryTransport()
orch = Orchestrator(store, transport, StubAgent(), operators={"don"})
orch.start()
return orch, store, transport
def test_restart_mid_gate_resumes_with_revisions_intact(tmp_path: Path) -> None:
db = tmp_path / "state.db"
_, store, transport = boot(db)
transport.receive("don", "use demo")
transport.receive("don", "needs work") # arch gate now at rev-2
task_id = store.oplog.task_id
store.close()
_, store2, transport2 = boot(db)
state = store2.get_state()
assert state is not None
assert state.state == AWAIT_ARCH_SIGNOFF
assert store2.oplog.task_id == task_id # correlation survives restart
notices = [m.text for m in transport2.outbound]
assert any("rev-2" in n and "still open" in n for n in notices)
transport2.receive("don", "approve rev-1") # stale across the restart too
assert store2.get_state().state == AWAIT_ARCH_SIGNOFF
transport2.receive("don", "approve rev-2")
assert store2.get_state().state == AWAIT_MILESTONE_SIGNOFF
def test_restart_with_unresolved_intent_gates_boot(tmp_path: Path) -> None:
db = tmp_path / "state.db"
_, store, transport = boot(db)
transport.receive("don", "use demo")
store.begin_intent("push_branch", {"branch": "m1"}, cause="test")
store.close()
_, store2, transport2 = boot(db)
assert store2.get_state().state == AWAIT_INPUT
question = transport2.outbound[-1]
assert question.kind == "question"
assert "unresolved work" in question.text
tests/integration/test_gitea_live.py
+1 -1
View File
@@ -9,11 +9,11 @@ from pathlib import Path
import pytest
from e2e_support import fresh_repo, operator_gitea
from nisse_dev.agent import remote_url
from nisse_dev.config import AgentConfig
from nisse_dev.execution import SSHHost
from nisse_dev.git import GitError, GitRepo, SSHKey
from nisse_dev.gitea import GiteaClient, pull_request_reconciler
from nisse_dev.loop.build import remote_url
from nisse_dev.state import Intent
pytestmark = pytest.mark.integration
tests/integration/test_llm_live.py
+3 -3
View File
@@ -11,7 +11,7 @@ from nisse_dev.llm import LiteLLMClient
pytestmark = pytest.mark.integration
def test_executor_model_completes_via_litellm(
def test_the_driver_model_completes_via_litellm(
litellm: LiteLLMClient, live_config: AgentConfig
) -> None:
events = [
@@ -23,7 +23,7 @@ def test_executor_model_completes_via_litellm(
),
)
]
result = litellm.complete(events, tools=None, model=live_config.roles.executor)
result = litellm.complete(events, tools=None, model=live_config.roles.driver)
texts = [e.text for e in result if isinstance(e, AssistantText)]
assert texts, f"no assistant text in {result!r}"
# A plumbing check, so any answer will do: what the model chooses to say
@@ -41,7 +41,7 @@ def test_declared_limits_reach_a_budgeted_request(
roles = live_config.roles.pinned()
catalog.require_limits(roles) # every test role has a declared window
model = live_config.roles.executor
model = live_config.roles.driver
facts = catalog.facts(model)
client = BudgetedClient(
litellm,
tests/integration/test_milestone_live.py
-99
View File
@@ -1,99 +0,0 @@
"""End-to-end milestone against real models: plan with the planner role,
approve through the gate, then let the executor role actually create and test
a file on disk.
Uses the test config's planner model for every role: it needs a model that
holds the tool-call contract on both ends.
"""
import io
from pathlib import Path
import pytest
from nisse_dev.control import (
AWAIT_ARCH_SIGNOFF,
AWAIT_INPUT,
AWAIT_MILESTONE_SIGNOFF,
EXECUTING,
PR_OPEN,
Orchestrator,
)
from nisse_dev.execution import SSHHost
from nisse_dev.executor import LLMExecutor
from nisse_dev.llm import LiteLLMClient, RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.planner import LLMPlanningAgent
from nisse_dev.state import StateStore
from nisse_dev.tools import Workspace
from nisse_dev.transport import InMemoryTransport
pytestmark = pytest.mark.integration
def _drive_to_execution(
store: StateStore, transport: InMemoryTransport, max_steps: int = 10
) -> None:
"""Play the operator. A scripted approve/approve/start sequence is not
robust against a live model: ask_human is a legitimate move at any gate,
and a question shifts the state machine under a fixed script. So respond
to whatever the agent actually did."""
for _ in range(max_steps):
state = store.get_state()
assert state is not None
if state.state == EXECUTING:
return
gate = state.data.get("gate") or {}
if state.state == AWAIT_INPUT:
transport.receive("don", "Use your best judgement and proceed.")
elif state.state == AWAIT_MILESTONE_SIGNOFF and gate.get("approved"):
# The milestone runs to completion inside this command.
transport.receive("don", "start milestone 1")
return
elif state.state in (AWAIT_ARCH_SIGNOFF, AWAIT_MILESTONE_SIGNOFF):
transport.receive("don", "approve")
else:
raise AssertionError(f"unexpected state {state.state}")
raise AssertionError(
f"never reached execution; last message: {transport.outbound[-1]}"
)
def test_plan_approve_execute_produces_a_real_file(
tmp_path: Path, litellm: LiteLLMClient, live_model: str, ssh_host: SSHHost
) -> None:
# Both roles use the capable model here: the point is the whole gated
# loop over a real proxy, not local-model competence (measured separately).
model = live_model
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
client = litellm
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
workspace = Workspace(ssh_host, tmp_path / "work", timeout_s=60)
transport = InMemoryTransport()
orchestrator = Orchestrator(
store,
transport,
LLMPlanningAgent(client, roles),
operators={"don"},
roles=roles,
executor=LLMExecutor(client, roles, workspace, max_turns=10),
)
orchestrator.start()
transport.receive(
"don",
"use greet. Milestone 1: create greet.py in the workspace root defining "
"greet(name) returning 'Hello, <name>!', and verify it by running python.",
)
_drive_to_execution(store, transport)
state = store.get_state()
assert state is not None
# Never EXECUTING at rest: with no publisher a finished milestone waits in
# PR_OPEN, and one that stopped short waits for the operator.
assert state.state in (PR_OPEN, AWAIT_INPUT)
created = list(workspace.root.rglob("*.py"))
assert created, f"executor created no Python file; said: {transport.outbound[-1]}"
combined = "\n".join(p.read_text() for p in created)
assert "def greet" in combined
tests/integration/test_model_capability.py
+17 -20
View File
@@ -30,11 +30,10 @@ from nisse_dev.canonical import (
)
from nisse_dev.catalog import ModelCatalog, select_client
from nisse_dev.config import AgentConfig
from nisse_dev.executor import FINISH_TOOL
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
from nisse_dev.loop.tools import ASK_HUMAN, FINISH_MILESTONE, RESTATE
from nisse_dev.tools import EXECUTION_TOOLS
pytestmark = pytest.mark.integration
@@ -99,44 +98,42 @@ def test_model_answers_at_all(model: str, catalog: ModelCatalog, routes) -> None
assert texts, f"{model} produced no text: {produced!r}"
def test_model_holds_the_planner_contract(
def test_model_holds_the_restatement_contract(
model: str, catalog: ModelCatalog, routes
) -> None:
"""L3's approvals are structural at the tool layer: a model that answers in
prose cannot be gated, because there is no plan to approve."""
"""L3's go-ahead is structural at the tool layer: a driver that answers
in prose cannot be gated, because there is no restatement to agree to."""
normalized, passthrough = routes
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Draft an architecture plan for a command-line slug generator."),
PLANNER_TOOLS,
user("Issue #1: add a command-line slug generator."),
[RESTATE, ASK_HUMAN],
model,
system=(
"You must answer by calling exactly one tool: submit_plan or "
"You must answer by calling exactly one tool: restate or "
"ask_human. Plain prose is not accepted."
),
)
assert_answered(client, model, catalog)
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", "")
assert calls[0].tool in {"restate", "ask_human"}
if calls[0].tool == "restate":
content = calls[0].arguments.get("text", "")
# 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}"
)
assert len(str(content)) > 80, f"{model} restated degenerately: {content!r}"
def test_model_round_trips_a_tool_result(
model: str, catalog: ModelCatalog, routes
) -> None:
"""The executor loop feeds results back; a model that cannot consume its
"""The driver's run feeds results back; a model that cannot consume its
own tool result cannot run a milestone."""
normalized, passthrough = routes
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.")
produced = client.complete(events, EXECUTION_TOOLS, model, system="Use a tool.")
calls = [e for e in produced if isinstance(e, ToolCall)]
assert calls, f"{model} made no tool call"
@@ -154,7 +151,7 @@ def test_model_round_trips_a_tool_result(
)
)
assert_answered(client, model, catalog)
follow_up = client.complete(events, EXECUTOR_TOOLS, model, system="Use a tool.")
follow_up = client.complete(events, EXECUTION_TOOLS, model, system="Use a tool.")
assert follow_up, f"{model} produced nothing after its tool result"
@@ -203,7 +200,7 @@ def test_route_selection_matches_the_model(
assert_answered(chosen, model, catalog)
FINISH_TOOLS = [*EXECUTOR_TOOLS, FINISH_TOOL]
FINISH_TOOLS = [*EXECUTION_TOOLS, FINISH_MILESTONE]
OFFERED = {tool["function"]["name"] for tool in FINISH_TOOLS}
@@ -212,7 +209,7 @@ def test_model_calls_only_the_tools_it_was_offered(
) -> None:
"""Measured on a server that did not enforce tool_choice: a model asked
to chat called "chat", "say_hello" and "respond", none of them offered.
The executor counts those as failures; this measures how often."""
The driver answers those as failures; this measures how often."""
normalized, passthrough = routes
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
tests/integration/test_passthrough_live.py
+3 -3
View File
@@ -16,7 +16,7 @@ from nisse_dev.config import AgentConfig
from nisse_dev.llm import AnthropicPassthroughClient
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.tools import EXECUTOR_TOOLS
from nisse_dev.tools import EXECUTION_TOOLS
pytestmark = pytest.mark.integration
@@ -54,7 +54,7 @@ def test_passthrough_round_trips_a_tool_call(
client = passthrough
produced = client.complete(
store.transcript("t", "helper").read(),
EXECUTOR_TOOLS,
EXECUTION_TOOLS,
live_model,
system="Use a tool. Do not answer in prose.",
)
@@ -69,7 +69,7 @@ def test_passthrough_round_trips_a_tool_call(
ToolResult(call_id=calls[0].call_id, result="app.py\ntests/", success=True)
)
follow_up = client.complete(
store.transcript("t", "helper").read(), EXECUTOR_TOOLS, live_model
store.transcript("t", "helper").read(), EXECUTION_TOOLS, live_model
)
assert follow_up, "the vendor rejected or ignored the tool_result turn"
tests/integration/test_planning_live.py
-47
View File
@@ -1,47 +0,0 @@
"""Live planning turn: a real model must answer with a submit_plan or
ask_human tool call, not prose.
Plans with the test config's planner role. The planner role is a frontier
model today (Development Plan §2), and small local models do not hold the tool-call
contract reliably, so that role must be one that does.
"""
import io
from pathlib import Path
import pytest
from nisse_dev.canonical import UserMessage
from nisse_dev.control import AskHuman, SubmitPlan
from nisse_dev.llm import LiteLLMClient, RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.planner import LLMPlanningAgent
from nisse_dev.state import StateStore
pytestmark = pytest.mark.integration
def test_live_planning_turn_returns_a_tool_action(
tmp_path: Path, litellm: LiteLLMClient, live_model: str
) -> None:
model = live_model
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
store.transcript("t", "operator").append(
UserMessage(
text=(
"Project 'hello-cli': a Python script that prints a greeting. "
"Draft the architecture plan."
),
source="live-test",
)
)
agent = LLMPlanningAgent(litellm, roles)
action = agent.plan(store.transcript("t", "planner"), "arch", "hello-cli")
assert isinstance(action, SubmitPlan | AskHuman)
text = action.content if isinstance(action, SubmitPlan) else action.question
assert text.strip(), "tool call carried empty content"
# The turn must be durable regardless of what the model said (L8/L7).
assert len(store.transcript("t", "planner").read()) > 1