CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m14s
Development Plan M1.1, as discussed. - Schema v4: a work_items table (id, scope, title, status, outcome, position, data) and an agent row whose active_item points at one (NULL: idle). Canonical events and intents carry their work item, and events their actor. Older stores are refused with a clear message: there was no production data, so no migration (the mechanism stays). - The log is reached through a Transcript bound to one work item and one actor: store.transcript(item, actor) reads only that item and writes with both. The planner, executor, delegation and expert desk get one per call from the orchestrator instead of holding the store, so a component can neither read another task's events (the planner read every earlier task's, #78) nor write without attribution. store.all_events() is kept for readers that span items (evaluation capture, dumps). Intents are begun through the transcript or with work_item=, so each belongs to its item. - The orchestrator opens a work item on 'use', closes it with its outcome (completed or abandoned), and carries what the operator said while idle into the new item as copies with their origin, so a task sent before 'use' is still planned from (D1). control.py keeps its state machine through get_state()/transition(), now over the active work item, until M2.6 retires it. - nisse_dev.dump shows work items and each event's item and actor. Tests: test_work_items.py (scoping, attribution, intent ownership, restart, outcomes, carried messages, a second task never seeing the first); the suites updated for the per-call transcript. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
295 lines
11 KiB
Python
295 lines
11 KiB
Python
"""Composition root: the one place that knows how the whole agent fits together.
|
|
|
|
Every component below has been unit-tested in isolation for some time; nothing
|
|
had ever assembled them. That gap hid two live bugs — the delegation harness
|
|
and the Gitea comment poller were both fully built and reachable from nothing
|
|
but their own tests — which is the argument for having a single root rather
|
|
than constructing pieces ad hoc.
|
|
"""
|
|
|
|
import threading
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
|
|
from nisse_dev.approvals import PlanBoard
|
|
from nisse_dev.cancel import Cancel
|
|
from nisse_dev.catalog import ModelCatalog, select_client
|
|
from nisse_dev.config import AgentConfig
|
|
from nisse_dev.context import BudgetedClient
|
|
from nisse_dev.control import Orchestrator
|
|
from nisse_dev.delegation import Caps, Delegation
|
|
from nisse_dev.executor import LLMExecutor
|
|
from nisse_dev.expert import ExpertDesk
|
|
from nisse_dev.git import GitRepo, authenticated_url
|
|
from nisse_dev.gitea import Comment, GiteaClient
|
|
from nisse_dev.llm import AnthropicPassthroughClient, LiteLLMClient
|
|
from nisse_dev.matrix import MatrixTransport
|
|
from nisse_dev.oplog import OpLog
|
|
from nisse_dev.planner import LLMPlanningAgent
|
|
from nisse_dev.poller import CommentPoller
|
|
from nisse_dev.publish import MilestonePublisher
|
|
from nisse_dev.state import StateStore
|
|
from nisse_dev.tools import Workspace
|
|
from nisse_dev.worker import SerialWorker
|
|
|
|
# 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 = ("planner", "executor", "helper")
|
|
|
|
|
|
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 Agent:
|
|
config: AgentConfig
|
|
store: StateStore
|
|
transport: MatrixTransport
|
|
orchestrator: Orchestrator
|
|
gitea: GiteaClient
|
|
account: str
|
|
worker: SerialWorker
|
|
_stopped: threading.Event
|
|
|
|
def start(self) -> None:
|
|
"""Boot in the order L8 requires: reconcile against external truth
|
|
before accepting a single operator command."""
|
|
self.worker.call(self.orchestrator.start)
|
|
self.transport.start()
|
|
state = self.worker.call(self.store.get_state)
|
|
where = state.state if state else "IDLE"
|
|
self.transport.send_message(
|
|
f"nisse-dev up as {self.account} — project {self.config.project}, "
|
|
f"state {where}, planner {self.config.roles.planner}, "
|
|
f"executor {self.config.roles.executor}"
|
|
)
|
|
|
|
def wait(self) -> None:
|
|
self._stopped.wait()
|
|
|
|
def stop(self) -> None:
|
|
"""Shut down outermost-first: stop accepting work, then stop talking,
|
|
then close the state store last so nothing writes to a closed file.
|
|
The store belongs to the worker, so the worker closes it."""
|
|
self.orchestrator.stop_watching()
|
|
self.transport.close()
|
|
self.gitea.close()
|
|
try:
|
|
self.worker.call(self.store.close, timeout=SHUTDOWN_WAIT_S)
|
|
except TimeoutError:
|
|
pass # still working; exiting mid-work is what L8 are for
|
|
self.worker.stop(timeout=SHUTDOWN_WAIT_S)
|
|
self._stopped.set()
|
|
|
|
|
|
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 prepare_workspace(config: AgentConfig, account: str) -> GitRepo:
|
|
"""Ensure the workspace is a clone of the project repo.
|
|
|
|
Cloning here rather than on `use <project>` is the single-project
|
|
shortcut: resolving a project name to a repo is issue #27's work.
|
|
"""
|
|
repo = GitRepo(config.workspace)
|
|
remote = authenticated_url(
|
|
config.gitea_url, config.gitea_repo, account, config.gitea_token
|
|
)
|
|
if not (config.workspace / ".git").exists():
|
|
config.workspace.parent.mkdir(parents=True, exist_ok=True)
|
|
GitRepo(config.workspace.parent).clone(remote, config.workspace)
|
|
repo.configure_identity(account, f"{account}@users.noreply.local")
|
|
return repo
|
|
|
|
|
|
def build_agent(
|
|
config: AgentConfig, *, notify: Callable[[str], None] | None = None
|
|
) -> Agent:
|
|
oplog = OpLog("nisse-dev")
|
|
worker = SerialWorker(
|
|
on_error=lambda exc: oplog.emit(
|
|
"worker.job_failed", cause="unhandled exception", error=repr(exc)
|
|
)
|
|
)
|
|
worker.start()
|
|
# Opened on the worker, which is the only thread allowed to use it.
|
|
store = worker.call(lambda: StateStore(config.state_db, oplog=oplog))
|
|
cancel = Cancel()
|
|
# Before the clone, before the token checks: a typo'd role name should
|
|
# cost a second, not a repository.
|
|
catalog = discover_models(config, oplog)
|
|
|
|
gitea = GiteaClient(config.gitea_url, config.gitea_token)
|
|
# Whoami rather than another config variable: it names the identity every
|
|
# Gitea permission is evaluated against, and proves the token works before
|
|
# anything depends on it.
|
|
account = gitea.whoami()
|
|
|
|
git_repo = prepare_workspace(config, account)
|
|
tuning = config.tuning
|
|
workspace = Workspace(
|
|
config.workspace,
|
|
timeout_s=tuning.tools.command_timeout_s,
|
|
max_output_chars=tuning.tools.max_output_chars,
|
|
)
|
|
|
|
normalized, passthrough = make_llm_clients(config, catalog)
|
|
|
|
transport = MatrixTransport(config.matrix)
|
|
announce = notify or transport.send_message
|
|
|
|
expert = ExpertDesk(normalized, config.roles, budget=tuning.expert.consult_budget)
|
|
delegate_model = config.roles.resolve("helper")
|
|
delegation = Delegation(
|
|
select_client(delegate_model, normalized, passthrough, catalog),
|
|
store,
|
|
workspace,
|
|
git_repo,
|
|
model=delegate_model,
|
|
backing_model=catalog.facts(delegate_model).backing,
|
|
caps=Caps(
|
|
turns=tuning.delegation.max_turns,
|
|
tokens=tuning.delegation.max_tokens,
|
|
wall_clock_s=tuning.delegation.wall_clock_s,
|
|
),
|
|
cancel=cancel,
|
|
)
|
|
executor = LLMExecutor(
|
|
normalized,
|
|
config.roles,
|
|
workspace,
|
|
expert=expert,
|
|
delegate=delegation,
|
|
notify=announce,
|
|
cancel=cancel,
|
|
max_turns=tuning.executor.max_turns,
|
|
failure_limit=tuning.executor.failure_limit,
|
|
)
|
|
publisher = MilestonePublisher(
|
|
store,
|
|
gitea,
|
|
config.gitea_repo,
|
|
git_repo,
|
|
base_url=config.gitea_url,
|
|
account=account,
|
|
token=config.gitea_token,
|
|
base_branch=config.base_branch,
|
|
)
|
|
|
|
def poller_factory(on_comment: Callable[[Comment], None]) -> CommentPoller:
|
|
return CommentPoller(
|
|
gitea,
|
|
config.gitea_repo,
|
|
on_comment,
|
|
oplog=oplog,
|
|
# Its own comments are not new input; without this the agent
|
|
# answers itself.
|
|
ignore_authors={account},
|
|
)
|
|
|
|
orchestrator = Orchestrator(
|
|
store,
|
|
transport,
|
|
LLMPlanningAgent(normalized, config.roles, cancel=cancel),
|
|
operators=config.operators,
|
|
roles=config.roles,
|
|
executor=executor,
|
|
publisher=publisher,
|
|
gitea_operators=config.gitea_operators,
|
|
board=PlanBoard(gitea, config.gitea_repo),
|
|
delegate=delegation,
|
|
poller_factory=poller_factory,
|
|
catalog=catalog,
|
|
worker=worker,
|
|
cancel=cancel,
|
|
)
|
|
return Agent(
|
|
config=config,
|
|
store=store,
|
|
transport=transport,
|
|
orchestrator=orchestrator,
|
|
gitea=gitea,
|
|
account=account,
|
|
worker=worker,
|
|
_stopped=threading.Event(),
|
|
)
|