"""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 re
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, make_world
from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture
from test_loop_projects import Router

from nisse_dev.cancel import Cancel
from nisse_dev.canonical import Fact, PlanSubmitted, ToolCall, UserMessage
from nisse_dev.execution import SSHHost
from nisse_dev.gitea import Comment, Issue
from nisse_dev.loop.agent import LoopAgent, Project
from nisse_dev.loop.driver import Driver
from nisse_dev.loop.items import (
    ANSWER,
    BLOCKED,
    CHAT,
    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)
TWO_PROJECT_SEEDS = range(100, 116)
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, projects: list[str]) -> None:
        self.rng = rng
        self.projects = projects
        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 "reply" in offered:  # a conversation: talk, or propose work
            if self.rng.random() < 0.4:
                return [
                    call(
                        "propose_work",
                        project=self.rng.choice(self.projects),
                        title=f"filed {self.n}",
                        summary=f"do thing {self.n}.",
                    )
                ]
            return [call("reply", text=f"an answer {self.n}")]
        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:
            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}",
                final=self.rng.choice(["yes", "no"]),
            ),
        ]


class Walk:
    """``worlds``: project name -> its world (forge, bare repo, clone,
    milestones), all over one store."""

    def __init__(self, worlds: dict[str, dict[str, Any]], seed: int) -> None:
        self.worlds = worlds
        self.world = next(iter(worlds.values()))  # the store's home
        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.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()
        router = Router({w["repo_name"]: w["forge"] for w in self.worlds.values()})
        for w in self.worlds.values():
            w["milestones"]._gitea = router
        agent = LoopAgent(
            self.world["store"],
            transport,
            router,  # type: ignore[arg-type]
            projects=[
                Project(
                    name=name,
                    repo=w["repo_name"],
                    workspace=Workspace(w["host"], w["repo"].root, timeout_s=30),
                    milestones=w["milestones"],
                )
                for name, w in self.worlds.items()
            ],
            account="nisse-dev",
            operators={DON},
            gitea_operators={"don"},
            driver=Driver(
                PolicyModel(self.rng, list(self.worlds)),  # type: ignore[arg-type]
                "m",
                cancel=cancel,
            ),
            worker=InlineWorker(),
            cancel=cancel,
            oplog=OpLog("nisse-dev", _Null()),
            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()

    def name(self, item: Any) -> str:
        """How the operator names an item: by project when there are two."""
        issue = loop_of(item).issue
        return f"#{issue}" if len(self.worlds) == 1 else f"{item.scope}#{issue}"

    def step(self) -> None:
        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,
            "chat": 1,
        }
        item = self.rng.choice(open_items) if open_items else None
        loop = loop_of(item) if item else None
        # The forge an item's project lives on; a new issue picks one.
        project = (
            item.scope
            if item and item.scope != "general"
            else self.rng.choice(list(self.worlds))
        )
        forge: Forge = self.worlds[project]["forge"]
        if item and loop:
            weights.update({"go": 3, "answer": 2, "abandon": 1, "go-stale": 1})
            weights.update({"plain-yes": 2, "wait": 2})
            if item.scope != "general":  # conversations aren't named by number
                weights.update({"first": 1, "named-go": 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}{self.name(item)}" if item else action)
        if action == "intake":
            self.agent.intake()
        elif action == "chat":
            # A top-level message: a conversation of its own (L1, L5).
            self.transport.receive(DON, f"thinking aloud {len(self.steps)}")
        elif action == "issue":
            # A new issue, sometimes depending on one that is still open.
            project = self.rng.choice(list(self.worlds))
            forge = self.worlds[project]["forge"]
            # Numbered by the forge, as filed records are: never reused.
            n = max([i.number for i in forge.issues] + [14]) + 1
            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 {self.name(item)}")
        elif action == "wait":
            # Time passes: bug countdowns may run out, conversations idle.
            self.now += self.rng.choice([60, 700, 90000])
            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 {self.name(item)} 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 = self.world["store"]
        items = self.items()
        issues = [(i.scope, 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 == CHAT and item.scope == "general")
                    or (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: {item.id} {gos} not in {self.shown_at_go}: {where}"
            )
            if item.scope == "general":
                continue  # no forge: nothing below applies
            forge = self.worlds[item.scope]["forge"]
            repo = self.worlds[item.scope]["repo_name"]
            # 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(repo, 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}"
        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}"
        )
        for item in items:
            if item.scope == "general":
                loop = loop_of(item)
                # L1: a conversation has no workspace, milestone or PR.
                assert not (loop.prepared or loop.milestone or loop.pr), (
                    f"conversations_do_no_work: {item.id}: {where}"
                )
        for name, w in self.worlds.items():
            self._records_have_one_item(name, w, where)
            self._own_repository_only(name, w, where)
            self._parked_work_safe(w, name, where)
            self._milestones_start_fresh(w, where)
            self._review_fixes_land(w, where)

    def _records_have_one_item(self, name: str, world: Any, where: str) -> None:
        """D1: a record issue filed from a conversation is its item's, from
        the moment it exists, and intake never takes it for another."""
        store = self.world["store"]
        for issue in world["forge"].issues:
            found = re.search(r"<!-- nisse-item: ([\w-]+) -->", issue.body or "")
            if not found:
                continue
            item = store.work_item(found.group(1))
            assert item is not None and item.scope == name, (
                f"records_have_one_item: {name}#{issue.number} has no item: {where}"
            )
            assert loop_of(item).issue == issue.number, (
                f"records_have_one_item: {item.id} isn't #{issue.number}: {where}"
            )
            assert loop_of(item).started, f"filed work starts agreed: {where}"

    def _own_repository_only(self, name: str, world: Any, where: str) -> None:
        """D8: an item's branches and PRs reach only its own project's
        repository."""
        others = [i.id for i in self.items() if i.scope != name]
        heads = [p["head"] for p in world["forge"].prs.values()]
        branches = _git(
            world["bare"], "for-each-ref", "--format=%(refname:short)", "refs/heads"
        ).split()
        for ref in [*heads, *branches]:
            assert not any(ref.startswith(f"{i}-") for i in others), (
                f"own_repository_only: {ref} in {name}: {where}"
            )

    def _review_fixes_land(self, world: Any, where: str) -> None:
        forge: Forge = world["forge"]
        heads = [p["head"] for p in forge.prs.values()]
        assert len(heads) == len(set(heads)), f"one_pr_per_milestone: {where}"
        bare = 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 _milestones_start_fresh(self, world: Any, where: str) -> None:
        """D4: a milestone starts from the latest default branch, so one
        redone after a closed PR never carries the rejected work."""
        forge: Forge = world["forge"]
        bare = world["bare"]
        for item in self.items():
            mine = sorted(
                (n, p)
                for n, p in forge.prs.items()
                if p["head"].startswith(f"{item.id}-milestone-")
                and "-fix-" not in p["head"]
            )
            for i, (n, earlier) in enumerate(mine):
                if earlier["merged"]:
                    continue
                rejected = _rev(bare, f"refs/heads/{earlier['head']}")
                if _is_ancestor(bare, rejected, "refs/heads/main"):
                    continue
                for m, later in mine[i + 1 :]:
                    assert not _is_ancestor(
                        bare, rejected, f"refs/heads/{later['head']}"
                    ), f"milestones_start_fresh: PR {m} carries PR {n}: {where}"

    def _parked_work_safe(self, world: Any, name: str, 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 its project's forge."""
        root, bare = world["repo"].root, world["bare"]
        current = _git(root, "rev-parse", "--abbrev-ref", "HEAD")
        ids = [i.id for i in self.items() if i.status == "open" and i.scope == name]
        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" or item.scope != name:
                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(
                    # Full history: a parked stash keeps new files in its
                    # third parent, which plain log simplifies away.
                    bare,
                    "log",
                    "--all",
                    "--full-history",
                    "--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({"sandbox": world}, seed)
    for _ in range(STEPS):
        walk.step()
        walk.check()


@pytest.mark.parametrize("seed", TWO_PROJECT_SEEDS)
def test_two_projects_keep_the_properties(
    ssh_host: SSHHost, tmp_path: Path, seed: int
) -> None:
    """D8: the same walk over two projects, whose issue numbers collide."""
    sandbox = make_world(ssh_host, tmp_path / "sandbox")
    other = make_world(
        ssh_host, tmp_path / "other", store=sandbox["store"], repo_name="don/other"
    )
    walk = Walk({"sandbox": sandbox, "other": other}, seed)
    for _ in range(STEPS):
        walk.step()
        walk.check()