dongarandClaude Opus 5.5
f32ccf9881
Shared parts, not a shared scheduler (agreed): nisse-review will have its own small scheduler; a common one waits until two exist to compare. - nisse_core.loop.driver: the driver run, moved; the agent names its waiting tools per run (nisse-dev's are WAITING_TOOLS), and execution goes to any Executor (ToolOutcome moves with it). - nisse_core.loop.tools: the tool helper and the tools any agent offers (notes, memory, memory_done, ask_expert); nisse_core.loop.expert. - nisse_core.knowledge.handlers: Knowledge, the vault and memory as control tools, the index and prompt text, out of LoopAgent. - nisse_core.config: file reading, references, people, tuning sections of the agent's own, LLM numbers, the vault. - New core tests: a driver run with a non-workspace executor and the agent's own waiting set; Knowledge alone; the shared config pieces. Docs: the Development Plan marks M7.2 done; Architecture Part L and the nisse-core README name the new parts. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
188 lines
6.1 KiB
Python
188 lines
6.1 KiB
Python
"""The trust dial (L3, D5; M3.3): a project set to `merge: approved` has its
|
|
PRs merged by the harness once an operator approved the PR's current commit
|
|
and its checks pass. Gitea enforces the approval (protection); the agent
|
|
checks at start that it does, and otherwise leaves merging to the operator."""
|
|
|
|
import io
|
|
from typing import Any
|
|
|
|
from test_loop_milestones import (
|
|
DON,
|
|
FINISH,
|
|
RESTATE,
|
|
WORK,
|
|
Forge,
|
|
ScriptedModel,
|
|
call,
|
|
say,
|
|
)
|
|
from test_loop_milestones import item as the_item
|
|
from test_loop_milestones import world as world # noqa: PLC0414 -- the fixture
|
|
|
|
from nisse_core.conversation.transport import InMemoryTransport
|
|
from nisse_core.loop.driver import Driver
|
|
from nisse_core.oplog import OpLog
|
|
from nisse_core.state.cancel import Cancel
|
|
from nisse_core.state.worker import InlineWorker
|
|
from nisse_dev.loop.agent import LoopAgent, Project
|
|
from nisse_dev.loop.items import MERGE, loop_of
|
|
from nisse_dev.tools import Workspace
|
|
|
|
CLOSE = [call("close_item", summary="added and merged")]
|
|
|
|
|
|
def agent_for(
|
|
world: dict[str, Any], model: ScriptedModel, merge: str = "approved"
|
|
) -> tuple[LoopAgent, InMemoryTransport]:
|
|
transport = InMemoryTransport()
|
|
cancel = Cancel()
|
|
project = Project(
|
|
"sandbox",
|
|
"don/sandbox",
|
|
Workspace(world["host"], world["repo"].root, timeout_s=30),
|
|
world["milestones"],
|
|
merge=merge,
|
|
)
|
|
loop = LoopAgent(
|
|
world["store"],
|
|
transport,
|
|
world["forge"],
|
|
projects=[project],
|
|
account="nisse-dev",
|
|
operators={DON},
|
|
gitea_operators={"don"},
|
|
driver=Driver(model, "m", cancel=cancel), # type: ignore[arg-type]
|
|
worker=InlineWorker(),
|
|
cancel=cancel,
|
|
oplog=OpLog("nisse-dev", io.StringIO()),
|
|
)
|
|
transport.on_message(loop.on_message)
|
|
return loop, transport
|
|
|
|
|
|
def to_review(world: dict[str, Any], merge: str = "approved", *more: Any) -> Any:
|
|
"""An item whose PR is up, waiting for the merge."""
|
|
world["forge"].protect()
|
|
model = ScriptedModel(RESTATE, [*WORK, *FINISH], *more)
|
|
agent, transport = agent_for(world, model, merge)
|
|
agent.boot()
|
|
agent.intake()
|
|
say(transport, world, "go")
|
|
item = the_item(world)
|
|
assert loop_of(item).wait == MERGE, (item.position, loop_of(item), said(transport))
|
|
return agent, transport
|
|
|
|
|
|
def said(transport: InMemoryTransport) -> list[str]:
|
|
return [m.text for m in transport.outbound]
|
|
|
|
|
|
def test_an_approved_pr_is_merged_and_the_item_goes_on(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, transport = to_review(world, "approved", CLOSE)
|
|
pr = loop_of(the_item(world)).pr
|
|
agent.watch()
|
|
assert forge.agent_merged == [] # not approved yet
|
|
forge.approve(pr)
|
|
agent.watch()
|
|
agent._schedule()
|
|
assert forge.agent_merged == [pr]
|
|
assert f"Merged PR #{pr} after don's approval." in said(transport)
|
|
assert the_item(world).status == "closed"
|
|
|
|
|
|
def test_the_operator_merges_by_default(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _ = to_review(world, "operator")
|
|
pr = loop_of(the_item(world)).pr
|
|
forge.approve(pr)
|
|
agent.watch()
|
|
assert forge.agent_merged == []
|
|
assert loop_of(the_item(world)).wait == MERGE
|
|
|
|
|
|
def test_an_approval_of_an_older_commit_doesnt_count(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _ = to_review(world)
|
|
pr = loop_of(the_item(world)).pr
|
|
forge.approve(pr, sha="0" * 40) # what the PR was before a fix
|
|
agent.watch()
|
|
assert forge.agent_merged == []
|
|
|
|
|
|
def test_only_an_operators_approval_counts(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _ = to_review(world)
|
|
pr = loop_of(the_item(world)).pr
|
|
forge.approve(pr, by="mallory")
|
|
agent.watch()
|
|
assert forge.agent_merged == []
|
|
|
|
|
|
def test_pending_checks_wait_and_failed_ones_are_said_once(
|
|
world: dict[str, Any],
|
|
) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, transport = to_review(world)
|
|
pr = loop_of(the_item(world)).pr
|
|
forge.approve(pr)
|
|
forge.status[forge.head_sha(pr)] = "pending"
|
|
agent.watch()
|
|
assert forge.agent_merged == []
|
|
forge.status[forge.head_sha(pr)] = "failure"
|
|
agent.watch()
|
|
agent.watch()
|
|
held = [t for t in said(transport) if "its checks failed" in t]
|
|
assert len(held) == 1 and forge.agent_merged == []
|
|
forge.status[forge.head_sha(pr)] = "success"
|
|
agent.watch()
|
|
assert forge.agent_merged == [pr]
|
|
|
|
|
|
def test_protection_is_checked_at_start(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
forge.protect()
|
|
agent, transport = agent_for(world, ScriptedModel())
|
|
agent.boot()
|
|
assert agent._merging == {"sandbox": True}
|
|
# The check's branch and PR are gone again.
|
|
assert all(p["state"] == "closed" for p in forge.prs.values())
|
|
assert forge.agent_merged == []
|
|
assert not [t for t in said(transport) if "leave merging to you" in t]
|
|
|
|
|
|
def test_unprotected_means_the_operator_merges(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"] # not protected: the agent could merge anything
|
|
model = ScriptedModel(RESTATE, [*WORK, *FINISH])
|
|
agent, transport = agent_for(world, model)
|
|
agent.boot()
|
|
assert agent._merging == {"sandbox": False}
|
|
assert any("I could merge my own PR" in t for t in said(transport))
|
|
agent.intake()
|
|
say(transport, world, "go")
|
|
pr = loop_of(the_item(world)).pr
|
|
forge.approve(pr)
|
|
agent.watch()
|
|
assert pr not in forge.agent_merged # left to the operator
|
|
|
|
|
|
def test_a_refused_merge_is_said_once_and_left_to_the_operator(
|
|
world: dict[str, Any],
|
|
) -> None:
|
|
from nisse_core.gitea import GiteaError
|
|
|
|
forge: Forge = world["forge"]
|
|
agent, transport = to_review(world)
|
|
pr = loop_of(the_item(world)).pr
|
|
forge.approve(pr)
|
|
|
|
def refuse(repo: str, index: int) -> None:
|
|
raise GiteaError("gitea POST merge returned 405: not mergeable")
|
|
|
|
forge.merge_pull_request = refuse # type: ignore[method-assign]
|
|
agent.watch()
|
|
agent.watch()
|
|
refused = [t for t in said(transport) if "Gitea refused my merge" in t]
|
|
assert len(refused) == 1
|
|
assert loop_of(the_item(world)).wait == MERGE
|