Make the config file the only home for settings
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 51s

With no production instance yet, there is no reason to carry two ways
of configuring the agent. Every environment variable that duplicated a
config key is gone; the environment now supplies only the secrets the
file references by ${…}, plus NISSE_CONFIG and NISSE_ENV to find the
files.

- AgentConfig loads from a file (load, from_file) or a parsed tree
  (from_tree, for tests); a missing file is an error naming the example
- The from_env constructors on LiteLLMClient, AnthropicPassthroughClient,
  GiteaClient, MatrixConfig and RoleMap are removed, with the env
  reading for max output tokens; clients take their settings from config
- The live integration tests share a committed agent config,
  tests/integration/nisse-test.yaml, aimed at test resources, and one
  conftest that loads it and builds clients. NISSE_LIVE_PLANNER_MODEL
  becomes its planner role; NISSE_CAPABILITY_MODELS stays as a per-run
  switch
- A missing secret now fails the live tests instead of skipping them
- CI passes only the three tokens
- .env.example is secrets only; spec A5 says config comes from the file

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Opus 5.5 committed 2026-09-23 23:25:58 +00:00
1 parent df449c7e34
commit 15d69799bf
28 files changed
+571 -777

No files matched your search

+62
View File
@@ -0,0 +1,62 @@
"""Shared setup for the live integration tests.
They run against real services, configured by nisse-test.yaml beside this
file: an ordinary agent config aimed at test resources, committed to the
repository. Its secrets are ${…} references, supplied by CI as environment
variables and locally by .env at the repository root.
A missing secret is a failure, not a skip. A half-configured environment must
not pass quietly as a skip: that is how a broken CI credential hides.
"""
from pathlib import Path
import pytest
from nisse_dev.config import AgentConfig
from nisse_dev.envfile import load_env_file
from nisse_dev.gitea import GiteaClient
from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient
REPO = Path(__file__).resolve().parents[2]
TEST_CONFIG = Path(__file__).parent / "nisse-test.yaml"
# At import, so per-run switches such as NISSE_CAPABILITY_MODELS can be set
# in .env too. Real environment variables win over the file.
load_env_file(REPO / ".env")
@pytest.fixture(scope="session")
def live_config() -> AgentConfig:
return AgentConfig.from_file(TEST_CONFIG)
@pytest.fixture
def live_model(live_config: AgentConfig) -> str:
"""The model the live tests plan, execute and delegate with."""
return live_config.roles.planner
@pytest.fixture
def litellm(live_config: AgentConfig) -> LiteLLMClient:
return LiteLLMClient(
live_config.llm_base_url,
live_config.llm_token,
max_tokens=live_config.max_output_tokens,
)
@pytest.fixture
def passthrough(live_config: AgentConfig) -> AnthropicPassthroughClient:
return AnthropicPassthroughClient(
live_config.llm_base_url,
live_config.llm_token,
max_tokens=live_config.max_output_tokens,
)
@pytest.fixture
def gitea(live_config: AgentConfig) -> GiteaClient:
client = GiteaClient(live_config.gitea_url, live_config.gitea_token)
yield client
client.close()
+40
View File
@@ -0,0 +1,40 @@
# Agent config for the live integration tests, aimed at test resources: the
# testing identities, the dedicated test room, and the sandbox repo. Never
# point it at production.
#
# Secrets are ${…} references. CI passes them as environment variables;
# locally they come from .env at the repository root. A missing one fails
# the live tests that need it, rather than skipping them, so a broken
# credential cannot hide as a skip.
# Unused by the live tests, which work in pytest's temporary directories;
# present because every agent config has them.
state_db: /tmp/nisse-test/state.db
workspace: /tmp/nisse-test/work
project: nisse-dev-testing
llm:
base_url: https://litellm.dongar.net
token: ${NISSE_LLM_TOKEN}
# The live tests exercise the gated loop and the routes over a real proxy,
# 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
delegate: claude-haiku-4-5
consultant: claude-haiku-4-5
matrix:
homeserver: https://matrix.dongar.net
user_id: "@nisse-dev-testing:dongar.net"
token: ${NISSE_MATRIX_TOKEN}
room_id: "!UxpXhTlQsFIeNZwkgv:dongar.net" # the test room, never production
operators: ["@dongar:dongar.net"]
gitea:
url: https://gitea.dongar.net
token: ${NISSE_GITEA_TOKEN}
repo: dongar/nisse-dev-testing-sandbox # the sandbox, never a real project
operators: [dongar]
+18 -20
View File
@@ -2,30 +2,18 @@
working code and a structured handback out."""
import io
import os
from pathlib import Path
import pytest
from nisse_dev.delegation import Caps, Delegation
from nisse_dev.envfile import load_env_file
from nisse_dev.git import GitRepo
from nisse_dev.llm import AnthropicPassthroughClient
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.tools import Workspace
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)",
),
]
pytestmark = pytest.mark.integration
BRIEF = """Create `slugify.py` in the workspace root containing a function
`slugify(text)` that lowercases the text and replaces every run of
@@ -34,7 +22,9 @@ hyphen. Then create `test_slugify.py` with at least two assertions covering
it, and run `python -m pytest test_slugify.py -q` until it passes."""
def build(tmp_path: Path, caps: Caps) -> tuple[Delegation, GitRepo, StateStore]:
def build(
tmp_path: Path, caps: Caps, client: AnthropicPassthroughClient, model: str
) -> tuple[Delegation, GitRepo, StateStore]:
repo = GitRepo(tmp_path / "work")
repo.init()
repo.configure_identity("nisse-dev-testing", "n@t.test")
@@ -42,18 +32,22 @@ def build(tmp_path: Path, caps: Caps) -> tuple[Delegation, GitRepo, StateStore]:
repo.commit_all("seed")
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
delegation = Delegation(
AnthropicPassthroughClient.from_env(),
client,
store,
Workspace(repo.root, timeout_s=60),
repo,
model=os.environ[SENTINEL],
model=model,
caps=caps,
)
return delegation, repo, store
def test_delegate_completes_a_scoped_subtask(tmp_path: Path) -> None:
delegation, repo, _store = build(tmp_path, Caps(turns=12, wall_clock_s=300.0))
def test_delegate_completes_a_scoped_subtask(
tmp_path: Path, passthrough: AnthropicPassthroughClient, live_model: str
) -> None:
delegation, repo, _store = build(
tmp_path, Caps(turns=12, wall_clock_s=300.0), passthrough, live_model
)
result = delegation.run(brief=BRIEF, branch="live/delegation-1")
@@ -76,10 +70,14 @@ def test_delegate_completes_a_scoped_subtask(tmp_path: Path) -> None:
assert check.success, f"delegate's own tests do not pass: {check.result[:400]}"
def test_a_cap_breach_discards_the_work(tmp_path: Path) -> None:
def test_a_cap_breach_discards_the_work(
tmp_path: Path, passthrough: AnthropicPassthroughClient, live_model: str
) -> None:
# One turn cannot finish the brief, so this exercises the abort path
# against a real model rather than a scripted one.
delegation, repo, _ = build(tmp_path, Caps(turns=1, wall_clock_s=120.0))
delegation, repo, _ = build(
tmp_path, Caps(turns=1, wall_clock_s=120.0), passthrough, live_model
)
result = delegation.run(brief=BRIEF, branch="live/delegation-2")
+11 -6
View File
@@ -2,6 +2,7 @@
import os
import subprocess
from pathlib import Path
import pytest
@@ -16,27 +17,31 @@ def test_ops_entry_point_runs_and_greets() -> None:
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.
def test_dev_entry_point_refuses_to_start_unconfigured(tmp_path: Path) -> None:
"""An incomplete config 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.
Run from an empty directory, with an empty config file and a .env path
that does not exist: the repository's .env holds live credentials, and a
test must never be one cwd away from starting a real agent.
"""
config = tmp_path / "nisse-dev.yaml"
config.write_text("")
result = subprocess.run(
["nisse-dev"],
capture_output=True,
text=True,
timeout=30,
check=False,
cwd=tmp_path,
env={
"PATH": os.environ["PATH"],
"HOME": os.environ["HOME"],
"NISSE_ENV": "/nonexistent/.env",
"NISSE_CONFIG": str(config),
},
)
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"):
for expected in ("state_db", "matrix.token", "gitea.repo"):
assert expected in result.stderr
+16 -22
View File
@@ -5,40 +5,34 @@ 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)",
),
]
pytestmark = pytest.mark.integration
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
@pytest.fixture
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, delegate=model, consultant=model)
store = StateStore(
tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())
)
return ExpertDesk(litellm, store, roles, budget=budget), store
return make
def test_live_consult_parses_against_the_local_schema(tmp_path: Path) -> None:
expert, store = desk(tmp_path)
def test_live_consult_parses_against_the_local_schema(desk) -> None:
expert, store = desk()
answer = expert.ask(
"For a single-process agent storing a few thousand rows of task state, "
@@ -53,8 +47,8 @@ def test_live_consult_parses_against_the_local_schema(tmp_path: Path) -> None:
assert record.exit_status == "answered"
def test_live_budget_stops_a_circling_caller(tmp_path: Path) -> None:
expert, _ = desk(tmp_path, budget=1)
def test_live_budget_stops_a_circling_caller(desk) -> None:
expert, _ = desk(budget=1)
expert.ask("Name one advantage of WAL mode in SQLite. Be brief.")
with pytest.raises(BudgetExhausted):
expert.ask("And another?")
+9 -24
View File
@@ -5,29 +5,17 @@ Cleans up after itself (closes the PR, deletes the branch) so repeated runs
do not silt up the sandbox.
"""
import os
import uuid
from pathlib import Path
import pytest
from nisse_dev.envfile import load_env_file
from nisse_dev.config import AgentConfig
from nisse_dev.git import GitError, GitRepo, authenticated_url
from nisse_dev.gitea import GiteaClient, pull_request_reconciler
from nisse_dev.state import Intent
load_env_file(Path(__file__).resolve().parents[2] / ".env")
SENTINEL = "NISSE_GITEA_URL"
ALSO_REQUIRED = ("NISSE_GITEA_TOKEN", "NISSE_GITEA_TEST_REPO")
pytestmark = [
pytest.mark.integration,
pytest.mark.skipif(
not os.environ.get(SENTINEL),
reason=f"no live Gitea configured ({SENTINEL} unset)",
),
]
pytestmark = pytest.mark.integration
def intent_for(branch: str) -> Intent:
@@ -41,16 +29,13 @@ def intent_for(branch: str) -> Intent:
)
def test_push_open_pr_and_reconcile(tmp_path: Path) -> None:
missing = [var for var in ALSO_REQUIRED if not os.environ.get(var)]
assert not missing, (
f"{SENTINEL} is set, so the live Gitea path is expected to work, "
f"but these are missing: {', '.join(missing)}"
)
base_url = os.environ[SENTINEL]
token = os.environ["NISSE_GITEA_TOKEN"]
repo_path = os.environ["NISSE_GITEA_TEST_REPO"]
client = GiteaClient.from_env()
def test_push_open_pr_and_reconcile(
tmp_path: Path, live_config: AgentConfig, gitea: GiteaClient
) -> None:
base_url = live_config.gitea_url
token = live_config.gitea_token
repo_path = live_config.gitea_repo # the sandbox
client = gitea
account = client.whoami()
assert "testing" in account, (
+7 -33
View File
@@ -1,41 +1,17 @@
"""Live smoke test against the real LiteLLM proxy.
NISSE_LLM_BASE_URL alone decides whether this runs: if it is set, the live
path is expected to work, and anything else missing is a *failure*, not a
skip. A half-configured environment must not pass quietly as a skip — that is
how a broken CI credential hides.
"""
import os
from pathlib import Path
"""Live smoke test against the real LiteLLM proxy."""
import pytest
from nisse_dev.canonical import AssistantText, StoredEvent, UserMessage
from nisse_dev.envfile import load_env_file
from nisse_dev.config import AgentConfig
from nisse_dev.llm import LiteLLMClient
load_env_file(Path(__file__).resolve().parents[2] / ".env")
SENTINEL = "NISSE_LLM_BASE_URL"
ALSO_REQUIRED = ("NISSE_LLM_TOKEN", "NISSE_ROLE_EXECUTOR")
pytestmark = [
pytest.mark.integration,
pytest.mark.skipif(
not os.environ.get(SENTINEL),
reason=f"no live LLM configured ({SENTINEL} unset)",
),
]
pytestmark = pytest.mark.integration
def test_executor_model_completes_via_litellm() -> None:
missing = [var for var in ALSO_REQUIRED if not os.environ.get(var)]
assert not missing, (
f"{SENTINEL} is set, so the live LLM path is expected to work, "
f"but these are missing: {', '.join(missing)}"
)
client = LiteLLMClient.from_env()
def test_executor_model_completes_via_litellm(
litellm: LiteLLMClient, live_config: AgentConfig
) -> None:
events = [
StoredEvent(
id=1,
@@ -45,9 +21,7 @@ def test_executor_model_completes_via_litellm() -> None:
),
)
]
result = client.complete(
events, tools=None, model=os.environ["NISSE_ROLE_EXECUTOR"]
)
result = litellm.complete(events, tools=None, model=live_config.roles.executor)
texts = [e.text for e in result if isinstance(e, AssistantText)]
assert texts, f"no assistant text in {result!r}"
assert "pong" in texts[0].lower()
+6 -33
View File
@@ -1,36 +1,14 @@
"""Live smoke test against the real homeserver, bound strictly to the
dedicated test room (never the production room).
dedicated test room that nisse-test.yaml names (never the production room)."""
NISSE_MATRIX_HOMESERVER alone decides whether this runs: if it is set, the
live path is expected to work, and anything else missing is a *failure*, not
a skip. A half-configured environment must not pass quietly as a skip — that
is how a broken CI credential hides."""
import os
import uuid
from pathlib import Path
import pytest
from nisse_dev.envfile import load_env_file
from nisse_dev.config import AgentConfig
from nisse_dev.matrix import MatrixConfig, MatrixTransport
load_env_file(Path(__file__).resolve().parents[2] / ".env")
SENTINEL = "NISSE_MATRIX_HOMESERVER"
ALSO_REQUIRED = (
"NISSE_MATRIX_USER_ID",
"NISSE_MATRIX_TOKEN",
"NISSE_MATRIX_TEST_ROOM_ID",
)
pytestmark = [
pytest.mark.integration,
pytest.mark.skipif(
not os.environ.get(SENTINEL),
reason=f"no live homeserver configured ({SENTINEL} unset)",
),
]
pytestmark = pytest.mark.integration
def _assert_testing_identity(transport: MatrixTransport, config: MatrixConfig) -> None:
@@ -46,7 +24,7 @@ def _assert_testing_identity(transport: MatrixTransport, config: MatrixConfig) -
whoami.raise_for_status()
actual = whoami.json()["user_id"]
assert actual == config.user_id, (
f"NISSE_MATRIX_USER_ID is {config.user_id} but the token belongs to "
f"matrix.user_id is {config.user_id} but the token belongs to "
f"{actual}; the self-message filter would not match"
)
assert "testing" in actual, (
@@ -54,13 +32,8 @@ def _assert_testing_identity(transport: MatrixTransport, config: MatrixConfig) -
)
def test_send_lands_in_test_room() -> None:
missing = [var for var in ALSO_REQUIRED if not os.environ.get(var)]
assert not missing, (
f"{SENTINEL} is set, so the live Matrix path is expected to work, "
f"but these are missing: {', '.join(missing)}"
)
config = MatrixConfig.from_env(room_var="NISSE_MATRIX_TEST_ROOM_ID")
def test_send_lands_in_test_room(live_config: AgentConfig) -> None:
config = live_config.matrix
marker = f"transport smoke test {uuid.uuid4().hex}"
transport = MatrixTransport(config)
try:
+8 -22
View File
@@ -2,12 +2,11 @@
approve through the gate, then let the executor role actually create and test
a file on disk.
Gated on NISSE_LIVE_PLANNER_MODEL like the planning test — it needs a model
that holds the tool-call contract on both ends.
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
import os
from pathlib import Path
import pytest
@@ -19,7 +18,6 @@ from nisse_dev.control import (
EXECUTING,
Orchestrator,
)
from nisse_dev.envfile import load_env_file
from nisse_dev.executor import LLMExecutor
from nisse_dev.llm import LiteLLMClient, RoleMap
from nisse_dev.oplog import OpLog
@@ -28,17 +26,7 @@ from nisse_dev.state import StateStore
from nisse_dev.tools import Workspace
from nisse_dev.transport import InMemoryTransport
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 planner route configured ({SENTINEL} unset)",
),
]
pytestmark = pytest.mark.integration
def _drive_to_execution(
@@ -67,16 +55,14 @@ def _drive_to_execution(
)
def test_plan_approve_execute_produces_a_real_file(tmp_path: Path) -> None:
assert os.environ.get("NISSE_LLM_TOKEN"), (
f"{SENTINEL} is set, so the live path is expected to work, "
"but NISSE_LLM_TOKEN is missing"
)
def test_plan_approve_execute_produces_a_real_file(
tmp_path: Path, litellm: LiteLLMClient, live_model: str
) -> 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 = os.environ[SENTINEL]
model = live_model
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
client = LiteLLMClient.from_env()
client = litellm
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
workspace = Workspace(tmp_path / "work", timeout_s=60)
transport = InMemoryTransport()
+24 -20
View File
@@ -14,7 +14,6 @@ because it spends real tokens across several models.
"""
import os
from pathlib import Path
import pytest
@@ -26,14 +25,12 @@ from nisse_dev.canonical import (
UserMessage,
)
from nisse_dev.catalog import ModelCatalog, select_client
from nisse_dev.envfile import load_env_file
from nisse_dev.config import AgentConfig
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()]
@@ -45,17 +42,18 @@ pytestmark = [
]
def clients() -> tuple[LiteLLMClient, AnthropicPassthroughClient]:
return LiteLLMClient.from_env(), AnthropicPassthroughClient.from_env()
@pytest.fixture
def routes(
litellm: LiteLLMClient, passthrough: AnthropicPassthroughClient
) -> tuple[LiteLLMClient, AnthropicPassthroughClient]:
return litellm, passthrough
@pytest.fixture(scope="module")
def catalog() -> ModelCatalog:
def catalog(live_config: AgentConfig) -> 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"]
)
return ModelCatalog.discover(live_config.llm_base_url, live_config.llm_token)
def user(text: str) -> list[StoredEvent]:
@@ -63,8 +61,8 @@ def user(text: str) -> list[StoredEvent]:
@pytest.mark.parametrize("model", MODELS)
def test_model_answers_at_all(model: str, catalog: ModelCatalog) -> None:
normalized, passthrough = clients()
def test_model_answers_at_all(model: str, catalog: ModelCatalog, routes) -> None:
normalized, passthrough = routes
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Reply with exactly one word: pong"), None, model, system="Be terse."
@@ -74,10 +72,12 @@ def test_model_answers_at_all(model: str, catalog: ModelCatalog) -> None:
@pytest.mark.parametrize("model", MODELS)
def test_model_holds_the_planner_contract(model: str, catalog: ModelCatalog) -> None:
def test_model_holds_the_planner_contract(
model: str, catalog: ModelCatalog, routes
) -> 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()
normalized, passthrough = routes
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Draft an architecture plan for a command-line slug generator."),
@@ -101,10 +101,12 @@ def test_model_holds_the_planner_contract(model: str, catalog: ModelCatalog) ->
@pytest.mark.parametrize("model", MODELS)
def test_model_round_trips_a_tool_result(model: str, catalog: ModelCatalog) -> None:
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
own tool result cannot run a milestone."""
normalized, passthrough = clients()
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.")
@@ -130,11 +132,11 @@ def test_model_round_trips_a_tool_result(model: str, catalog: ModelCatalog) -> N
@pytest.mark.parametrize("model", MODELS)
def test_model_holds_the_prompt_level_json_contract(
model: str, catalog: ModelCatalog
model: str, catalog: ModelCatalog, routes
) -> 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()
normalized, passthrough = routes
client = select_client(model, normalized, passthrough, catalog)
produced = client.complete(
user("Is SQLite sufficient for a single-process agent's task state?"),
@@ -153,11 +155,13 @@ def test_model_holds_the_prompt_level_json_contract(
@pytest.mark.parametrize("model", MODELS)
def test_route_selection_matches_the_model(model: str, catalog: ModelCatalog) -> None:
def test_route_selection_matches_the_model(
model: str, catalog: ModelCatalog, routes
) -> 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()
normalized, passthrough = routes
facts = catalog.facts(model)
chosen = select_client(model, normalized, passthrough, catalog)
if facts.provider == "anthropic":
+19 -26
View File
@@ -6,30 +6,19 @@ passthrough rather than being flattened by the proxy.
"""
import io
import os
from pathlib import Path
import httpx
import pytest
from nisse_dev.canonical import AssistantText, ToolCall, ToolResult, UserMessage
from nisse_dev.envfile import load_env_file
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
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)",
),
]
pytestmark = pytest.mark.integration
def store_with(tmp_path: Path, *events) -> StateStore:
@@ -39,18 +28,22 @@ def store_with(tmp_path: Path, *events) -> StateStore:
return store
def test_passthrough_completes_a_plain_turn(tmp_path: Path) -> None:
def test_passthrough_completes_a_plain_turn(
tmp_path: Path, passthrough: AnthropicPassthroughClient, live_model: str
) -> None:
store = store_with(
tmp_path, UserMessage(text="Reply with exactly: pong", source="live")
)
result = AnthropicPassthroughClient.from_env().complete(
store.read_events(), None, os.environ[SENTINEL], system="Be terse."
result = passthrough.complete(
store.read_events(), None, live_model, system="Be terse."
)
texts = [e.text for e in result if isinstance(e, AssistantText)]
assert texts and "pong" in texts[0].lower()
def test_passthrough_round_trips_a_tool_call(tmp_path: Path) -> None:
def test_passthrough_round_trips_a_tool_call(
tmp_path: Path, passthrough: AnthropicPassthroughClient, live_model: str
) -> None:
store = store_with(
tmp_path,
UserMessage(
@@ -58,11 +51,11 @@ def test_passthrough_round_trips_a_tool_call(tmp_path: Path) -> None:
source="live",
),
)
client = AnthropicPassthroughClient.from_env()
client = passthrough
produced = client.complete(
store.read_events(),
EXECUTOR_TOOLS,
os.environ[SENTINEL],
live_model,
system="Use a tool. Do not answer in prose.",
)
calls = [e for e in produced if isinstance(e, ToolCall)]
@@ -75,18 +68,18 @@ def test_passthrough_round_trips_a_tool_call(tmp_path: Path) -> None:
store.append_event(
ToolResult(call_id=calls[0].call_id, result="app.py\ntests/", success=True)
)
follow_up = client.complete(
store.read_events(), EXECUTOR_TOOLS, os.environ[SENTINEL]
)
follow_up = client.complete(store.read_events(), EXECUTOR_TOOLS, live_model)
assert follow_up, "the vendor rejected or ignored the tool_result turn"
def test_prompt_caching_survives_the_passthrough() -> None:
def test_prompt_caching_survives_the_passthrough(
live_config: AgentConfig, live_model: str
) -> None:
# A5 asks for this to be verified rather than assumed: a proxy that
# flattened the request would silently drop cache control and turn a
# cost optimisation into a per-call expense.
base = os.environ["NISSE_LLM_BASE_URL"].rstrip("/")
token = os.environ["NISSE_LLM_TOKEN"]
base = live_config.llm_base_url.rstrip("/")
token = live_config.llm_token
response = httpx.post(
f"{base}/v1/messages",
headers={
@@ -95,7 +88,7 @@ def test_prompt_caching_survives_the_passthrough() -> None:
"anthropic-version": "2023-06-01",
},
json={
"model": os.environ[SENTINEL],
"model": live_model,
"max_tokens": 16,
"messages": [{"role": "user", "content": "Say pong."}],
},
+9 -26
View File
@@ -1,47 +1,30 @@
"""Live planning turn: a real model must answer with a submit_plan or
ask_human tool call, not prose.
Deliberately opt-in via NISSE_LIVE_PLANNER_MODEL rather than the usual
NISSE_LLM_BASE_URL sentinel. Planning escalates to a frontier model
unconditionally (A12), and the small local model does not hold the tool-call
contract reliably — measured, see the PR — so this test needs a frontier
route that does not exist yet. Set the variable to the model to plan with
once one is configured; until then this skips on purpose, not by accident.
Plans with the test config's planner role. Planning escalates to a frontier
model unconditionally (A12), and small local models do not hold the tool-call
contract reliably, so that role must be one that does.
"""
import io
import os
from pathlib import Path
import pytest
from nisse_dev.canonical import UserMessage
from nisse_dev.control import AskHuman, SubmitPlan
from nisse_dev.envfile import load_env_file
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
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 frontier planner route configured ({SENTINEL} unset)",
),
]
pytestmark = pytest.mark.integration
def test_live_planning_turn_returns_a_tool_action(tmp_path: Path) -> None:
assert os.environ.get("NISSE_LLM_TOKEN"), (
f"{SENTINEL} is set, so the live planning path is expected to work, "
"but NISSE_LLM_TOKEN is missing"
)
model = os.environ[SENTINEL]
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, delegate=model, consultant=model)
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
store.append_event(
@@ -53,7 +36,7 @@ def test_live_planning_turn_returns_a_tool_action(tmp_path: Path) -> None:
source="live-test",
)
)
agent = LLMPlanningAgent(LiteLLMClient.from_env(), store, roles)
agent = LLMPlanningAgent(litellm, store, roles)
action = agent.plan("arch", "hello-cli")