Add the composition root and a runnable entry point
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m14s

Nothing had ever assembled the agent. main.py was still the hello-world
placeholder from the scaffold, and no code outside the test suite
constructed an Orchestrator — which is how two fully built, fully
tested components came to be reachable from nothing: the delegation
harness, and the Gitea comment poller. Unit tests that construct a
component directly cannot catch that; only assembly can.

- config.py validates every variable before anything is built and
  names all of them at once, rather than failing at first use twenty
  minutes into a milestone on the call that mattered. Whitespace-only
  counts as missing, since a token of spaces passes a truthiness check
  and fails at the first request instead
- agent.py builds the whole graph and owns startup and shutdown order:
  reconcile before accepting commands, and on stop, stop watching,
  then stop talking, then close the store last
- The Gitea account is read via whoami rather than configured: it names
  the identity every permission is evaluated against, and proves the
  token works before anything depends on it
- Poller lifecycle is now a state-machine concern: watch the plan issue
  while a gate stands, the PR while it is open, and stop on leaving.
  The orchestrator owns when to watch; an injected factory owns how to
  build one
- The executor reports each turn, because it runs synchronously inside
  the transport callback: without it the agent is silent for a whole
  milestone and a working run looks exactly like a hung one
- main.py loads .env, builds, and shuts down cleanly on SIGINT/SIGTERM

The entry-point tests changed shape: nisse-dev no longer greets, it
refuses to start unconfigured and says everything that is missing. Both
tests now run with a deliberately empty environment and a .env path
that does not exist, because the working directory's .env holds live
credentials and a test must never be one cwd away from starting a real
agent.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Opus 5 committed 2026-09-08 22:26:26 +00:00
1 parent 6331fcafd9
commit 2fb25b8195
11 files changed
+694 -15

No files matched your search

+179
View File
@@ -0,0 +1,179 @@
"""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, prepare_workspace
from nisse_dev.config import AgentConfig
from nisse_dev.git import GitRepo
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.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()
+91
View File
@@ -0,0 +1,91 @@
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
+21 -5
View File
@@ -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:
main()
assert (
capsys.readouterr().out == "Hello from nisse-dev, the house spirit is awake.\n"
)
def test_main_refuses_to_start_without_configuration(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
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