From a85e855fc200024dd177e66bd92d602bf8423369 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sat, 17 Oct 2026 15:41:55 +0000 Subject: [PATCH] nisse-review: a stop ends a review in progress; quieter repo failures SIGTERM used to stop only new rounds: a review in progress ran on until systemd killed it 90s later. Now the scheduler's stop sets the driver's cancel flag, so the run ends at its next turn, and a model call in flight is broken off (Interruptible, raising Stopped from the signal handler, only ever inside a model call). The item goes back to ready, with nothing drafted posted and no failure counted; replies already posted carry their markers and are not repeated. intake.repo_failed is logged when a repository starts failing or the reason changes, and intake.repo_ok when it lists again, not every round. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk --- docs/Architecture.md | 5 +- docs/Development-Plan.md | 5 + .../src/nisse_core/models/interruptible.py | 51 ++++++ nisse-review/src/nisse_review/build.py | 19 +- nisse-review/src/nisse_review/main.py | 2 +- nisse-review/src/nisse_review/scheduler.py | 41 ++++- nisse-review/tests/test_review.py | 4 +- nisse-review/tests/test_review_stop.py | 170 ++++++++++++++++++ 8 files changed, 287 insertions(+), 10 deletions(-) create mode 100644 nisse-core/src/nisse_core/models/interruptible.py create mode 100644 nisse-review/tests/test_review_stop.py diff --git a/docs/Architecture.md b/docs/Architecture.md index f9db335..1cae445 100644 --- a/docs/Architecture.md +++ b/docs/Architecture.md @@ -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 diff --git a/docs/Development-Plan.md b/docs/Development-Plan.md index 1537d15..afd6bbf 100644 --- a/docs/Development-Plan.md +++ b/docs/Development-Plan.md @@ -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). diff --git a/nisse-core/src/nisse_core/models/interruptible.py b/nisse-core/src/nisse_core/models/interruptible.py new file mode 100644 index 0000000..0fe94ea --- /dev/null +++ b/nisse-core/src/nisse_core/models/interruptible.py @@ -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) diff --git a/nisse-review/src/nisse_review/build.py b/nisse-review/src/nisse_review/build.py index 508c2cd..2b2e461 100644 --- a/nisse-review/src/nisse_review/build.py +++ b/nisse-review/src/nisse_review/build.py @@ -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) diff --git a/nisse-review/src/nisse_review/main.py b/nisse-review/src/nisse_review/main.py index bffa6d9..c142757 100644 --- a/nisse-review/src/nisse_review/main.py +++ b/nisse-review/src/nisse_review/main.py @@ -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) diff --git a/nisse-review/src/nisse_review/scheduler.py b/nisse-review/src/nisse_review/scheduler.py index 4bae9c6..70c6ca6 100644 --- a/nisse-review/src/nisse_review/scheduler.py +++ b/nisse-review/src/nisse_review/scheduler.py @@ -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: diff --git a/nisse-review/tests/test_review.py b/nisse-review/tests/test_review.py index 6fe0e74..7d916b6 100644 --- a/nisse-review/tests/test_review.py +++ b/nisse-review/tests/test_review.py @@ -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, diff --git a/nisse-review/tests/test_review_stop.py b/nisse-review/tests/test_review_stop.py new file mode 100644 index 0000000..b688d39 --- /dev/null +++ b/nisse-review/tests/test_review_stop.py @@ -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() -- 2.54.0