Merge pull request 'Threads, commands and labels on the new loop (M2.3)' (#102) from loop-labels into main
CI / checks (push) Successful in 3m42s

Reviewed-on: dongar/nisse#102
Reviewed-by: Don Garrett <dgarrett@acm.org>
This commit was merged in pull request #102.
This commit is contained in:
dongar committed 2026-10-04 20:15:08 +00:00
commit da32cac5cf
13 files changed
+793 -35

No files matched your search

+14 -5
View File
@@ -27,7 +27,7 @@ Only `nisse-dev` has real code; `nisse-core` and `nisse-ops` are placeholders.
| **L2** Agent loop | partial | `control.py`, a fixed state machine with no-stranding, `stop`, `abandon`. Replaced by the new loop (Phase 2) |
| **L3** Authority | partial | Revision binding and arrival stamping (`gates.py`); scope is still plan approval, not the issue and its conversation |
| **L4** Classifier | none | |
| **L5** Conversation | partial | Matrix transport, mentions, lenient commands; no threads, no question handling, little explanation |
| **L5** Conversation | partial | Matrix transport, mentions, lenient commands; on the new loop, a thread per item, named and ambiguous commands, plain-language go-aheads (M2.3); little explanation |
| **L6** Context | partial | Context fitting, per-role views and orchestrator facts (M1.2); no scope or agent memory |
| **L7** Canonical log | partial | Work item and actor on every event, `fact` (M1.1, M1.2); no `classifier_verdict`, `knowledge_retrieved` |
| **L8** Durability | mostly done | WAL store, intents, boot reconciliation, PR adoption, comment redelivery; to move onto the new loop |
@@ -36,7 +36,7 @@ Only `nisse-dev` has real code; `nisse-core` and `nisse-ops` are placeholders.
| **L11** Injection | partial | Operator allowlists on every channel; no classifier |
| **L12** Secrets, logging | partial | Structured state-change log; secrets in `.env`; no per-operation handout or redaction |
| **L13** Execution | none | Tools run locally in the agent's process. The token leaks into the clone (#89) |
| **D1–D2** Workflow | none | Plans posted to a Gitea issue and gated; no issue intake, labels, or documents in the repo |
| **D1–D2** Workflow | partial | On the new loop: issue intake, restatement and go-ahead (M2.1), labels (M2.3); no documents in the repo |
| **D3–D4** Workspace, git | partial | Idempotent publishing; no fresh workspace (#62), `git add -A` (#80) |
| **D5** Review | partial | PR conversation comments only (#69), watch stall (#70) |
| **D7–D8** Deployment, projects | none | Runs from a venv as a systemd unit on this box; one project in config |
@@ -154,9 +154,18 @@ dependencies, a closed PR handled (L1, D4). End-to-end: two issues and a depende
and checks that parked work is never lost (every file an open item wrote is on
the workspace or in the forge).
**M2.3 Threads and labels.** *Pre-approved.* Routing by thread, ambiguous commands
asked about (L5); `bug` grace periods, `discuss`, `design`; questions and
plain-language go-aheads (D1). End-to-end: a bug and a feature, labelled and not.
**M2.3 Threads and labels.** *Pre-approved.* **Done** (this PR). Routing by thread,
ambiguous commands asked about (L5); `bug` grace periods, `discuss`, `design`;
questions and plain-language go-aheads (D1). End-to-end: a bug and a feature,
labelled and not.
- Commands naming their item ("abandon #14") apply to it anywhere; `go`, `approve`
and `abandon` alone at top level are asked about. `stop` is never an answer.
- A plain-language go-ahead is read by one model call (`GoAheadReader`, on the
driver's model until the classifier role, M2.6); questions never reach it.
- `bug` waits `tuning.loop.bug_grace_s` (default 10 minutes) after its restatement;
any reply stops the countdown. `discuss` isn't offered `restate` until the
operator has spoken; `design` is steered by the driver's instructions.
- Labels are read at intake; a label changed later is not picked up yet.
**M2.4 Multi-milestone work and documents.** *Pre-approved.* Build plans in the
repository, `ARCHITECTURE.md` kept current, waiting for merges between milestones
+13
View File
@@ -513,3 +513,16 @@ Later work that will add knobs, each with its own section here:
model and a policy for what is safe to summarize.
- **Benchmarking the other roles:** planner and delegate, with delegation
turned back on for comparisons across frontier tiers.
### `tuning.loop.bug_grace_s` — default 600
How long the restatement of an issue labelled `bug` waits for a reply before
the agent goes ahead on its own (architecture D1). Any reply in the item's
thread stops the countdown, and `go` starts at once.
- **Too low:** you can't read the restatement before the fix starts, so a
misread report costs a wasted milestone.
- **Too high:** bugs wait as long as features, and the label buys nothing.
- **Deciding:** about the time it takes you to notice a message and read it
when you're around. A wrong start costs only time, since the fix still comes
as a PR you review.
+2
View File
@@ -69,3 +69,5 @@ tuning:
context: # fitting each request into the model's window
keep_recent_turns: 6 # turns never cut or dropped
safety_margin: 0.1 # fraction of the window held back
loop: # the new loop (loop: new)
bug_grace_s: 600 # a `bug` restatement proceeds after this long unanswered
+8
View File
@@ -104,6 +104,13 @@ class ContextTuning:
safety_margin: float = 0.1
@dataclass(frozen=True)
class LoopTuning:
# A restatement on a `bug` issue proceeds after this long without a
# reply (D1).
bug_grace_s: float = 600.0
@dataclass(frozen=True)
class Tuning:
executor: ExecutorTuning = field(default_factory=ExecutorTuning)
@@ -111,6 +118,7 @@ class Tuning:
delegation: DelegationTuning = field(default_factory=DelegationTuning)
tools: ToolsTuning = field(default_factory=ToolsTuning)
context: ContextTuning = field(default_factory=ContextTuning)
loop: LoopTuning = field(default_factory=LoopTuning)
_TUNING_SECTIONS: dict[str, type] = {
+182 -19
View File
@@ -17,6 +17,7 @@ run (L1).
import re
import threading
import time
import uuid
from collections.abc import Callable, Iterable
from dataclasses import replace
@@ -42,6 +43,7 @@ from nisse_dev.loop.items import (
save,
)
from nisse_dev.loop.milestones import OPEN_PR, Milestones
from nisse_dev.loop.reader import GoAheadReader, is_question
from nisse_dev.loop.tools import (
ASK_HUMAN,
CLOSE_ITEM,
@@ -59,6 +61,11 @@ from nisse_dev.worker import Worker
INTAKE_INTERVAL_S = 30.0
_GO = frozenset({"go", "approve"})
_NEXT = re.compile(r"(?:do|work on) #?(\d+) (?:next|first|now)")
# A command that names its item applies to it, wherever it is said (L5).
_NAMED = re.compile(r"(go|approve|abandon|stop) #(\d+)")
# At top level these need an item; without one they are ambiguous (L5).
_NEEDS_ITEM = frozenset({"go", "approve", "abandon"})
BUG_GRACE_S = 600.0
# Before the go-ahead the driver may look around (read, list, run a command
# to reproduce a bug) but not change anything (D1); after it, everything.
@@ -71,7 +78,8 @@ The issue and the conversation about it are below.
Before any work, find out what you need, then use restate to put your
understanding to the operator: what you take the request to mean, how it fits
the project, and your first step. Wait for their go-ahead. If the request is
unclear, ask_human instead of guessing.
unclear, ask_human instead of guessing: a restatement asks no questions, so
the operator can agree to it as it stands.
After the go-ahead, work in a milestone: call start_milestone before
changing anything (never create or switch branches yourself), make the
@@ -84,8 +92,25 @@ the PR is closed without merging, ask the operator what next. When a review
needs no change from you, or you are only waiting for the operator to merge
(you can't merge it yourself), call wait_for_review.
{labels}Never label the issue yourself; you may suggest a label to the operator.
{workspace}"""
# What an issue's labels add to the driver's instructions (D1).
LABELS = {
"bug": "This issue is labelled bug. Before restating, reproduce it with "
"your tools (run the code or the tests). If you can't reproduce it, "
"ask_human instead of restating: never fix a bug you couldn't see. Once "
"you restate, the fix goes ahead after a grace period unless the operator "
"replies; make it a failing test first, then the fix.",
"discuss": "This issue is labelled discuss: talk it through with the "
"operator (ask_human) before restating, and restate only once you agree "
"on what is wanted.",
"design": "This issue is labelled design: agree the design with the "
"operator before any code. Your restatement puts the design to them; "
"code waits for their go-ahead on it.",
}
class LoopAgent:
def __init__(
@@ -106,6 +131,9 @@ class LoopAgent:
oplog: OpLog,
roles: dict[str, str] | None = None,
milestones: Milestones | None = None,
reader: GoAheadReader | None = None,
bug_grace_s: float = BUG_GRACE_S,
clock: Callable[[], float] = time.time,
intake_interval_s: float = INTAKE_INTERVAL_S,
) -> None:
self._store = store
@@ -123,10 +151,16 @@ class LoopAgent:
self._oplog = oplog
self._roles = roles or {}
self._milestones = milestones
self._reader = reader
self._grace = bug_grace_s
self._clock = clock
self._interval = intake_interval_s
# thread root -> revision on screen there, read on the transport's
# thread to stamp replies on arrival (L3).
# thread to stamp replies on arrival (L3); issue -> thread root, for
# a command that names its item.
self._shown: dict[str, int] = {}
self._threads: dict[int, str] = {}
self._active_thread = ""
self._running = False
self._stop = threading.Event()
self._intake_thread: threading.Thread | None = None
@@ -173,6 +207,7 @@ class LoopAgent:
self._reconcile_publishing()
for item in self._store.work_items("open"):
loop = loop_of(item)
self._threads[loop.issue] = loop.thread
if item.position == RUNNING:
note = "I restarted while working on this. Reply to continue."
self._wait(item, replace(loop, wait=ANSWER, note=note), note)
@@ -249,6 +284,7 @@ class LoopAgent:
self._open(issue, tuple(blocked_by))
self.watch()
self.check_blocked()
self.check_grace()
self._schedule()
# -- PR review and merge (D4, D5) -------------------------------------------
@@ -354,6 +390,24 @@ class LoopAgent:
cause=said,
)
def check_grace(self) -> None:
"""D1: a `bug` restatement left unanswered for the grace period is
taken as agreed, bound to the revision that was up all along."""
now = self._clock()
for item in self._store.work_items("open"):
loop = loop_of(item)
if (
item.position == WAITING
and loop.wait == GO_AHEAD
and loop.grace_until
and now >= loop.grace_until
):
self._transport.post(
f"No reply on rev-{loop.rev}, so I'm going ahead.",
thread=loop.thread,
)
self._go(item, loop, "the bug grace period")
def _from_an_operator(self, issue: Issue) -> bool:
"""L11: only an operator's issue, or one an operator assigned, is
an instruction."""
@@ -375,7 +429,13 @@ class LoopAgent:
root = self._transport.post(
f"Queued {self._repo}#{issue.number}: {issue.title}{where}"
)
loop = Loop(thread=root, issue=issue.number, blocked_by=blocked_by)
loop = Loop(
thread=root,
issue=issue.number,
blocked_by=blocked_by,
labels=tuple(issue.labels),
)
self._threads[issue.number] = root
self._store.open_work_item(
item_id,
title=issue.title,
@@ -408,10 +468,21 @@ class LoopAgent:
"""On the transport's thread: stamp, then hand to the worker."""
if msg.sender not in self._operators:
return
if command_form(msg.text) == "stop" and self._running:
command = command_form(msg.text)
thread = msg.thread
named = _NAMED.fullmatch(command)
if named is not None:
command = named.group(1)
thread = self._threads.get(int(named.group(2)), thread)
if (
command == "stop"
and self._running
# Unnamed at top level, it stops the active item (L5).
and (not thread or thread == self._active_thread)
):
# Straight to the flag: the run is what it has to reach (L2).
self._cancel.set()
about = self._shown.get(msg.thread) if msg.thread else None
about = self._shown.get(thread) if thread else None
self._worker.submit(
self._settled("your message", lambda: self.handle(msg, about))
)
@@ -430,13 +501,38 @@ class LoopAgent:
)
self._schedule()
return
named = _NAMED.fullmatch(command)
if named is not None:
# L5: a command that names its item applies to it.
command, issue = named.group(1), int(named.group(2))
item = next(
(
i
for i in self._store.work_items("open")
if loop_of(i).issue == issue
),
None,
)
if item is None:
self._store.transcript(None, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
self._transport.post(
f"#{issue} isn't one of my open work items.", thread=msg.thread
)
return
if item is None:
self._store.transcript(None, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
if command == "status":
self._transport.post(self.status(), thread=msg.thread)
elif command != "stop":
elif command in _NEEDS_ITEM:
self._transport.post(self._which(command), thread=msg.thread)
elif command == "stop":
if not self._store.active_item():
self._transport.post("Nothing is running.", thread=msg.thread)
else:
self._transport.post(
f"I take work from issues assigned to me in {self._repo}: "
"assign one and I'll pick it up. Reply in its thread to "
@@ -447,12 +543,27 @@ class LoopAgent:
self._store.transcript(item.id, "operator").append(
UserMessage(text=text, source=f"transport:{msg.sender}")
)
loop = replace(loop_of(item), asked_on_pr=0, asked_where="", asked_text="")
# Any reply stops a grace period's countdown: the operator is here.
loop = replace(
loop_of(item),
asked_on_pr=0,
asked_where="",
asked_text="",
grace_until=0.0,
talked=True,
)
if loop != loop_of(item):
save(self._store, item.id, item.position, loop, cause="operator spoke")
item = self._store.work_item(item.id) or item
if command == "abandon":
self._close(item, loop, "abandoned", f"abandoned by {msg.sender}")
elif command == "stop":
# Stopping a run took effect on arrival; a stop is never an answer.
if not loop.note.startswith("stopped ("):
self._transport.post(
f"#{loop.issue} isn't running; nothing to stop.",
thread=loop.thread,
)
elif item.position == BLOCKED:
self._transport.post(
f"#{loop.issue} is blocked by {', '.join(loop.blocked_by)}; "
@@ -468,14 +579,18 @@ class LoopAgent:
thread=loop.thread,
)
return
self._fact(item, f"go-ahead on rev-{loop.rev} from {msg.sender}")
save(
self._store,
item.id,
READY,
replace(loop, started=True, wait=""),
cause=f"go-ahead on rev-{loop.rev}",
self._go(item, loop, msg.sender)
elif (
item.position == WAITING
and loop.wait == GO_AHEAD
and about == loop.rev
and self._plain_go(item, text)
):
# L5, D1: a plain-language yes, said so before acting.
self._transport.post(
f"Taking that as a go-ahead on rev-{loop.rev}.", thread=loop.thread
)
self._go(item, loop, f"{msg.sender} (in plain language)")
elif item.position == WAITING:
# An answer, or changes to a restatement: the driver reads it.
save(
@@ -487,6 +602,39 @@ class LoopAgent:
)
self._schedule()
def _go(self, item: WorkItem, loop: Loop, by: str) -> None:
self._fact(item, f"go-ahead on rev-{loop.rev} from {by}")
save(
self._store,
item.id,
READY,
replace(loop, started=True, wait="", grace_until=0.0),
cause=f"go-ahead on rev-{loop.rev}",
)
def _plain_go(self, item: WorkItem, text: str) -> bool:
"""Whether a reply that isn't a command agrees to the restatement up.
Questions never do (L5)."""
if self._reader is None or is_question(text):
return False
proposals = [
e.event.plan
for e in self._store.transcript(item.id, "driver").read()
if isinstance(e.event, PlanSubmitted)
]
return bool(proposals) and self._reader.agrees(proposals[-1], text)
def _which(self, command: str) -> str:
"""L5: an unnamed command at top level is ambiguous: ask."""
items = self._store.work_items("open")
if not items:
return f"{command} what? I have no open work items."
listed = ", ".join(f"#{loop_of(i).issue} ({i.title})" for i in items)
return (
f"Which item? Say it in the item's thread, or name it "
f'("{command} #{loop_of(items[0]).issue}"). Open: {listed}.'
)
def _put_first(self, issue: int) -> str:
"""The operator's "do #N next" (L1): it runs before every other
ready item, the latest such order first. It doesn't unblock."""
@@ -525,6 +673,9 @@ class LoopAgent:
what = item.position
if item.position == WAITING:
what = f"waiting for {loop.wait} ({loop.note or f'rev-{loop.rev}'})"
if loop.grace_until:
left = max(0, int(loop.grace_until - self._clock()) // 60)
what += f", going ahead in {left} min unless you reply"
elif item.position == BLOCKED:
what = f"blocked by {', '.join(loop.blocked_by)}"
elif item.position == READY:
@@ -574,8 +725,10 @@ class LoopAgent:
self._transport.post(said, thread=loop.thread)
loop = replace(loop, prepared=True)
save(self._store, item.id, RUNNING, loop, cause="driver run")
# D1: a `discuss` item talks before it restates.
hold = "discuss" in loop.labels and not loop.talked and not loop.started
tools = [
RESTATE,
*([] if hold else [RESTATE]),
ASK_HUMAN,
*(
_LOOK
@@ -593,8 +746,10 @@ class LoopAgent:
repo=self._repo,
issue=loop.issue,
title=item.title,
labels="".join(f"{LABELS[x]}\n\n" for x in loop.labels if x in LABELS),
workspace=self._workspace.describe() if self._workspace else "",
)
self._active_thread = loop.thread
self._running = True
try:
outcome = self._driver.run(
@@ -622,11 +777,19 @@ class LoopAgent:
)
self._fact(item, f"restatement rev-{rev} put up for a go-ahead")
self._shown[loop.thread] = rev
self._transport.post(
f"rev-{rev}: {text}\n\nSay go to start, or reply with changes.",
thread=loop.thread,
grace = self._clock() + self._grace if "bug" in loop.labels else 0.0
after = "Say go to start, or reply with changes."
if grace:
after = (
f"It's a bug, so I'll go ahead in {round(self._grace / 60)} min "
"unless you reply; say go to start now, or reply with changes."
)
self._transport.post(f"rev-{rev}: {text}\n\n{after}", thread=loop.thread)
self._wait(
item,
replace(loop, rev=rev, wait=GO_AHEAD, note="", grace_until=grace),
None,
)
self._wait(item, replace(loop, rev=rev, wait=GO_AHEAD, note=""), None)
elif outcome.kind == "wait" and outcome.tool == "ask_human":
question = str(args.get("question", "")).strip()
self._answer_where_asked(loop, question)
+5
View File
@@ -19,6 +19,7 @@ from nisse_dev.gitea import GiteaClient
from nisse_dev.loop.agent import LoopAgent
from nisse_dev.loop.driver import Driver
from nisse_dev.loop.milestones import Milestones
from nisse_dev.loop.reader import GoAheadReader
from nisse_dev.matrix import MatrixTransport
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
@@ -109,6 +110,10 @@ def build_loop_agent(config: AgentConfig) -> RunningLoop:
cancel=cancel,
oplog=oplog,
roles=config.roles.pinned(),
# The driver's model reads plain-language go-aheads until the
# classifier role arrives (M2.6).
reader=GoAheadReader(normalized, config.roles.resolve("planner")),
bug_grace_s=config.tuning.loop.bug_grace_s,
milestones=Milestones(
store,
git,
+4
View File
@@ -40,6 +40,9 @@ class Loop:
fixes: int = 0 # review fixes pushed, for follow-up branch names
blocked_by: tuple[str, ...] = () # open dependencies, "owner/repo#N" (L1)
first: int = 0 # the operator's "do #N next": higher runs sooner
labels: tuple[str, ...] = () # the issue's labels at intake: they steer (D1)
grace_until: float = 0.0 # a `bug` restatement proceeds then, unanswered
talked: bool = False # the operator has spoken in the item's thread
# Where the operator last spoke (D5: answer where asked): "" for the
# Matrix thread, or the PR it came from, with the inline location and a
# little of what they said, for quoting.
@@ -55,6 +58,7 @@ def loop_of(item: WorkItem) -> Loop:
data = dict(item.data.get("loop", {}))
data["notes"] = tuple(data.get("notes", ()))
data["blocked_by"] = tuple(data.get("blocked_by", ()))
data["labels"] = tuple(data.get("labels", ()))
return Loop(**data)
+81
View File
@@ -0,0 +1,81 @@
"""Reading a plain-language reply as a go-ahead (L5, D1).
"Yep, that's it" in reply to a restatement may be taken as `go`, because a
wrong reading costs only time. One short model call decides, and it decides
conservatively: only a clear yes counts. A question is never a go-ahead and
never reaches the model (L5: questions never change state).
The driver's model reads for now; the classifier role takes over in M2.6.
"""
from typing import Any
from nisse_dev.canonical import StoredEvent, ToolCall, UserMessage
from nisse_dev.llm import LLMClient, LLMError
VERDICT = {
"type": "function",
"function": {
"name": "verdict",
"description": "Your reading of the operator's reply.",
"parameters": {
"type": "object",
"properties": {
"answer": {
"type": "string",
"enum": ["yes", "no", "unclear"],
"description": "yes: a clear go-ahead on the proposal as it "
"stands. no: it asks for changes, disagrees or says to wait. "
"unclear: anything else.",
}
},
"required": ["answer"],
},
},
}
SYSTEM = """You read one reply from an operator to a software agent. The agent
put a proposal up for agreement; decide whether the reply clearly agrees to the
proposal as it stands, so the agent may start. A reply that asks for any
change, asks a question, adds a condition, or is about something else is not a
go-ahead. When in doubt, it is not. Answer with the verdict tool."""
def is_question(text: str) -> bool:
return text.rstrip().endswith("?")
class GoAheadReader:
def __init__(self, client: LLMClient, model: str) -> None:
self._client = client
self._model = model
def agrees(self, proposal: str, reply: str) -> bool:
if is_question(reply):
return False
prompt = f"The proposal:\n\n{proposal}\n\nThe operator's reply:\n\n{reply}"
try:
produced = self._client.complete(
[
StoredEvent(
id=0,
ts="",
event=UserMessage(text=prompt, source="go-ahead reader"),
actor="orchestrator",
)
],
tools=[VERDICT],
model=self._model,
system=SYSTEM,
)
except LLMError:
return False # unread: the driver reads it as an answer instead
return any(_yes(event) for event in produced)
def _yes(event: Any) -> bool:
return (
isinstance(event, ToolCall)
and event.tool == "verdict"
and event.arguments.get("answer") == "yes"
)
+2
View File
@@ -158,6 +158,7 @@ def test_tuning_is_read() -> None:
"expert": {"consult_budget": 5},
"delegation": {"max_turns": 20, "max_tokens": 100000, "wall_clock_s": 600},
"tools": {"command_timeout_s": 300, "max_output_chars": 2000},
"loop": {"bug_grace_s": 120},
}
)
tuning = config.tuning
@@ -166,6 +167,7 @@ def test_tuning_is_read() -> None:
assert tuning.delegation.max_tokens == 100_000
assert tuning.delegation.wall_clock_s == 600.0
assert tuning.tools.max_output_chars == 2000
assert tuning.loop.bug_grace_s == 120.0
def test_a_partly_given_tuning_section_keeps_the_other_defaults() -> None:
+11 -3
View File
@@ -72,12 +72,18 @@ def issue(number: int = 14, author: str = "don", **kw: Any) -> Issue:
title=kw.get("title", "Add count_words"),
body=kw.get("body", "count words in a string"),
author=author,
labels=(),
labels=kw.get("labels", ()),
)
class Harness:
def __init__(self, tmp_path: Path, model: ScriptedModel, gitea: FakeGitea) -> None:
def __init__(
self,
tmp_path: Path,
model: ScriptedModel,
gitea: FakeGitea,
**agent: Any, # more LoopAgent arguments: reader, clock, ...
) -> None:
tmp_path.mkdir(parents=True, exist_ok=True)
self.store = StateStore(
tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())
@@ -100,6 +106,7 @@ class Harness:
worker=InlineWorker(),
cancel=self.cancel,
oplog=OpLog("nisse-dev", io.StringIO()),
**agent,
)
self.transport.on_message(self.agent.on_message)
@@ -215,7 +222,8 @@ def test_a_go_at_top_level_is_not_a_go_ahead(h: Harness) -> None:
h.agent.intake()
h.say("go", thread=None)
assert not loop_of(h.item()).started
assert "assign one and I'll pick it up" in h.posted()[-1]
# L5: an unnamed command at top level is ambiguous, so it asks.
assert "Which item?" in h.posted()[-1] and "#14" in h.posted()[-1]
# --- waiting points (L2) -------------------------------------------------------------
+282
View File
@@ -0,0 +1,282 @@
"""Threads, commands and labels in the new loop (M2.3): commands that name
their item, ambiguous ones asked about (L5); plain-language go-aheads; `bug`
grace periods, `discuss` and `design` (D1)."""
from pathlib import Path
from typing import Any
import pytest
from test_loop import DON, RESTATED, FakeGitea, Harness, ScriptedModel, call, issue
from test_loop import h as h # noqa: PLC0414 -- the fixture
from nisse_dev.canonical import Fact, ToolCall
from nisse_dev.llm import LLMError
from nisse_dev.loop.items import ANSWER, GO_AHEAD, READY, WAITING, loop_of
from nisse_dev.loop.reader import GoAheadReader
STARTED = [call("ask_human", question="started; anything else?")]
class Clock:
def __init__(self) -> None:
self.now = 1000.0
def __call__(self) -> float:
return self.now
class Reader:
"""A scripted go-ahead reader: agrees with what it's told to."""
def __init__(self, *yes: str) -> None:
self.yes = set(yes)
self.asked: list[tuple[str, str]] = []
def agrees(self, proposal: str, reply: str) -> bool:
self.asked.append((proposal, reply))
return reply in self.yes
def facts(h: Harness) -> list[str]:
return [
e.event.text
for e in h.store.transcript(h.item().id, "orchestrator").read()
if isinstance(e.event, Fact)
]
def two(tmp_path: Path, *turns: list[Any]) -> Harness:
return Harness(
tmp_path,
ScriptedModel(*turns),
FakeGitea(issue(14), issue(15, title="Add count_lines")),
)
# --- commands that name their item, and ambiguous ones (L5) -------------------
def test_a_named_go_at_top_level_applies_to_its_item(tmp_path: Path) -> None:
h = two(tmp_path, RESTATED, RESTATED, STARTED)
h.agent.intake()
h.say("go #15", thread=None)
items = {loop_of(i).issue: i for i in h.store.work_items()}
assert loop_of(items[15]).started and not loop_of(items[14]).started
def test_a_named_abandon_in_another_thread_still_names_its_item(
tmp_path: Path,
) -> None:
h = two(tmp_path, RESTATED, RESTATED)
h.agent.intake()
items = {loop_of(i).issue: i for i in h.store.work_items()}
h.transport.receive(DON, "Abandon #15.", thread=loop_of(items[14]).thread)
items = {loop_of(i).issue: i for i in h.store.work_items()}
assert items[15].status == "closed" and items[14].status == "open"
def test_an_unnamed_command_at_top_level_is_asked_about(tmp_path: Path) -> None:
h = two(tmp_path, RESTATED, RESTATED)
h.agent.intake()
h.say("abandon", thread=None)
assert all(i.status == "open" for i in h.store.work_items())
asked = h.posted()[-1]
assert "Which item?" in asked and "#14" in asked and "#15" in asked
def test_naming_an_item_that_isnt_open_says_so(h: Harness) -> None:
h.agent.intake()
h.say("go #99", thread=None)
assert "#99 isn't one of my open work items" in h.posted()[-1]
def test_stop_in_a_waiting_items_thread_is_not_an_answer(h: Harness) -> None:
h.agent.intake()
h.say("stop")
assert (h.item().position, loop_of(h.item()).wait) == (WAITING, GO_AHEAD)
assert "nothing to stop" in h.posted()[-1]
# --- plain-language go-aheads (L5, D1) ------------------------------------------
def test_a_plain_yes_is_a_go_ahead_said_so_first(tmp_path: Path) -> None:
reader = Reader("Yep, that's it")
h = Harness(
tmp_path, ScriptedModel(RESTATED, STARTED), FakeGitea(issue()), reader=reader
)
h.agent.intake()
h.say("Yep, that's it")
assert loop_of(h.item()).started
assert "Taking that as a go-ahead on rev-1" in " ".join(h.posted())
assert reader.asked == [
("I'll add count_words to text.py, with tests.", "Yep, that's it")
]
assert any(f.startswith("go-ahead on rev-1 from") for f in facts(h))
def test_anything_else_goes_to_the_driver(tmp_path: Path) -> None:
h = Harness(
tmp_path,
ScriptedModel(RESTATED, [call("restate", text="v2")]),
FakeGitea(issue()),
reader=Reader(),
)
h.agent.intake()
h.say("Mostly, but call it word_count")
loop = loop_of(h.item())
assert not loop.started and loop.rev == 2
def test_a_question_never_reaches_the_reader(tmp_path: Path) -> None:
reader = Reader("sounds good?")
h = Harness(
tmp_path,
ScriptedModel(RESTATED, [call("restate", text="v2")]),
FakeGitea(issue()),
reader=reader,
)
h.agent.intake()
h.say("sounds good?")
assert not loop_of(h.item()).started and reader.asked == []
def test_a_plain_yes_to_an_older_revision_is_not_a_go(tmp_path: Path) -> None:
reader = Reader("yes")
h = Harness(
tmp_path,
ScriptedModel(RESTATED, [call("restate", text="v2")], STARTED),
FakeGitea(issue()),
reader=reader,
)
h.agent.intake()
h.say("change it a bit") # rev-2 goes up
h.agent.handle(_msg("yes", h.thread()), about=1)
assert not loop_of(h.item()).started
assert [reply for _, reply in reader.asked] == ["change it a bit"]
class _Model:
def __init__(self, result: Any) -> None:
self.result = result
def complete(self, events, tools, model, system=None) -> list[Any]:
if isinstance(self.result, Exception):
raise self.result
return self.result
@pytest.mark.parametrize(
("result", "agrees"),
[
([ToolCall(call_id="v", tool="verdict", arguments={"answer": "yes"})], True),
(
[ToolCall(call_id="v", tool="verdict", arguments={"answer": "unclear"})],
False,
),
([], False),
(LLMError("down"), False),
],
)
def test_the_reader_agrees_only_on_a_clear_yes(result: Any, agrees: bool) -> None:
reader = GoAheadReader(_Model(result), "m") # type: ignore[arg-type]
assert reader.agrees("add count_words", "ok") is agrees
# --- labels (D1) ----------------------------------------------------------------
def bug(tmp_path: Path, *turns: list[Any], **kw: Any) -> tuple[Harness, Clock]:
clock = Clock()
h = Harness(
tmp_path,
ScriptedModel(*turns),
FakeGitea(issue(labels=("bug",))),
clock=clock,
bug_grace_s=600,
**kw,
)
return h, clock
def test_a_bug_goes_ahead_after_the_grace_period(tmp_path: Path) -> None:
h, clock = bug(tmp_path, RESTATED, STARTED)
h.agent.intake()
assert "go ahead in 10 min unless you reply" in h.posted()[-1]
assert "going ahead in 10 min" in h.agent.status()
clock.now += 599
h.agent.intake()
assert not loop_of(h.item()).started
clock.now += 1
h.agent.intake()
assert loop_of(h.item()).started
assert "go-ahead on rev-1 from the bug grace period" in facts(h)
def test_a_reply_stops_the_bugs_countdown(tmp_path: Path) -> None:
h, clock = bug(tmp_path, RESTATED, [call("ask_human", question="which file?")])
h.agent.intake()
h.say("hmm, what about unicode")
clock.now += 3600
h.agent.intake()
loop = loop_of(h.item())
assert not loop.started and loop.grace_until == 0 and loop.wait == ANSWER
def test_a_new_revision_of_a_bug_gets_its_own_countdown(tmp_path: Path) -> None:
h, clock = bug(tmp_path, RESTATED, [call("restate", text="v2")], STARTED)
h.agent.intake()
clock.now += 300
h.say("use the other file")
assert loop_of(h.item()).grace_until == clock.now + 600
clock.now += 600
h.agent.intake()
assert "go-ahead on rev-2 from the bug grace period" in facts(h)
def test_the_bug_label_reaches_the_driver(tmp_path: Path) -> None:
seen: list[str] = []
class Spy(ScriptedModel):
def complete(self, events, tools, model, system=None): # type: ignore[no-untyped-def]
seen.append(system or "")
return super().complete(events, tools, model, system)
h = Harness(tmp_path, Spy(RESTATED), FakeGitea(issue(labels=("bug", "ui"))))
h.agent.intake()
assert "never fix a bug you couldn't see" in seen[0]
assert "labelled discuss" not in seen[0]
def test_discuss_talks_before_it_restates(tmp_path: Path) -> None:
offered: list[set[str]] = []
class Spy(ScriptedModel):
def complete(self, events, tools, model, system=None): # type: ignore[no-untyped-def]
offered.append({t["function"]["name"] for t in tools})
return super().complete(events, tools, model, system)
h = Harness(
tmp_path,
Spy([call("ask_human", question="what's it for?")], RESTATED),
FakeGitea(issue(labels=("discuss",))),
)
h.agent.intake()
assert "restate" not in offered[0]
assert loop_of(h.item()).wait == ANSWER
h.say("for the stats page")
assert "restate" in offered[1]
assert loop_of(h.item()).wait == GO_AHEAD
def test_an_unlabelled_issue_waits_for_go_with_no_countdown(h: Harness) -> None:
h.agent.intake()
loop = loop_of(h.item())
assert loop.wait == GO_AHEAD and loop.grace_until == 0
assert h.item().position != READY
def _msg(text: str, thread: str) -> Any:
from nisse_dev.transport import InboundMessage
return InboundMessage(DON, text, thread=thread)
+70 -4
View File
@@ -19,7 +19,7 @@ from test_loop_milestones import DON, Forge, call
from test_loop_milestones import world as world # noqa: PLC0414 -- the fixture
from nisse_dev.cancel import Cancel
from nisse_dev.canonical import Fact, ToolCall
from nisse_dev.canonical import Fact, PlanSubmitted, ToolCall, UserMessage
from nisse_dev.gitea import Comment, Issue
from nisse_dev.loop.agent import LoopAgent
from nisse_dev.loop.driver import Driver
@@ -60,6 +60,8 @@ class PolicyModel:
return [call("ask_human", question=f"question {self.n}?")]
if roll < 0.12:
return [ToolCall(call_id=f"p{self.n}", tool="nonsense", arguments={})]
if "restate" not in offered and "finish_milestone" not in offered:
return [call("ask_human", question=f"talk first {self.n}?")] # discuss
if "finish_milestone" not in offered:
return [call("restate", text=f"restatement {self.n}")]
if "wait_for_review" in offered and self.rng.random() < 0.2:
@@ -89,8 +91,14 @@ class Walk:
self.steps: list[str] = []
self.shown_at_go: list[str] = [] # item:revision on screen at each go
self.next_issue = 15
self.now = 1000.0
self.agent, self.transport = self._agent()
def reads_yes(self, proposal: str, reply: str) -> bool:
"""The scripted go-ahead reader: a plain yes usually reads as one,
and nothing else ever does."""
return reply == "yep, that's right" and self.rng.random() < 0.7
def _agent(self) -> tuple[LoopAgent, InMemoryTransport]:
transport = InMemoryTransport()
cancel = Cancel()
@@ -110,10 +118,16 @@ class Walk:
cancel=cancel,
oplog=OpLog("nisse-dev", _Null()),
milestones=w["milestones"],
reader=self, # type: ignore[arg-type]
bug_grace_s=600,
clock=lambda: self.now,
)
transport.on_message(agent.on_message)
return agent, transport
def agrees(self, proposal: str, reply: str) -> bool:
return self.reads_yes(proposal, reply)
def items(self) -> list[Any]:
return self.world["store"].work_items()
@@ -127,7 +141,7 @@ class Walk:
loop = loop_of(item) if item else None
if item and loop:
weights.update({"go": 3, "answer": 2, "abandon": 1, "go-stale": 1})
weights.update({"first": 1})
weights.update({"first": 1, "plain-yes": 2, "named-go": 1, "wait": 2})
if loop.pr:
weights.update(
{"comment": 4, "merge": 2, "close-pr": 1, "comment-merge": 2}
@@ -139,11 +153,33 @@ class Walk:
elif action == "issue":
# A new issue, sometimes depending on one that is still open.
n, self.next_issue = self.next_issue, self.next_issue + 1
forge.issues.append(Issue(n, f"task {n}", f"do task {n}", "don", ()))
labels = self.rng.choice([(), (), ("bug",), ("discuss",), ("design",)])
forge.issues.append(Issue(n, f"task {n}", f"do task {n}", "don", labels))
live = [i.number for i in forge.assigned_issues("", "") if i.number != n]
if live and self.rng.random() < 0.5:
forge.deps[n] = [self.rng.choice(live)]
self.agent.intake()
elif action == "plain-yes" and loop:
# Read as a go-ahead or not; either way bound to what's up now.
if item.position == WAITING and loop.wait == GO_AHEAD:
self.shown_at_go.append(f"{item.id}:{loop.rev}")
self.transport.receive(DON, "yep, that's right", thread=loop.thread)
elif action == "named-go" and loop:
if item.position == WAITING and loop.wait == GO_AHEAD:
self.shown_at_go.append(f"{item.id}:{loop.rev}")
self.transport.receive(DON, f"go #{loop.issue}")
elif action == "wait":
# Time passes: bug countdowns may run out.
self.now += self.rng.choice([60, 700])
for other in open_items:
o = loop_of(other)
if (
other.position == WAITING
and o.wait == GO_AHEAD
and 0 < o.grace_until <= self.now
):
self.shown_at_go.append(f"{other.id}:{o.rev}")
self.agent.intake()
elif action == "first" and loop:
self.agent.handle(InboundMessage(DON, f"do #{loop.issue} next"), None)
elif action == "crash":
@@ -213,7 +249,7 @@ class Walk:
and f.event.text.startswith("go-ahead on rev-")
]
assert all(f"{item.id}:{rev}" in self.shown_at_go for rev in gos), (
f"go_binds: {gos}: {where}"
f"go_binds: {item.id} {gos} not in {self.shown_at_go}: {where}"
)
# Blocked comes from external truth (L1): an item whose issue
# had an open dependency from the start never ran before it closed.
@@ -238,6 +274,36 @@ class Walk:
assert not [
i for i in store.pending_intents() if i.work_item == item.id
], f"closed_items_leave_nothing (intent): {where}"
for item in items:
# D1: a bug's grace period runs out only on an unanswered
# restatement: any reply since it went up stops the countdown.
replied = False
for e in store.transcript(item.id, "orchestrator").read():
text = getattr(e.event, "text", "")
if isinstance(e.event, Fact) and text.startswith("restatement rev-"):
replied = False
elif isinstance(e.event, UserMessage) and e.actor == "operator":
replied = replied or e.event.source.startswith("transport:")
elif text.endswith("from the bug grace period"):
assert not replied, (
f"grace_only_unanswered: #{loop_of(item).issue}: {where}"
)
for item in items:
# D1: a `discuss` item talks before it restates.
if "discuss" not in loop_of(item).labels:
continue
events = store.transcript(item.id, "driver").read()
plans = [e.id for e in events if isinstance(e.event, PlanSubmitted)]
spoke = [
e.id
for e in events
if e.actor == "operator"
and isinstance(e.event, UserMessage)
and e.event.source.startswith("transport:")
]
assert not plans or (spoke and spoke[0] < plans[0]), (
f"discuss_talks_first: #{loop_of(item).issue}: {where}"
)
assert not [i for i in items if i.position == RUNNING], (
f"one_active (nothing left running): {where}"
)
+119 -4
View File
@@ -9,6 +9,7 @@ the item's Matrix thread through the in-memory transport. Run with
scripts/e2e.sh after each milestone.
"""
import base64
import contextlib
import os
import time
@@ -22,7 +23,7 @@ from typing import Any
import pytest
from e2e_support import base_branch, dependent_task, operator_gitea, task
from nisse_dev.config import AgentConfig
from nisse_dev.config import AgentConfig, LoopTuning
from nisse_dev.execution import SSHHost
from nisse_dev.gitea import GiteaClient, GiteaError
from nisse_dev.loop.build import RunningLoop, build_loop_agent
@@ -42,6 +43,7 @@ pytestmark = [pytest.mark.integration, pytest.mark.e2e]
OPERATOR = "@dongar:dongar.net" # an allowlisted Matrix operator in nisse-test.yaml
DEADLINE_S = 1200
MAX_ANSWERS = 5 # per wait for something else, before giving up
BUG_GRACE_S = 20
@pytest.fixture
@@ -69,6 +71,8 @@ def running(
state_db=keep / "state.db",
workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
build_host=build_host(ssh_host),
# Short, so a bug's grace period runs out within the test.
tuning=replace(live_config.tuning, loop=LoopTuning(bug_grace_s=BUG_GRACE_S)),
)
transport = InMemoryTransport()
monkeypatch.setattr("nisse_dev.loop.build.MatrixTransport", lambda cfg: transport)
@@ -143,8 +147,9 @@ class Operator:
self.transport.receive(OPERATOR, text, thread=thread)
self.settle()
def drive_until(self, wait: str, issue: int = 0) -> Any:
"""Answer whatever the agent asks until it waits for ``wait``."""
def drive_until(self, wait: str, issue: int = 0, *, go: bool = True) -> Any:
"""Answer whatever the agent asks until it waits for ``wait``.
Without ``go``, a restatement is left for its grace period."""
deadline = time.monotonic() + DEADLINE_S
answers = 0
while time.monotonic() < deadline:
@@ -155,7 +160,13 @@ class Operator:
return item
if item.status == "closed":
raise AssertionError(f"closed early: {self.last()}")
if item.position == WAITING and loop.wait == GO_AHEAD:
if wait == GO_AHEAD and loop.started:
raise AssertionError(f"went past the go-ahead: {self.last()}")
if item.position == WAITING and loop.wait == GO_AHEAD and not go:
assert loop.grace_until, f"no grace period: {self.last()}"
time.sleep(2)
self.tick() # what the agent's timer would do
elif item.position == WAITING and loop.wait == GO_AHEAD:
self.say("go", issue)
elif item.position == WAITING and loop.wait == ANSWER:
answers += 1
@@ -310,3 +321,107 @@ def test_an_issue_blocked_on_another_waits_for_it(
assert f"\n+def vowels_{run}" not in diff
finally:
operator.tidy(issues, prs)
def label(gitea: GiteaClient, repo: str, name: str) -> int:
"""The repository's label ``name``, made if it's missing."""
for existing in gitea._request("GET", f"/repos/{repo}/labels"):
if existing["name"] == name:
return int(existing["id"])
made = gitea._request(
"POST", f"/repos/{repo}/labels", json={"name": name, "color": "#ee0701"}
)
return int(made["id"])
def test_a_bug_goes_ahead_after_its_grace_period(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.3: a `bug` issue is reproduced and restated, and work starts when
the grace period runs out, with no go from the operator."""
agent, transport, run = running
operator = Operator(agent, transport)
repo, base = agent.config.gitea_repo, agent.config.base_branch
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
try:
# A real bug on this run's base branch, for the agent to reproduce.
code = (
f"def last_word_{run}(text):\n"
' """The last word of text, or "" if it has none."""\n'
" return text.split()[-1]\n"
)
as_operator._request(
"POST",
f"/repos/{repo}/contents/sandbox/last_word_{run}.py",
json={
"branch": base,
"content": base64.b64encode(code.encode()).decode(),
"message": f"last_word_{run}, with a bug",
},
)
data = as_operator._request(
"POST",
f"/repos/{repo}/issues",
json={
"title": f"last_word_{run} crashes on empty text",
"body": f"`last_word_{run}('')` in sandbox/last_word_{run}.py "
'raises IndexError; it should return "" as its docstring says.',
"labels": [label(as_operator, repo, "bug")],
"assignees": [agent.account],
},
)
issues.append(int(data["number"]))
operator.tick()
assert loop_of(operator.item(issues[0])).labels == ("bug",)
item = operator.drive_until(MERGE, issues[0], go=False)
prs.append(loop_of(item).pr)
orchestrator = agent.worker.call(
lambda: agent.store.transcript(item.id, "orchestrator").read()
)
said = [getattr(e.event, "text", "") for e in orchestrator]
assert any(t.endswith("from the bug grace period") for t in said), said
finally:
operator.tidy(issues, prs)
def test_a_plain_yes_is_a_go_ahead(
running: tuple[RunningLoop, InMemoryTransport, str],
) -> None:
"""M2.3: "yep, that's right" to a restatement is read as a go-ahead,
and the agent says so before acting."""
agent, transport, run = running
operator = Operator(agent, transport)
repo = agent.config.gitea_repo
as_operator = operator_gitea(agent.config.gitea_url)
issues: list[int] = []
prs: list[int] = []
try:
issues.append(as_operator.create_issue(repo, f"vowels_{run}", task(run)))
as_operator._request(
"PATCH",
f"/repos/{repo}/issues/{issues[0]}",
json={"assignees": [agent.account]},
)
operator.tick()
for _ in range(MAX_ANSWERS):
operator.drive_until(GO_AHEAD, issues[0])
if "?" not in operator.last():
break
# A restatement with a question in it can't be agreed to as it
# stands: answer first (the reader rightly says no to a yes).
operator.say(
f"Not yet: a new function vowels_{run}, not reusing anything "
"that exists. Put up a new restatement with that.",
issues[0],
)
operator.say("Yep, that's right, sounds good.", issues[0])
said = [m.text for m in transport.outbound]
assert any(t.startswith("Taking that as a go-ahead on rev-") for t in said)
assert loop_of(operator.item(issues[0])).started
pr = loop_of(operator.item(issues[0])).pr
if pr:
prs.append(pr)
finally:
operator.tidy(issues, prs)