nisse/nisse-dev/tests/test_config.py

import copy
import re
from pathlib import Path
from typing import Any

import pytest

from nisse_core.config import ConfigError
from nisse_core.models.llm import DEFAULT_REQUEST_TIMEOUT_S
from nisse_dev.config import AgentConfig, Project

# A complete config as the YAML file would give it, with secrets by reference.
COMPLETE: dict[str, Any] = {
    "state_db": "~/demo/state.db",
    "projects": {
        "greet": {
            "repo": "don/demo",
            "workspace": "~/demo/work",
            "build_host": {"address": "build.test", "key": "~/.ssh/nisse-dev"},
        }
    },
    "llm": {"base_url": "https://litellm.test", "token": "${TEST_LLM_TOKEN}"},
    "roles": {
        "driver": "claude-opus-5",
        "helper": "claude-haiku-4-5",
        "expert": "claude-haiku-4-5",
    },
    "matrix": {
        "homeserver": "https://matrix.test/",
        "user_id": "@nisse-dev-testing:test",
        "token": "${TEST_MATRIX_TOKEN}",
        "room_id": "!room:test",
        "operators": ["@don:test"],
    },
    "gitea": {
        "url": "https://gitea.test",
        "token": "${TEST_GITEA_TOKEN}",
        "operators": ["don"],
    },
}
P = "projects__greet__"  # the one project's keys, for tree()
SECRETS = {
    "TEST_LLM_TOKEN": "sk-x",
    "TEST_MATRIX_TOKEN": "syt_x",
    "TEST_GITEA_TOKEN": "gt",
}


def tree(**changes: Any) -> dict[str, Any]:
    """COMPLETE with dotted keys changed; a value of None removes the key."""
    result = copy.deepcopy(COMPLETE)
    for dotted, value in changes.items():
        *parents, leaf = dotted.split("__")
        node = result
        for part in parents:
            node = node.setdefault(part, {})
        if value is None:
            node.pop(leaf, None)
        else:
            node[leaf] = value
    return result


def build(**changes: Any) -> AgentConfig:
    return AgentConfig.from_tree(tree(**changes), SECRETS)


def only(config: AgentConfig) -> Project:
    [project] = config.projects
    return project


def test_complete_config_builds_with_secrets_by_reference() -> None:
    config = build()
    assert only(config).name == "greet"
    assert config.llm_token == "sk-x"
    assert config.gitea_token == "gt"
    assert config.roles.driver == "claude-opus-5"
    assert config.matrix.room_id == "!room:test"
    assert config.matrix.homeserver == "https://matrix.test"
    assert only(config).repo == "don/demo"
    assert only(config).base_branch == "main"
    assert config.tuning.driver.max_turns == 40  # absent section: defaults


def test_paths_expand_the_home_directory() -> None:
    config = build()
    assert only(config).workspace.is_absolute()
    assert "~" not in str(config.state_db)


def test_every_missing_setting_is_reported_at_once() -> None:
    # Failing at first use would surface a missing token twenty minutes into
    # a milestone, on the call that mattered.
    with pytest.raises(ConfigError) as excinfo:
        build(matrix__token=None, **{f"{P}repo": None})
    message = str(excinfo.value)
    assert "matrix.token" in message
    assert "projects.greet: unset or empty — repo" in message


def test_empty_is_treated_as_missing() -> None:
    with pytest.raises(ConfigError, match="projects.greet: unset or empty — repo"):
        build(**{f"{P}repo": "   "})


def test_a_reference_to_an_unset_variable_is_an_error() -> None:
    # Never an empty string: an empty token would pass for a value and fail
    # at the first request.
    env = {k: v for k, v in SECRETS.items() if k != "TEST_GITEA_TOKEN"}
    with pytest.raises(ConfigError, match=r"gitea.token: references \$\{TEST_GITEA"):
        AgentConfig.from_tree(tree(), env)


def test_operator_lists_drop_blanks() -> None:
    config = build(matrix__operators=[" @don:test ", "@other:test", " "])
    assert config.operators == {"@don:test", "@other:test"}
    # A blank entry must not admit an empty identity into an allowlist.
    assert "" not in config.operators


def test_an_operator_list_of_only_blanks_is_missing() -> None:
    with pytest.raises(ConfigError, match="gitea.operators"):
        build(gitea__operators=[" "])


def test_base_branch_is_configurable() -> None:
    assert only(build(**{f"{P}base_branch": "trunk"})).base_branch == "trunk"


def test_max_output_tokens_is_configurable_and_checked() -> None:
    assert build(llm__max_output_tokens=8192).max_output_tokens == 8192
    with pytest.raises(ConfigError, match="llm.max_output_tokens"):
        build(llm__max_output_tokens=0)


def test_request_timeout_is_configurable_and_checked() -> None:
    assert build().request_timeout_s == DEFAULT_REQUEST_TIMEOUT_S
    assert build(llm__request_timeout_s=900).request_timeout_s == 900.0
    for bad in (0, -5, "soon"):
        with pytest.raises(ConfigError, match="llm.request_timeout_s"):
            build(llm__request_timeout_s=bad)


def test_the_request_timeout_reaches_both_clients() -> None:
    from nisse_core.models.catalog import ModelCatalog
    from nisse_dev.loop.build import make_llm_clients

    normalized, passthrough = make_llm_clients(
        build(llm__request_timeout_s=42), ModelCatalog({})
    )
    for client in (normalized, passthrough):
        assert client._inner._http.timeout.read == 42.0


def test_redacted_summary_carries_no_secrets() -> None:
    summary = build().redacted()
    flat = " ".join(summary.values())
    for secret in SECRETS.values():
        assert secret not in flat
    assert "greet (don/demo, build.test:" in summary["projects"]


def test_tuning_is_read() -> None:
    config = build(
        tuning={
            "driver": {"max_turns": 40, "failure_limit": 3},
            "expert": {"consult_budget": 5},
            "delegation": {"max_turns": 20, "max_tokens": 100000, "wall_clock_s": 600},
            "tools": {"command_timeout_s": 300, "max_output_chars": 2000},
            "loop": {"bug_grace_s": 120},
            "build_vm": {"retry_s": 60},
        }
    )
    tuning = config.tuning
    assert (tuning.driver.max_turns, tuning.driver.failure_limit) == (40, 3)
    assert tuning.expert.consult_budget == 5
    assert tuning.delegation.max_tokens == 100_000
    assert tuning.delegation.wall_clock_s == 600.0
    assert tuning.tools.max_output_chars == 2000
    assert tuning.loop.bug_grace_s == 120.0
    assert (tuning.build_vm.retry_s, tuning.build_vm.settle_s) == (60.0, 180.0)


def test_a_partly_given_tuning_section_keeps_the_other_defaults() -> None:
    tuning = build(tuning={"driver": {"max_turns": 40}}).tuning
    assert tuning.driver.failure_limit == 2


def test_every_problem_is_reported_at_once() -> None:
    # A misspelt knob that was silently ignored would look exactly like a knob
    # with no effect, so unknown keys are errors.
    bad = tree(llm_extra=1, **{f"{P}workspace": None})
    bad["tuning"] = {
        "driver": {"max_turn": 40, "failure_limit": 0},
        "expert": {"consult_budget": True},
        "bogus": {"x": 1},
    }
    with pytest.raises(ConfigError) as excinfo:
        AgentConfig.from_tree(bad, SECRETS)
    message = str(excinfo.value)
    for expected in (
        "projects.greet: unset or empty — workspace",
        "llm_extra: unknown setting",
        "tuning.driver.max_turn: unknown setting",
        "tuning.driver.failure_limit: must be a positive number",
        "tuning.expert.consult_budget: must be a positive number",
        "tuning.bogus: unknown section",
    ):
        assert expected in message


def test_a_whole_number_is_required_where_one_is_meant() -> None:
    with pytest.raises(ConfigError, match="tuning.driver.max_turns"):
        build(tuning={"driver": {"max_turns": 2.5}})


# --- the file --------------------------------------------------------------

FILE = """
state_db: ~/demo/state.db
projects:
  greet:
    repo: don/demo
    workspace: ~/demo/work
    build_host:
      address: build.test
      key: ~/.ssh/nisse-dev
llm: {base_url: "https://litellm.test", token: "${TEST_LLM_TOKEN}"}
roles:
  driver: local-fast
  helper: claude-opus-5
  expert: claude-haiku-4-5
matrix:
  homeserver: https://matrix.test
  user_id: "@nisse-dev-testing:test"
  token: ${TEST_MATRIX_TOKEN}
  room_id: "!room:test"
  operators: ["@don:test"]
gitea:
  url: https://gitea.test
  token: ${TEST_GITEA_TOKEN}
  operators: [don]
"""


def write(directory: Path, text: str = FILE) -> Path:
    path = directory / "nisse-dev.yaml"
    path.write_text(text)
    return path


def test_a_file_is_read(tmp_path: Path) -> None:
    config = AgentConfig.from_file(write(tmp_path), SECRETS)
    assert config.roles.driver == "local-fast"
    assert config.config_file == tmp_path / "nisse-dev.yaml"


def test_invalid_yaml_is_reported(tmp_path: Path) -> None:
    with pytest.raises(ConfigError, match="not valid YAML"):
        AgentConfig.from_file(write(tmp_path, "roles: [unclosed"), SECRETS)


def test_load_uses_nisse_config(tmp_path: Path) -> None:
    path = write(tmp_path)
    assert (
        only(AgentConfig.load({**SECRETS, "NISSE_CONFIG": str(path)})).name == "greet"
    )


def test_load_defaults_to_the_working_directory(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    write(tmp_path)
    monkeypatch.chdir(tmp_path)
    assert only(AgentConfig.load(SECRETS)).name == "greet"


def test_load_without_a_file_says_how_to_make_one(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.chdir(tmp_path)
    with pytest.raises(ConfigError, match="nisse-dev.example.yaml"):
        AgentConfig.load(SECRETS)


def test_the_example_file_is_valid() -> None:
    # The example documents every knob; it must not drift out of loading.
    example = Path(__file__).resolve().parents[2] / "nisse-dev.example.yaml"
    env = {
        "NISSE_LLM_TOKEN": "sk-x",
        "NISSE_MATRIX_TOKEN": "syt_x",
        "NISSE_GITEA_TOKEN": "gt",
    }
    config = AgentConfig.from_file(example, env)
    assert config.tuning.driver.max_turns == 40
    assert config.roles.driver == "claude-opus-5-5"


# --- nothing that carries secrets may sit inside the clone -----------------


def test_state_db_and_workspace_are_separate() -> None:
    # L8: build artefacts must not share a volume with agent state.
    config = build()
    assert config.state_db.parent != only(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"):
        build(state_db=str(work / "state.db"), **{f"{P}workspace": str(work)})


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_tree(
            tree(state_db=str(tmp_path / "state.db"), **{f"{P}workspace": str(work)}),
            {**SECRETS, "NISSE_ENV": str(work / ".env")},
        )


def test_a_config_file_inside_the_clone_is_refused(tmp_path: Path) -> None:
    work = tmp_path / "project"
    work.mkdir()
    text = FILE.replace("workspace: ~/demo/work", f"workspace: {work}")
    assert text != FILE
    with pytest.raises(ConfigError, match="NISSE_CONFIG"):
        AgentConfig.from_file(write(work, text), SECRETS)


def test_sibling_paths_are_fine(tmp_path: Path) -> None:
    config = AgentConfig.from_tree(
        tree(
            state_db=str(tmp_path / "state.db"),
            **{f"{P}workspace": str(tmp_path / "project")},
        ),
        {**SECRETS, "NISSE_ENV": str(tmp_path / ".env")},
    )
    assert only(config).workspace.name == "project"


def test_context_tuning_is_read_and_checked() -> None:
    config = build(tuning={"context": {"keep_recent_turns": 4, "safety_margin": 0.2}})
    assert config.tuning.context.keep_recent_turns == 4
    assert config.tuning.context.safety_margin == 0.2
    assert build().tuning.context.keep_recent_turns == 6
    with pytest.raises(ConfigError, match="safety_margin: must be less than 1"):
        build(tuning={"context": {"safety_margin": 1}})


def test_compaction_tuning_is_read_and_checked() -> None:
    compaction = build().tuning.context.compaction()
    assert (compaction.compact_at, compaction.compact_to) == (0.6, 0.25)
    assert compaction.keep_turns == 3
    off = build(tuning={"context": {"compact_at": 0}}).tuning.context
    assert off.compaction().compact_at == 0
    with pytest.raises(ConfigError, match="compact_to must be less than compact_at"):
        build(tuning={"context": {"compact_at": 0.3, "compact_to": 0.5}})
    with pytest.raises(ConfigError, match="compact_to must be less than compact_at"):
        build(tuning={"context": {"compact_at": 1.5}})


def test_the_build_host_is_required_and_defaults_to_root_on_22() -> None:
    host = only(build()).build_host
    assert (host.address, host.user) == ("build.test", "root")
    assert host.port == 22
    assert host.key.is_absolute()
    with pytest.raises(ConfigError, match="build_host.address"):
        build(**{f"{P}build_host__address": None})


def test_a_bad_build_host_port_is_reported() -> None:
    with pytest.raises(ConfigError, match="build_host.port"):
        build(**{f"{P}build_host__port": "ssh"})


def test_the_push_key_is_optional_and_a_path() -> None:
    assert only(build()).push_key is None
    assert only(build(**{f"{P}push_key": "~/k"})).push_key == Path("~/k").expanduser()


def test_a_push_token_needs_its_user_and_excludes_a_key() -> None:
    token = {f"{P}push_user": "pusher", f"{P}push_token": "${TEST_GITEA_TOKEN}"}
    project = only(build(**token))
    assert (project.push_user, project.push_token) == ("pusher", "gt")
    with pytest.raises(ConfigError, match="set both or neither"):
        build(**{f"{P}push_user": "pusher"})
    with pytest.raises(ConfigError, match="not both"):
        build(**token, **{f"{P}push_key": "~/k"})


def test_no_credential_appears_in_the_config_repr() -> None:
    config = build(
        **{f"{P}push_user": "pusher", f"{P}push_token": "${TEST_GITEA_TOKEN}"},
        **{
            f"{P}secrets": {"DB": {"value": "${TEST_LLM_TOKEN}", "visible": "commands"}}
        },
    )
    text = repr(config)
    for secret in SECRETS.values():
        assert secret not in text


def test_the_old_loops_settings_are_refused_with_what_to_do_instead() -> None:
    old = tree(loop="new", roles__planner="claude-opus-5", roles__driver=None)
    old["roles"]["executor"] = "local-fast"
    old["tuning"] = {"executor": {"max_turns": 40}}
    with pytest.raises(ConfigError) as excinfo:
        AgentConfig.from_tree(old, SECRETS)
    message = str(excinfo.value)
    for expected in (
        "loop: retired — the loop is the only one now; remove it",
        "roles.planner: retired — use roles.driver",
        "roles.executor: retired",
        "tuning.executor: retired — it is tuning.driver now",
        "unset or empty — roles.driver",
    ):
        assert expected in message


# --- projects (D8) -----------------------------------------------------------

SECOND = {
    "repo": "don/other",
    "workspace": "~/demo/other",
    "build_host": {"address": "build2.test", "key": "~/.ssh/nisse-dev"},
}


def test_a_second_project_is_a_second_entry() -> None:
    config = build(projects__other=SECOND)
    assert [p.name for p in config.projects] == ["greet", "other"]
    other = config.projects[1]
    assert (other.repo, other.build_host.address) == ("don/other", "build2.test")
    assert other.base_branch == "main" and other.labels == () and other.secrets == {}


def test_at_least_one_project_is_needed() -> None:
    with pytest.raises(ConfigError, match="projects: must name at least one"):
        build(projects={})


def test_project_problems_are_named_by_project() -> None:
    bad = tree(projects__other={**SECOND, "repo": "don/demo", "colour": "red"})
    with pytest.raises(ConfigError) as excinfo:
        AgentConfig.from_tree(bad, SECRETS)
    message = str(excinfo.value)
    assert "projects.other.colour: unknown setting" in message
    assert "projects.other.repo: don/demo is also project greet's" in message


def test_a_project_name_is_a_slug() -> None:
    with pytest.raises(ConfigError, match="lowercase letters, digits"):
        build(projects__Other_Project=SECOND)


def test_labels_and_egress_are_lists_of_names() -> None:
    config = build(**{f"{P}labels": ["discuss"], f"{P}egress": ["pypi.org"]})
    assert only(config).labels == ("discuss",)
    assert only(config).egress == ("pypi.org",)
    with pytest.raises(ConfigError, match="projects.greet.labels: must be a list"):
        build(**{f"{P}labels": "discuss"})


def test_secrets_are_by_reference_with_a_visibility() -> None:
    config = build(
        **{
            f"{P}secrets": {
                "DB_URL": {"value": "${TEST_LLM_TOKEN}", "visible": "commands"},
                "DEPLOY": {"value": "${TEST_GITEA_TOKEN}", "visible": "harness"},
            }
        }
    )
    project = only(config)
    assert project.command_secrets() == {"DB_URL": "sk-x"}
    assert project.secrets["DEPLOY"].visible == "harness"


def test_a_secret_written_inline_is_refused() -> None:
    with pytest.raises(ConfigError, match="must be a \\$\\{VAR\\} reference"):
        build(**{f"{P}secrets": {"DB": {"value": "hunter2", "visible": "commands"}}})


def test_a_secret_needs_a_known_visibility() -> None:
    with pytest.raises(ConfigError, match="visible: must be one of commands, harness"):
        build(
            **{f"{P}secrets": {"DB": {"value": "${TEST_LLM_TOKEN}", "visible": "all"}}}
        )


def test_a_workspace_of_any_project_inside_is_refused(tmp_path: Path) -> None:
    work = tmp_path / "other"
    with pytest.raises(ConfigError, match="project other"):
        build(
            state_db=str(work / "state.db"),
            projects__other={**SECOND, "workspace": str(work)},
        )


# --- roles and profiles (L9) -------------------------------------------------


def test_the_new_loops_roles_fall_back() -> None:
    config = build()
    assert config.roles.resolve("driver") == "claude-opus-5"  # the planner's
    assert config.roles.resolve("classifier") == "claude-haiku-4-5"  # the expert's
    named = build(roles__driver="qwen3.8-27b", roles__classifier="gemma-4-12b")
    assert named.roles.resolve("driver") == "qwen3.8-27b"
    assert named.roles.pinned()["classifier"] == "gemma-4-12b"


def test_capability_profiles_are_read() -> None:
    config = build(
        profiles={"qwen3.8-27b": {"tools": "reliable", "weaknesses": "speed"}}
    )
    profile = config.profiles["qwen3.8-27b"]
    assert profile.describe() == "tool calls: reliable; weak at speed"
    assert build().profiles == {}


def test_a_bad_profile_is_reported() -> None:
    with pytest.raises(ConfigError, match="profiles.m.colour: unknown setting"):
        build(profiles={"m": {"colour": "red"}})


def test_the_trust_dial_defaults_to_the_operator_merging() -> None:
    assert only(build()).merge == "operator"
    assert only(build(**{f"{P}trust__merge": "approved"})).merge == "approved"
    with pytest.raises(ConfigError, match="trust.merge: must be one of operator"):
        build(**{f"{P}trust__merge": "always"})


def test_preflight_tuning_is_read() -> None:
    tuning = build(tuning={"preflight": {"max_turns": 8}}).tuning
    assert (tuning.preflight.max_turns, tuning.preflight.command_timeout_s) == (8, 900)


def test_the_vault_is_read() -> None:
    config = build(
        vault={
            "root": "/var/lib/nisse/vault",
            "folders": {"memory/nisse-dev": "rw", "notes/": "ro"},
        }
    )
    assert config.vault is not None
    assert config.vault.root == Path("/var/lib/nisse/vault")
    assert dict(config.vault.folders) == {"memory/nisse-dev": "rw", "notes": "ro"}
    assert config.vault.ignore == (".*", "_plug")
    assert build().vault is None  # optional: no vault, no note tools


@pytest.mark.parametrize(
    ("vault", "said"),
    [
        ({"folders": {"notes": "ro"}}, "vault.root"),
        ({"root": "/v", "folders": {}}, "vault.folders: a mapping"),
        ({"root": "/v", "folders": {"notes": "write"}}, "must be one of rw, ro"),
        ({"root": "/v", "folders": {"../x": "ro"}}, "a folder inside the vault"),
        (
            {"root": "/v", "folders": {"memory": "ro", "memory/nisse-dev": "rw"}},
            "memory/nisse-dev is inside memory",
        ),
        ({"root": "/v", "folders": {"n": "ro"}, "mounts": 1}, "vault.mounts: unknown"),
        ({"root": "/v", "folders": {"n": "ro"}, "ignore": "x"}, "vault.ignore"),
    ],
)
def test_a_bad_vault_is_refused(vault: dict[str, Any], said: str) -> None:
    with pytest.raises(ConfigError, match=re.escape(said)):
        build(vault=vault)