nisse-review: a stop ends a review in progress; quieter repo failures #159

Merged
dev-agent merged 1 commits from review-stop into main 2026-10-17 22:12:32 +00:00
8 changed files with 287 additions and 10 deletions

No files matched your search

+4 -1
View File
@@ -962,7 +962,10 @@ Two agents can argue forever, so a PR's exchanges are capped.
room, it gets one too. Until then the operator stops it on a PR by commenting
`stop` there (a review is asked for again with `review`: a fresh one of the
whole PR, with the round count reset), and stops it entirely by stopping its
container. A command is an operator's whole top-level comment, acknowledged
container. Stopping it ends a review in progress at once: at its next turn,
or by breaking off the model call it is waiting on. Nothing drafted is
posted, and the PR is reviewed again when it starts, with no failure
counted. A command is an operator's whole top-level comment, acknowledged
with a reaction rather than a comment the PR's author would have to answer.
- **Its own identity** (L12): a Gitea account (`nisse-review`) with read access to
the repositories it reviews, its own token and LiteLLM key, and its own
+5
View File
@@ -694,6 +694,11 @@ of earlier points, replies on threads, the round cap and its reset, `stop` and
- **Tests:** `test_review_rounds.py`, including a no-dead-end walk over every
position; the live smoke test adds a push and a question, and checks the
later review at the new head.
- **Follow-ups:** a stop (SIGTERM) ends a review in progress at its next turn,
or breaks off its model call (`Interruptible`, `nisse_core.models`), and the
PR goes back to ready with no failure counted; a repository Gitea won't
list is logged when that starts or changes (`intake.repo_failed`), and when
it ends (`intake.repo_ok`), not every round. Tests: `test_review_stop.py`.
**M7.6 Deployment.** *Waits for the agent host (Phase 6).* Its image, container,
Gitea account and secrets (R5).
@@ -0,0 +1,51 @@
"""A model call that a stop can break off (L2).
A ``Cancel`` takes effect at the next tool boundary, but one model call can
run for many minutes (``request_timeout_s``), longer than a service manager
waits before it kills the process. ``Interruptible`` wraps the client and
knows when a call is in flight; ``interrupt``, called from the signal
handler, raises ``Stopped`` inside that call so it ends at once.
Only a model call is ever broken off this way, never a post or a write:
outside a call, ``interrupt`` does nothing and the run ends at its next
boundary instead. Python runs signal handlers on the main thread, so this
works when the calls are made there.
"""
from typing import Any
from nisse_core.log.canonical import CanonicalEvent, StoredEvent
from nisse_core.models.llm import LLMClient
class Stopped(BaseException):
"""A model call broken off by a stop. A ``BaseException``, so no handler
meant for failures (``except Exception``, ``except LLMError``) takes it
for one."""
class Interruptible:
def __init__(self, client: LLMClient) -> None:
self._client = client
self._calling = False
def complete(
self,
events: list[StoredEvent],
tools: list[dict[str, Any]] | None,
model: str,
system: str | None = None,
) -> list[CanonicalEvent]:
self._calling = True
try:
return self._client.complete(events, tools, model, system)
finally:
self._calling = False
def interrupt(self) -> None:
"""From a signal handler: breaks off the call in flight, if any."""
if self._calling:
raise Stopped("a stop broke off the model call")
def __getattr__(self, name: str) -> Any:
return getattr(self._client, name)
+15 -4
View File
@@ -12,6 +12,7 @@ from nisse_core.loop.driver import Driver
from nisse_core.loop.expert import ExpertDesk
from nisse_core.models.catalog import ModelCatalog
from nisse_core.models.context import BudgetedClient
from nisse_core.models.interruptible import Interruptible
from nisse_core.models.llm import LiteLLMClient
from nisse_core.oplog import OpLog
from nisse_core.state.cancel import Cancel
@@ -36,6 +37,13 @@ class Built:
repositories: list[str]
store: StateStore
gitea: GiteaClient
model: Interruptible
def stop(self) -> None:
"""From the signal handler: no more rounds, the run in progress
ends at its next turn, and a model call in flight is broken off."""
self.scheduler.stop()
self.model.interrupt()
def close(self) -> None:
self.gitea.close()
@@ -97,7 +105,8 @@ def build(
oplog = oplog or OpLog(AGENT)
gitea = gitea or GiteaClient(config.gitea_url, config.gitea_token)
account, repos = check_access(gitea, config.repositories, oplog)
client = client if client is not None else llm_client(config)
# Every model call, the reviewer's and ask_expert's, can be broken off.
model = Interruptible(client if client is not None else llm_client(config)) # type: ignore[arg-type]
config.state_db.parent.mkdir(parents=True, exist_ok=True)
store = StateStore(config.state_db, oplog=oplog)
vault = (
@@ -116,16 +125,17 @@ def build(
else None
)
tuning = config.tuning
cancel = Cancel()
reviewer = Reviewer(
store,
gitea,
Driver(client, config.roles.resolve("driver"), cancel=Cancel()), # type: ignore[arg-type]
Driver(model, config.roles.resolve("driver"), cancel=cancel), # type: ignore[arg-type]
account=account,
gitea_url=config.gitea_url,
oplog=oplog,
knowledge=Knowledge(vault, memory, oplog),
expert=ExpertDesk(
client, # type: ignore[arg-type]
model, # type: ignore[arg-type]
config.roles,
budget=tuning.expert.consult_budget,
),
@@ -147,5 +157,6 @@ def build(
operators=config.gitea_operators,
max_rounds=tuning.review.max_rounds,
patterns=config.repositories,
cancel=cancel,
)
return Built(scheduler, account, repos, store, gitea)
return Built(scheduler, account, repos, store, gitea, model)
+1 -1
View File
@@ -26,7 +26,7 @@ def main() -> int:
def shutdown(_signum: int, _frame: FrameType | None) -> None:
print("nisse-review: stopping", file=sys.stderr)
built.scheduler.stop()
built.stop()
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
+38 -3
View File
@@ -18,7 +18,9 @@ comment quiets it until their `review` (R5), which asks for a fresh review.
Crash safety (L8): an item found ``running`` was interrupted. Whether its
review got posted is asked of Gitea, never assumed; until Gitea can say, the
item waits rather than risk a second review. A reply carries the key of the
point it answers, so a re-run never answers it twice.
point it answers, so a re-run never answers it twice. A stop (SIGTERM) ends
a run at its next turn, or breaks off its model call, and the item goes back
to ready, with nothing drafted posted and no failure counted.
"""
import re
@@ -30,7 +32,9 @@ from dataclasses import replace
from pathlib import Path
from nisse_core.gitea import GiteaClient, GiteaError, PullRequest
from nisse_core.models.interruptible import Stopped
from nisse_core.oplog import OpLog
from nisse_core.state.cancel import Cancel
from nisse_core.state.store import WorkItem
from nisse_review.access import check, reviewed
from nisse_review.reader import Repository
@@ -96,6 +100,7 @@ class Scheduler:
ci_grace_s: float = 120.0,
operators: frozenset[str] = frozenset(),
max_rounds: int = 4,
cancel: Cancel | None = None,
clock: Callable[[], float] = time.time,
) -> None:
self._store = store
@@ -114,6 +119,9 @@ class Scheduler:
self._clock = clock
self._said_waiting: set[str] = set() # "item@head", logged once
self._stop = threading.Event()
# The reviewer's driver's flag: a stop ends a run at its next turn.
self._cancel = cancel or Cancel()
self._repo_failing: dict[str, str] = {} # repo: why, logged on change
# -- the loop ---------------------------------------------------------------
@@ -123,7 +131,10 @@ class Scheduler:
self._stop.wait(self._poll_s)
def stop(self) -> None:
"""No more rounds, and a run in progress ends at its next turn,
back to ready: a stop is no failure of the review."""
self._stop.set()
self._cancel.set()
def round(self) -> bool:
"""The wall, intake, then one run if one is due; True if one ran.
@@ -158,9 +169,14 @@ class Scheduler:
except GiteaError as exc:
# One repository Gitea won't list (an empty one, one without
# pull requests) holds up none of the others; its items stay
# as they are.
self._oplog.emit("intake.repo_failed", cause=str(exc)[:300], repo=repo)
# as they are. Said when it starts, or the reason changes.
why = str(exc)[:300]
if self._repo_failing.get(repo) != why:
self._repo_failing[repo] = why
self._oplog.emit("intake.repo_failed", cause=why, repo=repo)
continue
if self._repo_failing.pop(repo, None) is not None:
self._oplog.emit("intake.repo_ok", cause="listed again", repo=repo)
seen: set[str] = set()
for pr in prs:
key = item_id(repo, pr.number)
@@ -392,6 +408,9 @@ class Scheduler:
"review.partial", cause=outcome.note, work_item=item.id
)
tool, summary = "post_review", PARTIAL.format(n=len(draft.comments))
elif outcome.kind == "stopped":
self._interrupted(item, known, outcome.note)
return
elif outcome.kind != "wait":
self._failed(item, known, pr, outcome.note or outcome.kind)
return
@@ -404,6 +423,10 @@ class Scheduler:
posted = self._reviewer.post(
item.id, known.repo, pr, summary, kept, n=known.posts + 1
)
except Stopped as exc:
# A stop broke off a model call: a stop all the same.
self._interrupted(item, known, str(exc))
return
except GiteaError as exc:
self._failed(item, known, None, f"Gitea: {exc}")
return
@@ -462,6 +485,18 @@ class Scheduler:
cause=f"{tool}",
)
def _interrupted(self, item: WorkItem, known: PR, why: str) -> None:
"""A stop ended the run: nothing drafted is posted, no attempt is
counted, and it runs again as it was. Replies it posted carry their
markers, so the next run doesn't repeat them."""
self._oplog.emit("review.interrupted", cause=why[:300], work_item=item.id)
self._store.update_work_item(
item.id,
position=READY,
data=data(replace(known, run="")),
cause="stopped mid-run",
)
def _failed(
self, item: WorkItem, known: PR, pr: PullRequest | None, why: str
) -> None:
+3 -1
View File
@@ -15,6 +15,7 @@ from nisse_core.knowledge.vault import Vault
from nisse_core.log.canonical import AssistantText
from nisse_core.loop.driver import Driver
from nisse_core.oplog import OpLog
from nisse_core.state.cancel import Cancel
from nisse_review.access import AccessRefused, check, reviewed
from nisse_review.build import StartupError, check_access
from nisse_review.review import Reviewer, _diff_lines, marker
@@ -38,12 +39,13 @@ def world(
) -> dict[str, Any]:
gitea = gitea or FakeGitea(repos={REPO: [pr()]})
oplog = OpLog("nisse-review", io.StringIO())
cancel = more.setdefault("cancel", Cancel()) # the driver's and the scheduler's
store = StateStore(tmp_path / "state.db", oplog=oplog)
vault = Vault(tmp_path / "vault", {"memory/nisse-review": "rw"}, agent="r")
reviewer = Reviewer(
store,
gitea, # type: ignore[arg-type]
Driver(model, "m"), # type: ignore[arg-type]
Driver(model, "m", cancel=cancel), # type: ignore[arg-type]
account=gitea.account,
gitea_url="https://gitea.test",
oplog=oplog,
+170
View File
@@ -0,0 +1,170 @@
"""A clean stop (L2): SIGTERM ends a review run at its next turn, or breaks
off its model call, and the item goes back to ready with no failure counted;
a repository Gitea won't list is logged when that changes, not every round."""
import io
import json
from collections.abc import Callable
from pathlib import Path
from typing import Any
from review_fakes import FakeGitea, Scripted, call, pr
from test_review import REPO, REVIEW, item, world
from nisse_core.models.interruptible import Interruptible, Stopped
from nisse_core.oplog import OpLog
from nisse_review.build import build
from nisse_review.config import ReviewConfig
from nisse_review.review import reply_marker
from nisse_review.state import READY, REVIEWED, pr_of
class Stopping(Scripted):
"""Scripted, but at call ``at`` (0-based) it first runs ``then``: a stop
arriving while the model works on that turn."""
def __init__(self, *turns: list[Any], at: int) -> None:
super().__init__(*turns)
self.at = at
self.then: Callable[[], None] = lambda: None
self.calls = 0
def complete(self, events, tools, model, system=None): # type: ignore[no-untyped-def]
now, self.calls = self.calls, self.calls + 1
if now == self.at:
self.then()
return super().complete(events, tools, model, system)
def events(out: io.StringIO, name: str) -> list[dict[str, Any]]:
found = (json.loads(line) for line in out.getvalue().splitlines())
return [e for e in found if e["event"] == name]
def logged(w: dict[str, Any]) -> io.StringIO:
out = io.StringIO()
w["scheduler"]._oplog = OpLog("t", out)
return out
def test_a_stop_ends_the_run_at_its_next_turn(tmp_path: Path) -> None:
model = Stopping(*REVIEW, at=1)
w = world(tmp_path, model)
model.then = w["scheduler"].stop
out = logged(w)
assert w["scheduler"].round() is True
assert model.calls == 2 # the turn in flight finished; no third
assert (item(w).position, pr_of(item(w)).attempts) == (READY, 0)
assert pr_of(item(w)).run == "" and w["gitea"].posted == []
assert len(events(out, "review.interrupted")) == 1
assert events(out, "review.failed") == []
w["scheduler"].run_forever() # stopped: returns at once
def test_a_stop_breaks_off_the_model_call_in_flight(tmp_path: Path) -> None:
model = Stopping(*REVIEW, at=1)
wrapped = Interruptible(model)
w = world(tmp_path, wrapped) # type: ignore[arg-type]
def sigterm() -> None: # what Built.stop does, in the signal handler
w["scheduler"].stop()
wrapped.interrupt()
model.then = sigterm
out = logged(w)
assert w["scheduler"].round() is True
assert (item(w).position, pr_of(item(w)).attempts) == (READY, 0)
assert w["gitea"].posted == []
[said] = events(out, "review.interrupted")
assert "broke off the model call" in said["cause"]
def test_outside_a_model_call_interrupt_does_nothing() -> None:
wrapped = Interruptible(Scripted([]))
wrapped.interrupt() # nothing in flight: no Stopped
assert wrapped.complete([], None, "m") == []
wrapped.interrupt() # the call is over: still nothing
assert wrapped.offered == [set()] # the rest reaches the client
def test_stopped_is_no_failure_handlers_would_take() -> None:
assert not issubclass(Stopped, Exception)
def test_replies_posted_before_a_stop_are_not_posted_again(tmp_path: Path) -> None:
model = Stopping(
*REVIEW,
[call("reply", point="P1", text="Tabs never occur here.")],
[call("reply", point="P2", text="It counts words, yes.")], # stopped here
[call("reply", point="P1", text="It counts words, yes.")],
[call("done")],
at=5,
)
gitea = FakeGitea(repos={REPO: [pr()]})
w = world(tmp_path, model, gitea, operators=frozenset({"don"}))
model.then = w["scheduler"].stop
assert w["scheduler"].round() is True
first = w["gitea"].say("nisse-dev", "Why flag split()?")
second = w["gitea"].say("nisse-dev", "Does it count words?")
assert w["scheduler"].round() is True # replied to P1, then stopped
assert item(w).position == READY
[(_, _, posted)] = w["gitea"].comments
assert reply_marker(f"comment-{first}") in posted
w["scheduler"]._stop.clear()
w["scheduler"]._cancel.clear()
assert w["scheduler"].round() is True # only the second is left
bodies = [body for _, _, body in w["gitea"].comments]
assert sum(reply_marker(f"comment-{first}") in b for b in bodies) == 1
assert sum(reply_marker(f"comment-{second}") in b for b in bodies) == 1
assert item(w).position == REVIEWED
# --- a repository Gitea won't list ---------------------------------------------
def test_a_repository_that_keeps_failing_is_logged_once(tmp_path: Path) -> None:
gitea = FakeGitea(repos={"a/empty": [], REPO: []}, unlisted={"a/empty"})
w = world(tmp_path, Scripted(), gitea)
out = logged(w)
for _ in range(3):
w["scheduler"].round()
assert len(events(out, "intake.repo_failed")) == 1
gitea.unlisted.clear()
w["scheduler"].round()
w["scheduler"].round()
[ok] = events(out, "intake.repo_ok")
assert ok["repo"] == "a/empty"
gitea.unlisted.add("a/empty") # failing again is news again
w["scheduler"].round()
assert len(events(out, "intake.repo_failed")) == 2
def test_build_wires_a_stop_to_every_model_call(tmp_path: Path) -> None:
config = ReviewConfig.from_tree(
{
"state_db": str(tmp_path / "state.db"),
"llm": {"base_url": "http://llm.test", "token": "t"},
"roles": {"reviewer": "m", "expert": "e"},
"gitea": {
"url": "https://gitea.test",
"token": "t",
"operators": ["don"],
},
"repositories": [REPO],
},
{},
)
built = build(
config,
gitea=FakeGitea(repos={REPO: []}), # type: ignore[arg-type]
client=Scripted(),
oplog=OpLog("t", io.StringIO()),
)
try:
driver = built.scheduler._reviewer._driver
assert driver._client is built.model
assert built.scheduler._reviewer._expert._client is built.model
built.stop() # no call in flight: nothing raised
assert driver._cancel.requested and built.scheduler._stop.is_set()
finally:
built.close()