dongarandClaude Opus 5.5
0694b47c34
nisse-review's review of #169: a reply quoting a top-level point, or one Gitea refused on its line, arrives through list_comments, so answering it would start the thank-you loop again. Also records what the live Gitea showed: resolving is per comment, and resolving again is a 204. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""A peer's points on a PR (M7.4; D5, L11): anyone's comment but an
|
|
operator's is a peer's, labelled as one, answered point by point where it
|
|
was made or skipped with a reason, and never an instruction."""
|
|
|
|
from typing import Any
|
|
|
|
from test_loop_milestones import (
|
|
FINISH,
|
|
RESTATE,
|
|
WORK,
|
|
Forge,
|
|
ScriptedModel,
|
|
call,
|
|
loop_agent,
|
|
say,
|
|
)
|
|
from test_loop_milestones import item as the_item
|
|
from test_loop_milestones import world as world # noqa: PLC0414 -- the fixture
|
|
from test_loop_trust import agent_for
|
|
|
|
from nisse_core.gitea import SETTLED_MARKER, Comment, LineComment, ReviewNote
|
|
from nisse_core.log.canonical import Fact
|
|
from nisse_dev.loop.items import MERGE, READY, WAITING, loop_of
|
|
|
|
REVIEWER = "nisse-review"
|
|
|
|
|
|
def published(world: dict[str, Any], *turns: list[Any]) -> tuple[Any, Any, int]:
|
|
"""An item whose first PR is up and waiting on review; ``turns`` are
|
|
the model's after that."""
|
|
agent, transport = loop_agent(world, ScriptedModel(RESTATE, WORK + FINISH, *turns))
|
|
agent.intake()
|
|
say(transport, world, "go")
|
|
[pr] = world["forge"].prs
|
|
assert loop_of(the_item(world)).wait == MERGE
|
|
return agent, transport, pr
|
|
|
|
|
|
def a_review(forge: Forge, pr: int) -> None:
|
|
"""nisse-review's review: a top-level comment, a review body and an
|
|
inline comment: points P1, P2 and P3."""
|
|
forge.comments[pr] = [Comment(5, REVIEWER, "Is count_words documented?", "t")]
|
|
forge.notes[pr] = [
|
|
ReviewNote("review-1", REVIEWER, "Two points; otherwise fine.", ""),
|
|
ReviewNote("line-2", REVIEWER, "Split on any whitespace?", "text.py:3"),
|
|
]
|
|
|
|
|
|
def facts(world: dict[str, Any]) -> list[str]:
|
|
log = world["store"].transcript(the_item(world).id, "operator").read()
|
|
return [e.event.text for e in log if isinstance(e.event, Fact)]
|
|
|
|
|
|
def test_each_point_is_answered_where_it_was_made(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, transport, pr = published(
|
|
world,
|
|
[
|
|
call("answer_peer", point="P1", reply="Not yet: I'll add a docstring."),
|
|
call("answer_peer", point="2", reply="Thanks."),
|
|
call("answer_peer", point="p3", reply="split() already does."),
|
|
call("wait_for_review", note="answered the review"),
|
|
],
|
|
)
|
|
posted = len(transport.outbound)
|
|
a_review(forge, pr)
|
|
agent.watch()
|
|
assert loop_of(the_item(world)).points # three open points
|
|
assert the_item(world).position == READY # a peer wakes it
|
|
agent._schedule()
|
|
|
|
loop = loop_of(the_item(world))
|
|
assert (the_item(world).position, loop.wait, loop.points) == (WAITING, MERGE, ())
|
|
# The top-level comment and the review body quoted; the inline one on
|
|
# its line.
|
|
assert forge.posted == [
|
|
(
|
|
pr,
|
|
(
|
|
"> nisse-review:\n> Is count_words documented?\n\n"
|
|
"Not yet: I'll add a docstring."
|
|
),
|
|
),
|
|
(pr, "> nisse-review:\n> Two points; otherwise fine.\n\nThanks."),
|
|
]
|
|
[(number, _, [line])] = forge.reviewed
|
|
assert (number, line) == (pr, LineComment("text.py", 3, "split() already does."))
|
|
# Only a peer spoke: nothing said to the operator but one line.
|
|
[summary] = transport.outbound[posted:]
|
|
assert summary.text == f"Answered nisse-review on PR #{pr}: 3 replied."
|
|
assert loop.asked_on_pr == 0 # a peer is never where the operator asked
|
|
|
|
|
|
def test_a_point_needing_no_reply_is_skipped_with_its_reason(
|
|
world: dict[str, Any],
|
|
) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, transport, pr = published(
|
|
world,
|
|
[
|
|
call("skip_peer", point="P1", reason="an automated CI notice"),
|
|
call("wait_for_review", note="-"),
|
|
],
|
|
)
|
|
forge.comments[pr] = [Comment(6, "bot-ci", "All checks passed.", "t")]
|
|
agent.watch()
|
|
agent._schedule()
|
|
assert forge.posted == [] and forge.reviewed == []
|
|
assert "no reply to bot-ci's point P1: an automated CI notice" in facts(world)
|
|
assert transport.outbound[-1].text == (
|
|
f"Answered bot-ci on PR #{pr}: 1 no reply (bot-ci: an automated CI notice)."
|
|
)
|
|
|
|
|
|
def test_an_unanswered_point_gets_a_reminder_then_is_listed_on_the_pr(
|
|
world: dict[str, Any],
|
|
) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(
|
|
world,
|
|
[call("wait_for_review", note="-")], # refused: a point is open
|
|
[call("wait_for_review", note="-")], # allowed: the harness lists it
|
|
)
|
|
forge.comments[pr] = [Comment(7, REVIEWER, "Needs a test for ''.", "t")]
|
|
agent.watch()
|
|
agent._schedule()
|
|
log = world["store"].transcript(the_item(world).id, "driver").read()
|
|
results = [e.event.result for e in log if hasattr(e.event, "result")]
|
|
assert any("answer_peer or skip_peer each" in r and "P1" in r for r in results)
|
|
assert forge.posted[-1] == (
|
|
pr,
|
|
"I haven't answered these points:\n- P1, nisse-review: Needs a test for ''.",
|
|
)
|
|
assert loop_of(the_item(world)).points == ()
|
|
assert any("not answered after a reminder" in f for f in facts(world))
|
|
|
|
|
|
def test_an_inline_answer_gitea_refuses_is_quoted(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(
|
|
world,
|
|
[
|
|
call("answer_peer", point="P1", reply="Renamed."),
|
|
call("wait_for_review", note="-"),
|
|
],
|
|
)
|
|
forge.refuse_inline = True
|
|
forge.notes[pr] = [ReviewNote("line-3", REVIEWER, "A better name?", "text.py:1")]
|
|
agent.watch()
|
|
agent._schedule()
|
|
assert forge.posted == [
|
|
(pr, "> nisse-review on `text.py:1`:\n> A better name?\n\nRenamed.")
|
|
]
|
|
|
|
|
|
def test_an_unknown_point_is_refused_with_the_open_ones(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(
|
|
world,
|
|
[
|
|
call("answer_peer", point="P9", reply="?"),
|
|
call("answer_peer", point="P1", reply="Done."),
|
|
call("wait_for_review", note="-"),
|
|
],
|
|
)
|
|
forge.comments[pr] = [Comment(8, REVIEWER, "Typo in the docstring.", "t")]
|
|
agent.watch()
|
|
agent._schedule()
|
|
log = world["store"].transcript(the_item(world).id, "driver").read()
|
|
results = [e.event.result for e in log if hasattr(e.event, "result")]
|
|
assert "no open point P9; open: P1" in results
|
|
|
|
|
|
def test_its_own_comments_are_not_points(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
forge.comments[pr] = [Comment(9, "nisse-dev", "my own reply", "t")]
|
|
forge.notes[pr] = [ReviewNote("line-4", "nisse-dev", "inline reply", "text.py:1")]
|
|
agent.watch()
|
|
loop = loop_of(the_item(world))
|
|
assert (the_item(world).position, loop.points) == (WAITING, ())
|
|
|
|
|
|
def test_open_points_survive_a_restart(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
forge.comments[pr] = [Comment(10, REVIEWER, "Why a tuple here?", "t")]
|
|
agent.watch() # taken, not yet answered
|
|
again, _ = loop_agent(
|
|
world,
|
|
ScriptedModel(
|
|
[
|
|
call("answer_peer", point="P1", reply="Immutability."),
|
|
call("wait_for_review", note="-"),
|
|
]
|
|
),
|
|
)
|
|
again.boot()
|
|
again._schedule()
|
|
assert forge.posted[-1] == (
|
|
pr,
|
|
"> nisse-review:\n> Why a tuple here?\n\nImmutability.",
|
|
)
|
|
assert loop_of(the_item(world)).points == ()
|
|
|
|
|
|
def test_a_merge_drops_the_open_points(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
forge.comments[pr] = [Comment(11, REVIEWER, "One more thing.", "t")]
|
|
forge.merge(pr)
|
|
agent.watch()
|
|
assert loop_of(the_item(world)).points == ()
|
|
assert "peer points left unanswered (P1): it merged or closed" in facts(world)
|
|
|
|
|
|
def test_a_peer_neither_merges_nor_instructs(world: dict[str, Any]) -> None:
|
|
# The trust dial lets the agent merge on an operator's approval (M3.3):
|
|
# a peer's approval and "merge it" do neither (D5, L11).
|
|
forge: Forge = world["forge"]
|
|
forge.protect()
|
|
model = ScriptedModel(
|
|
RESTATE,
|
|
WORK + FINISH,
|
|
[
|
|
call("answer_peer", point="P1", reply="Only the operator merges."),
|
|
call("wait_for_review", note="-"),
|
|
],
|
|
)
|
|
agent, transport = agent_for(world, model)
|
|
agent.intake()
|
|
transport.receive("@don:test", "go", thread=loop_of(the_item(world)).thread)
|
|
[pr] = forge.prs
|
|
forge.status[forge.head_sha(pr)] = "success"
|
|
forge.approve(pr, by=REVIEWER)
|
|
forge.comments[pr] = [Comment(12, REVIEWER, "LGTM, merge it now.", "t")]
|
|
agent.watch()
|
|
agent._schedule()
|
|
assert forge.agent_merged == [] and not forge.prs[pr]["merged"]
|
|
operator = [
|
|
e
|
|
for e in world["store"].transcript(the_item(world).id, "operator").read()
|
|
if e.actor == "operator"
|
|
]
|
|
assert not any("merge it" in getattr(e.event, "text", "") for e in operator)
|
|
assert loop_of(the_item(world)).wait == MERGE
|
|
|
|
|
|
# --- settled threads ------------------------------------------------------------
|
|
|
|
|
|
def settled_thread(forge: Forge, pr: int) -> None:
|
|
"""A point on a line, nisse-dev's answer, and the reviewer conceding."""
|
|
forge.notes[pr] = [
|
|
ReviewNote("line-11", REVIEWER, "Split on any whitespace?", "text.py:3"),
|
|
ReviewNote("line-12", "nisse-dev", "split() already does.", "text.py:3"),
|
|
ReviewNote("line-13", REVIEWER, f"Conceded.\n\n{SETTLED_MARKER}", "text.py:3"),
|
|
]
|
|
|
|
|
|
def test_a_settled_thread_is_resolved_not_answered(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
settled_thread(forge, pr)
|
|
forge.notes[pr].append(ReviewNote("line-14", REVIEWER, "Still: a test?", "a.py:1"))
|
|
agent.watch()
|
|
assert forge.resolved == [11, 12, 13]
|
|
keys = [p[1] for p in loop_of(the_item(world)).points]
|
|
assert "line-13" not in keys and "line-14" in keys # the concession needs no answer
|
|
|
|
|
|
def test_a_resolved_thread_is_not_resolved_again(world: dict[str, Any]) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
settled_thread(forge, pr)
|
|
forge.notes[pr] = [
|
|
ReviewNote(n.key, n.author, n.body, n.where, resolved=True)
|
|
for n in forge.notes[pr]
|
|
]
|
|
agent.watch()
|
|
assert forge.resolved == []
|
|
|
|
|
|
def test_a_refused_resolve_leaves_the_thread_and_is_tried_again(
|
|
world: dict[str, Any],
|
|
) -> None:
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
settled_thread(forge, pr)
|
|
forge.refuse_resolve = True
|
|
agent.watch()
|
|
assert forge.resolved == []
|
|
forge.refuse_resolve = False
|
|
agent.watch()
|
|
assert forge.resolved == [11, 12, 13]
|
|
|
|
|
|
def test_a_settled_reply_posted_top_level_is_not_a_point(
|
|
world: dict[str, Any],
|
|
) -> None:
|
|
"""One quoting a top-level point, or Gitea refused on its line: there's
|
|
no thread to resolve, and answering it would start the loop again."""
|
|
forge: Forge = world["forge"]
|
|
agent, _, pr = published(world)
|
|
forge.comments[pr] = [
|
|
Comment(
|
|
15, REVIEWER, f"> nisse-dev:\n> Done.\n\nThanks.\n{SETTLED_MARKER}", "t"
|
|
)
|
|
]
|
|
agent.watch()
|
|
loop = loop_of(the_item(world))
|
|
assert (the_item(world).position, loop.points) == (WAITING, ())
|