The new loop's core: intake, restatement, go-ahead, threads (M2.1, PR B)

nisse_dev.loop runs alongside control.py, chosen with `loop: new`.
Issues assigned to the agent (by an operator, or filed by one) become work
items, each with a Matrix thread. A driver run is one model with tools
until it calls a waiting tool: restate puts rev-N up for a go-ahead, bound
to the revision on screen when the reply arrived; ask_human waits for an
answer; close_item closes the item and its issue. Before the go-ahead the
driver can look but not change or close anything. One item is active at a
time; replies in an item's thread go to it. On boot, an item cut off
mid-run waits on the operator, and waiting points are put up again.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

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

@@ -0,0 +1,5 @@
+"""The conversational loop (architecture L1, L2, L5; Development Plan M2.1).
+
+Built alongside control.py, which keeps running until the switch-over
+(M2.8); the `loop: new` setting chooses this one.
+"""

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

@@ -0,0 +1,409 @@
+"""The loop's scheduler (L1, L2, L5, D1): intake, messages and driver runs.
+
+Everything durable happens on one worker thread, in order. Inputs (an
+operator's message, an issue assigned to the agent) are handed to it as they
+arrive; a message is stamped with the revision on screen in its thread first,
+because by the time the worker reaches it a newer one may be up (L3).
+
+One work item at a time is active. An item runs until its driver calls a
+waiting tool; it then waits on the operator, the active pointer is cleared
+(a safe point, L1), and the next ready item runs. Conversation is not work:
+a message is recorded and changes its item's state, but never interrupts a
+run (L1).
+"""
+
+import threading
+import uuid
+from collections.abc import Callable, Iterable
+from dataclasses import replace
+
+from nisse_dev.cancel import Cancel
+from nisse_dev.canonical import Fact, PlanSubmitted, UserMessage
+from nisse_dev.gates import command_form
+from nisse_dev.gitea import GiteaClient, GiteaError, Issue
+from nisse_dev.loop.driver import Driver, RunOutcome
+from nisse_dev.loop.items import (
+    ANSWER,
+    CLOSED,
+    GO_AHEAD,
+    READY,
+    RUNNING,
+    WAITING,
+    Loop,
+    item_for_issue,
+    item_for_thread,
+    loop_of,
+    save,
+)
+from nisse_dev.loop.tools import ASK_HUMAN, CLOSE_ITEM, RESTATE
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore, WorkItem
+from nisse_dev.tools import EXECUTOR_TOOLS, Workspace
+from nisse_dev.transport import InboundMessage, Transport
+from nisse_dev.worker import Worker
+
+INTAKE_INTERVAL_S = 30.0
+_GO = frozenset({"go", "approve"})
+
+# 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.
+_LOOK = [t for t in EXECUTOR_TOOLS if t["function"]["name"] != "write_file"]
+
+SYSTEM = """You are nisse-dev, a software agent working with the operator on
+the repository {repo}. You are on one work item: issue #{issue}, "{title}".
+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.
+
+After the go-ahead, do the work. When it is done, call close_item with a
+summary of what was done, what was verified, and what is left open.
+
+{workspace}"""
+
+
+class LoopAgent:
+    def __init__(
+        self,
+        store: StateStore,
+        transport: Transport,
+        gitea: GiteaClient,
+        *,
+        repo: str,
+        project: str,
+        account: str,
+        operators: Iterable[str],
+        gitea_operators: Iterable[str],
+        driver: Driver,
+        workspace: Workspace | None,
+        worker: Worker,
+        cancel: Cancel,
+        oplog: OpLog,
+        roles: dict[str, str] | None = None,
+        intake_interval_s: float = INTAKE_INTERVAL_S,
+    ) -> None:
+        self._store = store
+        self._transport = transport
+        self._gitea = gitea
+        self._repo = repo
+        self._project = project
+        self._account = account
+        self._operators = frozenset(operators)
+        self._gitea_operators = frozenset(gitea_operators)
+        self._driver = driver
+        self._workspace = workspace
+        self._worker = worker
+        self._cancel = cancel
+        self._oplog = oplog
+        self._roles = roles or {}
+        self._interval = intake_interval_s
+        # thread root -> revision on screen there, read on the transport's
+        # thread to stamp replies on arrival (L3).
+        self._shown: dict[str, int] = {}
+        self._running = False
+        self._stop = threading.Event()
+        self._intake_thread: threading.Thread | None = None
+
+    # -- lifecycle ----------------------------------------------------------
+
+    def start(self) -> None:
+        self._worker.call(self.boot)
+        self._transport.on_message(self.on_message)
+        start = getattr(self._transport, "start", None)
+        if start is not None:
+            start()
+        self._intake_thread = threading.Thread(target=self._intake_loop, daemon=True)
+        self._intake_thread.start()
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._intake_thread is not None:
+            self._intake_thread.join(timeout=5)
+
+    def _intake_loop(self) -> None:
+        while not self._stop.is_set():
+            self._worker.submit(self._settled("intake", self.intake))
+            self._stop.wait(self._interval)
+
+    def _settled(self, what: str, job: Callable[[], None]) -> Callable[[], None]:
+        """A worker job that reports a failure instead of losing it."""
+
+        def run() -> None:
+            try:
+                job()
+            except Exception as exc:  # noqa: BLE001 -- reported, the loop goes on
+                self._oplog.emit("loop.job_failed", cause=what, error=repr(exc))
+                self._transport.post(f"something went wrong with {what}: {exc}")
+
+        return run
+
+    # -- boot (L8) --------------------------------------------------------------
+
+    def boot(self) -> None:
+        """Resume waiting points as they were, and never resume interrupted
+        work blindly: an item cut off mid-run waits on the operator."""
+        self._store.set_active(None, cause="boot")
+        for item in self._store.work_items("open"):
+            loop = loop_of(item)
+            if item.position == RUNNING:
+                note = "I restarted while working on this. Reply to continue."
+                self._wait(item, replace(loop, wait=ANSWER, note=note), note)
+            elif item.position == WAITING and loop.thread:
+                if loop.wait == GO_AHEAD:
+                    self._shown[loop.thread] = loop.rev
+                    self._transport.post(
+                        f"(after a restart) rev-{loop.rev} is still up; say go "
+                        "to start, or reply with changes.",
+                        thread=loop.thread,
+                    )
+                else:
+                    self._transport.post(
+                        f"(after a restart) still waiting: {loop.note}",
+                        thread=loop.thread,
+                    )
+
+    # -- intake (D1) ---------------------------------------------------------
+
+    def intake(self) -> None:
+        """Issues assigned to the agent become work items, once each."""
+        try:
+            issues = self._gitea.assigned_issues(self._repo, self._account)
+        except GiteaError as exc:
+            # L5: an outage delays intake, never fails anything.
+            self._oplog.emit("loop.intake_unavailable", cause=str(exc))
+            return
+        for issue in issues:
+            if item_for_issue(self._store, issue.number) is not None:
+                continue
+            if not self._from_an_operator(issue):
+                self._oplog.emit(
+                    "loop.intake_ignored",
+                    cause="not filed or assigned by an operator",
+                    issue=issue.number,
+                )
+                continue
+            self._open(issue)
+        self._schedule()
+
+    def _from_an_operator(self, issue: Issue) -> bool:
+        """L11: only an operator's issue, or one an operator assigned, is
+        an instruction."""
+        if issue.author in self._gitea_operators:
+            return True
+        try:
+            who = self._gitea.assigned_by(self._repo, issue.number, self._account)
+        except GiteaError:
+            return False
+        return who in self._gitea_operators
+
+    def _open(self, issue: Issue) -> None:
+        item_id = f"{self._project}-{issue.number}-{uuid.uuid4().hex[:6]}"
+        cause = f"assigned {self._repo}#{issue.number}"
+        root = self._transport.post(
+            f"Queued {self._repo}#{issue.number}: {issue.title}"
+        )
+        loop = Loop(thread=root, issue=issue.number)
+        self._store.open_work_item(
+            item_id,
+            title=issue.title,
+            scope=self._project,
+            position=READY,
+            data={"loop": loop.to_data(), "roles": self._roles},
+            cause=cause,
+        )
+        log = self._store.transcript(item_id, "orchestrator")
+        log.append(
+            Fact(
+                text=f"work item opened from {self._repo}#{issue.number} "
+                f"({issue.title}), filed by {issue.author}"
+                + (f", labels: {', '.join(issue.labels)}" if issue.labels else "")
+            )
+        )
+        text = f"{issue.title}\n\n{issue.body}".strip()
+        if issue.author in self._gitea_operators:
+            log.as_actor("operator").append(
+                UserMessage(text=text, source=f"gitea:issue-body:{issue.number}")
+            )
+        else:
+            # L11: someone else's text is data; the operator's assignment is
+            # the instruction.
+            log.append(Fact(text=f"the issue as written by {issue.author}:\n{text}"))
+
+    # -- messages (L5) -------------------------------------------------------
+
+    def on_message(self, msg: InboundMessage) -> None:
+        """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:
+            # 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
+        self._worker.submit(
+            self._settled("your message", lambda: self.handle(msg, about))
+        )
+
+    def handle(self, msg: InboundMessage, about: int | None) -> None:
+        item = item_for_thread(self._store, msg.thread) if msg.thread else None
+        text = (msg.original or msg.text).strip()
+        command = command_form(msg.text)
+        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":
+                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 "
+                    "talk about it.",
+                    thread=msg.thread,
+                )
+            return
+        self._store.transcript(item.id, "operator").append(
+            UserMessage(text=text, source=f"transport:{msg.sender}")
+        )
+        loop = loop_of(item)
+        if command == "abandon":
+            self._close(item, loop, "abandoned", f"abandoned by {msg.sender}")
+        elif item.position == WAITING and loop.wait == GO_AHEAD and command in _GO:
+            if about != loop.rev:
+                # L3: a go-ahead is about what was on screen when it arrived.
+                self._transport.post(
+                    f"that go was for rev-{about}, but rev-{loop.rev} is up now; "
+                    "say go again if it still stands.",
+                    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}",
+            )
+        elif item.position == WAITING:
+            # An answer, or changes to a restatement: the driver reads it.
+            save(
+                self._store,
+                item.id,
+                READY,
+                replace(loop, wait="", note=""),
+                cause="the operator answered",
+            )
+        self._schedule()
+
+    def status(self) -> str:
+        items = self._store.work_items("open")
+        if not items:
+            return f"idle: no open work items. I take issues assigned to me in {self._repo}."
+        lines = []
+        for item in items:
+            loop = loop_of(item)
+            what = item.position
+            if item.position == WAITING:
+                what = f"waiting for {loop.wait} ({loop.note or f'rev-{loop.rev}'})"
+            lines.append(f"#{loop.issue} {item.title}: {what}")
+        return "\n".join(lines)
+
+    # -- running (L1, L2) ------------------------------------------------------
+
+    def _schedule(self) -> None:
+        """Run ready items until none is left. Only on the worker."""
+        if self._running:
+            return
+        while True:
+            ready = [i for i in self._store.work_items("open") if i.position == READY]
+            if not ready:
+                self._store.set_active(None, cause="nothing ready")
+                return
+            self._run(ready[0])  # oldest first (L1)
+
+    def _run(self, item: WorkItem) -> None:
+        loop = loop_of(item)
+        self._store.set_active(item.id, cause="ready")
+        save(self._store, item.id, RUNNING, loop, cause="driver run")
+        tools = [
+            RESTATE,
+            ASK_HUMAN,
+            *(_LOOK if not loop.started else [CLOSE_ITEM, *EXECUTOR_TOOLS]),
+        ]
+        system = SYSTEM.format(
+            repo=self._repo,
+            issue=loop.issue,
+            title=item.title,
+            workspace=self._workspace.describe() if self._workspace else "",
+        )
+        self._running = True
+        try:
+            outcome = self._driver.run(
+                self._store.transcript(item.id, "driver"),
+                system=system,
+                tools=tools,
+                workspace=self._workspace,
+            )
+        finally:
+            self._running = False
+            self._cancel.clear()
+        self._settle(item.id, outcome)
+
+    def _settle(self, item_id: str, outcome: RunOutcome) -> None:
+        item = self._store.work_item(item_id)
+        assert item is not None
+        loop = loop_of(item)
+        args = outcome.arguments or {}
+        if outcome.kind == "wait" and outcome.tool == "restate":
+            text = str(args.get("text", "")).strip()
+            rev = loop.rev + 1
+            self._store.transcript(item.id, "driver").append(
+                PlanSubmitted(plan=text, model=self._roles.get("planner", ""))
+            )
+            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,
+            )
+            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._wait(item, replace(loop, wait=ANSWER, note=question), question)
+        elif outcome.kind == "wait" and outcome.tool == "close_item":
+            summary = str(args.get("summary", "")).strip()
+            self._transport.post(f"Done: {summary}", thread=loop.thread)
+            self._close(item, loop, "completed", summary)
+        elif outcome.kind == "stopped":
+            note = f"stopped ({outcome.note}). Why did you stop it? Reply to continue."
+            self._wait(item, replace(loop, wait=ANSWER, note=note), note)
+        else:
+            note = f"I stopped short: {outcome.note}. Reply to continue."
+            self._wait(item, replace(loop, wait=ANSWER, note=note), note)
+        self._store.set_active(None, cause="item waiting")
+
+    def _wait(self, item: WorkItem, loop: Loop, say: str | None) -> None:
+        if say:
+            self._transport.post(say, thread=loop.thread)
+        save(self._store, item.id, WAITING, loop, cause=f"waiting for {loop.wait}")
+
+    def _close(self, item: WorkItem, loop: Loop, outcome: str, note: str) -> None:
+        self._fact(item, f"work item {outcome}: {note}")
+        save(self._store, item.id, CLOSED, loop, cause=outcome)
+        self._store.close_work_item(item.id, outcome=outcome, cause=note)
+        if loop.issue:
+            try:
+                self._gitea.post_comment(self._repo, loop.issue, f"{outcome}: {note}")
+                self._gitea.close_issue(self._repo, loop.issue)
+            except GiteaError as exc:
+                self._transport.post(
+                    f"closed here, but could not close #{loop.issue}: {exc}",
+                    thread=loop.thread,
+                )
+        if outcome == "abandoned":
+            self._transport.post(f"abandoned: {note}", thread=loop.thread)
+
+    def _fact(self, item: WorkItem, text: str) -> None:
+        self._store.transcript(item.id, "orchestrator").append(Fact(text=text))

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

@@ -0,0 +1,121 @@
+"""Assembling the new loop (M2.1): the same services as the old agent, with
+LoopAgent in place of the Orchestrator. Chosen with `loop: new` in config."""
+
+import threading
+from dataclasses import dataclass
+
+from nisse_dev.agent import (
+    SHUTDOWN_WAIT_S,
+    build_host,
+    discover_models,
+    make_llm_clients,
+    prepare_workspace,
+    push_credential,
+    remote_url,
+)
+from nisse_dev.cancel import Cancel
+from nisse_dev.config import AgentConfig
+from nisse_dev.gitea import GiteaClient
+from nisse_dev.loop.agent import LoopAgent
+from nisse_dev.loop.driver import Driver
+from nisse_dev.matrix import MatrixTransport
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+from nisse_dev.tools import Workspace
+from nisse_dev.worker import SerialWorker
+
+
+@dataclass
+class RunningLoop:
+    config: AgentConfig
+    account: str
+    loop: LoopAgent
+    store: StateStore
+    transport: MatrixTransport
+    gitea: GiteaClient
+    worker: SerialWorker
+    _stopped: threading.Event
+
+    def start(self) -> None:
+        self.loop.start()
+        self.transport.post(
+            f"nisse-dev up as {self.account} (new loop) on {self.config.gitea_repo}; "
+            f"{self.loop_status()}"
+        )
+
+    def loop_status(self) -> str:
+        return self.worker.call(self.loop.status)
+
+    def stop(self) -> None:
+        self.loop.stop()
+        self.transport.close()
+        self.gitea.close()
+        try:
+            self.worker.call(self.store.close, timeout=SHUTDOWN_WAIT_S)
+        except TimeoutError:
+            pass  # still working; L8 covers exiting mid-work
+        self.worker.stop(timeout=SHUTDOWN_WAIT_S)
+        self._stopped.set()
+
+    def wait(self) -> None:
+        self._stopped.wait()
+
+
+def build_loop_agent(config: AgentConfig) -> RunningLoop:
+    oplog = OpLog("nisse-dev")
+    worker = SerialWorker(
+        on_error=lambda exc: oplog.emit(
+            "worker.job_failed", cause="unhandled exception", error=repr(exc)
+        )
+    )
+    worker.start()
+    store = worker.call(lambda: StateStore(config.state_db, oplog=oplog))
+    catalog = discover_models(config, oplog)
+    gitea = GiteaClient(config.gitea_url, config.gitea_token)
+    account = gitea.whoami()
+    host = build_host(config)
+    credential = push_credential(config)
+    remote = remote_url(gitea, config.gitea_repo, credential)
+    prepare_workspace(host, config, account, remote, credential)
+    workspace = Workspace(
+        host,
+        config.workspace,
+        timeout_s=config.tuning.tools.command_timeout_s,
+        max_output_chars=config.tuning.tools.max_output_chars,
+    )
+    normalized, _passthrough = make_llm_clients(config, catalog)
+    cancel = Cancel()
+    # The driver is the planner's model for now (M2.6 brings the driver role).
+    driver = Driver(
+        normalized,
+        config.roles.resolve("planner"),
+        max_turns=config.tuning.executor.max_turns,
+        cancel=cancel,
+    )
+    transport = MatrixTransport(config.matrix)
+    loop = LoopAgent(
+        store,
+        transport,
+        gitea,
+        repo=config.gitea_repo,
+        project=config.project,
+        account=account,
+        operators=config.operators,
+        gitea_operators=config.gitea_operators,
+        driver=driver,
+        workspace=workspace,
+        worker=worker,
+        cancel=cancel,
+        oplog=oplog,
+        roles=config.roles.pinned(),
+    )
+    return RunningLoop(
+        config=config,
+        account=account,
+        loop=loop,
+        store=store,
+        transport=transport,
+        gitea=gitea,
+        worker=worker,
+        _stopped=threading.Event(),
+    )

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

@@ -0,0 +1,110 @@
+"""A driver run (L2): one model with tools, looping until it calls a waiting
+tool, runs out of turns, or is stopped.
+
+Everything the run does is written to the item's transcript as the driver
+(L7), and every model call sees the driver's view of it (L6). Execution tools
+go to the build VM (L13); control tools are answered here, and a waiting one
+ends the run, so the scheduler can put it to the operator.
+"""
+
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from nisse_dev.cancel import Cancel, skip_calls
+from nisse_dev.canonical import ToolCall, ToolResult
+from nisse_dev.llm import LLMClient, LLMError
+from nisse_dev.loop.tools import WAITING, names
+from nisse_dev.state import Transcript
+from nisse_dev.tools import Workspace
+
+DEFAULT_MAX_TURNS = 30
+
+# A reply in prose breaks the tool contract; the model is told once.
+_PROSE_NUDGE = (
+    "Your last reply was prose. Every reply must be a tool call: use one of "
+    "your tools, or ask_human if you need the operator."
+)
+
+
+@dataclass(frozen=True)
+class RunOutcome:
+    kind: Literal["wait", "capped", "stopped", "failed"]
+    tool: str = ""  # the waiting tool called
+    arguments: dict[str, Any] | None = None
+    note: str = ""
+
+
+class Driver:
+    def __init__(
+        self,
+        client: LLMClient,
+        model: str,
+        *,
+        max_turns: int = DEFAULT_MAX_TURNS,
+        cancel: Cancel | None = None,
+    ) -> None:
+        self._client = client
+        self._model = model
+        self._max_turns = max_turns
+        self._cancel = cancel or Cancel()
+
+    def run(
+        self,
+        transcript: Transcript,
+        *,
+        system: str,
+        tools: list[dict[str, Any]],
+        workspace: Workspace | None,
+    ) -> RunOutcome:
+        offered = names(tools)
+        nudged = False
+        for turn in range(self._max_turns):
+            # L2: stop takes effect at the next tool boundary.
+            if self._cancel.requested:
+                return RunOutcome("stopped", note=f"before turn {turn + 1}")
+            try:
+                produced = self._client.complete(
+                    transcript.view(), tools=tools, model=self._model, system=system
+                )
+            except LLMError as exc:
+                return RunOutcome("failed", note=f"the driver's call failed: {exc}")
+            calls: list[ToolCall] = []
+            for event in produced:
+                transcript.append(event)
+                if isinstance(event, ToolCall):
+                    calls.append(event)
+            if not calls:
+                if nudged:
+                    return RunOutcome("failed", note="kept replying in prose")
+                nudged = True
+                system = f"{system}\n\n{_PROSE_NUDGE}"
+                continue
+            for index, call in enumerate(calls):
+                if self._cancel.requested:
+                    skip_calls(transcript, calls[index:])
+                    return RunOutcome("stopped", note=f"during turn {turn + 1}")
+                if call.tool in WAITING and call.tool in offered:
+                    transcript.append(
+                        ToolResult(
+                            call_id=call.call_id,
+                            result="posted; your turn has ended",
+                            success=True,
+                        )
+                    )
+                    skip_calls(
+                        transcript, calls[index + 1 :], "not run: the turn had ended"
+                    )
+                    return RunOutcome("wait", call.tool, dict(call.arguments))
+                if call.tool not in offered or workspace is None:
+                    result, ok = (
+                        f"no such tool: {call.tool}; the tools are "
+                        + ", ".join(sorted(offered)),
+                        False,
+                    )
+                else:
+                    outcome = workspace.dispatch(call.tool, call.arguments)
+                    result, ok = outcome.result, outcome.success
+                transcript.append(
+                    ToolResult(call_id=call.call_id, result=result, success=ok)
+                )
+        return RunOutcome("capped", note=f"hit the {self._max_turns}-turn cap")

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

@@ -0,0 +1,66 @@
+"""A work item's place in the loop (L1), kept in its row in the state store.
+
+The row's ``position`` is the item's state; what the loop needs besides
+(the thread its conversation lives in, the revision of what is put up for
+agreement, what it is waiting for) is in ``data["loop"]``.
+"""
+
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from nisse_dev.state import StateStore, WorkItem
+
+READY = "ready"
+RUNNING = "running"  # a driver run is under way; on boot it means interrupted
+WAITING = "waiting"
+CLOSED = "closed"
+
+# What a waiting item waits for.
+GO_AHEAD = "go-ahead"  # a go on the restatement up (L3, D1)
+ANSWER = "answer"  # an answer to a question
+
+
+@dataclass(frozen=True)
+class Loop:
+    thread: str = ""  # the root of the item's Matrix thread (L5)
+    issue: int = 0  # its record issue (D1)
+    wait: str = ""  # GO_AHEAD / ANSWER, when waiting
+    rev: int = 0  # the latest proposal put up (L3)
+    started: bool = False  # scope agreed: a go-ahead was given (D1)
+    note: str = ""  # what a waiting item said it is waiting on
+
+    def to_data(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def loop_of(item: WorkItem) -> Loop:
+    return Loop(**item.data.get("loop", {}))
+
+
+def save(
+    store: StateStore, item_id: str, state: str, loop: Loop, *, cause: str
+) -> None:
+    item = store.work_item(item_id)
+    assert item is not None
+    store.update_work_item(
+        item_id,
+        position=state,
+        data={**item.data, "loop": loop.to_data()},
+        cause=cause,
+    )
+
+
+def item_for_thread(store: StateStore, thread: str) -> WorkItem | None:
+    for item in store.work_items("open"):
+        if loop_of(item).thread == thread:
+            return item
+    return None
+
+
+def item_for_issue(store: StateStore, issue: int) -> WorkItem | None:
+    """Any item, open or closed, recorded by this issue: intake never takes
+    an issue twice (D1)."""
+    for item in store.work_items():
+        if loop_of(item).issue == issue:
+            return item
+    return None

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

@@ -0,0 +1,49 @@
+"""Control tools (L13): handled in the harness, acting only on the agent's
+own state, conversation and records. A *waiting* tool ends the driver's run:
+the item then waits on the operator."""
+
+from typing import Any
+
+
+def _tool(name: str, description: str, **params: str) -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": name,
+            "description": description,
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    p: {"type": "string", "description": d} for p, d in params.items()
+                },
+                "required": list(params),
+            },
+        },
+    }
+
+
+RESTATE = _tool(
+    "restate",
+    "Put your understanding of the request to the operator, in their item's "
+    "thread: what you take it to mean, how it fits the project, and your "
+    "first step. Ends your turn; work starts when they agree.",
+    text="Your restatement, in plain language.",
+)
+ASK_HUMAN = _tool(
+    "ask_human",
+    "Ask the operator a question in the item's thread. Ends your turn; you "
+    "continue when they answer.",
+    question="The question.",
+)
+CLOSE_ITEM = _tool(
+    "close_item",
+    "The work is done: close the work item with a summary of what was done, "
+    "what was verified, and what is left open.",
+    summary="What was done, verified, and left open.",
+)
+
+WAITING = frozenset({"restate", "ask_human", "close_item"})
+
+
+def names(tools: list[dict[str, Any]]) -> set[str]:
+    return {tool["function"]["name"] for tool in tools}