nisse/nisse-dev/tests/test_loop_properties.py

"""specs/loop.qnt's properties, checked against the real loop (M2.1, PR D).

The Quint model checks the design; this checks the code. A seeded random
walk interleaves what the operator and Gitea can do (new issues, some
depending on others, go-aheads, answers, "do #N next", comments, merges,
closes, abandon, a crash and restart) with an adaptive scripted model, and
checks every property after every step. A failure prints
the seed and the steps, so it replays exactly.
"""

import random
import subprocess
from pathlib import Path
from typing import Any

import pytest
from model_contract import assert_renderable
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.gitea import Comment, Issue
from nisse_dev.loop.agent import LoopAgent
from nisse_dev.loop.driver import Driver
from nisse_dev.loop.items import (
    ANSWER,
    BLOCKED,
    GO_AHEAD,
    MERGE,
    RUNNING,
    WAITING,
    loop_of,
)
from nisse_dev.oplog import OpLog
from nisse_dev.tools import Workspace
from nisse_dev.transport import InboundMessage, InMemoryTransport
from nisse_dev.worker import InlineWorker

SEEDS = range(32)
STEPS = 30


class PolicyModel:
    """Plays a plausible driver: restate first, then a milestone, fixes
    after comments, close after a merge; sometimes asks a question or
    replies in prose. Seeded, so a walk replays exactly."""

    def __init__(self, rng: random.Random) -> None:
        self.rng = rng
        self.n = 0

    def complete(self, events, tools, model, system=None) -> list[Any]:
        assert_renderable(events)  # as the real API would
        self.n += 1
        offered = {t["function"]["name"] for t in tools}
        facts = [getattr(e.event, "text", "") for e in events]
        roll = self.rng.random()
        if roll < 0.08:
            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 "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:
            return [call("wait_for_review", note=f"waiting {self.n}")]
        if self.rng.random() < 0.15:
            # Half done, then a question: work a switch must park.
            return [
                call("start_milestone", name=f"m{self.n}"),
                call("write_file", path=f"f{self.n}.py", content=f"x = {self.n}\n"),
                call("ask_human", question=f"halfway {self.n}?"),
            ]
        merged = any("merged" in t for t in facts[-6:])
        if merged and self.rng.random() < 0.8:
            return [call("close_item", summary=f"done {self.n}")]
        return [
            call("start_milestone", name=f"m{self.n}"),
            call("write_file", path=f"f{self.n}.py", content=f"x = {self.n}\n"),
            call("finish_milestone", summary=f"work {self.n}"),
        ]


class Walk:
    def __init__(self, world: dict[str, Any], seed: int) -> None:
        self.world = world
        self.rng = random.Random(seed)
        self.seed = seed
        self.steps: list[str] = []
        self.shown_at_go: list[str] = []  # item:revision on screen at each go
        self.next_issue = 15
        self.agent, self.transport = self._agent()

    def _agent(self) -> tuple[LoopAgent, InMemoryTransport]:
        transport = InMemoryTransport()
        cancel = Cancel()
        w = self.world
        agent = LoopAgent(
            w["store"],
            transport,
            w["forge"],
            repo="don/sandbox",
            project="sandbox",
            account="nisse-dev",
            operators={DON},
            gitea_operators={"don"},
            driver=Driver(PolicyModel(self.rng), "m", cancel=cancel),  # type: ignore[arg-type]
            workspace=Workspace(w["host"], w["repo"].root, timeout_s=30),
            worker=InlineWorker(),
            cancel=cancel,
            oplog=OpLog("nisse-dev", _Null()),
            milestones=w["milestones"],
        )
        transport.on_message(agent.on_message)
        return agent, transport

    def items(self) -> list[Any]:
        return self.world["store"].work_items()

    def step(self) -> None:
        forge: Forge = self.world["forge"]
        open_items = [i for i in self.items() if i.status == "open"]
        # Weighted towards what an item can actually do, so review paths
        # are walked often, not drowned out by intake and crashes.
        weights = {"intake": 2 if not open_items else 1, "crash": 1, "issue": 1}
        item = self.rng.choice(open_items) if open_items else None
        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})
            if loop.pr:
                weights.update(
                    {"comment": 4, "merge": 2, "close-pr": 1, "comment-merge": 2}
                )
        action = self.rng.choices(list(weights), list(weights.values()))[0]
        self.steps.append(f"{action}#{loop.issue}" if loop else action)
        if action == "intake":
            self.agent.intake()
        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", ()))
            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 == "first" and loop:
            self.agent.handle(InboundMessage(DON, f"do #{loop.issue} next"), None)
        elif action == "crash":
            # Volatile state is lost: a new agent over the same store.
            self.agent, self.transport = self._agent()
            self.agent.boot()
            self.agent._schedule()
        elif action in ("go", "answer", "abandon") and loop:
            text = {"go": "go", "answer": "an answer", "abandon": "abandon"}[action]
            if action == "go" and item.position == WAITING and loop.wait == GO_AHEAD:
                self.shown_at_go.append(f"{item.id}:{loop.rev}")
            self.transport.receive(DON, text, thread=loop.thread)
        elif action == "go-stale" and loop and loop.rev > 1:
            # A go sent while an older revision was on screen.
            self.agent.handle(
                InboundMessage(DON, "go", thread=loop.thread), about=loop.rev - 1
            )
        elif action == "comment" and loop:
            forge.comments.setdefault(loop.pr, []).append(
                Comment(len(self.steps) * 10, "don", f"comment {len(self.steps)}", "t")
            )
            self.agent.watch()
            self.agent._schedule()
        elif (
            action == "comment-merge" and loop and forge.prs[loop.pr]["state"] == "open"
        ):
            # The race: a comment is seen, then the PR merges before the
            # driver's fix lands.
            forge.comments.setdefault(loop.pr, []).append(
                Comment(len(self.steps) * 10, "don", f"comment {len(self.steps)}", "t")
            )
            self.agent.watch()
            forge.merge(loop.pr)
            self.agent._schedule()
        elif action == "merge" and loop and forge.prs[loop.pr]["state"] == "open":
            forge.merge(loop.pr)
            self.agent.watch()
            self.agent._schedule()
        elif action == "close-pr" and loop and forge.prs[loop.pr]["state"] == "open":
            forge.prs[loop.pr]["state"] = "closed"
            self.agent.watch()
            self.agent._schedule()

    # -- the properties (specs/loop.qnt) -------------------------------------

    def check(self) -> None:
        where = f"seed {self.seed}, steps {self.steps}"
        store, forge = self.world["store"], self.world["forge"]
        items = self.items()
        issues = [loop_of(i).issue for i in items if loop_of(i).issue]
        assert len(issues) == len(set(issues)), f"one_item_per_issue: {where}"
        for item in items:
            loop = loop_of(item)
            if item.status == "open" and item.position == WAITING:
                explained = (
                    (loop.wait == GO_AHEAD and loop.rev > 0)
                    or (loop.wait == ANSWER and bool(loop.note))
                    or (loop.wait == MERGE and loop.pr > 0)
                )
                assert explained, (
                    f"waits_are_explainable: {item.position} {loop}: {where}"
                )
            gos = [
                int(f.event.text.split("rev-")[1].split()[0])
                for f in store.transcript(item.id, "orchestrator").read()
                if isinstance(f.event, Fact)
                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}"
            )
            # Blocked comes from external truth (L1): an item whose issue
            # had an open dependency from the start never ran before it closed.
            if forge.open_dependencies("don/sandbox", loop.issue):
                assert item.position == BLOCKED or item.status == "closed", (
                    f"blocked_never_runs: #{loop.issue} {item.position}: {where}"
                )
                ran = [
                    e
                    for e in store.transcript(item.id, "driver").read()
                    if e.actor == "driver"
                ]
                assert not ran, f"blocked_never_runs (driver): #{loop.issue}: {where}"
            mine = [p for p in forge.prs.values() if p["head"].startswith(item.id)]
            assert sum(p["state"] == "open" for p in mine) <= 1, (
                f"milestone_after_merge: {where}"
            )
            if item.status == "closed":
                assert not any(p["state"] == "open" for p in mine), (
                    f"closed_items_leave_nothing (PR): {where}"
                )
                assert not [
                    i for i in store.pending_intents() if i.work_item == item.id
                ], f"closed_items_leave_nothing (intent): {where}"
        assert not [i for i in items if i.position == RUNNING], (
            f"one_active (nothing left running): {where}"
        )
        self._parked_work_safe(where)
        heads = [p["head"] for p in forge.prs.values()]
        assert len(heads) == len(set(heads)), f"one_pr_per_milestone: {where}"
        bare = self.world["bare"]
        for n, pr in forge.prs.items():
            if not pr["merged"]:
                continue
            head = _rev(bare, f"refs/heads/{pr['head']}")
            # A fix pushed after the merge must reach the operator as a PR
            # (loop.qnt's lostFix): in main, or in a later PR, whatever they
            # then decide about it.
            landed = _is_ancestor(bare, head, "refs/heads/main") or any(
                _is_ancestor(bare, head, f"refs/heads/{other['head']}")
                for m, other in forge.prs.items()
                if m > n
            )
            assert landed, f"review_fixes_land (PR {n}): {where}"

    def _parked_work_safe(self, where: str) -> None:
        """L1, D4: nothing of a parked item lives only on the build VM: what
        every branch of an open item that isn't on the workspace points at is
        in the forge."""
        root, bare = self.world["repo"].root, self.world["bare"]
        current = _git(root, "rev-parse", "--abbrev-ref", "HEAD")
        ids = [i.id for i in self.items() if i.status == "open"]
        for line in _git(
            root, "for-each-ref", "--format=%(refname:short)", "refs/heads"
        ).splitlines():
            if line == current or not any(line.startswith(f"{i}-") for i in ids):
                continue
            local = _git(root, "rev-parse", line)
            pushed = subprocess.run(
                [
                    "git",
                    "--git-dir",
                    str(bare),
                    "cat-file",
                    "-e",
                    f"{local}^{{commit}}",
                ],
                capture_output=True,
                check=False,
            )
            assert pushed.returncode == 0, (
                f"parked_work_safe: {line} isn't pushed: {where}"
            )
        # And no write is lost: every file an open item's driver wrote is on
        # the workspace while it's that item's, or in the forge.
        for item in self.items():
            if item.status != "open":
                continue
            written = {
                str(e.event.arguments.get("path"))
                for e in self.world["store"].transcript(item.id, "driver").read()
                if isinstance(e.event, ToolCall) and e.event.tool == "write_file"
            }
            here = current.startswith(f"{item.id}-")
            for path in written:
                if here and (root / path).exists():
                    continue
                history = _git(bare, "log", "--all", "--format=%H", "--", path)
                assert history, f"parked_work_safe: {path} of #{item.id} lost: {where}"


def _git(root: Path, *args: str) -> str:
    where = ["--git-dir", str(root)] if root.suffix == ".git" else ["-C", str(root)]
    return subprocess.run(
        ["git", *where, *args], capture_output=True, text=True, check=True
    ).stdout.strip()


class _Null:
    def write(self, _: str) -> None: ...

    def flush(self) -> None: ...


def _rev(bare: Path, ref: str) -> str:
    return subprocess.run(
        ["git", "--git-dir", str(bare), "rev-parse", ref],
        capture_output=True,
        text=True,
        check=True,
    ).stdout.strip()


def _is_ancestor(bare: Path, commit: str, ref: str) -> bool:
    return (
        subprocess.run(
            ["git", "--git-dir", str(bare), "merge-base", "--is-ancestor", commit, ref],
            capture_output=True,
            check=False,
        ).returncode
        == 0
    )


@pytest.mark.parametrize("seed", SEEDS)
def test_the_loop_keeps_its_properties(world: dict[str, Any], seed: int) -> None:
    walk = Walk(world, seed)
    for _ in range(STEPS):
        walk.step()
        walk.check()