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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk

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

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).

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)

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

@@ -2111,6 +2111,7 @@ class LoopAgent:
         result = delegation.run(
             self._store.transcript(item.id, "helper"), brief=brief, branch=branch
         )
+        lost: GitError | None = None
         try:
             # The milestone takes the helper's work (or, on failure, only
             # its own back) as uncommitted changes: harness branches never
@@ -2122,6 +2123,8 @@ class LoopAgent:
                 git._git("branch", "-q", "-D", branch)
         except GitError as exc:
             self._fact(item, f"after helper {number}: {exc}")
+            if isinstance(exc.__cause__, HostUnavailable):
+                lost = exc  # the VM went: the driver's run ends too (#137)
         for intent in self._store.pending_intents():
             # Cut off rather than finished: discarded all the same, never
             # resumed blindly (L8, L9).
@@ -2130,6 +2133,8 @@ class LoopAgent:
                 and intent.payload.get("branch") == branch
             ):
                 self._store.abort_intent(intent.id, cause=f"delegation {result.status}")
+        if lost is not None:
+            raise VMUnreachable(str(lost)) from lost
         if result.usable and result.handback is not None:
             self._fact(item, f"helper {number} finished: {result.handback.render()}")
             return f"the helper finished:\n{result.handback.render()}", True

nisse-dev/tests/test_loop_vm.py

@@ -338,3 +338,42 @@ def test_a_vm_lost_just_before_the_step_blocks_before_any_model_call(
     agent, _ = agent_for(world, project, model, clock=Clock())
     agent.intake()
     assert list(blocks(world).values()) == ["unreachable"]
+
+
+def test_a_vm_lost_as_a_helper_hands_back_ends_the_drivers_run(
+    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Found by the property walk (seed 468): taking the helper's work back
+    # onto the milestone's branch failed for the lost VM, was noted as a git
+    # error, and the driver's model was called with the VM gone.
+    from test_loop_help import Spy, hand_back, helpers
+
+    faults = Faults(monkeypatch, world["host"])
+
+    class Leaving(Spy):
+        def complete(self, events, tools, model, system=None) -> list[Any]:
+            out = super().complete(events, tools, model, system)
+            if not self.turns:  # gone as its work is taken back
+                faults.drop_on = "checkout -q -f -B"
+            return out
+
+    helper = Leaving(
+        [call("write_file", path="helped.py", content="H = 1\n")], [hand_back()]
+    )
+    model = ScriptedModel(
+        ready(),
+        RESTATE,
+        [
+            call("start_milestone", name="m"),
+            call("delegate", task="write helped.py"),
+        ],
+        [call("run_command", command="echo never")],  # never asked for
+    )
+    agent, transport = agent_for(
+        world, project, model, clock=Clock(), helpers=helpers(helper)
+    )
+    agent.intake()
+    say(transport, world, "go")
+    assert list(blocks(world).values()) == ["unreachable"]
+    assert the_item(world).position == READY
+    assert len(model.turns) == 1  # the driver's model wasn't called again

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)

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)

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:

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,

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()