nisse-review: a stop ends a review in progress; quieter repo failures #159
No files matched your search
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user