#!/usr/bin/env python3
"""Issue operations dispatcher — create, get, list, transition, assign, update, and more."""

import argparse
import functools
import json
import re
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
from flydocs_api import (
    get_client,
    output_json,
    fail,
    resolve_text_input,
    description_hash,
    DEFAULT_LIST_LIMIT,
    RelayError,
)
from atomic_io import atomic_write_text
from status_vocab import (
    CANONICAL_STATUSES,
    CLOSED_STATUSES,
    DEFAULT_STATUS_MAPPING,
    PROVIDER_SUGGESTIONS,
    status_list,
    transition_hint,
)


def _is_clear_sentinel(value: str | None) -> bool:
    """FLY-804: detect 'clear this field' sentinels for nullable reference fields.

    Recognizes: 'clear', 'none', 'null', and empty string. Case-insensitive.
    Used by `update` to support unsetting --assignee, --milestone, --project,
    and --due-date in a single consistent way across Linear and Jira backends.
    """
    if value is None:
        return False
    return value.strip().lower() in ("clear", "none", "null", "")


def _resolve_assignee(assignee: str | None) -> str | None:
    """FLY-783: Resolve 'me' / 'self' to the local user's provider UUID.

    Reads identity from ~/.flydocs/me.json (with project-local fallback),
    matched against the active provider in .flydocs/config.json. Other
    assignee strings (UUIDs, emails) pass through unchanged.

    Fails with a clear error when the shortcut is used but no identity is
    available, so the caller doesn't hit a USER_NOT_FOUND from the relay.
    """
    if assignee not in ("me", "self"):
        return assignee
    client = get_client()
    provider_id, _ = client._resolve_local_identity()
    if not provider_id:
        fail(
            "Cannot resolve 'me' — no local identity found. "
            "Run: python3 .claude/skills/flydocs-workflow/scripts/workspace.py get-me"
        )
    return provider_id


WORKSPACE_FILENAME = ".flydocs-workspace.json"


def find_workspace_root(start: Path | None = None) -> Path | None:
    """Walk up for the workspace file (FLY-1084).

    Scripts run from inside a child repo, so a CWD-only check never finds the
    workspace file that sits one level up. Hooks, running at the root, did find
    it — and that disagreement is what let the writer and the reader of session
    state land in different directories.
    """
    base = (start or Path.cwd()).resolve()
    for candidate in (base, *base.parents):
        if (candidate / WORKSPACE_FILENAME).is_file():
            return candidate
    return None


def _resolve_session_dir() -> Path:
    """Resolve the session directory (FLY-674, FLY-1084).

    Scoped by workspaceId to keep workspaces isolated. In a multi-repo
    workspace the directory is anchored at the **workspace root**, because
    active-issue state is a workspace concern — the same issue is worked across
    repos in one session. Single-repo layouts are unchanged.
    """
    from flydocs_api import find_project_root
    root = find_project_root()
    config_file = root / ".flydocs" / "config.json"
    workspace_id: str | None = None
    if config_file.exists():
        try:
            config = json.loads(config_file.read_text())
            workspace_id = config.get("workspaceId")
        except (json.JSONDecodeError, OSError):
            pass
    scope = workspace_id if workspace_id else "default"
    workspace_root = find_workspace_root(root)
    base = workspace_root if workspace_root else root
    return base / ".flydocs" / "session" / scope


def synthesize_active_context(config: dict) -> dict | None:
    """Build an active context from the flat config keys (FLY-1097).

    `activeContexts` is written by `init` and `update`, but plenty of live v3
    configs carry only the flat `activeSprintId` / `activeProjectId` keys. Those
    hold everything needed to focus, so treating their absence as "no focus
    configured" silently disables the feature on workspaces that are in fact
    fully configured.

    Returns None only when there is genuinely nothing to focus on.
    """
    sprint_id = config.get("activeSprintId")
    project_id = config.get("activeProjectId") or next(
        iter(config.get("activeProjects") or []), None
    )
    if not sprint_id and not project_id:
        return None
    context: dict = {"type": "project", "synthesized": True}
    if project_id:
        context["id"] = project_id
    if sprint_id:
        context["sprintId"] = sprint_id
    return context


def _get_active_context() -> dict | None:
    """Read the primary active context from config (FLY-692).

    Prefers an explicit `activeContexts` entry; falls back to synthesizing one
    from the flat config keys (FLY-1097).
    """
    from flydocs_api import find_project_root
    root = find_project_root()
    config_file = root / ".flydocs" / "config.json"
    if config_file.exists():
        try:
            config = json.loads(config_file.read_text())
            # FLY-699: v3 has activeContexts at top level; v1 nests under workspace
            contexts = config.get("activeContexts") or config.get("workspace", {}).get("activeContexts", [])
            if contexts:
                return contexts[0]
            return synthesize_active_context(config)
        except (json.JSONDecodeError, OSError):
            pass
    return None


def verify_sprint_id(client, sprint_id: str) -> tuple[str | None, str | None]:
    """Check a cached sprint id against the provider's real sprints (FLY-1097).

    `activeSprintId` is a cached value that goes stale the moment a sprint
    closes, and the relay **silently ignores an unknown sprint filter** — it
    returns the unfiltered project list rather than an error. So a stale id does
    not fail loudly; it quietly turns focus off and hands back everything, which
    is the worst possible outcome for a narrowing feature.

    Observed on this workspace: config held `03a53d05-…`, which matched no
    sprint at all, while the genuinely active sprint was Cycle 33. `--focused`
    reported "Focused on active sprint" and returned 50 issues, 35 of them
    closed.

    Same defect class as FLY-1064 — trusting a local mirror over authoritative
    state. Returns `(sprint_id, note)`; the id is repaired to the real active
    sprint when the cached one is stale, or None when nothing is active.
    """
    try:
        sprints = client.list_sprints(active=True)
    except Exception:
        # Provider unreachable — trust the cached id rather than dropping focus.
        return sprint_id, None

    live = next(
        (
            s
            for s in sprints
            if s.get("id") and str(s.get("state", "")).lower() == "active"
        ),
        None,
    )
    if not live:
        return None, (
            "Note: No sprint is currently active — showing your open issues."
        )
    if str(live["id"]) == str(sprint_id):
        return sprint_id, None

    # Existence is not the test — a closed cycle is a perfectly valid id and
    # completely wrong for focus. Observed here: config pointed at a closed
    # cycle, so "focused" returned 50 issues, 35 of them Done.
    return str(live["id"]), (
        f"Note: Configured sprint is not the active one — using "
        f"{live.get('name') or live['id']}. Run `flydocs update` to refresh config."
    )


def resolve_focus(
    ctx: dict | None,
    sprint_filter: str | None = None,
    board_filter: str | None = None,
) -> tuple[str | None, str | None, bool, str | None]:
    """Resolve `--focused` to the narrowest available filter (FLY-1097).

    Returns `(sprint, board, narrow_fallback, note)`.

    Focus signals are layered and each is independently optional: a sprint is
    narrower than a board, a board narrower than a project. Resolution picks
    the narrowest signal actually present rather than branching on `boardType`.
    That matters because FLY-691 specifies Linear contexts carry `type:
    "project"` with **no** board metadata, so a `boardType` switch sends every
    correct Linear context to its unrecognized branch.

    A sprint id resolves the same way for both providers — a Linear cycle and a
    Jira sprint are the same concept.

    When nothing resolves, `narrow_fallback` is True: the caller must narrow to
    the developer's own open work rather than widening to the whole project.
    Returning *more* work is the one answer that cannot be right, since the
    caller explicitly asked to focus.
    """
    if sprint_filter or board_filter:
        return sprint_filter, board_filter, False, None

    if not ctx:
        return None, None, True, (
            "Note: No active sprint or project configured — showing your open "
            "issues. Run /start-session to set focus, or --all for everything."
        )

    sprint_id = ctx.get("sprintId")
    board_id = ctx.get("id") if ctx.get("type") == "board" else ctx.get("boardId")
    board_type = (ctx.get("boardType") or "").lower()
    label = ctx.get("sprintName") or ctx.get("name") or ""

    if sprint_id:
        note = f"Note: Focused on active sprint{f' ({label})' if label else ''}."
        return str(sprint_id), None, False, note

    if board_id:
        # A Scrum board between sprints still has a board, which is narrower
        # than the project — use it, but say why the sprint wasn't used.
        reason = " (no active sprint)" if board_type == "scrum" else ""
        note = (
            f"Note: Focused on board{f' ({label})' if label else ''}{reason}."
        )
        return None, str(board_id), False, note

    # A Scrum board between sprints has a board but no active sprint. Say so
    # plainly — the previous message claimed it was showing "all project
    # issues", which is both wider than intended and not what focus means.
    if board_type == "scrum":
        return None, None, True, (
            "Note: Scrum board has no active sprint — showing your open issues."
        )

    if ctx.get("id"):
        return None, None, True, (
            "Note: No sprint or board on the active context — showing your "
            "open issues in the active project."
        )

    return None, None, True, (
        "Note: Active context has no usable focus signal — showing your open issues."
    )


# ---------------------------------------------------------------------------
# Subcommand handlers
# ---------------------------------------------------------------------------

def cmd_create(args: argparse.Namespace) -> None:
    """Create a new issue.

    Enforces required fields to prevent incomplete issues:
    - Description must be non-empty unless --triage is set
    - Title is already required by argparse
    - Type is already required by argparse

    Auto-resolves from config when not explicitly passed:
    - Category labels from issueLabels.category
    - Repo labels from repoDefaults (falls back to deprecated issueLabels.repo)
    - Project from activeProjectId
    - Component from repoDefaults.component (Jira)
    """
    # FLY-1309: match cmd_estimate — any non-negative integer on the provider
    # scale is accepted; scale membership is the provider's call.
    if args.estimate is not None and args.estimate < 0:
        fail("Estimate points must be non-negative")

    # Description resolution: --description-file > stdin > --description > --template
    # FLY-699: Pass --description as text_arg so explicit flag is preferred
    # over stdin. Prevents hang when subprocess leaves stdin open (Claude Code
    # and similar harnesses) — resolve_text_input would otherwise block
    # indefinitely on sys.stdin.read().
    description = resolve_text_input(
        text_arg=args.description,
        file_arg=args.description_file,
    )
    if description is None:
        description = ""

    # Template fallback — read type template if no description provided
    if not description.strip() and args.template:
        template_path = Path(f".flydocs/templates/{args.type}.md")
        if template_path.exists():
            try:
                raw = template_path.read_text()
                # Strip agent instruction comments
                import re as _re
                description = _re.sub(r'<!--\s*AGENT:.*?-->\s*\n?', '', raw).strip()
            except OSError:
                pass

    # Enforce non-empty description — triage bypasses with warning
    if not description.strip():
        if args.triage:
            description = f"[Quick capture — needs refinement via /refine]\n\n{args.title}"
            import sys as _sys
            print(
                "Note: Triage issue created with minimal description. "
                "Run /refine to add full description and AC.",
                file=_sys.stderr,
            )
        else:
            template_path = f".flydocs/templates/{args.type}.md"
            fail(
                "Description is required when creating an issue. "
                "Use --description, --description-file, or pipe to stdin.\n"
                f"Read the template at {template_path} for the expected format, "
                "then populate all sections before creating the issue."
            )

    # FLY-783: Resolve 'me' / 'self' to local provider UUID
    assignee = _resolve_assignee(args.assignee)

    client = get_client()
    result = client.create_issue(
        title=args.title,
        issue_type=args.type,
        description=description,
        priority=args.priority,
        estimate=args.estimate,
        assignee=assignee,
        project=args.project,
        milestone=args.milestone,
        triage=args.triage,
    )
    output_json(result)


def cmd_get(args: argparse.Namespace) -> None:
    """Get a single issue by reference."""
    client = get_client()
    result = client.get_issue(args.ref, fields=args.fields)
    output_json(result)


# FLY-1115: DEFAULT_LIST_LIMIT now lives in flydocs_api.py — it was duplicated
# here, in the cloud client and in the local file store as three separate
# literals, which is how the local tier stayed at 50 after FLY-1105 raised the
# CLI to 250.


def cmd_list(args: argparse.Namespace) -> None:
    """List issues with optional filters."""
    sprint_filter = getattr(args, "sprint", None)
    board_filter = getattr(args, "board", None)
    focused = getattr(args, "focused", False)

    # FLY-692 + FLY-1097: --focused resolves to the narrowest available signal.
    narrow_fallback = False
    if focused:
        sprint_filter, board_filter, narrow_fallback, note = resolve_focus(
            _get_active_context(), sprint_filter, board_filter
        )
        if note:
            print(note, file=sys.stderr)

    # FLY-692: --sprint active resolves to active context's sprintId
    if sprint_filter == "active":
        ctx = _get_active_context()
        if ctx and ctx.get("sprintId"):
            sprint_filter = ctx["sprintId"]
        else:
            print(
                "Note: No active sprint configured — falling back to project-scoped list. "
                "Use --sprint ID for a specific sprint.",
                file=sys.stderr,
            )
            sprint_filter = None

    # FLY-692: --board active resolves to active context's board ID
    if board_filter == "active":
        ctx = _get_active_context()
        board_id = ctx.get("id") or ctx.get("boardId") if ctx else None
        if board_id:
            board_filter = board_id
        else:
            print(
                "Note: No active board configured — falling back to project-scoped list. "
                "Use --board ID for a specific board.",
                file=sys.stderr,
            )
            board_filter = None

    client = get_client()

    # FLY-1097: a cached sprint id that no longer exists is silently ignored by
    # the relay, so verify it before relying on it.
    if focused and sprint_filter:
        sprint_filter, repair_note = verify_sprint_id(client, sprint_filter)
        if repair_note:
            print(repair_note, file=sys.stderr)
        if not sprint_filter:
            narrow_fallback = True

    # FLY-1097: an unresolvable --focused narrows to the developer's own open
    # work. Widening to the entire project — including closed issues — is the
    # one outcome that cannot be correct when focus was explicitly requested.
    mine = args.mine
    active = args.active
    if narrow_fallback:
        mine = True
        active = True

    result = client.list_issues(
        status=args.status,
        active=active,
        project=args.project,
        assignee=args.assignee,
        milestone=args.milestone,
        mine=mine,
        show_all=args.show_all,
        limit=args.limit,
        sprint=sprint_filter,
        board=board_filter,
    )

    # FLY-1105 / FLY-1115: a truncated list reads as a complete one, so say so.
    #
    # FLY-1105 could only warn on page saturation, which cannot distinguish 6
    # remaining from 600. The relay now returns pagination headers, so when the
    # provider supplies an exact total the note states returned-vs-total.
    #
    # Linear exposes hasNextPage but no cheap exact count, so `total` is often
    # absent. The wording must not imply a number we do not have — the fallback
    # says "more exist" without quantifying, which is weaker but true.
    pagination = client.last_list_pagination()
    total = pagination.get("total")
    has_more = pagination.get("has_more")
    saturated = bool(args.limit) and len(result) >= args.limit

    if total is not None and len(result) < total:
        remaining = total - len(result)
        print(
            f"Note: returned {len(result)} of {total} issues "
            f"({remaining} not shown). Re-run with --limit to widen.",
            file=sys.stderr,
        )
    elif has_more or saturated:
        limit_txt = f"the {args.limit} limit was reached, so " if saturated else ""
        print(
            f"Note: returned {len(result)} issues — {limit_txt}more exist "
            f"(the provider did not report a total). "
            f"Re-run with --limit to widen.",
            file=sys.stderr,
        )

    # FLY-779: empty-result staleness check.
    # When list returns 0 with implicit active-project scoping, validate the
    # active project's status. If it's stale (closed/missing/etc.), surface
    # a stderr warning so the user knows results are silently filtered to a
    # dead project — and how to recover. We don't auto-retry: the user
    # should pick the right project explicitly.
    if (
        client.is_cloud
        and not result
        and not args.project
        and not args.show_all
        and not args.milestone
        and not args.assignee
    ):
        active_id = client.config.get("activeProjectId")
        if active_id:
            try:
                status_info = client.get_project_status(active_id)
                if status_info["status"] != "active":
                    print(
                        f"Note: Active project '{status_info['name']}' is "
                        f"{status_info['status']} (state: {status_info['raw_state']}). "
                        f"Results are scoped to this project and may be incomplete. "
                        f"Run: workspace.py validate-active-project to see active candidates.",
                        file=sys.stderr,
                    )
            except Exception:
                # Validation is advisory — never block the list response.
                pass

    output_json(result)


# ---------------------------------------------------------------------------
# Canonical status vocabulary (FLY-688, consolidated FLY-1272)
# ---------------------------------------------------------------------------
# Core only speaks canonical FlyDocs statuses. The relay translates to
# provider-native names (Linear, Jira, etc.) automatically.
#
# CANONICAL_STATUSES and PROVIDER_SUGGESTIONS now come from status_vocab.py —
# the one client-side source the hooks read too. They are imported at module
# scope above and re-exported here by name so existing callers and tests keep
# working.
#
# FLY-1265: the transition table itself is no longer imported here. This module
# asks `transition_hint()` whether an edge looks unusual and prints the answer;
# it never reads the table to decide anything, because it no longer decides —
# the relay does.

# FLY-1186: issue-ref shape, matching the server contract in
# usage-attribution.ts. focus.md's only job is to hold a ref this pattern
# accepts — it once held the literal string "--help" for a day and every
# attribution tuple that day recorded issue: null.
ISSUE_REF_RE = re.compile(r"[A-Z][A-Z0-9]{1,9}-[0-9]{1,6}")


@functools.lru_cache(maxsize=1)
def _status_mapping() -> dict[str, str]:
    """The workspace's canonical -> provider-native status mapping (FLY-1356).

    **A cache, not the authority.** The server's workspace mapping decides what
    a transition resolves to; this file is a copy of it, written only by
    `flydocs init` (`init.ts` ~664) and `flydocs update` (`runSync`, `sync.ts`
    ~339) from a server response, and optional in the schema (`types.ts` ~157).
    A team that changes Status Mapping in the dashboard leaves every un-synced
    checkout holding a stale copy, and a workspace that has never synced holds
    none at all — in which case this falls back to `DEFAULT_STATUS_MAPPING`,
    which is Linear's default template names and may match nothing the
    workspace uses. So this mapping may confirm the relay and may fail to; it
    must never be allowed to *contradict* it. See `reached_status` for where
    that line falls.

    `post-transition-check.py` has a similar `_load_status_mapping`, and the
    two are deliberately not the same function. That hook is advisory — it
    prints notes about an edge after the fact and writes nothing (FLY-1186) —
    so it is free to differ, and it does, on four axes: it finds the config
    through `repo_context.resolve_repo_dir` rather than `find_project_root`,
    it filters keys with `normalize()` (so `ALL_STATUSES`, TRIAGE included),
    it tries an exact canonical spelling before the mapping, and it resolves an
    ambiguous provider name to the first match. Only this writer's answer
    becomes a file other tools read as fact, which is why only this one refuses
    ambiguity. Changing the hook to match is out of scope for FLY-1356.

    Keys outside `CANONICAL_STATUSES` are dropped rather than trusted —
    deliberately the narrow set, not `ALL_STATUSES`: the source-only statuses
    are names an issue can be *in*, never names this writer may put in the
    mirror, and widening the filter is how TRIAGE would have become writable.

    Cached for the process. A single `issues.py transition` run resolves at
    most one reply, but `bridge.py` composes several dispatcher calls in one
    interpreter (`issue_activate` = assign + transition), and re-reading and
    re-parsing the same config file per lookup bought nothing. Config is read
    once per process, which is also how long the mapping is allowed to be
    stale for — call `_status_mapping.cache_clear()` if that ever changes.
    """
    try:
        from flydocs_api import find_project_root
        config = json.loads(
            (find_project_root() / ".flydocs" / "config.json").read_text()
        )
        mapping = config.get("statusMapping")
        if isinstance(mapping, dict):
            known = {
                str(canonical).strip().upper(): provider
                for canonical, provider in mapping.items()
                if str(canonical).strip().upper() in CANONICAL_STATUSES
                and isinstance(provider, str)
            }
            if known:
                return known
    except (OSError, ValueError, SystemExit):
        pass
    return DEFAULT_STATUS_MAPPING


def _canonical_name(name: object) -> str | None:
    """`name` as a canonical FlyDocs status, or None (FLY-1356).

    An exact spelling check, not a translation: local tier stores canonical
    names verbatim and a provider is free to call a column "Blocked", so
    recognising the exact name is not a guess. Restricted to
    `CANONICAL_STATUSES` — see `_status_mapping` for why the source-only
    statuses are excluded.
    """
    if not isinstance(name, str) or not name.strip():
        return None
    upper = name.strip().upper()
    return upper if upper in CANONICAL_STATUSES else None


def canonical_status(name: object) -> str | None:
    """Canonical FlyDocs status for a provider-native status name, or None.

    Precedence mirrors the relay's own `canonicalize.ts` (FLY-1253 §5): the
    workspace's explicit `statusMapping` first, then an exact canonical
    spelling. What differs is the answer to ambiguity. The server picks the
    first candidate in lifecycle order and reports the rest in
    `ambiguousWith`; this client has nowhere to report them, and its answer
    becomes a file that every hook then reads as fact. So a provider name that
    more than one canonical status maps to resolves to **None** here.

    Ambiguity is not hypothetical: the dashboard's auto-propose fills gaps from
    `STATUS_FALLBACK`, which points REVIEW *and* TESTING *and* COMPLETE at one
    provider column in a workflow that has only "Done" — and nothing validates
    uniqueness. Scanning for the first canonical key whose provider name
    matched therefore answered by dict order: on such a workspace a completed
    issue recorded REVIEW (and never got its session state cleared), or a
    TESTING transition recorded COMPLETE and deleted the whole mirror.

    Returns None — never a guess — for a name that resolves to neither, so the
    caller can stay silent instead of recording something it invented.
    """
    if not isinstance(name, str) or not name.strip():
        return None
    raw = name.strip()
    candidates = {
        canonical.upper()
        for canonical, provider in _status_mapping().items()
        if isinstance(provider, str) and provider.strip().lower() == raw.lower()
    }
    if len(candidates) == 1:
        return candidates.pop()
    if candidates:
        return None
    return _canonical_name(raw)


def reached_status(result: dict) -> str | None:
    """The canonical status the relay CONFIRMS the issue reached (FLY-1356).

    Two reply shapes, because the relay has two.

    **The adapter shape** carries `mappedFromFlydocsStatus`, and it is not what
    its name suggests: both adapters set it from the *request*
    (`status.toUpperCase()`) before resolving anything, and echo it verbatim on
    every path — same-state no-op, forced, and ordinary (`linear.ts` ~876/921/
    965-975, `jira.ts` ~1256/1293/1345-1348). It is the canonical target that
    was asked for. `newStatus` and `actualStatus` are provider-native names of
    the state the issue is in ("In Review", "Done"), never canonical ones.

    So the relay has already stated, in this reply, which canonical status it
    resolved and moved the issue to. **That statement is recorded.** The local
    `statusMapping` is a cache of the server's (see `_status_mapping`), so a
    provider name it does not recognise means only that this checkout is behind
    a dashboard change or has never synced — a reason to say nothing about the
    provider name, never a reason to overrule the relay. Vetoing on that
    mismatch wiped session state on every transition for any team that had
    re-mapped a column, which is the ordinary case, not the exotic one.

    The single exception is `forceUsed`. A force override is documented as
    bypassing the canonical mapping and resolving the provider state by name
    (`linear.ts` ~925/970, `jira.ts` ~1297/1351), so there — and only there —
    the relay itself says it did not honour the canonical target. If the
    provider name the reply carries is also not the one the mapping gives for
    `mappedFromFlydocsStatus`, the two accounts of where the issue landed
    disagree for a reason the reply states outright, and a writer that picks a
    side is guessing: None.

    **The reconciliation shape** (`service.ts` ~513-522) carries only
    `success`, `previousStatus` and `newStatus` — no canonical answer at all.
    There, and only there, the provider name is reverse-looked-up through
    `canonical_status`, which refuses ambiguity. The force exception cannot
    apply on this shape: `transitionOperationOutcome` builds a fresh body and
    drops `forceUsed` along with `mappedFromFlydocsStatus`, so a forced
    transition whose audit comment is still pending arrives looking like any
    other reconciliation reply and resolves through the provider name.

    Returns None when nothing in the reply confirms a canonical status; the
    caller clears the mirror rather than recording the request. `fallbackUsed`
    is not consulted: every shipped adapter path hard-codes it `false` (FLY-685
    removed runtime fallback), so branching on it was branching on a constant.
    """
    provider_name = result.get("actualStatus") or result.get("newStatus")
    if not isinstance(provider_name, str) or not provider_name.strip():
        return None

    requested = _canonical_name(result.get("mappedFromFlydocsStatus"))
    if requested:
        if not result.get("forceUsed"):
            return requested
        mapped_target = _status_mapping().get(requested)
        if (isinstance(mapped_target, str)
                and mapped_target.strip().lower() == provider_name.strip().lower()):
            # Forced at the column the mapping already named — the override
            # changed how the target was resolved, not what it was.
            return requested
        return None

    return canonical_status(provider_name)


# Session state holds two independent facts about two different issues, and
# each has its own record of whose it is (FLY-1407):
#
#   status + status-ref        — what status an issue is in. `status-ref` names
#                                the issue, and speaks for `status` ALONE.
#   focus.md + acceptance-...  — which issue is the attributed subject, and its
#                                criteria. `focus.md` names its own issue.
#
# They can name different issues, because they answer different questions and
# are written on different occasions: focus.md moves only on activation, the
# pair moves on every transition the session's own issue makes, and a close
# clears whichever of the two it speaks for. Clearing on close therefore has to
# ask each pair its own question. Keying the whole clear on `status-ref` — as
# the first cut of this fix did — deletes two files it has no authority over
# (completing FLY-200 takes FLY-100's focus.md and snapshot with it) and keeps
# two it should have cleared (closing the issue focus.md names while the pair
# tracks another leaves the attributed subject pointing at closed work).
#
# Until FLY-1471 the drift was manufactured here rather than merely tolerated:
# a transition to ANY open status claimed the pair without touching focus.md,
# so reviewing FLY-200 while FLY-100 was the focus left `status-ref` saying
# FLY-200 and focus.md saying FLY-100 — and that is the state the wipe needed.
# The open-status write now asks focus.md first (see `cmd_transition`), so
# ordinary work no longer produces it; legacy and half-cleared state still can,
# which is why both guards below stay.
#
# Both readers return None for absent, empty or unreadable, which every caller
# treats as "not about some other issue" rather than "about nobody" — there is
# nothing there to protect, so the unlink is a no-op either way. Both catch
# UnicodeDecodeError alongside OSError: these run AFTER the relay confirmed the
# transition, so an exception here would lose the mirror write and exit
# non-zero on a move that already landed.


def mirror_ref(session_dir: Path) -> str | None:
    """The issue ref `status-ref` describes, or None (FLY-1407).

    Authoritative for `status` and `status-ref`, and for nothing else. The
    transition-hint block above already refuses to *judge* a mirror describing
    another issue (FLY-1064); this is how the clear refuses to delete one.
    """
    try:
        tracked = (session_dir / "status-ref").read_text().strip().upper()
    except (OSError, UnicodeDecodeError):
        return None
    return tracked or None


def focus_ref(session_dir: Path) -> str | None:
    """The issue ref `focus.md` names, or None (FLY-1407).

    Authoritative for `focus.md` and the acceptance snapshot beside it. The
    file is prose — /activate writes a title, criteria and context around the
    ref — so the ref is searched for rather than parsed, matching how
    `prompt-submit.py` (~126) and `auto-approve.py` (~168) read the same file
    for attribution.

    Matching them means searching the RAW text. `ISSUE_REF_RE` is
    case-sensitive and both of those readers apply it as-is, so searching an
    uppercased copy here made this the only reader that could see a lowercase
    ref in prose — and the two then disagreed about whose file it was.
    "see fly-050 for context" above the real "FLY-100" subject handed the file
    to FLY-050, and closing FLY-050 deleted the focus.md and snapshot that
    attribution was charging to FLY-100 (FLY-1407). Nothing is lost by
    matching: every ref the system writes here is already uppercase — the
    write below gates on `fullmatch(ref_upper)`, and /activate writes the same
    canonical form.
    """
    try:
        text = (session_dir / "focus.md").read_text()
    except (OSError, UnicodeDecodeError):
        return None
    found = ISSUE_REF_RE.search(text)
    return found.group(0) if found else None


def gates_off_clause(session_dir: Path) -> str:
    """Say that the focused issue has no status left, or say nothing (FLY-1471).

    Read AFTER the writes and clears, because that is the state the caller has
    to report. Three branches of `cmd_transition` can end a call with a
    focused issue and no `status` describing it: the open-status refusal (the
    pair stays with, or is cleared from, another issue), the closed-status
    clear when `status-ref` named an issue other than the focus, and the
    unresolvable reply that deletes the pair as unknown (FLY-1356).

    In all three the local state is correct and the session is worse off: every
    consumer reads a missing status as "unknown" and stands down, so the Stop
    gate, the edit gate and the workflow directive are all off for the issue
    being worked. Silence there is the exact failure FLY-1471 is about, one
    door along — so each branch says it, and says how to end it.

    Returns "" when there is nothing to report: no focused issue (an idle or
    just-closed session has no subject to strand), or a pair that describes
    the focus, which means the gates are armed.
    """
    focused = focus_ref(session_dir)
    if not focused:
        return ""
    if mirror_ref(session_dir) == focused and (session_dir / "status").exists():
        return ""
    return (f"the session has no recorded status for {focused}, so the gates "
            f"are off until you transition it")


def cmd_transition(args: argparse.Namespace) -> None:
    """Transition an issue to a new status with a comment.

    Validates:
    - Comment is non-empty (not just whitespace) — hard failure, the workflow
      contract is that no status moves silently
    - Target is a canonical FlyDocs status — hard failure, a name the relay
      cannot resolve is a typo, not a policy question
    - The edge itself is only *hinted* (FLY-1265): an unlisted from->to prints
      a warning and proceeds. The relay's lifecycle service owns rejection.
    """
    # Enforce non-empty comment
    if not args.comment.strip():
        fail(
            "Transition comment cannot be empty. Every status transition "
            "requires a meaningful comment describing what changed."
        )

    target = args.status.upper()

    # FLY-688: Validate target is a canonical FlyDocs status before any API call.
    # Core only speaks canonical statuses — the relay handles provider translation.
    if target not in CANONICAL_STATUSES:
        suggestion = PROVIDER_SUGGESTIONS.get(args.status.lower())
        if suggestion:
            fail(
                f"'{args.status}' is a provider-specific status name. "
                f"Use the canonical FlyDocs status instead: {suggestion}\n"
                f"Valid statuses: {status_list()}"
            )
        else:
            fail(
                f"'{args.status}' is not a valid FlyDocs status.\n"
                f"Valid statuses: {status_list()}"
            )

    # FLY-1407: strip before upper. Without it a padded " fly-100 "
    # compared unequal to the mirror's FLY-100, so the ownership guards
    # below read the session's own issue as someone else's and skipped
    # the clear they exist to perform.
    ref_upper = args.ref.strip().upper()

    # FLY-674: Resolve workspace-scoped session directory so that multiple
    # workspaces sharing the same repo directory don't bleed session state.
    session_dir = _resolve_session_dir()

    # FLY-645: Only hint on transitions when the local session tracks THIS
    # issue. The status file holds the current status; status-ref holds the
    # issue ref it belongs to. If the refs don't match, the cached status is
    # for a different issue and says nothing about this one.
    #
    # FLY-1265 (spec §12.1): this used to `fail()`. It no longer does. The
    # source it judged against is a *cached local file*, and the relay reads the
    # live provider state — so on every disagreement the side that refused was
    # the side with the staler information, and a legal move (a revival out of
    # ARCHIVED, a resume the cache had not caught up with) died on the client
    # with no way past it. The relay's lifecycle service is authoritative: it
    # answers 409 TRANSITION_ILLEGAL with the real current state and the real
    # allowed targets. What survives here is the fast-feedback half — a typo
    # gets named in milliseconds instead of a round trip.
    status_file = session_dir / "status"
    status_ref_file = session_dir / "status-ref"
    if status_file.exists() and status_ref_file.exists():
        try:
            tracked_ref = status_ref_file.read_text().strip().upper()
            current = status_file.read_text().strip().upper()
            if tracked_ref == ref_upper:
                hint = transition_hint(current, target)
                if hint:
                    print(f"  Note: {hint}", file=sys.stderr)
        # FLY-1407: UnicodeDecodeError alongside OSError. A non-UTF-8 mirror
        # file is unreadable in exactly the sense this handler already forgives,
        # but it is not an OSError, so it escaped and killed the command before
        # the transition was ever attempted — over a cached hint the caller is
        # free to do without.
        #
        # This is a behaviour change, not only a crash fix, and the change is
        # the point: one corrupt byte in `status` used to brick EVERY
        # transition in the workspace, with no way past it short of deleting
        # the file by hand. Now the transition runs, and the write block below
        # replaces the corrupt file with a good one — so the state repairs
        # itself on the next move instead of blocking every move.
        except (OSError, UnicodeDecodeError):
            pass

    # FLY-689: Pass --force override if provided
    force = getattr(args, "force", None)

    client = get_client()
    result = client.transition(args.ref, target, args.comment, force=force)

    # FLY-1356: answering is not succeeding. An HTTP failure already exits
    # through the API layer, but a 200 carrying `success: false` — or carrying
    # no `success` at all — reached this block and wrote the mirror for a
    # transition that never happened. Absent counts as failed: the writer's
    # whole contract is to record what the relay confirmed, and an omitted
    # field confirms nothing.
    if result.get("success") is not True:
        detail = result.get("error") or result.get("message") or ""
        # The body goes to stdout before the exit. `post-transition-check.py`
        # reads the failure signal off this script's stdout
        # (`read_authoritative_statuses`, ~161-162) and defaults `succeeded` to
        # True when it finds nothing to read — so failing straight out of
        # `fail()` left an empty buffer and the hook audited the failed
        # transition as a successful one.
        output_json(result)
        fail(
            f"Transition of {ref_upper} to {target} was not confirmed by the "
            f"relay (success={result.get('success')!r})"
            + (f": {detail}" if detail else "")
            + ". Session state left unchanged."
        )

    # FLY-645 + FLY-653: After a successful transition, update both files so
    # subsequent validations see the fresh state. Write the ACTUAL status
    # from the response, not the requested target — when a fallback or provider
    # mapping produces a different end state, the local tracker must reflect it.
    try:
        status_file.parent.mkdir(parents=True, exist_ok=True)
        # FLY-1356: the reached status comes from the reply and only from the
        # reply, canonicalized first. This used to accept a *canonical*
        # `newStatus` and otherwise fall back to the requested target, so a
        # provider-native reply ("In Review" against a requested TESTING) made
        # the mirror record TESTING — a state the issue was never in.
        effective = reached_status(result)

        # FLY-1186: this script is the single writer of session state. The
        # post-transition hook used to duplicate these writes, but it only
        # fired when it could regex the transition out of the Bash command
        # and re-parse its stdout — piped or truncated output silently
        # defeated it, which is how focus.md kept a stale value for a day
        # while attribution recorded issue: null.
        if effective is None:
            # Nothing in the reply resolves to a canonical status this client
            # can stand behind. Recording the request here is what made the
            # mirror lie — but *leaving* the mirror is the same lie one move
            # older: `status` would still read IMPLEMENTING after a TESTING
            # transition that landed, and `bridge.py` reports that file back to
            # the MCP caller as `sessionState.status`. Every consumer treats a
            # missing status/status-ref pair as "unknown" and degrades
            # gracefully (stop-gate.py ~172-179, prompt-submit.py ~116-160,
            # auto-approve.py ~160-180, post-transition-check.py ~195-202), so
            # deleting the pair is the one available way to say so. focus.md
            # stays: it records which issue is being worked on, which this
            # reply does not put in doubt.
            #
            # Exit stays 0. The transition itself happened — the relay
            # confirmed `success` — and exiting non-zero would make agents
            # retry a move that already landed and would break
            # `issue_activate`'s assign-then-transition composition.
            # A closed target clears the whole mirror even unresolved. The
            # relay confirmed the move; the one thing this reply fails to say
            # is *which* closed state it landed in, and that distinction
            # changes nothing about session state — the issue is off the board
            # either way, so leaving focus.md and the acceptance snapshot
            # behind would keep a closed issue as the session's subject.
            # Both branches read the same `CLOSED_STATUSES` (FLY-1407): this
            # one used to face a resolved branch testing only ("COMPLETE",
            # "CANCELED"), so the same question got two answers and the
            # narrower one was wrong.
            #
            # ...except under `forceUsed`, where reading `target` as "the issue
            # is off the board" contradicts the reason we are in this branch at
            # all: `reached_status` returned None precisely because the relay
            # said it did NOT honour that target. Trusting the same target one
            # line later closed the session on a `--force COMPLETE` whose reply
            # put the issue in a review column.
            #
            # None of it applies to state describing a DIFFERENT issue
            # (FLY-1407). "Nothing here resolves to a canonical status" is a
            # statement about THIS ref; it puts nothing in doubt about an issue
            # the session is tracking elsewhere, and clearing that pair would
            # report an unrelated issue as unknown to every consumer above. So
            # each pair answers for itself, per the note on `mirror_ref` and
            # `focus_ref`: the status pair goes when `status-ref` is this ref
            # or absent, the focus pair only on a close and only when
            # `focus.md` names this ref or names nobody.
            tracked = mirror_ref(session_dir)
            focused = focus_ref(session_dir)
            closing = target in CLOSED_STATUSES and not result.get("forceUsed")
            cleared = []
            if tracked in (None, ref_upper):
                cleared += ["status", "status-ref"]
            if closing and focused in (None, ref_upper):
                cleared += ["focus.md", "acceptance-criteria.md"]
            for name in cleared:
                (session_dir / name).unlink(missing_ok=True)
            owners = ", ".join(sorted({r for r in (tracked, focused) if r}))
            outcome = (
                f"cleared {', '.join(cleared)}" if cleared
                else f"the session state describes {owners}, so it is "
                     f"left alone"
            )
            # FLY-1471: what it cleared is half the account. Deleting the pair
            # as unknown can leave a focused issue with no status — under
            # `forceUsed`, where `closing` is false and focus.md survives, or
            # whenever the pair described the focus itself.
            stranded = gates_off_clause(session_dir)
            if stranded:
                outcome += f"; {stranded}"
            # One physical line: bridge.py's `_warnings_from` (~340-346) turns
            # every stderr line into a separate MCP warning.
            print(
                "  Note: the relay's reply does not identify a canonical "
                f"status for this transition (newStatus="
                f"{result.get('newStatus')!r}, actualStatus="
                f"{result.get('actualStatus')!r}, mappedFromFlydocsStatus="
                f"{result.get('mappedFromFlydocsStatus')!r}) — the transition "
                f"landed, but the requested {target} is not recorded; "
                f"{outcome}.",
                file=sys.stderr,
            )
        elif effective in CLOSED_STATUSES:
            # A closed issue stops being the session's subject, whichever way
            # it closed — but only the session's OWN subject.
            #
            # FLY-1407, first half: this read ("COMPLETE", "CANCELED") while
            # the unresolved branch above read `CLOSED_STATUSES`, so a
            # transition the relay resolved cleanly to ARCHIVED or DUPLICATE
            # left the mirror pointing at work that is off the board.
            # `auto-approve.py` then read that stale `status` and, because
            # `EDIT_OK_STATUSES` covers IMPLEMENTING/REVIEW/TESTING/COMPLETE/
            # CANCELED and not those two, told the agent on every edit that
            # "Issue X is in DUPLICATE, not IMPLEMENTING. Transition before
            # making changes" — for an issue that was correctly closed.
            # Attribution went on charging turns to it too. (Not the Stop gate:
            # `GATED_STATUSES` excludes both, so it was silent.) The set is the
            # one definition of "off the board"; asking the question twice is
            # how the two halves drifted apart in the first place.
            #
            # FLY-1407, second half: the clear was also unconditional, so
            # closing FLY-200 deleted the state of whichever issue the session
            # was actually working — and widening it to ARCHIVED/DUPLICATE
            # moves that onto the bulk-triage path, where closing five
            # duplicates in a row is ordinary and none of the five is the
            # focused issue. State describing someone else is not this
            # transition's to delete, so each pair is asked about its own
            # owner: `status-ref` for the status pair (FLY-1064), `focus.md`
            # for the attributed subject and its snapshot. One guard over both
            # was the first cut of this fix and was wrong in both directions —
            # it deleted a focus.md `status-ref` does not speak for, and kept
            # one naming an issue that had just closed.
            #
            # Note the deliberate asymmetry with the write below: moving an
            # issue to IMPLEMENTING does claim the mirror, because that is
            # what starting to track an issue looks like (FLY-1471 narrowed
            # the claim to exactly that status). Closing an issue may not
            # un-track a different one, because that is not a claim about it.
            #
            # acceptance-criteria.md stays in the list to clear snapshots left
            # by earlier versions.
            #
            # The guards sit inside the branch, not in its `elif`: a closed
            # status that failed them must fall through to *nothing*, never to
            # the write below, which would replace the focused issue's mirror
            # with the closed one's — the same wipe wearing a different hat.
            #
            # Two guards, because the two pairs answer for different issues and
            # routinely name different ones (see the note on `mirror_ref` and
            # `focus_ref` above).
            if mirror_ref(session_dir) in (None, ref_upper):
                for name in ("status", "status-ref"):
                    (session_dir / name).unlink(missing_ok=True)
            if focus_ref(session_dir) in (None, ref_upper):
                for name in ("focus.md", "acceptance-criteria.md"):
                    (session_dir / name).unlink(missing_ok=True)
            # FLY-1471: this branch said nothing at all, and one of its
            # outcomes is a session left without enforcement — the pair
            # described the issue just closed, so clearing it (correctly)
            # leaves the DIFFERENT issue in focus.md with no status any
            # consumer will act on. Bulk-closing a review sweep is exactly
            # how a session used to arrive there.
            stranded = gates_off_clause(session_dir)
            if stranded:
                print(f"  Note: {ref_upper} is now {effective}; {stranded}.",
                      file=sys.stderr)
        else:
            # FLY-1471: an open-status move claims the status pair only when
            # it is about the session's own subject — or when it IS the
            # session changing subject.
            #
            # The claim used to be unconditional, and that is how the mirror
            # came to describe one issue while focus.md named another:
            # `transition FLY-200 REVIEW` during work on FLY-100 left
            # `status-ref` saying FLY-200 and focus.md saying FLY-100, with
            # nothing said about it. The mismatch is not cosmetic. Every
            # consumer of the pair fails safe on it and goes quiet —
            # stop-gate.py skips the gate entirely (~305-315), auto-approve.py
            # drops its edit nudge (~185-190), prompt-submit.py reports the
            # issue with no status (~155-162) — so a review sweep across
            # other issues silently disarms the workflow for the issue the
            # session is actually working. It is also the step that armed
            # FLY-1407's cross-issue wipe: the close guard asks `status-ref`
            # whose the pair is, and this write had already made it answer
            # FLY-200, so completing FLY-200 took FLY-100's focus.md and
            # acceptance snapshot with it.
            #
            # IMPLEMENTING is the exception, and the only one. It is
            # activation (reference/status-workflow.md): starting work on an
            # issue is exactly the moment the session's subject changes, and
            # the focus.md write below is the same statement. Every other open
            # status — READY, BLOCKED, REVIEW, TESTING, BACKLOG — is a move on
            # the board that says nothing about who the session works for.
            #
            # An absent or unnamed focus.md means no subject to protect, not
            # an unknown one: `focus_ref` returns None for missing, empty,
            # unreadable and garbage alike, and refusing the write there would
            # leave a transition-only workspace with no mirror at all.
            #
            # Whether it IS the activation is asked of `effective`, not of
            # the requested `target`: the relay decides where an issue lands,
            # and the mirror's whole job is to record that. A request for
            # IMPLEMENTING that the provider resolved into a review column is
            # not an activation and may not take the session with it; a
            # request the provider resolved INTO IMPLEMENTING is one, whatever
            # was asked for. Reading `target` here would also re-open FLY-1356
            # from the side: the status recorded and the status the claim was
            # judged on could disagree.
            #
            # The refusal is not the whole answer, because the pair can
            # already be describing the ref being transitioned (focus.md is
            # written by /activate over a pair another issue left; older
            # clients wrote it unguarded). Leaving that pair alone would leave
            # it WRONG rather than merely foreign — it would say FLY-200 is
            # IMPLEMENTING while FLY-200 sits in REVIEW — and two consumers
            # read the pair as the issue's own before-state and warn on legal
            # moves because of it: the FLY-645 hint block ~150 lines above,
            # and `post-transition-check.py`'s `read_cached_status` (~379-390,
            # "Unusual transition ... Verify this is intentional", injected as
            # additionalContext). So it is cleared, on exactly the authority
            # the close path uses: `status-ref` names this ref, so this
            # transition speaks for it. Unknown degrades gracefully in all
            # four consumers; wrong does not.
            #
            # A pair naming some THIRD issue is neither ours to take nor ours
            # to clear, and is left untouched.
            #
            # The issue itself moved either way — the relay confirmed it
            # before this block. What is refused here is only the local claim.
            focused = focus_ref(session_dir)
            if (effective != "IMPLEMENTING"
                    and focused is not None and focused != ref_upper):
                tracked = mirror_ref(session_dir)
                if tracked == ref_upper:
                    for name in ("status", "status-ref"):
                        (session_dir / name).unlink(missing_ok=True)
                    kept = (f"session focus stays on {focused}, and the stale "
                            f"mirror entry for {ref_upper} is cleared")
                elif tracked == focused:
                    kept = f"session mirror stays on {focused}"
                else:
                    # The pair is absent, or about a third issue. Either way
                    # the note may not claim it says anything about `focused`.
                    kept = f"session focus stays on {focused}"
                stranded = gates_off_clause(session_dir)
                if stranded:
                    kept += f"; {stranded}"
                # One physical line: bridge.py's `_warnings_from` (~340-346)
                # turns every stderr line into a separate MCP warning.
                print(
                    f"  Note: {ref_upper} -> {effective} recorded on the "
                    f"issue; {kept} — only a transition to IMPLEMENTING "
                    f"claims the mirror for another issue.",
                    file=sys.stderr,
                )
            else:
                # FLY-1293: write-temp-replace. A call killed between these two
                # writes used to be able to leave a truncated `status`, and the
                # bridge makes mid-call kills ordinary (client cancel, timeout).
                atomic_write_text(status_file, effective)
                atomic_write_text(status_ref_file, ref_upper)
                # Transitioning to IMPLEMENTING makes this issue the attributed
                # focus, so transition — not just /activate — writes focus.md.
                # Only a ref the pattern accepts is ever written, and a richer
                # /activate-written file already naming this issue is left alone.
                if effective == "IMPLEMENTING" and ISSUE_REF_RE.fullmatch(ref_upper):
                    focus_file = session_dir / "focus.md"
                    existing = focus_file.read_text() if focus_file.exists() else ""
                    if not re.search(rf"\b{re.escape(ref_upper)}\b", existing):
                        atomic_write_text(focus_file, f"# Active Issue\n\n{ref_upper}\n")
                        # FLY-1471: the acceptance snapshot sits beside
                        # focus.md and belongs to whoever focus.md names.
                        # Taking the focus and leaving the snapshot makes
                        # `session-start.py` (~82-92) count the PREVIOUS
                        # issue's boxes and print them under this issue's ref.
                        # Nothing regenerates it (FLY-1065 removed the write —
                        # criteria are read live), so the honest move is to
                        # drop it. Only when the subject actually changes: a
                        # BLOCKED -> IMPLEMENTING resume of the focused issue
                        # takes the branch above and keeps everything.
                        (session_dir / "acceptance-criteria.md").unlink(
                            missing_ok=True)
    except OSError:
        pass

    # FLY-1407: a "fallback used" note stood here and could not fire. Its two
    # arms were `fallbackUsed` — hard-coded `false` on every shipped adapter
    # path since FLY-685 removed runtime fallback — and
    # `actual != mapped_from and mapped_from != target`, where `mapped_from`
    # is `mappedFromFlydocsStatus`: the REQUEST echoed back verbatim
    # (`status.toUpperCase()`, captured before any resolution), defaulting to
    # `target` when absent. So the second conjunct compared `target` with
    # `target`. Nothing the relay can emit satisfied either arm.
    #
    # Nothing replaces it, because there is no real condition left to report
    # here: where the issue actually landed is now `reached_status`'s answer
    # and is recorded in the mirror above, and the one case where the relay
    # says it did not honour the canonical target is `forceUsed`, which keeps
    # its own note below.

    # FLY-689: Surface force override info
    if result.get("forceUsed"):
        print(
            f"  Note: Force override used — transitioned to provider-native "
            f"'{result.get('forceTarget', force)}' instead of mapped status",
            file=sys.stderr,
        )

    output_json(result)


def cmd_assign(args: argparse.Namespace) -> None:
    """Assign or unassign an issue."""
    assignee = None if args.unassign else _resolve_assignee(args.assignee)
    client = get_client()
    result = client.assign(args.ref, assignee)
    output_json(result)


def cmd_update(args: argparse.Namespace) -> None:
    """Update one or more fields on an issue — never the description (FLY-1469).

    `--description` / `--description-file` are retired here. They wrote the
    whole document through `PATCH /issues/:ref`, which is the same replacement
    `issues.py description` performs but with none of the §8 protection: no
    `expectedRevision` on the wire, so no mismatch check, and — before FLY-1469
    fixed the route — no would-block verdict either. Two spellings for one
    dangerous write is the hole, not the ergonomics: FLY-1468 refuses a
    tokenless `--file` rewrite precisely so a stale document cannot silently
    overwrite a concurrent edit, and an agent that took that refusal and
    re-ran it as `update --file` would have won. A guard with a second door
    is advice.

    So there is one description writer, `issues.py description`, and it is the
    one that carries the token. The flags stay on the parser only so this
    redirect can be the answer; argparse's `unrecognized arguments` tells an
    agent nothing about where the write went.
    """
    if args.description is not None or args.description_file is not None:
        fail(
            f"{args.ref}: `update` no longer writes descriptions — use "
            "`issues.py description` (operation `issue.description`).\n"
            "It is the same whole-document replace, under the revision check "
            "this path never had:\n"
            f"  issues.py get {args.ref}            # note its `revision`\n"
            f"  issues.py description {args.ref} --file BODY.md "
            "--expected-revision <REVISION>\n"
            "<REVISION> is the `revision` field that get returns.\n"
            "Nothing was updated — re-run without the description flag for the "
            "other fields."
        )

    fields: dict = {}
    if args.title is not None:
        fields["title"] = args.title
    if args.priority is not None:
        fields["priority"] = args.priority
    if args.estimate is not None:
        if args.estimate < 0:
            fail("Estimate points must be non-negative")
        fields["estimate"] = args.estimate
    if args.assignee is not None:
        # FLY-804: clear sentinel ('clear', 'none', 'null', '') unsets the field
        if _is_clear_sentinel(args.assignee):
            fields["assignee"] = None
        else:
            fields["assignee"] = _resolve_assignee(args.assignee)
    if args.state is not None:
        fields["state"] = args.state
        # FLY-1265 / FLY-1257 (spec §9): `update --state` is a real status
        # change on the server now — it runs the same lifecycle pipeline as
        # `transition`, and the `--comment` sent alongside it BECOMES the audit
        # comment rather than being a second, separate note. Two consequences
        # worth stating where the agent will read them:
        #
        # 1. The status must be canonical. The server answers 400 INVALID_STATUS
        #    with a `suggestedStatus` for provider-native spellings; nothing is
        #    translated silently.
        # 2. A comment is still not *required* here. Requiring one would break
        #    every existing `update --state` caller and every script that batches
        #    a status into a field update, and the server already supplies its
        #    own fallback text ("Status updated via issue update.") when none
        #    arrives — so the audit trail exists either way. What the agent
        #    wants is the richer path, hence the nudge rather than the gate.
        if args.comment is None:
            print(
                "  Note: --state moves the issue and posts an audit comment. "
                "Prefer `issues.py transition REF STATUS \"why\"` for workflow "
                "moves, or pass --comment to say why in the audit trail.",
                file=sys.stderr,
            )
    if args.labels is not None:
        # FLY-661: --labels accepts JSON array (e.g. '["bug","chore"]') or
        # comma-separated list (e.g. 'bug,chore'). Parse to a list before sending
        # so the relay receives a proper array, not a literal string.
        labels_arg = args.labels.strip()
        try:
            parsed = json.loads(labels_arg)
            if isinstance(parsed, list):
                fields["labels"] = [str(x).strip() for x in parsed if str(x).strip()]
            else:
                fail(
                    "--labels must be a JSON array (e.g. '[\"bug\",\"chore\"]') "
                    "or comma-separated list (e.g. 'bug,chore')"
                )
        except json.JSONDecodeError:
            # Fall back to comma-separated
            fields["labels"] = [s.strip() for s in labels_arg.split(",") if s.strip()]
    if args.milestone is not None:
        # FLY-804: clear sentinel ('clear', 'none', 'null', '') unsets the milestone
        fields["milestone"] = (
            None if _is_clear_sentinel(args.milestone) else args.milestone
        )
    # FLY-663: --due-date accepts YYYY-MM-DD or ISO 8601 datetime; "clear" / empty
    # string removes the due date. Validate format client-side so users fail fast.
    if getattr(args, "due_date", None) is not None:
        due_arg = args.due_date.strip()
        if due_arg in ("", "clear", "none", "null"):
            fields["dueDate"] = None
        else:
            # Accept YYYY-MM-DD or full ISO 8601 datetime
            import re as _re
            iso_date_pattern = _re.compile(
                r"^\d{4}-\d{2}-\d{2}(T\d{2}:\d{2}(:\d{2}(\.\d+)?)?(Z|[+-]\d{2}:?\d{2})?)?$"
            )
            if not iso_date_pattern.match(due_arg):
                fail(
                    f"Invalid date format '{due_arg}'. "
                    "Expected ISO 8601: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SSZ. "
                    "Use 'clear' to remove the current due date."
                )
            fields["dueDate"] = due_arg
    if args.comment is not None:
        fields["comment"] = args.comment
    if getattr(args, "project", None) is not None:
        # FLY-804: clear sentinel ('clear', 'none', 'null', '') unsets the project
        fields["projectId"] = (
            None if _is_clear_sentinel(args.project) else args.project
        )

    if not fields:
        fail("No fields to update")

    client = get_client()
    result = client.update_issue(args.ref, **fields)
    output_json(result)


def cmd_description(args: argparse.Namespace) -> None:
    """Replace an issue's description under the §8 revision check (FLY-1468).

    There is one write shape here and it is whole-document: `--file`, `--text`
    and stdin all resolve to a complete replacement body. So this command never
    retries. A `REVISION_MISMATCH` means the description changed between the
    read the new text was written against and this write; re-sending that text
    is exactly the clobber the check exists to prevent, and only the author can
    say what the merged document should be.

    `issues.py acceptance` is the path that may retry (§10): its edits are
    criterion-addressed, so the relay re-merges them line-surgically against
    the fresh text and a retry cannot lose anyone's prose.

    The guard has to come from the read the text was written against, and on
    the cloud tier `--file` is required to carry one. A file is text composed
    earlier — by an agent that read the issue, drafted, and came back — so
    fetching a fresh token here would authenticate the write against a read
    that happened milliseconds ago and prove nothing. Worse, it made the
    refusal advisory: re-running without the flag would sail through and
    overwrite the very edit the 409 was protecting. `--text` and stdin keep the
    self-read (short inline edits, composed in the same breath) and say on
    stderr what it does not cover.

    FLY-1470 adds a second, better guard: `--expected-description-hash`, a
    digest of the description the text was written against. `--expected-revision`
    is a last-modified timestamp, so it moves when someone changes the status
    and (on the current mapping) does not move when someone comments — an agent
    that reads an issue, drafts a description and transitions the issue to
    IMPLEMENTING then cannot write the description it just drafted. The hash
    compares the document itself, so that write goes through and a real prose
    edit is still refused, as `DESCRIPTION_CHANGED`.

    Either flag satisfies the `--file` refusal; neither present on cloud is
    still refused. On the `--text`/stdin self-read the command sends both, and
    prefers the `descriptionHash` the read returned over recomputing it — the
    server publishes the normalization, but agreeing on a digest both sides
    already hold beats agreeing on an algorithm.
    """
    # Text resolution: --file > --text > stdin
    text = resolve_text_input(text_arg=args.text, file_arg=args.file)
    if text is None:
        fail("No description text provided (use --text, --file, or pipe to stdin)")

    ref = args.ref
    client = get_client()
    revision = getattr(args, "expected_revision", None)
    content_hash = getattr(args, "expected_description_hash", None)

    if not client.is_cloud:
        # Nothing to race: the local backend writes the file directly, and both
        # guards are relay concepts. Say so rather than dropping a safety flag
        # on the floor.
        for flag, value in (
            ("--expected-revision", revision),
            ("--expected-description-hash", content_hash),
        ):
            if value:
                print(
                    f"{ref}: {flag} is ignored on the local tier "
                    f"(current tier: {client.tier}) — there is no relay to "
                    "check it against, and the file write is not concurrent.",
                    file=sys.stderr,
                )
        revision = None
        content_hash = None
    elif not revision and not content_hash:
        if args.file:
            fail(
                "a file rewrite must carry the guard it was written against:\n"
                f"  issues.py description {ref} --file {args.file} "
                "--expected-description-hash <HASH>\n"
                f"  issues.py description {ref} --file {args.file} "
                "--expected-revision <REVISION>\n"
                f"<HASH> is the `descriptionHash` field `issues.py get {ref}` "
                "returned and <REVISION> is its `revision` field — both on the "
                "basic field set, so the read you drafted this file from "
                "already had them.\n"
                "Reading a fresh one here would only prove the description "
                "did not change in the last millisecond, which is not what "
                "this file was written against.\n"
                "Prefer the hash: the revision is a last-modified timestamp, "
                "so it also trips on a status change that touched no prose."
            )
        issue = client.get_issue(ref, fields="basic")
        current = issue.get("revision") if isinstance(issue, dict) else None
        revision = current if isinstance(current, str) and current else None
        content_hash = _self_read_description_hash(issue)
        if not revision and not content_hash:
            # §8: an empty token is "unknown", never "unchanged" — sending it
            # would fail every comparison. Same for a digest of a description
            # this read did not return. Say the write is unprotected.
            print(
                f"{ref}: this read supplied neither a revision token nor a "
                "description to hash — the write is going out unprotected "
                "against a concurrent edit.",
                file=sys.stderr,
            )
        else:
            print(
                f"{ref}: no --expected-revision / --expected-description-hash: "
                "wrote against a read taken moments ago; an edit made before "
                "that read is not detected. Pass the guard from the read this "
                "text was based on to close that window.",
                file=sys.stderr,
            )

    try:
        result = client.update_description(ref, text, revision, content_hash)
    except RelayError as err:
        _fail_description(err, ref)
        return  # unreachable — _fail_description exits
    output_json(result)


def _self_read_description_hash(issue: object) -> str | None:
    """Digest for the description a `fields=basic` read just returned (FLY-1470).

    Server-provided first. `GET /issues/:ref` returns `descriptionHash` beside
    `revision`, and taking it removes the one way the client and the relay can
    disagree about a document they both hold — an old relay that omits the
    field falls back to computing it here, under the same normalization.

    An **empty** description is a document like any other and hashes to
    sha256(""), on both branches — that digest is exactly what the relay
    computes for the same empty text, so the guard matches and the write
    proceeds. It is not a "no guard" case.

    `None` is returned only when the read carried no `description` field at
    all, which is a different thing: a degraded response is not evidence that
    the description IS empty, and sending the empty-document digest there would
    409 a perfectly good write against prose nobody touched. No guard beats a
    wrong one.
    """
    if not isinstance(issue, dict):
        return None
    served = issue.get("descriptionHash")
    if isinstance(served, str) and served:
        return served
    description = issue.get("description")
    if isinstance(description, str):
        return description_hash(description)
    return None


def _fail_description(err: RelayError, ref: str) -> None:
    """Render a rejected description write and exit (spec §8, §11)."""
    if err.code == "REVISION_MISMATCH":
        fresh = (err.body or {}).get("revision")
        recovery = (
            f"Re-run `issues.py get {ref}`, redo the edit against what it "
            "returns"
        )
        recovery += (
            f", and pass --expected-revision {fresh}."
            if isinstance(fresh, str) and fresh else "."
        )
        fail(
            f"{ref} changed since you read it — re-read and retry. Nothing was "
            "written.\n"
            f"  {err.message}\n"
            "This command does not retry: it replaces the whole description, "
            "so re-sending text written against the old one would overwrite "
            "whatever just landed.\n"
            "Do not re-run this without --expected-revision to get past it. On "
            "the --text path that succeeds — against a token read a moment "
            "later — and silently overwrites the change this refusal just "
            "told you about.\n"
            f"{recovery}"
        )
    if err.code == "DESCRIPTION_CHANGED":
        # The content guard's own 409 (FLY-1470). Distinct from
        # REVISION_MISMATCH in exactly one way that matters to the caller: this
        # one is never a false alarm. The prose on the server really did move,
        # so re-reading is not a formality — it is where the merged document
        # has to come from.
        fresh = (err.body or {}).get("currentHash")
        recovery = (
            f"Re-run `issues.py get {ref}`, redo the edit against the "
            "description it returns"
        )
        recovery += (
            f", and pass --expected-description-hash {fresh}."
            if isinstance(fresh, str) and fresh else "."
        )
        fail(
            f"{ref}: the description changed since you read it — re-read and "
            "redo the edit. Nothing was written.\n"
            f"  {err.message}\n"
            "This command does not retry: it replaces the whole description, "
            "so re-sending text written against the old one would overwrite "
            "whatever just landed.\n"
            "Unlike a revision mismatch this is never a false alarm — the "
            "guard compares the document itself, so someone really did edit "
            "the prose.\n"
            f"{recovery}"
        )
    if err.code == "REVISION_REQUIRED":
        fail(
            f"{ref}: this workspace requires a revision token on description "
            "writes and none could be sent. Nothing was written.\n"
            f"  {err.message}\n"
            f"Re-run `issues.py get {ref}` and pass its `revision` as "
            "--expected-revision. For checkbox edits use `issues.py "
            "acceptance`, which carries the token itself."
        )
    # FLY-634 parity: raise_on_error routes provider auth failures here too,
    # and "reconnect your provider" beats a bare error code.
    if err.code in ("PROVIDER_AUTH_FAILED", "PROVIDER_TOKEN_REFRESH_FAILED"):
        fail(
            f"Provider credentials expired ({err.code}). "
            "Reconnect your provider in the FlyDocs dashboard at app.flydocs.ai"
        )
    provider = (err.body or {}).get("provider_error", "")
    msg = f"Relay API error ({err.code}): {err.message}"
    if provider:
        msg += f" — provider: {provider}"
    fail(msg)


# ---------------------------------------------------------------------------
# Acceptance criteria (FLY-1265 — relay-lifecycle-authority-spec.md §10)
# ---------------------------------------------------------------------------
#
# Ticking a checkbox used to mean: read the whole description, edit the markdown
# by hand, write the whole description back. Two agents doing that at once, or
# one agent working from a description it read five minutes ago, silently
# reverted the other's prose — a read-modify-write race with the entire issue
# body as the unit of conflict.
#
# `acceptance` addresses criteria instead of documents. The client says which
# ordinals to change and what it believes they say; the relay re-reads, merges
# line-surgically, and writes — all inside one operation, under one revision
# check. Everything outside the touched criterion is byte-preserved.

# §10: the client echoes the first 80 characters of the criterion it read as
# `expectedText`. The server compares against exactly this many characters
# (`acceptance.ts` EXPECTED_TEXT_GUARD_CHARS), so this constant is a contract,
# not a display choice.
ACCEPTANCE_GUARD_CHARS = 80

# FLY-1087 deferral destination — the same shape the server enforces on the
# write side. Checked client-side so a typo costs no round trip.
DEFER_REF_RE = re.compile(r"^[A-Z][A-Z0-9]{1,9}-\d{1,6}$")


def _parse_ordinals(raw_values: list[str] | None, flag: str) -> list[int]:
    """Parse repeated/comma-joined ordinals: `--check 1,3 --check 5` -> [1,3,5]."""
    ordinals: list[int] = []
    for raw in raw_values or []:
        for piece in raw.split(","):
            piece = piece.strip()
            if not piece:
                continue
            try:
                value = int(piece)
            except ValueError:
                fail(f"{flag} expects criterion numbers, got '{piece}'")
                return []
            if value < 1:
                fail(f"{flag} criterion numbers are 1-based, got '{piece}'")
            ordinals.append(value)
    return ordinals


def _parse_ordinal_pair(raw: str, flag: str) -> tuple[int, str]:
    """Parse `N:value` — `--defer 5:FLY-1234`, `--note 2:regressed in review`."""
    ordinal_part, sep, value = raw.partition(":")
    if not sep or not value.strip():
        example = "FLY-1234" if flag == "--defer" else "why it changed"
        fail(f"{flag} expects N:VALUE (e.g. {flag} 2:{example}), got '{raw}'")
    try:
        ordinal = int(ordinal_part.strip())
    except ValueError:
        fail(f"{flag} expects a criterion number before the colon, got '{raw}'")
        return (0, "")
    if ordinal < 1:
        fail(f"{flag} criterion numbers are 1-based, got '{raw}'")
    return ordinal, value.strip()


def _render_criteria(criteria: list[dict]) -> str:
    """The acceptance list as the ordinals a caller addresses."""
    lines = []
    for c in criteria:
        if not isinstance(c, dict):
            continue
        box = "x" if c.get("checked") else " "
        deferred = c.get("deferredTo")
        marker = f" (deferred: {deferred})" if deferred else ""
        lines.append(f"  {c.get('id')}. [{box}]{marker} {c.get('text', '')}")
    return "\n".join(lines) if lines else "  (none)"


def _fail_acceptance(err: RelayError, ref: str, lead: str | None = None) -> None:
    """Render a structured acceptance rejection and exit (spec §10, §11).

    Every branch ends with the criteria as they now read, when the relay sent
    them — a rejection whose recovery is "re-run against the current numbers"
    should hand over the current numbers.
    """
    body = err.body or {}
    lines: list[str] = []

    if lead:
        lines.append(lead)
        lines.append(f"  {err.message}")
    elif err.code == "CRITERION_MISMATCH":
        criterion_id = body.get("criterionId")
        lines.append(
            f"Criterion guard failed on {ref}"
            + (f" (criterion {criterion_id})" if criterion_id else "")
            + ": the description does not say what this command was built from."
        )
        lines.append(f"  {err.message}")
    elif err.code == "REVISION_MISMATCH":
        lines.append(f"{ref} changed since it was read. Nothing was written.")
        lines.append(f"  {err.message}")
    elif err.code.startswith("ACCEPTANCE_"):
        lines.append(f"Acceptance gate: {err.message}")
    else:
        lines.append(f"Relay API error ({err.code}): {err.message}")
        criterion_id = body.get("criterionId")
        if criterion_id is not None:
            lines.append(f"  Criterion: {criterion_id}")

    fresh = body.get("acceptance")
    if isinstance(fresh, list) and fresh:
        lines.append("")
        lines.append(f"Acceptance criteria on {ref} right now:")
        lines.append(_render_criteria(fresh))
        lines.append("")
        lines.append("Re-run against these numbers.")

    fail("\n".join(lines))


def cmd_acceptance(args: argparse.Namespace) -> None:
    """Check, uncheck, defer or annotate acceptance criteria by ordinal.

    Flow (§10): read the issue (the response carries `revision` and the parsed
    `acceptance` list) -> build changes against the ordinals the server just
    returned, each carrying an `expectedText` guard -> POST. `expectedRevision`
    is required, so a description that moved between the read and the write is
    rejected rather than silently overwritten.
    """
    ref = args.ref.upper()

    # Build the intent first: an argument mistake should cost no API calls.
    # `claimed` doubles as the duplicate check — the relay rejects a batch that
    # names one criterion twice (ambiguous intent), and catching it here
    # explains why, which a 400 cannot.
    intents: list[dict] = []
    claimed: dict[int, str] = {}

    def _claim(ordinal: int, flag: str) -> None:
        previous = claimed.get(ordinal)
        if previous == flag:
            fail(f"Criterion {ordinal} is listed twice under {flag}.")
        if previous:
            fail(
                f"Criterion {ordinal} appears in both {previous} and {flag}. "
                "The relay applies one status per criterion in a batch, and a "
                "note is its own status — so this batch has no single meaning. "
                "Run them as two commands (note first, then the box change)."
            )
        claimed[ordinal] = flag

    for ordinal in _parse_ordinals(args.check, "--check"):
        _claim(ordinal, "--check")
        intents.append({"criterionId": ordinal, "status": "checked"})
    for ordinal in _parse_ordinals(args.uncheck, "--uncheck"):
        _claim(ordinal, "--uncheck")
        intents.append({"criterionId": ordinal, "status": "unchecked"})
    for raw in args.defer or []:
        ordinal, destination = _parse_ordinal_pair(raw, "--defer")
        destination = destination.upper()
        if not DEFER_REF_RE.match(destination):
            fail(
                f"--defer needs an issue ref to defer to (e.g. 5:FLY-1234), "
                f"got '{destination}'. A deferral with no destination is an "
                "unfinished criterion with better manners (FLY-1087)."
            )
        _claim(ordinal, "--defer")
        intents.append({
            "criterionId": ordinal,
            "status": "deferred",
            "deferredTo": destination,
        })
    for raw in args.note or []:
        ordinal, text = _parse_ordinal_pair(raw, "--note")
        if "\n" in text or "\r" in text:
            fail("--note must be a single line — it is appended to the criterion.")
        _claim(ordinal, "--note")
        intents.append({"criterionId": ordinal, "status": "note", "note": text})

    if not intents:
        fail(
            "Nothing to change. Use --check N, --uncheck N, --defer N:REF or "
            "--note N:text (N is the criterion number from `issues.py get`)."
        )

    client = get_client()
    if not client.is_cloud:
        fail(
            "issues.py acceptance requires cloud tier — the criterion merge runs "
            f"on the relay. Current tier: {client.tier}.\n"
            "On local tier, edit the description directly: "
            f"issues.py description {ref} --file <path>"
        )

    # `basic` carries description, revision and the parsed view — the criterion
    # ordinals do not need comments or relations to be addressable.
    issue = client.get_issue(ref, fields="basic")
    criteria = issue.get("acceptance")
    revision = issue.get("revision")

    if not isinstance(criteria, list):
        # Deliberately NOT the revision this command just read (FLY-1468
        # review). Handing over a token from T0 invites an agent holding the
        # description from an earlier read to edit that text and write it under
        # the newer token — clobbering whatever landed in between, with the
        # hint's blessing. This command never prints the description, so a
        # re-read is required regardless; the token has to come from it.
        fail(
            f"This relay did not return a parsed acceptance list for {ref}. "
            "The acceptance route needs a relay that serves `acceptance` on "
            "issue reads; until then edit the description directly — re-read "
            "it first, and pass the `revision` that read returns:\n"
            f"  issues.py description {ref} --file <PATH> "
            "--expected-revision <REVISION>\n"
            "<PATH> holds the FULL updated description; <REVISION> is the "
            "`revision` field from that re-read — not one this command could "
            "hand you, since it never printed the description."
        )
    if not criteria:
        fail(f"{ref} has no acceptance criteria — there is nothing to change.")
    if not revision or not isinstance(revision, str):
        fail(
            f"The provider supplied no revision token for {ref}, and the "
            "acceptance route requires one — it is what makes the edit safe "
            "against a concurrent description change. Use `issues.py "
            "description` if this provider never supplies one."
        )

    by_id = {
        c.get("id"): c for c in criteria
        if isinstance(c, dict) and isinstance(c.get("id"), int)
    }
    unknown = sorted(o for o in claimed if o not in by_id)
    if unknown:
        plural = "criteria" if len(criteria) != 1 else "criterion"
        fail(
            f"{ref} has {len(criteria)} {plural}; no criterion "
            f"{', '.join(str(o) for o in unknown)}.\n"
            f"{_render_criteria(criteria)}"
        )

    # The text guard closes the window the ordinal leaves open: an insertion
    # between the read and the write shifts every ordinal after it, and without
    # the guard the wrong box would flip with a perfectly successful response.
    changes: list[dict] = []
    for intent in intents:
        text = str(by_id[intent["criterionId"]].get("text", ""))
        changes.append({**intent, "expectedText": text[:ACCEPTANCE_GUARD_CHARS]})

    try:
        result = client.acceptance(ref, changes, revision)
    except RelayError as err:
        if err.code != "REVISION_MISMATCH":
            _fail_acceptance(err, ref)
        result = _retry_acceptance_once(client, ref, changes, err)

    output_json({
        "success": bool(result.get("success", True)),
        "issue": result.get("issue", ref),
        "applied": [
            {k: v for k, v in change.items() if k != "expectedText"}
            for change in changes
        ],
    })


def _retry_acceptance_once(client: object, ref: str, changes: list[dict],
                           err: RelayError) -> dict:
    """Re-check the intent against the fresh state and re-issue once (§10).

    A `REVISION_MISMATCH` body carries both the fresh revision and the fresh
    acceptance list, so the recovery costs no extra read. The retry is a new
    intent against new state — the operation id is regenerated per call by
    `_request`, which is exactly right: reusing the old one would be
    `OPERATION_ID_REUSED`, because the input hash has changed.

    One retry, not a loop. A second conflict means something else is actively
    writing to this description, and the honest answer is to say so.
    """
    body = err.body or {}
    fresh = body.get("acceptance")
    fresh_revision = body.get("revision")

    if not isinstance(fresh, list) or not isinstance(fresh_revision, str) or not fresh_revision:
        _fail_acceptance(
            err, ref,
            lead=(
                f"{ref} changed since it was read, and the relay returned no "
                "fresh state to re-check the intent against. Nothing was "
                f"written — re-run `issues.py get {ref}` and try again."
            ),
        )

    fresh_by_id = {
        c.get("id"): c for c in fresh
        if isinstance(c, dict) and isinstance(c.get("id"), int)
    }
    drifted: list[str] = []
    for change in changes:
        criterion_id = change["criterionId"]
        current = fresh_by_id.get(criterion_id)
        if current is None:
            drifted.append(f"criterion {criterion_id} no longer exists")
            continue
        if str(current.get("text", ""))[:ACCEPTANCE_GUARD_CHARS] != change["expectedText"]:
            drifted.append(f"criterion {criterion_id} now reads different text")

    if drifted:
        fail(
            f"{ref} changed while this command was running, and the criteria it "
            "targets moved with it: " + "; ".join(drifted) + ".\n"
            "Nothing was written. The criteria now read:\n"
            f"{_render_criteria(fresh)}\n"
            "Re-run against these numbers."
        )

    print(
        f"{ref} changed since it was read; the targeted criteria are unmoved — "
        "retrying once against the fresh revision.",
        file=sys.stderr,
    )
    try:
        return client.acceptance(ref, changes, fresh_revision)
    except RelayError as retry_err:
        lead = None
        if retry_err.code == "REVISION_MISMATCH":
            lead = (
                f"{ref} kept changing underneath this command — the retry hit a "
                "second conflict. Nothing was written. Something else is "
                "writing to this description right now."
            )
        _fail_acceptance(retry_err, ref, lead=lead)
        return {}  # unreachable — _fail_acceptance exits


def cmd_comment(args: argparse.Namespace) -> None:
    """Add a comment to an issue."""
    body = args.body
    # FLY-699: Use stdin_has_data() instead of isatty() to avoid blocking
    # on open-but-empty pipes from subprocess harnesses.
    if body is None:
        from flydocs_api import stdin_has_data
        if stdin_has_data():
            body = sys.stdin.read().strip()
    if not body:
        fail("No comment body provided (pass as argument or pipe to stdin)")

    client = get_client()
    result = client.comment(args.ref, body)
    output_json(result)


def cmd_estimate(args: argparse.Namespace) -> None:
    """Set estimate points on an issue."""
    if args.points < 0:
        fail("Estimate points must be non-negative")

    client = get_client()
    result = client.estimate(args.ref, args.points)
    output_json(result)


def cmd_priority(args: argparse.Namespace) -> None:
    """Set priority level on an issue."""
    if args.level < 0 or args.level > 4:
        fail("Priority level must be 0-4")

    client = get_client()
    result = client.priority(args.ref, args.level)
    output_json(result)


def cmd_link(args: argparse.Namespace) -> None:
    """Link two issues together."""
    client = get_client()
    result = client.link(args.ref, args.related_ref, args.type)
    output_json(result)


def cmd_assign_milestone(args: argparse.Namespace) -> None:
    """Assign an issue to a milestone."""
    client = get_client()
    result = client.assign_milestone(args.ref, args.milestone_id)
    output_json(result)


def cmd_assign_cycle(args: argparse.Namespace) -> None:
    """@deprecated FLY-655: use cmd_assign_sprint."""
    print(
        "Note: 'assign-cycle' is deprecated — use 'assign-sprint' instead.",
        file=sys.stderr,
    )
    client = get_client()
    result = client.assign_cycle(args.ref, args.cycle_id)
    output_json(result)


def _resolve_sprint_alias(client: object, alias: str) -> str | None:
    """
    FLY-656: Resolve a sprint alias (current / next / previous) to a concrete sprint ID.

    Returns None if no sprint matches the alias. Caller should fail gracefully
    with a clear message when the alias can't be resolved.
    """
    if not alias or alias.lower() not in ("current", "next", "previous", "prev"):
        return None

    alias_lower = alias.lower()
    # Ask for active and future sprints, sorted by startsAt ascending (default)
    sprints = client.list_sprints(active=True, future=True)  # type: ignore[attr-defined]

    if alias_lower == "current":
        # Find the first active sprint
        for s in sprints:
            if s.get("state") == "active":
                return str(s.get("id", ""))
        # Fallback: if no state field, try the one whose window includes now
        # (can't know without startsAt/endsAt parsing — defer to App via ?current=true)
        current_sprints = client.list_sprints(current=True)  # type: ignore[attr-defined]
        if current_sprints:
            return str(current_sprints[0].get("id", ""))
        return None

    if alias_lower == "next":
        # First future sprint (not active)
        for s in sprints:
            if s.get("state") == "future":
                return str(s.get("id", ""))
        return None

    if alias_lower in ("previous", "prev"):
        # Most recently closed sprint
        closed = client.list_sprints(closed=True)  # type: ignore[attr-defined]
        if closed:
            # Assume sorted by startsAt ascending; take the last
            return str(closed[-1].get("id", ""))
        return None

    return None


def cmd_assign_sprint(args: argparse.Namespace) -> None:
    """
    FLY-656: Assign an issue to a sprint, with support for aliases.

    Accepts:
    - Numeric/UUID sprint ID: assign-sprint FLOCK-123 2118
    - Named aliases: assign-sprint FLOCK-123 current|next|previous
    - Unassign: assign-sprint FLOCK-123 (no id)
    """
    # FLY-692: Gate sprint assignment on Kanban boards
    ctx = _get_active_context()
    if ctx and ctx.get("boardType") == "kanban":
        fail(
            "Sprints are not available on Kanban boards. "
            "This board uses column-based workflow."
        )

    client = get_client()
    sprint_id = args.sprint_id

    # Resolve aliases to concrete IDs
    if sprint_id and sprint_id.lower() in ("current", "next", "previous", "prev"):
        resolved = _resolve_sprint_alias(client, sprint_id)
        if not resolved:
            fail(
                f"No sprint matches alias '{sprint_id}'. "
                f"Run 'flydocs list-sprints' to see available sprints."
            )
        print(
            f"Note: Resolved '{sprint_id}' → sprint ID {resolved}",
            file=sys.stderr,
        )
        sprint_id = resolved

    result = client.assign_sprint(args.ref, sprint_id)
    output_json(result)


# ---------------------------------------------------------------------------
# PR creation
# ---------------------------------------------------------------------------

def _detect_platform() -> str:
    """Detect git hosting platform from remote URL. Returns github|gitlab|bitbucket|unknown."""
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=5,
        )
        url = result.stdout.strip().lower()
        if "github" in url:
            return "github"
        if "gitlab" in url:
            return "gitlab"
        if "bitbucket" in url:
            return "bitbucket"
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
        pass
    return "unknown"


def _bitbucket_repo_from_remote(url: str) -> str | None:
    """`workspace/repo_slug` from a Bitbucket remote URL, or None (FLY-1543).

    Handles `git@bitbucket.org:ws/repo.git`, `ssh://git@bitbucket.org/ws/repo.git`
    and `https://user@bitbucket.org/ws/repo.git`.
    """
    m = re.search(r"bitbucket\.org[:/]([^/\s]+)/([^/\s]+?)(?:\.git)?/?$", url.strip())
    if not m:
        return None
    return f"{m.group(1)}/{m.group(2)}"


def _get_remote_url() -> str:
    try:
        result = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.strip()
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
        return ""


def _create_bitbucket_pr(
    *, title: str, body: str, branch: str, base: str, draft: bool
) -> str:
    """Open a Bitbucket PR through the relay and return its URL (FLY-1543).

    Bitbucket has no maintained CLI, so instead of shelling out the way the
    GitHub and GitLab branches do, this posts to the relay, which opens the PR
    with the org's Bitbucket connection. No credential touches this script;
    the PR is authored by the connection owner and the body carries the
    issue and developer attribution.
    """
    repo = _bitbucket_repo_from_remote(_get_remote_url())
    if not repo:
        fail("Could not read `workspace/repo_slug` from the origin remote URL.")
    client = get_client()
    client.require_cloud("Bitbucket PR creation")
    result = client.relay.post(
        "/scm/pull-requests",
        {
            "repo": repo,
            "title": title,
            "description": body,
            "sourceBranch": branch,
            "destinationBranch": base,
            "draft": bool(draft),
        },
    )
    url = result.get("url") if isinstance(result, dict) else None
    if not url:
        fail(f"Bitbucket PR creation returned no URL: {result}")
    return str(url)


def _get_current_branch() -> str:
    """Get current git branch name."""
    try:
        result = subprocess.run(
            ["git", "branch", "--show-current"],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.strip()
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
        return ""


def _get_default_branch() -> str:
    """Detect the default branch (main or master)."""
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--verify", "refs/heads/main"],
            capture_output=True, timeout=5,
        )
        if result.returncode == 0:
            return "main"
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
        pass
    return "master"


def _get_branch_commit_subjects(base: str) -> list[str]:
    """Commit subjects on the current branch not in base (newest last)."""
    if not base:
        return []
    try:
        result = subprocess.run(
            ["git", "log", f"{base}..HEAD", "--format=%s", "--no-merges", "--reverse"],
            capture_output=True, text=True, timeout=5,
        )
        if result.returncode != 0:
            return []
        return [ln.strip() for ln in result.stdout.splitlines() if ln.strip()]
    except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
        return []


def _split_items(raw: str | None) -> list[str]:
    """Split a --changes/--test-plan value on newlines or semicolons."""
    if not raw:
        return []
    parts = re.split(r"[\n;]+", raw)
    return [p.strip().lstrip("-").strip() for p in parts if p.strip()]


def render_pr_body(
    template: str,
    issue_ref: str | None,
    summary: str | None,
    acceptance: str | None,
    changes: list[str],
    test_plan: list[str],
    notes: str | None,
) -> str:
    """Pure PR-body renderer (FLY-998).

    Fills Summary/Acceptance from issue data, Changes from provided items or
    branch commits, and collapses empty sections instead of leaving hollow
    bullets/checkboxes. Test Plan with no items shows an italic hint; empty
    Notes drops the trailing section entirely.
    """
    body = template
    body = body.replace("{ISSUE_REF}", issue_ref or "")
    body = body.replace("{ISSUE_SUMMARY}", summary or "")
    body = body.replace("{ACCEPTANCE_CRITERIA}", acceptance or "")

    # Changes: real bullets, or drop the placeholder bullet lines entirely.
    changes_md = "\n".join(f"- {c}" for c in changes) if changes else ""
    body = body.replace("- {CHANGE_1}\n- {CHANGE_2}", changes_md)
    # Defensive: collapse any leftover individual placeholders / bare dashes.
    body = re.sub(r"^- \{CHANGE_\d+\}\n?", "", body, flags=re.MULTILINE)

    # Test Plan: checkboxes, or a single italic hint (never an empty checkbox).
    test_md = (
        "\n".join(f"- [ ] {t}" for t in test_plan)
        if test_plan
        else "_Describe how to verify this works._"
    )
    body = body.replace("- [ ] {TEST_1}\n- [ ] {TEST_2}", test_md)
    body = re.sub(r"^- \[ \] \{TEST_\d+\}\n?", "", body, flags=re.MULTILINE)

    # Notes: keep if provided, else drop the whole trailing Notes section.
    if notes and notes.strip():
        body = body.replace("{NOTES}", notes.strip())
    else:
        body = body.replace("{NOTES}", "")
        body = re.sub(r"\n#{1,6}\s+Notes\s*\n(?:\s*<!--.*?-->\s*)?\s*$", "\n", body, flags=re.DOTALL)

    # Collapse 3+ blank lines left by removed content; trim trailing space.
    body = re.sub(r"\n{3,}", "\n\n", body)
    return body.rstrip() + "\n"


def _build_pr_body(
    issue_ref: str | None,
    summary: str | None,
    changes: list[str] | None = None,
    test_plan: list[str] | None = None,
    notes: str | None = None,
    base: str | None = None,
) -> str:
    """Build PR body from template + issue data + branch commits."""
    script_dir = Path(__file__).parent
    template_path = script_dir.parent / "templates" / "pr" / "default.md"

    if template_path.exists():
        template = template_path.read_text()
    else:
        template = (
            "## Summary\n\nResolves {ISSUE_REF}\n\n{ISSUE_SUMMARY}\n\n"
            "## Changes\n\n- {CHANGE_1}\n- {CHANGE_2}\n\n"
            "## Test Plan\n\n- [ ] {TEST_1}\n- [ ] {TEST_2}\n\n"
            "## Acceptance Criteria\n\n{ACCEPTANCE_CRITERIA}\n\n## Notes\n\n{NOTES}\n"
        )

    acceptance: str | None = None
    if issue_ref:
        try:
            client = get_client()
            issue = client.get_issue(issue_ref, fields="full")
            if isinstance(issue, dict):
                summary = issue.get("title", "") or summary
                description = issue.get("description", "")
                ac_lines = [
                    line
                    for line in (description or "").splitlines()
                    if re.match(r"^\s*-\s*\[[ x]\]", line, re.IGNORECASE)
                ]
                acceptance = (
                    "\n".join(ac_lines)
                    if ac_lines
                    else "See issue for acceptance criteria."
                )
        except Exception:
            acceptance = "See issue for acceptance criteria."

    # Auto-fill Changes from branch commits when none were supplied (FLY-998).
    resolved_changes = changes if changes else _get_branch_commit_subjects(base or "")

    return render_pr_body(
        template=template,
        issue_ref=issue_ref,
        summary=summary,
        acceptance=acceptance,
        changes=resolved_changes,
        test_plan=test_plan or [],
        notes=notes,
    )


def cmd_pr(args: argparse.Namespace) -> None:
    """Create a pull/merge request with standard template."""
    platform = _detect_platform()
    branch = _get_current_branch()
    base = args.base or _get_default_branch()

    if not branch or branch == base:
        fail(f"Cannot create PR from branch '{branch}' — switch to a feature branch first.")

    # Build title
    title = args.title
    if not title and args.issue:
        try:
            client = get_client()
            issue = client.get_issue(args.issue, fields="basic")
            if isinstance(issue, dict):
                title = issue.get("title", args.issue)
        except Exception:
            title = args.issue
    if not title:
        title = branch

    # Build body from template
    body = _build_pr_body(
        issue_ref=args.issue,
        summary=title,
        changes=_split_items(getattr(args, "changes", None)),
        test_plan=_split_items(getattr(args, "test_plan", None)),
        notes=getattr(args, "notes", None),
        base=base,
    )

    if args.dry_run:
        output_json({
            "platform": platform,
            "branch": branch,
            "base": base,
            "title": title,
            "body": body,
        })
        return

    # Bitbucket goes through the relay (FLY-1543): no CLI to shell out to.
    if platform == "bitbucket":
        pr_url = _create_bitbucket_pr(
            title=title, body=body, branch=branch, base=base, draft=bool(args.draft)
        )
        if args.issue:
            try:
                get_client().comment(args.issue, f"PR created: {pr_url}")
            except Exception:
                pass  # Non-blocking — PR was created successfully
        output_json({
            "success": True,
            "platform": platform,
            "url": pr_url,
            "branch": branch,
            "base": base,
        })
        return

    # Create PR via platform CLI
    if platform == "github":
        cmd = ["gh", "pr", "create", "--title", title, "--body", body, "--base", base]
        if args.draft:
            cmd.append("--draft")
    elif platform == "gitlab":
        cmd = ["glab", "mr", "create", "--title", title, "--description", body, "--target-branch", base]
        if args.draft:
            cmd.append("--draft")
    else:
        fail(f"Could not detect git platform from remote URL. Detected: {platform}")

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        if result.returncode != 0:
            fail(f"PR creation failed: {result.stderr.strip()}")

        pr_url = result.stdout.strip()

        # Post PR link as comment on the issue
        if args.issue:
            try:
                client = get_client()
                client.comment(args.issue, f"PR created: {pr_url}")
            except Exception:
                pass  # Non-blocking — PR was created successfully

        output_json({
            "success": True,
            "platform": platform,
            "url": pr_url,
            "branch": branch,
            "base": base,
        })
    except FileNotFoundError:
        cli = "gh" if platform == "github" else "glab"
        fail(f"{cli} CLI not found. Install it to create PRs from the command line.")
    except subprocess.TimeoutExpired:
        fail("PR creation timed out.")


# ---------------------------------------------------------------------------
# Audit
# ---------------------------------------------------------------------------

CHECKBOX_PATTERN = re.compile(r"^\s*-\s*\[[ x]\]", re.MULTILINE | re.IGNORECASE)

# What each check has to be able to READ before it has an answer (FLY-1436).
#
# The audit runs over a LIST, and a list is a projection. The relay's list
# route returns `IssueListItem` (flydocs-app `src/lib/relay/adapters/types.ts`)
# — id, identifier, title, status, assignee, priority, estimate, dates,
# milestone, project — and no `description`, no `labels`. The local tier's
# `_local/file_store.list_issues` returns a narrower set still.
#
# So `issue.get("description", "")` never read an empty description. It read a
# column that was never sent, on every row, and a 60-issue audit reported all
# 60 as `missing_description` and `no_labels` — including issues whose
# single-issue read returns 4KB of description and a label. `no_priority`
# discriminated correctly for the only reason that mattered: priority IS in the
# projection.
#
# A check therefore declares what it needs, and a check whose fields the record
# does not carry is reported as **not evaluated**. Not as a pass: a silent pass
# is the same bug wearing the other sign, and it is what hid
# `no_acceptance_criteria` doing nothing for as long as this one existed.
AUDIT_CHECK_FIELDS: dict[str, tuple[str, ...]] = {
    "missing_description": ("description",),
    "no_acceptance_criteria": ("description", "status"),
    "unassigned_active": ("assignee", "status"),
    "no_labels": ("labels",),
    # `no_priority` needs only the priority. Its status term is a negative
    # filter (skip CANCELED), so an unrecognised status still leaves the check
    # answerable — the issue is not canonically canceled either way.
    "no_priority": ("priority",),
}

AUDIT_CHECKS: tuple[str, ...] = tuple(AUDIT_CHECK_FIELDS)

AUDIT_FIELDS: frozenset[str] = frozenset(
    field for fields in AUDIT_CHECK_FIELDS.values() for field in fields
)

# Statuses that make each status-dependent check mean something. Canonical
# names — the record carries provider-native ones ("In Progress", "Done"),
# which is the second half of the same bug: the old code upper-cased the
# provider string and compared it to "IMPLEMENTING", so on any cloud-tier
# workspace `unassigned_active` and `no_acceptance_criteria` could not fire at
# all. `canonical_status` is the translation, and it already exists here for
# `reached_status`.
AC_EXPECTED_STATUSES: frozenset[str] = frozenset(
    {"IMPLEMENTING", "REVIEW", "TESTING", "COMPLETE"}
)
ASSIGNEE_EXPECTED_STATUSES: frozenset[str] = frozenset({"IMPLEMENTING", "REVIEW"})


def audit_supplied_fields(issue: dict) -> set[str]:
    """Which audited fields this record actually carries.

    An absent key — or one present as `null` — was not supplied by the read
    that produced the record. That is a different fact from a field the
    provider returned empty, and collapsing the two is the whole of FLY-1436.

    `status` is the exception: it is always present in both projections, but a
    provider-native name this checkout cannot translate is unusable in the same
    way an absent field is. So it counts as supplied only when
    `canonical_status` resolves it, and the checks that read it fall to "not
    evaluated" rather than silently comparing against a name that never matches.
    """
    supplied = {
        field
        for field in AUDIT_FIELDS
        if field != "status" and issue.get(field) is not None
    }
    if canonical_status(issue.get("status")) is not None:
        supplied.add("status")
    return supplied


def _audit_check_fires(check: str, issue: dict, status: str | None) -> bool:
    """Whether one check finds a problem. Only called when its fields are there."""
    description = str(issue.get("description") or "")
    if check == "missing_description":
        return not description.strip()
    if check == "no_acceptance_criteria":
        # An empty description is already reported as `missing_description`;
        # adding "and it has no checkboxes" says the same thing twice.
        return (
            status in AC_EXPECTED_STATUSES
            and bool(description.strip())
            and not CHECKBOX_PATTERN.search(description)
        )
    if check == "unassigned_active":
        return status in ASSIGNEE_EXPECTED_STATUSES and not issue.get("assignee")
    if check == "no_labels":
        return not issue.get("labels")
    if check == "no_priority":
        return issue.get("priority") == 0 and status != "CANCELED"
    return False


def audit_issue(issue: dict) -> tuple[list[str], list[str]]:
    """Run every check against one issue record.

    Returns `(findings, not_evaluated)` — problems found, and checks that could
    not be run because the record did not carry what they read. Both lists
    follow `AUDIT_CHECKS` order. Pure: no client, no I/O.
    """
    supplied = audit_supplied_fields(issue)
    status = canonical_status(issue.get("status"))

    findings: list[str] = []
    not_evaluated: list[str] = []
    for check, required in AUDIT_CHECK_FIELDS.items():
        if not supplied.issuperset(required):
            not_evaluated.append(check)
        elif _audit_check_fires(check, issue, status):
            findings.append(check)
    return findings, not_evaluated


def build_audit_report(issues: list[dict], deep: bool = False) -> dict:
    """The audit report for a set of issue records. Pure — the tested seam."""
    findings: list[dict] = []
    skipped: dict[str, int] = {}

    for issue in issues:
        issue_findings, issue_skipped = audit_issue(issue)
        for check in issue_skipped:
            skipped[check] = skipped.get(check, 0) + 1
        if issue_findings:
            findings.append({
                "ref": issue.get("identifier", "?"),
                "title": issue.get("title", ""),
                "status": str(issue.get("status") or "").upper(),
                "findings": issue_findings,
            })

    report: dict = {
        "total_checked": len(issues),
        "issues_with_findings": len(findings),
        "findings": findings,
        "checks": list(AUDIT_CHECKS),
    }
    if skipped:
        report["not_evaluated"] = [
            {
                "check": check,
                "issues": skipped[check],
                "needs": list(AUDIT_CHECK_FIELDS[check]),
            }
            for check in AUDIT_CHECKS
            if check in skipped
        ]
        if not deep:
            report["hint"] = (
                "Some checks read fields the issue list does not carry "
                "(description, labels). Re-run with --deep to read each issue "
                "in full — one request per issue — and evaluate them."
            )
    return report


def _hydrate_issues(client: object, issues: list[dict]) -> tuple[list[dict], list[str]]:
    """Re-read each listed issue in full so every check has its fields.

    One request per issue, which is why `--deep` is opt-in rather than the
    default: at the default limit this is hundreds of round trips. A read that
    fails leaves the list record in place — that issue's description and label
    checks stay "not evaluated", which is the honest answer for a record we
    could not complete.
    """
    hydrated: list[dict] = []
    failed: list[str] = []
    for issue in issues:
        ref = str(issue.get("identifier") or "").strip()
        if not ref:
            hydrated.append(issue)
            continue
        try:
            full = client.get_issue(ref, fields="full")  # type: ignore[attr-defined]
        except (RelayError, OSError, ValueError):
            full = None
        if isinstance(full, dict):
            # Merge over the list record, dropping nulls: a field the full read
            # omits must not erase one the list supplied.
            hydrated.append({**issue, **{k: v for k, v in full.items() if v is not None}})
        else:
            failed.append(ref)
            hydrated.append(issue)
    return hydrated, failed


def cmd_audit(args: argparse.Namespace) -> None:
    """Audit issues for workflow compliance.

    Checks for: missing descriptions, missing labels, missing AC checkboxes,
    unassigned in-progress issues, and unset priorities. Checks whose fields
    the read did not supply are reported under `not_evaluated` rather than
    guessed at in either direction (FLY-1436).
    """
    client = get_client()
    issues_data = client.list_issues(
        status=args.status,
        show_all=True,
        limit=args.limit,
    )

    # Handle both list and dict response shapes
    issues = issues_data if isinstance(issues_data, list) else issues_data.get("issues", [])

    deep = bool(getattr(args, "deep", False))
    failed: list[str] = []
    if deep and issues:
        print(f"Reading {len(issues)} issues in full...", file=sys.stderr)
        issues, failed = _hydrate_issues(client, issues)

    report = build_audit_report(issues, deep=deep)
    if failed:
        report["hydration_failed"] = failed
    output_json(report)


def cmd_fix(args: argparse.Namespace) -> None:
    """Fix missing fields on an issue using config defaults.

    Auto-applies: category labels from type, project from activeProjectId.
    """
    client = get_client()
    issue = client.get_issue(args.ref, fields="full")
    if not isinstance(issue, dict):
        fail(f"Could not fetch issue {args.ref}")

    fixes: dict = {}
    applied: list[str] = []

    # Fix missing labels (if we can detect type from title/description)
    labels = issue.get("labels", [])
    if not labels and client.is_cloud:
        # Try to detect type from title keywords
        title_lower = issue.get("title", "").lower()
        detected_type = None
        if any(w in title_lower for w in ["bug", "fix", "broken", "error"]):
            detected_type = "bug"
        elif any(w in title_lower for w in ["add", "implement", "create", "new"]):
            detected_type = "feature"
        elif any(w in title_lower for w in ["refactor", "clean", "update", "upgrade", "remove"]):
            detected_type = "chore"
        if detected_type:
            cat_id = client.relay.get_category_label_id(detected_type)
            if cat_id:
                fixes["labels"] = cat_id
                applied.append(f"label:{detected_type}")

    # Fix missing project (ADR-011: activeProjectId is singular string)
    project = issue.get("project")
    if not project and client.is_cloud:
        active = client.relay.workspace.get("activeProjectId")
        if active:
            fixes["projectId"] = active
            applied.append("project:activeProjectId")

    if not fixes:
        output_json({"ref": args.ref, "fixed": [], "message": "No fixes needed"})
        return

    result = client.update_issue(args.ref, **fixes)
    output_json({
        "ref": args.ref,
        "fixed": applied,
        "result": result,
    })


# ---------------------------------------------------------------------------
# Argument parser
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(description="FlyDocs issue operations")
    sub = parser.add_subparsers(dest="command", required=True)

    # -- create --
    p = sub.add_parser("create", help="Create a new issue")
    p.add_argument("--title", required=True)
    p.add_argument("--type", required=True, choices=["feature", "bug", "chore", "idea"])
    p.add_argument("--description", default=None)
    # `--file` is the catalog-wide spelling for "the body is in this file"
    # (FLY-929). Alias, not a rename: `--description-file` keeps working.
    p.add_argument("--description-file", "--file", dest="description_file", default=None)
    p.add_argument("--priority", type=int, default=None, choices=[0, 1, 2, 3, 4])
    p.add_argument("--estimate", type=int, default=None,
                   help="Points on the provider scale (non-negative); "
                        "`flydocs run workspace.get-estimate-scale` reports it")
    p.add_argument("--assignee", default=None, help="Assignee name/ID, or 'self'/'me' for current user")
    p.add_argument("--project", default=None)
    p.add_argument("--milestone", default=None, help="Milestone ID")
    p.add_argument("--template", action="store_true", help="Read type template as description skeleton")
    p.add_argument("--triage", action="store_true", help="Quick capture — bypasses description enforcement")

    # -- get --
    p = sub.add_parser("get", help="Get a single issue")
    p.add_argument("ref")
    p.add_argument("--fields", default="full", choices=["basic", "full"])

    # -- list --
    p = sub.add_parser("list", help="List issues")
    p.add_argument("--status", default=None)
    p.add_argument("--active", action="store_true")
    p.add_argument("--project", default=None)
    p.add_argument("--assignee", default=None)
    p.add_argument("--milestone", default=None)
    p.add_argument("--mine", action="store_true")
    p.add_argument("--all", action="store_true", dest="show_all", help="Bypass product scope cascade")
    p.add_argument(
        "--limit", type=int, default=DEFAULT_LIST_LIMIT,
        help=f"Max issues to return (default {DEFAULT_LIST_LIMIT})",
    )
    p.add_argument("--sprint", default=None, help="Filter by sprint ID or 'active' for current sprint")
    p.add_argument("--board", default=None, help="Filter by board ID or 'active' for current board")
    p.add_argument("--focused", action="store_true", help="Smart filter: sprint for Scrum, board for Kanban")

    # -- transition --
    p = sub.add_parser("transition", help="Transition issue status")
    p.add_argument("ref")
    p.add_argument("status")
    p.add_argument("comment")
    p.add_argument(
        "--force",
        default=None,
        help="Provider-native status override (escape hatch for STATUS_NOT_REACHABLE)",
    )

    # -- assign --
    p = sub.add_parser("assign", help="Assign or unassign an issue")
    p.add_argument("ref")
    p.add_argument("assignee", nargs="?", default=None)
    p.add_argument("--unassign", action="store_true")

    # -- update --
    p = sub.add_parser("update", help="Update issue fields")
    p.add_argument("ref")
    p.add_argument("--title", default=None)
    p.add_argument("--priority", type=int, default=None)
    p.add_argument("--estimate", type=int, default=None)
    p.add_argument(
        "--assignee",
        default=None,
        help="Assignee name/ID/email, 'self'/'me' for current user, or 'clear' to unassign",
    )
    p.add_argument("--state", default=None)
    # FLY-1469: retired. Still parsed so `cmd_update` can answer with the
    # command that replaced it — argparse would answer `unrecognized
    # arguments`, which names no destination. See `cmd_update`.
    p.add_argument(
        "--description", default=None,
        help="RETIRED — use `issues.py description` (it carries the revision check)",
    )
    p.add_argument(
        "--description-file", "--file", dest="description_file", default=None,
        help="RETIRED — use `issues.py description --file` with --expected-revision",
    )
    p.add_argument("--labels", default=None)
    p.add_argument(
        "--milestone",
        default=None,
        help="Milestone ID or 'clear' to remove",
    )
    p.add_argument(
        "--due-date",
        dest="due_date",
        default=None,
        help="Due date in ISO 8601 format (YYYY-MM-DD) or 'clear' to remove",
    )
    p.add_argument("--comment", default=None)
    p.add_argument(
        "--project",
        default=None,
        help="Move issue to project (ID or name), or 'clear' to remove project",
    )

    # -- description --
    p = sub.add_parser("description", help="Update issue description")
    p.add_argument("ref")
    p.add_argument("--text", default=None)
    p.add_argument("--file", default=None)
    # FLY-1468: the §8 optimistic-concurrency token. Pass the `revision` from
    # the read this text was written against; omit it and the command reads the
    # current one, which still catches a writer landing mid-flight.
    p.add_argument(
        "--expected-revision", dest="expected_revision", default=None,
        help="Revision from the read this text was written against "
             "(`issues.py get REF` -> `revision`). The write is refused if the "
             "issue has changed since.",
    )
    # FLY-1470: the content guard. Stronger than the revision token, because
    # the token is a last-modified timestamp that also moves on a status change
    # touching no prose. Either flag satisfies the cloud `--file` refusal.
    p.add_argument(
        "--expected-description-hash", dest="expected_description_hash",
        default=None,
        help="Digest of the description this text was written against "
             "(`issues.py get REF` -> `descriptionHash`). The write is refused "
             "with DESCRIPTION_CHANGED if the prose has changed since — and, "
             "unlike --expected-revision, it is NOT refused when only the "
             "status or a field moved.",
    )

    # -- acceptance (FLY-1265) --
    p = sub.add_parser(
        "acceptance",
        help="Check / uncheck / defer / annotate acceptance criteria by number",
        description=(
            "Edit acceptance criteria by their number in the description "
            "(the numbering `issues.py get` returns under `acceptance`). "
            "The relay merges line-surgically under a revision check, so "
            "concurrent prose edits are preserved. Cloud tier only."
        ),
    )
    p.add_argument("ref")
    p.add_argument(
        "--check", action="append", default=None, metavar="N[,N...]",
        help="Mark criteria met (removes any deferral marker)",
    )
    p.add_argument(
        "--uncheck", action="append", default=None, metavar="N[,N...]",
        help="Mark criteria unmet",
    )
    p.add_argument(
        "--defer", action="append", default=None, metavar="N:REF",
        help="Defer a criterion to another issue, e.g. 5:FLY-1234 (box stays unchecked)",
    )
    p.add_argument(
        "--note", action="append", default=None, metavar="N:TEXT",
        help="Append a single-line note to a criterion (its own change — cannot share a criterion with a box change)",
    )

    # -- comment --
    p = sub.add_parser("comment", help="Add a comment to an issue")
    p.add_argument("ref")
    p.add_argument("body", nargs="?", default=None)

    # -- estimate --
    p = sub.add_parser("estimate", help="Set estimate points")
    p.add_argument("ref")
    p.add_argument("points", type=int)

    # -- priority --
    p = sub.add_parser("priority", help="Set priority level")
    p.add_argument("ref")
    p.add_argument("level", type=int)

    # -- link --
    p = sub.add_parser("link", help="Link two issues")
    p.add_argument("ref")
    p.add_argument("related_ref")
    p.add_argument("type", choices=["blocks", "related", "duplicate"])

    # -- assign-milestone --
    p = sub.add_parser("assign-milestone", help="Assign issue to milestone")
    p.add_argument("ref")
    p.add_argument("milestone_id")

    # -- assign-cycle (deprecated — use assign-sprint) --
    p = sub.add_parser("assign-cycle", help="[deprecated] Assign issue to cycle — use assign-sprint")
    p.add_argument("ref")
    p.add_argument("cycle_id", nargs="?", default=None)

    # -- assign-sprint (FLY-656) --
    p = sub.add_parser(
        "assign-sprint",
        help="Assign issue to sprint (supports id or current|next|previous)",
    )
    p.add_argument("ref")
    p.add_argument(
        "sprint_id",
        nargs="?",
        default=None,
        help="Sprint ID or alias: current, next, previous (omit to unassign)",
    )

    # -- pr --
    p = sub.add_parser("pr", help="Create pull/merge request with standard template")
    p.add_argument("--issue", default=None, help="Issue ref to link (e.g. FLY-123)")
    p.add_argument("--title", default=None, help="PR title (defaults to issue title or branch name)")
    p.add_argument("--base", default=None, help="Base branch (defaults to main/master)")
    p.add_argument("--changes", default=None, help="Changes items (newline- or semicolon-separated). Auto-filled from branch commits if omitted.")
    p.add_argument("--test-plan", dest="test_plan", default=None, help="Test Plan items (newline- or semicolon-separated)")
    p.add_argument("--notes", default=None, help="Notes for reviewers (trade-offs, follow-ups, risks)")
    p.add_argument("--draft", action="store_true", help="Create as draft PR")
    p.add_argument("--dry-run", action="store_true", help="Show what would be created without creating")

    # -- audit --
    p = sub.add_parser("audit", help="Audit issues for workflow compliance")
    p.add_argument("--status", default=None, help="Filter by status (default: all)")
    # FLY-1115: was a bare 50 — an audit that silently checks the first 50
    # issues reports a clean bill of health for a project it never fully read.
    p.add_argument("--limit", type=int, default=DEFAULT_LIST_LIMIT,
                   help=f"Max issues to check (default {DEFAULT_LIST_LIMIT})")
    # FLY-1436: the list projection carries no description and no labels, so
    # those checks report "not evaluated" without this. One request per issue,
    # hence opt-in.
    p.add_argument("--deep", action="store_true",
                   help="Read each issue in full so the description, acceptance "
                        "criteria and label checks can be evaluated")

    # -- fix --
    p = sub.add_parser("fix", help="Fix missing fields on an issue using config defaults")
    p.add_argument("ref")

    args = parser.parse_args()

    commands = {
        "create": cmd_create,
        "get": cmd_get,
        "list": cmd_list,
        "transition": cmd_transition,
        "assign": cmd_assign,
        "update": cmd_update,
        "description": cmd_description,
        "acceptance": cmd_acceptance,
        "comment": cmd_comment,
        "estimate": cmd_estimate,
        "priority": cmd_priority,
        "link": cmd_link,
        "assign-milestone": cmd_assign_milestone,
        "assign-cycle": cmd_assign_cycle,
        "assign-sprint": cmd_assign_sprint,
        "pr": cmd_pr,
        "audit": cmd_audit,
        "fix": cmd_fix,
    }
    commands[args.command](args)


if __name__ == "__main__":
    main()