#!/usr/bin/env python3
"""Answer-free autopilot for `[Question]` intakes: triage, receipt, and one digest.

Why "answer-free". The mechanical half of this loop is already automatic — a maintainer answers, the
`answered` label fires `sync-question-answers.yml`, the answer lands in D1 and everything after that is
retrieval (`misakanet_search` serves it as `type=faq`, re-submission returns it). What does **not** scale
is the part in the middle: somebody has to read every question, work out whether the corpus already
covers it, notice that four of them are one question, and reply. With ~8 new questions a day and a
historical total of 4 answers, that middle step is the whole backlog.

This script automates everything around the answer and nothing of it:

  1. **coverage** — asks the *public* MCP endpoint whether the corpus answers the question. That is the
     production retrieval path (BM25 + the IDF-weighted relevance floor), which is the only one that can
     be trusted for a 伪盲区/真盲区 verdict: the local Python engine's `_rank_docs` has no floor and
     scores "knitting pattern stitch count" 1.007 against "pip install timeout behind corporate proxy"
     0.905, so a verdict built on it is noise (measured 2026-09-25).
  2. **clustering** — groups open questions that share their distinctive vocabulary, so N questions
     become M clusters and the digest asks for M answers rather than N.
  3. **a receipt** — one idempotent comment per question, stating the measured state and the single next
     step that would make it answerable. The SOP's §3 receipt rule applies to *closing* an intake; this
     applies its spirit to receiving one, because an anonymous submitter who hears nothing has no way to
     tell "ignored" from "queued".
  4. **one digest** — a single issue listing the unresolved clusters, instead of a scrolling wall of
     identically-shaped open questions.

What it will not do, by construction:

  * **it never writes an answer.** Not because answers are hard to generate, but because a generated
    answer is an invented source unless it comes from someone who observed the failure.
    `docs/maintainer/provenance-gate-2026-09-16.md` exists because four lesson PRs once passed 24/24
    checks while citing a repository that returns 404.
  * **it never adds the `answered` label and never closes an issue.** `answered` is the switch that puts
    text into the FAQ, which is served to agents as a maintainer answer. That stays a sourced judgement.
  * **it never posts to a pull request or a lesson.** Questions only.

Usage
-----
    python3 scripts/question_autopilot.py                # dry run (default): print the plan
    python3 scripts/question_autopilot.py --json         # the plan as JSON
    python3 scripts/question_autopilot.py --post         # write the receipts and the digest

Environment:
    GITHUB_TOKEN / GH_TOKEN  (issues: read+write)
    MISAKANET_ENDPOINT       (default https://misakanet.org/mcp)

Stdlib only.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import pathlib
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

REPO_SLUG = os.environ.get("GITHUB_REPOSITORY", "Ikalus1988/MisakaNet")
API = "https://api.github.com"
ENDPOINT = os.environ.get("MISAKANET_ENDPOINT", "https://misakanet.org/mcp")

# Never mistaken for a maintainer answer: `sync_answered_questions.py` skips any comment carrying one of
# its AUTOMATED_MARKERS, and this marker is registered there for exactly that reason.
RECEIPT_MARKER = "<!-- misakanet-question-autopilot -->"
DIGEST_TITLE = "Question intake digest — unresolved clusters"
# A digest is a *report*: it has no acceptance criteria and must not be labelled `needs-ac`, which is
# the "phantom missing AC" churn `issue-quality-gate.yml` was already burned by (its own comment
# records five of six needs-ac issues once being intakes, "including a salvage digest the workflow
# itself opened"). That gate exempts a fixed label list, so the digest carries one of its own and
# the gate learns it — an agreement pinned by `tests/test_question_autopilot.py`.
DIGEST_LABEL = "question-digest"

# Pagination for every open-issue read. A single `per_page=100` request reads the newest 100 issues and
# stops — with 355 open issues that is 22% of the questions, so the other 78% could never get a receipt,
# a cluster or a bounty, and the digest reported "12 open question(s)" as if that were the backlog
# (measured 2026-10-30: 54 open questions, 12 visible). The cap is a runaway guard, not a limit: at
# 100/page it is 2,000 issues.
ISSUE_PAGE_SIZE = 100
ISSUE_PAGE_LIMIT = 20

# Coverage is measured against the live endpoint, which drops slow requests (#4072: the response is fully
# buffered, and a first byte later than ~3 s is cut by the link). One attempt turned 7 of 12 questions
# into "not measured yet" on 2026-10-30 — a read failure rendered as an absence, which is the one shape
# this script is written to avoid. Three attempts with a short backoff, and the count travels with the
# error so a receipt can say how hard it tried.
COVERAGE_ATTEMPTS = 3
COVERAGE_BACKOFF_SECONDS = (2, 5)

# Distinctive-token floor for calling two questions one cluster. Tuned low on purpose: an over-eager
# cluster costs a merge suggestion a human can ignore, an under-eager one costs the duplication this
# script exists to prevent.
# Swept against the 18 open questions on 2026-09-25 (complete linkage, measured token filtering):
#   2 -> 5 clusters, one of them a spurious 3-way macOS/PDF mix
#   3 -> 5 clusters: {2254,2257,2263} {2264,2266} {2255,2259} correct, plus two questionable pairs
#   4 -> 2 clusters, and it drops #2254 out of the secret-safe cluster it belongs to
#   5+ -> 1 or 0 clusters
# 3 is the only setting that recovers the three clusters a human reading the backlog also produced, so
# the two questionable pairs are accepted as *suggestions* — which is why the receipt says "may be the
# same question" and invites the reader to ignore the line.
# Re-swept after the Opire-banner fix (see `signature()`), same result: 3 is the only floor that
# recovers every cluster a human reading the backlog also produced. Known imprecision, stated
# rather than hidden: `{2171, 2260}` groups on `returns`/`source`/`text`, which are filler, so that
# pair is wrong. It is left in because the alternative — raising the floor to 4 — loses #2254 from
# the `secret-safe` cluster, and because both the receipt and the bounty task say explicitly that
# the grouping is a suggestion the reader may reject.
CLUSTER_MIN_SHARED = 3
# A token present in more than this share of the open questions distinguishes nothing, however
# topical it looks. Measured, not guessed — see `distinctive()`.
CLUSTER_MAX_DF_RATIO = 0.25
# Tokens that appear in most questions and therefore cluster nothing.
STOPWORDS = {
    "the", "and", "for", "with", "that", "this", "from", "not", "are", "was", "were", "has", "have",
    "how", "what", "which", "when", "where", "does", "should", "can", "may", "must", "its", "their",
    "than", "then", "into", "over", "under", "after", "before", "while", "still", "only", "also",
    "question", "problem", "issue", "reports", "reported", "without", "about", "would", "been",
}
TOKEN_RE = re.compile(r"[a-z][a-z0-9_.-]{2,}")


def _token_request(url: str, payload: dict | None = None, method: str | None = None,
                   headers: dict | None = None):
    data = json.dumps(payload).encode() if payload is not None else None
    request = urllib.request.Request(url, data=data, method=method)
    request.add_header("User-Agent", "misakanet-question-autopilot")
    request.add_header("Accept", "application/vnd.github+json")
    for key, value in (headers or {}).items():
        request.add_header(key, value)
    with urllib.request.urlopen(request, timeout=60) as response:
        return json.loads(response.read().decode() or "{}")


# ── GitHub ──────────────────────────────────────────────────────────────────

def token() -> str:
    """`$GITHUB_TOKEN` / `$GH_TOKEN`, else the `github.com` entry in `~/.git-credentials`.

    Same order as `scripts/gh_push_via_api.py`, and the same host-anchored parse as
    `scripts/push_preflight.py` (a substring test for `github.com` also accepts `notgithub.com` — CodeQL
    alert #285).
    """
    for key in ("GITHUB_TOKEN", "GH_TOKEN"):
        if os.environ.get(key):
            return os.environ[key]
    credentials = pathlib.Path.home() / ".git-credentials"
    if credentials.is_file():
        for line in credentials.read_text(encoding="utf-8", errors="replace").splitlines():
            try:
                parsed = urllib.parse.urlsplit(line.strip())
            except ValueError:
                continue
            if parsed.hostname == "github.com" and parsed.password:
                return urllib.parse.unquote(parsed.password)
    raise SystemExit("no GitHub token: set $GITHUB_TOKEN / $GH_TOKEN, or add a github.com entry "
                     "to ~/.git-credentials")


def gh(path: str, payload: dict | None = None, method: str | None = None):
    return _token_request(f"{API}/repos/{REPO_SLUG}{path}", payload, method,
                          {"Authorization": f"Bearer {token()}"})


def open_issues() -> list[dict]:
    """Every open issue, paginated, pull requests excluded.

    See `ISSUE_PAGE_SIZE`: this used to be one request, and every reader of "the open backlog" below
    inherited its blind spot — the question list, the digest's own de-duplication, and (before this
    was fixed here) the salvage digest's view of what still needed triage.
    """
    out: list[dict] = []
    for page in range(1, ISSUE_PAGE_LIMIT + 1):
        batch = gh(f"/issues?state=open&per_page={ISSUE_PAGE_SIZE}&page={page}")
        if not isinstance(batch, list):
            break
        out.extend(i for i in batch if "pull_request" not in i)
        if len(batch) < ISSUE_PAGE_SIZE:
            break
    return out


def open_questions() -> list[dict]:
    out = []
    for issue in open_issues():
        if "[Question]" not in issue["title"]:
            continue
        labels = {l["name"] for l in issue["labels"]}
        if "auto-rejected" in labels:
            continue
        out.append(issue)
    return out


def digest_issues() -> list[dict]:
    """Every open digest this script owns, oldest first (the number order is creation order)."""
    return sorted(
        (i for i in open_issues() if str(i.get("title", "")).startswith(DIGEST_TITLE)),
        key=lambda i: i["number"],
    )


# ── the technical signature (same shape the triage report used) ─────────────

def _section(body: str, name: str) -> str:
    m = re.search(rf"(?ims)^#+\s*{name}\s*$(.*?)(?=^#+\s|\Z)", body or "")
    return (m.group(1).strip() if m else "")


def signature(issue: dict) -> str:
    """Title + `## Problem` + `## Error` — the error signature, not the solution narrative.

    Deliberately not the whole body: `## Fix` and `## Verification` use the same vocabulary as the
    lessons that document the topic, so ranking on them makes every question look covered. The first
    version of the 2026-09-25 triage did that and put 54 of 91 issues in "corpus already covers this".
    """
    body = re.sub(r"<!--.*?-->", " ", issue.get("body") or "", flags=re.S)
    # Every issue in this repository carries an Opire banner inside `<details>` ("Everyone can add
    # rewards … `/reward 100` … `/try` … `/claim #N`"). Left in, it contributes `everyone`, `add`,
    # `reward`, `claim`, `amount` to *every* signature — tokens shared by all questions, which is the
    # definition of noise for a clustering step. Found by reading the generated bounty body: the
    # "shared vocabulary" it printed was `add`, `everyone`, `fields`.
    body = re.sub(r"(?is)<details.*?</details>", " ", body)
    body = re.sub(r"(?is)<[^>]{1,80}>", " ", body)
    # `re.M` is load-bearing: without it `$` only matches at the very end of the string, so the
    # submitter footer survived even though the pattern looked right (caught by its own test).
    body = re.sub(r"(?im)^_?submitted via remote mcp.*$", " ", body)
    body = re.sub(r"(?im)^\s*\*{0,2}(source|kind|dedup|submitted by|intake id)\*{0,2}\s*[:：].*$", " ", body)
    title = issue["title"].replace("[Question]", "").replace("Problem", "").strip()
    parts = [title, _section(body, "Problem")[:400], _section(body, "Error")[:200]]
    return " ".join(p for p in parts if p).strip()[:700]


def tokens(text: str) -> set[str]:
    return {t for t in TOKEN_RE.findall(text.lower()) if t not in STOPWORDS and len(t) > 2}


def distinctive(items: list[dict]) -> None:
    """Keep only the tokens that actually distinguish one question from another, in place.

    Measured over the open questions rather than guessed: a token that appears in more than
    `CLUSTER_MAX_DF_RATIO` of them separates nothing, however topical it looks ("macos", "check",
    "configuration"). The first version used a hand-written stopword list instead and produced one
    12-member cluster — every macOS question "shared" three generic words with every other.
    """
    df: dict[str, int] = {}
    for item in items:
        for token in item["tokens"]:
            df[token] = df.get(token, 0) + 1
    # Floor of 2, not 1: with only two questions in play, `int(2 * 0.25) == 0` would drop exactly
    # the tokens the two share — i.e. the filter would prevent the very clustering it feeds. A
    # token shared by two questions out of two is not evidence of anything until there are enough
    # questions to measure against. (Caught by a two-item test.)
    ceiling = max(2, int(len(items) * CLUSTER_MAX_DF_RATIO))
    for item in items:
        item["tokens"] = {t for t in item["tokens"] if df.get(t, 0) <= ceiling}
        item["df_ceiling"] = ceiling


def cluster(items: list[dict]) -> list[dict]:
    """Group questions that share enough **distinctive** tokens, with complete linkage.

    Complete linkage is the load-bearing part: single linkage ("shares 3 tokens with anyone already in the
    group") merged this backlog into one 12-member cluster on the first run, because the union of the
    shared vocabulary across 12 pairs clears any small floor. Requiring the floor against *every* member
    keeps a group meaning "all of these are the same question", which is what the digest asks a human to
    answer once.
    """
    groups: list[list[int]] = []
    for i, item in enumerate(items):
        for group in groups:
            if all(len(item["tokens"] & items[j]["tokens"]) >= CLUSTER_MIN_SHARED for j in group):
                group.append(i)
                break
        else:
            groups.append([i])
    out = []
    for group in groups:
        if len(group) < 2:
            continue
        shared = set.intersection(*[items[i]["tokens"] for i in group]) if group else set()
        out.append({"members": sorted(items[i]["number"] for i in group), "shared": sorted(shared)[:8]})
    return sorted(out, key=lambda g: -len(g["members"]))


# ── coverage, through the public endpoint ───────────────────────────────────

def corpus_answers(query: str) -> dict:
    """Ask production whether the corpus answers `query`, retrying a transport failure.

    Three attempts (see `COVERAGE_ATTEMPTS`). The retry is here rather than in the caller because every
    caller needs the same four keys back and none of them can tell a dropped connection from a real
    `no_match` — so the one place that *can* tell is the one place that must retry. A final failure still
    returns `ok: False` (never `no_match: True`), and now carries `attempts` so the receipt can say how
    many tries it took before it gave up.
    """
    last: dict = {}
    for attempt in range(COVERAGE_ATTEMPTS):
        last = _corpus_answers_once(query)
        if last.get("ok"):
            return last
        if attempt + 1 < COVERAGE_ATTEMPTS:
            time.sleep(COVERAGE_BACKOFF_SECONDS[min(attempt, len(COVERAGE_BACKOFF_SECONDS) - 1)])
    last["attempts"] = COVERAGE_ATTEMPTS
    return last


def _corpus_answers_once(query: str) -> dict:
    """Ask production whether the corpus answers `query`. Returns {no_match, lessons, faq, ok}."""
    payload = {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
               "params": {"name": "misakanet_search", "arguments": {"query": query, "top": 5}}}
    request = urllib.request.Request(ENDPOINT, data=json.dumps(payload).encode(), method="POST")
    request.add_header("Content-Type", "application/json")
    request.add_header("Accept", "application/json")
    request.add_header("MCP-Protocol-Version", "2025-06-18")
    request.add_header("Origin", "https://misakanet.org")
    request.add_header("User-Agent", "misakanet-question-autopilot")
    try:
        with urllib.request.urlopen(request, timeout=45) as response:
            body = json.loads(response.read().decode() or "{}")
    except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, ConnectionError, OSError) as exc:
        # An unreachable endpoint is `unknown`, never `uncovered`: a network failure must not be reported
        # as "the corpus has nothing on this", which is the one verdict a human would act on.
        return {"ok": False, "no_match": None, "lessons": [], "faq": [], "error": f"{type(exc).__name__}: {exc}"}
    text = (body.get("result", {}).get("content") or [{}])[0].get("text", "{}")
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        return {"ok": False, "no_match": None, "lessons": [], "faq": [], "error": "unparseable result"}
    results = parsed.get("results") or []
    return {
        "ok": True,
        "no_match": bool(parsed.get("no_match")),
        "lessons": [r for r in results if (r.get("type") or r.get("kind")) != "faq"],
        "faq": [r for r in results if (r.get("type") or r.get("kind")) == "faq"],
    }


# ── cluster ─────────────────────────────────────────────────────────────────

# ── bounty tasks: the answer step, delegated to the people who had the problem ──
#
# The one part of this loop that cannot be automated is writing the answer, because an answer generated
# by someone who never observed the failure is an invented source — the failure
# `docs/maintainer/provenance-gate-2026-09-16.md` exists to prevent. So it is *delegated* instead, to the
# repository's existing bounty machinery: `bounty` label, `/try` + `/claim` for claiming, merge credit and
# a leaderboard entry as the reward, and Opire's `/reward` for anyone who wants to fund one with money.
# See `JOIN.md` § "Bounties & payment" — `zero-bounty` is the design, not an omission.
#
# Why a **separate** issue rather than writing acceptance criteria into the question: the quality gate
# (`issue-quality-gate.yml`) deliberately skips anything labelled `intake`/`mcp-intake` — "intakes are
# *reports*, not specs" — and every question carries those labels. Writing AC into 22 intake issues was
# tried on 2026-09-12 and only produced `needs-ac` churn; the exemption is the fix for that. So the task is
# a new issue that *references* the question, which is also why the gate can label it `ready`: an
# AC section plus a checkbox list is exactly what `ready` requires.
#
# One bounty per **cluster**, not per question: the whole point of the clustering is that N questions are
# one answer, and a bounty per question would recreate the pile in a new place.
BOUNTY_MARKER = "<!-- misakanet-question-bounty:anchor-{} -->"
BOUNTY_LABEL = "bounty"


def bounty_body(anchor_number: int, members: list[dict], shared: list[str]) -> str:
    """The task the autopilot delegates: **an answer in the question thread**, not a lesson PR.

    The previous shape asked for a lesson. Measured on 2026-10-17, that produced exactly the failure the
    owner named: every submission arrived as a fork PR that could not land (seven of them, all blocked on
    DCO plus the derived artifacts). The answer is what the asker needs, the FAQ is built from answers
    rather than files, and an answer needs no branch, no signature and no generators — so the whole
    landing problem disappears for this class of contribution.

    A lesson remains *optional* (two lines at the end), because the corpus can keep it — but it is not
    the deliverable, and the task says so.
    """
    lines = [
        BOUNTY_MARKER.format(anchor_number),
        f"# [Bounty] Answer the {len(members)} linked question(s)",
        "",
        "## Context",
        "",
        "These intakes ask about something no lesson covers. Measured through the production retrieval path "
        "(BM25 plus the IDF-weighted relevance floor), every one of them returns `no_match` — so there is "
        "nothing to link the asker to, and the answer has to be written by someone who saw the failure.",
        "",
    ]
    for member in members:
        lines.append(f"- #{member['number']} — {member['title']}")
    if shared:
        lines += ["", f"_Shared vocabulary that grouped them: `{'`, `'.join(shared)}`._"]
    lines += [
        "",
        "## Deliverable",
        "",
        "**An answer, posted as a comment on each question issue above.** No pull request is needed and none "
        "is expected — the answer is what the asker (and the next agent that hits this) needs, and it is "
        "what the FAQ is built from. `sync-question-answers.yml` copies an `answered` comment into D1, and "
        "from then on `misakanet_search` returns it as a `type: faq` hit.",
        "",
        "The answer has to be *observed*, not generated. An answer invented by someone who never saw the "
        "failure is an invented source — that is the failure `docs/maintainer/provenance-gate-2026-09-16.md` "
        "exists to prevent. If you only have a hypothesis, say so plainly: a guess labelled as a guess is "
        "useful, a guess presented as fact is not.",
        "",
        "## How your answer reaches readers",
        "",
        "1. Post it on **each** linked question issue as a comment — not only here. The sync reads the "
        "question's own thread.",
        "2. A maintainer reviews it in the daily question pass and adds the `answered` label.",
        "3. `sync-question-answers.yml` copies the labelled comment into the FAQ (D1) within a day.",
        "",
        "## MANDATORY ACCEPTANCE CRITERIA (AC)",
        "",
        "**0. REPRODUCIBLE EVIDENCE — the condition most submissions fail**",
        "",
        "- [ ] Paste the command you ran and the output you got. Not a description of what you did — the "
        "command and its output. If you name an endpoint it must be `misakanet.org`.",
        "- [ ] Say what you observed yourself versus what you read somewhere.",
        "",
        "**1. The answer is complete**",
        "",
        "- [ ] It states the root cause, the fix (with the exact command or setting), and how to check it "
        "worked — the last one is a command plus the result that means success, not \"it should work\".",
        "- [ ] It names the version / environment it was observed on. \"Works on my machine\" without that "
        "line is unfalsifiable.",
        "",
        "**2. It is posted where the question lives**",
        "",
        "- [ ] A comment on **every** linked question issue above. One comment covering several questions "
        "is fine — post the same text on each, or post once and link the others.",
        "",
        "**3. Claiming**",
        "",
        "- [ ] Comment `/try` (or `/claim`) here so others know you are on it; the claim rules are in "
        "`JOIN.md` and `CONTRIBUTING.md`.",
        "",
        "## What the reward is, stated plainly",
        "",
        "**This is a `zero-bounty` task: $0.** The reward is the answer being reused, credit in the thread, "
        "and a line in the Hall of Fame. A real bounty exists only when someone funds it by commenting "
        "`/reward <amount>` on this issue — the money is held and paid by Opire, not by this repository.",
        "",
        "## Optional: also land it as a lesson",
        "",
        "Never required for this task. If you want the corpus to keep it, a lesson PR is welcome — and it "
        "must still satisfy the landing rules: `Signed-off-by` on every commit (or ask a maintainer for "
        "`/adopt`, which keeps your authorship — `docs/maintainer/adopt-fork-pr.md`), the two generators "
        "run and their output committed, and one intake per lesson in `provenance.issue` (the rest in "
        "`provenance.related`, per #3361). Say here first if you plan to open one.",
        "",
        "## Closes",
        "",
        "An answer closes this task once a maintainer labels it `answered`; the questions above close the "
        "same way. If a lesson lands instead, `Fixes #N` closes them through the receipt rule. If you "
        "cannot finish it, comment here and it goes back to the pool.",
        "",
        "<sub>Opened automatically by `scripts/question_autopilot.py` from measured coverage of the open "
        "questions. The grouping is a suggestion: if these are not really one question, say so and this "
        "task will be split.</sub>",
    ]
    return "\n".join(lines)


def planned_bounties(plan: dict) -> list[dict]:
    """One bounty per cluster. A cluster is the unit a single answer clears."""
    by_number = {i["number"]: i for i in plan["items"]}
    out = []
    for group in plan["groups"]:
        members = [by_number[n] for n in group["members"] if n in by_number]
        if len(members) < 2:
            continue
        out.append({"anchor": min(group["members"]), "members": members, "shared": group["shared"]})
    return out


def existing_bounty(anchor_number: int) -> dict | None:
    wanted = BOUNTY_MARKER.format(anchor_number)
    for issue in gh(f"/issues?state=all&labels={BOUNTY_LABEL}&per_page=100"):
        if "pull_request" in issue:
            continue
        if wanted in (issue.get("body") or ""):
            return issue
    return None


def open_bounties(items: list[dict], groups: list[dict]) -> list[dict]:
    """One entry per cluster; a question with no cluster is its own single-member task."""
    expected = []
    by_number = {i["number"]: i for i in items}
    if groups:
        expected = [{"anchor": min(g["members"]), "shared": g["shared"],
                     "members": [by_number[n] for n in g["members"] if n in by_number]} for g in groups]
    return [e for e in expected if len(e["members"]) >= 2 or True]


# ── receipt text ────────────────────────────────────────────────────────────

def receipt(item: dict) -> str:
    """One comment per question: the measured state, and the single next step."""
    cov = item["coverage"]
    if not cov["ok"]:
        tries = cov.get("attempts")
        state = (f"**State: not measured yet.** The retrieval endpoint did not answer "
                 f"{'after ' + str(tries) + ' attempts' if tries else 'this run'}, so nothing is claimed "
                 f"about coverage — a failed check is not a verdict.")
        nxt = "No action needed from you; this is re-checked daily."
    elif cov["lessons"]:
        tops = ", ".join(f'`{r.get("id") or r.get("title") or "?"}`' for r in cov["lessons"][:3])
        state = (f"**State: the corpus appears to answer this already** — the search that answers it "
                 f"returns {tops}. So this is a retrieval question, not a missing lesson.")
        nxt = ("Open the lesson(s) above first. If one of them does not actually cover your case, reply "
               "with what differs — that difference is the lesson worth writing, and writing a second "
               "lesson on the same ground is what the corpus is trying to avoid.")
    else:
        state = ("**State: no lesson in the corpus covers this.** Measured through the production search "
                 "path (relevance floor included), which returns `no_match` — so there is nothing to link "
                 "you to, and this is a genuine gap rather than a ranking miss.")
        nxt = ("Three things make it answerable, and you are the only one who has them: the concrete "
               "symptom you saw (`## Error`), what you already tried, and how you would check a fix "
               "worked (a command plus its expected output). With those this becomes either an answer "
               "or — better — a lesson, and the corpus keeps it for the next agent that hits it.")
    # No "related answered question" line, deliberately. The FAQ matcher is token overlap, and measured
    # on this backlog it attached two unrelated answered questions (#2099 site-offline, #1724 a bounty
    # status question) to a macOS PDF question. A receipt that points the asker at an unrelated answer is
    # worse than one that says nothing: it is the only claim in the comment, and it is wrong.
    faq = ""
    cluster_line = ""
    if item.get("cluster_with"):
        cluster_line = (
            f"\n\n**Possibly the same question as {', '.join('#' + str(n) for n in item['cluster_with'])}** "
            f"— grouped automatically by shared vocabulary, so treat it as a suggestion: if it is wrong, "
            f"ignore this line. When it is right, one answer covers all of them.")
    return (
        f"{RECEIPT_MARKER}\n"
        f"### Automated triage\n\n"
        f"{state}{cluster_line}{faq}\n\n"
        f"**Next step.** {nxt}\n\n"
        f"<sub>Machine-generated by `scripts/question_autopilot.py` on every open question; it triages and "
        f"reports, and never writes the answer itself (an answer must come from someone who observed the "
        f"failure — see `docs/maintainer/provenance-gate-2026-09-16.md`). Re-runs update this comment in "
        f"place; a maintainer can delete it or answer above it.</sub>"
    )


def digest_body(items: list[dict], groups: list[dict], bounties: list[dict] | None = None) -> str:
    lines = [
        "# Question intake digest — unresolved clusters",
        "",
        f"{len(items)} open question intake(s), of which **{sum(len(g['members']) for g in groups)} "
        f"form {len(groups)} cluster(s)**. One answer per cluster clears the cluster; answering per "
        "question is how the pile grows faster than it drains.",
        "",
        "## Clusters (answer once)",
        "",
        "| cluster | questions | shared vocabulary |",
        "|---|---|---|",
    ]
    if not groups:
        lines.append("| — | — | _none this run_ |")
    for g in groups:
        lines.append("| " + " · ".join("@" + "" for _ in [0]) + ", ".join(f"#{n}" for n in g["members"])
                     + " | " + str(len(g["members"])) + " | `" + "`, `".join(g["shared"]) + "` |")
    uncovered = [i for i in items if i["coverage"]["ok"] and not i["coverage"]["lessons"]]
    covered = [i for i in items if i["coverage"]["ok"] and i["coverage"]["lessons"]]
    unmeasured = [i for i in items if not i["coverage"]["ok"]]
    lines += [
        "",
        "## Claimable tasks",
        "",
        "Each genuine-gap cluster has a `[Bounty]` task linked to it. **$0 / `zero-bounty` by design**: the "
        "reward is merge credit, a leaderboard entry and a line in the Hall of Fame, and a real bounty "
        "exists only when someone funds it with `/reward <amount>` (`JOIN.md` § Bounties & payment).",
        "",
    ]
    if bounties:
        for entry in bounties:
            lines.append(f"- #{entry['anchor']} anchor · {len(entry['members'])} question(s) · "
                         f"`{', '.join('#' + str(m['number']) for m in entry['members'])}`")
    else:
        lines.append("- _none this run_")
    lines += [
        "",
        "## By measured coverage",
        "",
        f"- **genuine gaps** (no lesson reaches them): {len(uncovered)}"
        + (": " + ", ".join(f"#{i['number']}" for i in uncovered) if uncovered else ""),
        f"- **already covered** (a lesson is reached, so this is retrieval or phrasing): {len(covered)}"
        + (": " + ", ".join(f"#{i['number']}" for i in covered) if covered else ""),
        f"- **not measured** (endpoint unreachable — *not* a coverage claim): {len(unmeasured)}"
        + (": " + ", ".join(f"#{i['number']}" for i in unmeasured) if unmeasured else ""),
        "",
        "## What automation cannot do here",
        "",
        "Writing the answer. A generated answer to a question about a platform nobody here has run is an "
        "invented source, which is the one failure this repository has a gate for. The automation's job "
        "is to make the human step as small as possible: cluster it, measure it, ask the reporter for the "
        "missing pieces, and never leave them in silence.",
        "",
        "<sub>Regenerated by `scripts/question_autopilot.py` (`.github/workflows/question-autopilot.yml`).</sub>",
    ]
    return "\n".join(lines)


# ── plan / post ─────────────────────────────────────────────────────────────

def build_plan() -> dict:
    raw = open_questions()
    items = []
    for issue in raw:
        sig = signature(issue)
        items.append({
            "number": issue["number"],
            "title": issue["title"][:110],
            "signature": sig,
            "tokens": tokens(sig),
            "coverage": corpus_answers(sig),
            "comments": issue.get("comments", 0),
        })
    distinctive(items)
    groups = cluster(items)
    member_to_group = {n: [m for m in g["members"] if m != n] for g in groups for n in g["members"]}
    for item in items:
        item["cluster_with"] = member_to_group.get(item["number"], [])
    return {"items": items, "groups": groups}


def receipt_already_current(existing_comments: list[dict], body: str) -> bool:
    """True when this exact receipt is already the newest autopilot comment.

    Content hash, not presence: the coverage verdict changes as the corpus grows, and a receipt that
    keeps claiming "nothing covers this" after a lesson lands would be worse than no receipt.
    """
    digest = hashlib.sha256(body.encode("utf-8")).hexdigest()[:12]
    for comment in existing_comments:
        if RECEIPT_MARKER in (comment.get("body") or ""):
            return f"<!-- autopilot-digest:{digest} -->" in (comment.get("body") or "")
    return False


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--post", action="store_true", help="write the receipts and the digest")
    parser.add_argument("--json", action="store_true", help="print the plan as JSON and stop")
    parser.add_argument("--limit", type=int, default=0, help="only the first N questions (debugging)")
    parser.add_argument("--bounties", action="store_true",
                        help="also open/refresh one bounty task per cluster (requires --post to write; "
                             "dry run prints the bodies)")
    args = parser.parse_args(argv)

    plan = build_plan()
    if args.limit:
        plan["items"] = plan["items"][: args.limit]
    if args.json:
        # `tokens` is a set (the clustering needs set algebra); JSON does not take sets, and the first
        # version of this flag therefore crashed instead of printing a plan.
        print(json.dumps(plan, ensure_ascii=False, indent=1,
                         default=lambda o: sorted(o) if isinstance(o, (set, frozenset)) else str(o)))
        return 0

    print(f"{len(plan['items'])} open question(s); {len(plan['groups'])} cluster(s)")
    for item in plan["items"]:
        cov = item["coverage"]
        verdict = ("unmeasured" if not cov["ok"] else
                   f"covered by {cov['lessons'][0].get('id')}" if cov["lessons"] else "genuine gap")
        print(f"  #{item['number']:5} {verdict[:44]:44} cluster={item['cluster_with']}")

    if args.bounties:
        for entry in planned_bounties(plan):
            print(f"\n--- bounty for anchor #{entry['anchor']} "
                  f"({len(entry['members'])} question(s)) ---")
            print(bounty_body(entry["anchor"], entry["members"], entry["shared"]))

    if not args.post:
        print("\ndry run — nothing written. Pass --post to write the receipts and the digest.")
        return 0

    written = 0
    for item in plan["items"]:
        body = receipt(item)
        body += f"\n<!-- autopilot-digest:{hashlib.sha256(body.encode()).hexdigest()[:12]} -->"
        comments = gh(f"/issues/{item['number']}/comments?per_page=100")
        if receipt_already_current(comments, body):
            continue
        gh(f"/issues/{item['number']}/comments", {"body": body})
        written += 1
    print(f"receipts written: {written} (unchanged: {len(plan['items']) - written})")

    if args.bounties:
        created = updated = 0
        for entry in planned_bounties(plan):
            body = bounty_body(entry["anchor"], entry["members"], entry["shared"])
            found = existing_bounty(entry["anchor"])
            if found:
                gh(f"/issues/{found['number']}", {"body": body}, method="PATCH")
                updated += 1
            else:
                issue = gh("/issues", {"title": f"[Bounty] Answer the {len(entry['members'])} linked "
                                                  f"question(s)",
                                       "body": body, "labels": [BOUNTY_LABEL]})
                print(f"  bounty opened: #{issue['number']}")
                created += 1
        print(f"bounties: {created} created, {updated} refreshed")

    body = digest_body(plan["items"], plan["groups"], planned_bounties(plan))
    write_digest(body)
    return 0


DIGEST_SUPERSEDED = (
    "Superseded by #{keep} — this digest is one of several that were created because the "
    "de-duplication step could only see the newest 100 open issues (fixed 2026-10-30). "
    "#{keep} is now the living digest: it is updated on every run, and this one is frozen at the day "
    "it was written."
)


def write_digest(body: str) -> str:
    """Update the living digest, folding any older duplicates into it. Returns what happened.

    One digest, not five: the digest is the surface a human reads to decide what to answer, and five
    open copies with the same title and different numbers is worse than none — a reader cannot tell
    which is current. The de-duplication used to look for an existing digest with the same single-page
    fetch as `open_questions`, so once the first one fell off page 1 every later run created another.
    Now every copy is found (paginated), the **oldest** is kept (it already has the comments), and the
    rest are closed with a pointer to it.
    """
    existing = digest_issues()
    if not existing:
        created = gh("/issues", {"title": DIGEST_TITLE, "body": body, "labels": [DIGEST_LABEL]})
        print(f"digest created: #{created['number']}")
        return f"created #{created['number']}"
    keep, extras = existing[0], existing[1:]
    gh(f"/issues/{keep['number']}", {"body": body}, method="PATCH")
    print(f"digest updated: #{keep['number']}")
    for extra in extras:
        gh(f"/issues/{extra['number']}/comments", {"body": DIGEST_SUPERSEDED.format(keep=keep["number"])})
        gh(f"/issues/{extra['number']}", {"state": "closed", "state_reason": "not_planned"}, method="PATCH")
        print(f"digest folded into #{keep['number']} and closed: #{extra['number']}")
    return f"updated #{keep['number']}" + (f"; folded {len(extras)}" if extras else "")


if __name__ == "__main__":
    sys.exit(main())