nisse/nisse-dev/src/nisse_dev/loop/build.py

"""Composition root: the one place that knows how the whole agent fits
together. Every component is unit-tested in isolation; this is where they meet,
which is what keeps a built piece from being reachable only from its own tests
(the delegation harness and the comment poller once were)."""

import threading
from collections.abc import Callable
from dataclasses import dataclass

from nisse_core.config import ConfigError
from nisse_core.conversation.matrix import MatrixTransport
from nisse_core.execution.ssh import SSHHost
from nisse_core.gitea import GiteaClient
from nisse_core.knowledge.memory import Memory
from nisse_core.knowledge.vault import Vault
from nisse_core.loop.driver import Driver
from nisse_core.models.catalog import ModelCatalog, select_client
from nisse_core.models.context import BudgetedClient
from nisse_core.models.llm import AnthropicPassthroughClient, LiteLLMClient
from nisse_core.oplog import OpLog
from nisse_core.state.cancel import Cancel
from nisse_core.state.worker import SerialWorker
from nisse_dev.config import AgentConfig
from nisse_dev.config import Project as ProjectEntry
from nisse_dev.delegation import Caps
from nisse_dev.git import GitRepo, HTTPSToken, PushCredential, SSHKey
from nisse_dev.loop.agent import LoopAgent, Project
from nisse_dev.loop.classifier import ActionClassifier
from nisse_dev.loop.help import Helpers
from nisse_dev.loop.milestones import Milestones
from nisse_dev.loop.reader import GoAheadReader
from nisse_dev.state import StateStore
from nisse_dev.tools import Workspace

# How long shutdown waits for running work before leaving the store to the
# durability layer (L8). A milestone can run far longer than a service
# manager allows for a stop, and a hard exit mid-work is already safe.
SHUTDOWN_WAIT_S = 10.0

TOOL_DRIVING_ROLES = ("driver", "helper")
MEMORY_FOLDER = "memory/nisse-dev"  # L10: the agent's own, in its vault


def discover_models(config: AgentConfig, oplog: OpLog) -> ModelCatalog:
    """Resolve every configured role against LiteLLM before anything runs.

    Refusing to boot is the whole point: a role pointed at a model this key
    cannot see is a misconfiguration, and the cheapest place to learn that is
    here rather than ten minutes into a milestone. An unreachable proxy is
    reported separately -- it is an outage, not a mistake, and telling them
    apart is what lets an operator know whether to edit config or wait.
    """
    catalog = ModelCatalog.discover(config.llm_base_url, config.llm_token)
    catalog.validate(config.roles.pinned())
    # Every role's context window must be known before a request is built
    # against it (L6): a guess that is too large reaches a server that may
    # silently cut the prompt.
    catalog.require_limits(config.roles.pinned())
    for role in TOOL_DRIVING_ROLES:
        facts = catalog.facts(config.roles.resolve(role))
        if facts.supports_tools is False:
            # A warning, not a refusal: models LiteLLM declares tool-less are
            # observed driving tool loops anyway, and L9 expects weak local
            # models to be accommodated rather than excluded.
            oplog.emit(
                "model_tools_unsupported",
                cause="startup",
                role=role,
                model=facts.alias,
                backing=facts.backing,
            )
    return catalog


@dataclass
class RunningLoop:
    config: AgentConfig
    account: str
    loop: LoopAgent
    store: StateStore
    transport: MatrixTransport
    gitea: GiteaClient
    worker: SerialWorker
    _stopped: threading.Event

    def start(self) -> None:
        self.loop.start()
        self.transport.post(
            f"nisse-dev up as {self.account} on "
            f"{', '.join(p.repo for p in self.config.projects)}; "
            f"{self.loop_status()}"
        )

    def loop_status(self) -> str:
        return self.worker.call(self.loop.status)

    def stop(self) -> None:
        self.loop.stop()
        self.transport.close()
        self.gitea.close()
        try:
            self.worker.call(self.store.close, timeout=SHUTDOWN_WAIT_S)
        except TimeoutError:
            pass  # still working; L8 covers exiting mid-work
        self.worker.stop(timeout=SHUTDOWN_WAIT_S)
        self._stopped.set()

    def wait(self) -> None:
        self._stopped.wait()


def make_llm_clients(
    config: AgentConfig, catalog: ModelCatalog
) -> tuple[BudgetedClient, BudgetedClient]:
    """The two LLM routes, as every model call in the agent uses them.

    Shared with the benchmark, so that what it measures is exactly what the
    agent would do: the same context budget, the same declared limits, and
    tool_choice asked only of models declared to enforce it.
    """
    tuning = config.tuning

    def limits_for(model: str) -> tuple[int, int | None]:
        facts = catalog.facts(model)
        assert facts.max_input_tokens is not None  # require_limits checked
        return facts.max_input_tokens, facts.max_output_tokens

    def budgeted(client: object) -> BudgetedClient:
        # Every request, on either route, is fitted to its model's window.
        return BudgetedClient(
            client,
            limits_for,
            max_output_tokens=config.max_output_tokens,
            keep_recent_turns=tuning.context.keep_recent_turns,
            safety_margin=tuning.context.safety_margin,
            enforces_tool_choice=lambda m: (
                catalog.facts(m).supports_tool_choice is True
            ),
        )

    normalized = budgeted(
        LiteLLMClient(
            config.llm_base_url,
            config.llm_token,
            max_tokens=config.max_output_tokens,
            timeout_s=config.request_timeout_s,
        )
    )
    # Delegations prefer the vendor-native route; fidelity is cheap
    # elsewhere (L9). Which route a given model gets is decided by
    # select_client, not by where the caller happens to sit.
    passthrough = budgeted(
        AnthropicPassthroughClient(
            config.llm_base_url,
            config.llm_token,
            max_tokens=config.max_output_tokens,
            timeout_s=config.request_timeout_s,
        )
    )
    return normalized, passthrough


def build_host(config: AgentConfig, project: ProjectEntry) -> SSHHost:
    """A project's build VM (L13). Its known-hosts file and connection socket
    live beside the state store."""
    host = project.build_host
    return SSHHost(
        host.address,
        key=host.key,
        user=host.user,
        port=host.port,
        state_dir=config.state_db.parent / "ssh",
    )


def push_credential(project: ProjectEntry) -> PushCredential | None:
    """What a project's pushes go by: a push-only account's token over
    HTTPS, or a deploy key over SSH (D4)."""
    if project.push_user and project.push_token:
        return HTTPSToken(project.push_user, project.push_token)
    if project.push_key is None:
        return None
    try:
        return SSHKey(project.push_key.read_text())
    except OSError as exc:
        raise RuntimeError(
            f"cannot read the push key {project.push_key}: {exc}"
        ) from exc


def remote_url(gitea: GiteaClient, repo: str, credential: PushCredential | None) -> str:
    """The credential-free URL the workspace clones from and pushes to."""
    if isinstance(credential, SSHKey):
        return gitea.ssh_url(repo)
    return gitea.clone_url(repo)


def prepare_workspace(
    host: SSHHost,
    config: AgentConfig,
    account: str,
    remote: str,
    credential: PushCredential | None,
    project: ProjectEntry,
) -> GitRepo:
    """Ensure the workspace on the build VM is a clone of the project repo.

    The clone uses the repository's SSH URL, so no credential is ever stored
    in it (#89); the push credential is handed to the clone and nothing else.
    """
    repo = GitRepo(host, project.workspace)
    if not repo.exists():
        repo.clone(remote, credential=credential)
    repo.configure_identity(account, f"{account}@users.noreply.local")
    repo.install_excludes()  # build artifacts are never the work (#80)
    return repo


def project_runtime(
    config: AgentConfig,
    entry: ProjectEntry,
    store: StateStore,
    gitea: GiteaClient,
    account: str,
) -> Project:
    """One project's part of the loop: its clone on its build VM, workspace
    (with the secrets its commands may see, redacted on return), push
    credential and milestones."""
    host = build_host(config, entry)
    credential = push_credential(entry)
    remote = remote_url(gitea, entry.repo, credential)
    git = prepare_workspace(host, config, account, remote, credential, entry)
    return Project(
        name=entry.name,
        repo=entry.repo,
        labels=entry.labels,
        base=entry.base_branch,
        merge=entry.merge,
        workspace=Workspace(
            host,
            entry.workspace,
            timeout_s=config.tuning.tools.command_timeout_s,
            max_output_chars=config.tuning.tools.max_output_chars,
            secrets=entry.command_secrets(),
        ),
        milestones=Milestones(
            store,
            git,
            gitea,
            repo=entry.repo,
            remote=remote,
            credential=credential,
            base=entry.base_branch,
        ),
    )


def build_loop_agent(
    config: AgentConfig, load: Callable[[], AgentConfig] | None = None
) -> RunningLoop:
    """``load`` reads the configuration again, for a new project's entry
    (D8); by default from the file this one came from."""
    oplog = OpLog("nisse-dev")
    worker = SerialWorker(
        on_error=lambda exc: oplog.emit(
            "worker.job_failed", cause="unhandled exception", error=repr(exc)
        )
    )
    worker.start()
    store = worker.call(lambda: StateStore(config.state_db, oplog=oplog))
    catalog = discover_models(config, oplog)
    gitea = GiteaClient(config.gitea_url, config.gitea_token)
    account = gitea.whoami()
    # D8: each project its own build VM, clone, push credential and milestones.
    projects = [
        project_runtime(config, entry, store, gitea, account)
        for entry in config.projects
    ]
    normalized, passthrough = make_llm_clients(config, catalog)
    cancel = Cancel()
    # The configured driver; each work item runs on the one it pinned (L9).
    driver = Driver(
        normalized,
        config.roles.resolve("driver"),
        max_turns=config.tuning.driver.max_turns,
        cancel=cancel,
        failure_limit=config.tuning.driver.failure_limit,
    )
    tuning = config.tuning
    helpers = Helpers(
        client=normalized,
        roles=config.roles,
        client_for=lambda model: select_client(model, normalized, passthrough, catalog),
        backing=lambda model: catalog.facts(model).backing,
        caps=Caps(
            turns=tuning.delegation.max_turns,
            tokens=tuning.delegation.max_tokens,
            wall_clock_s=tuning.delegation.wall_clock_s,
        ),
        expert_budget=tuning.expert.consult_budget,
        profiles=config.profiles,
    )
    transport = MatrixTransport(config.matrix)
    vault = (
        Vault(
            config.vault.root,
            config.vault.folders,
            agent="nisse-dev",
            ignore=config.vault.ignore,
        )
        if config.vault
        else None
    )
    loop = LoopAgent(
        store,
        transport,
        gitea,
        projects=projects,
        account=account,
        operators=config.operators,
        gitea_operators=config.gitea_operators,
        driver=driver,
        worker=worker,
        cancel=cancel,
        oplog=oplog,
        roles=config.roles.pinned(),
        # The classifier role reads plain-language go-aheads (L5, L9).
        reader=GoAheadReader(normalized, config.roles.resolve("classifier")),
        # L4: what the fast path doesn't know, judged on the classifier role.
        classifier=ActionClassifier(
            normalized,
            config.roles.resolve("classifier"),
            backing=lambda model: catalog.facts(model).backing,
        ),
        helpers=helpers,
        reload=lambda known: [
            project_runtime(config, entry, store, gitea, account)
            for entry in (load or _from_file(config))().projects
            if entry.name not in known
        ],
        # D3, M4.1: the build VM checked, and the toolchain set up, before
        # the driver plans.
        preflight=True,
        preflight_turns=config.tuning.preflight.max_turns,
        preflight_timeout_s=config.tuning.preflight.command_timeout_s,
        bug_grace_s=config.tuning.loop.bug_grace_s,
        general_idle_s=config.tuning.loop.general_idle_s,
        vm_retry_s=config.tuning.build_vm.retry_s,
        vm_settle_s=config.tuning.build_vm.settle_s,
        # L10: the knowledge vault, when configured; L6.4: the agent's
        # memory, when its own folder in it is read-write.
        vault=vault,
        memory=(
            Memory(vault, MEMORY_FOLDER)
            if vault and vault.folders().get(MEMORY_FOLDER) == "rw"
            else None
        ),
        memory_turns=config.tuning.memory.max_turns,
    )
    return RunningLoop(
        config=config,
        account=account,
        loop=loop,
        store=store,
        transport=transport,
        gitea=gitea,
        worker=worker,
        _stopped=threading.Event(),
    )


def _from_file(config: AgentConfig) -> Callable[[], AgentConfig]:
    def load() -> AgentConfig:
        if config.config_file is None:
            raise ConfigError("this agent wasn't started from a config file")
        return AgentConfig.from_file(config.config_file)

    return load