#!/usr/bin/env python3
"""FlyDocs bridge — one JSON envelope in, one JSON envelope out (FLY-1293).

The MCP server (FLY-923/924/925) needs to reach the workflow dispatchers
without shelling out to `issues.py transition FLY-1 REVIEW "…"` and scraping
prose off stdout. This module is that boundary: a versioned, schema-checked
request/result protocol over a subprocess, spoken by `src/lib/bridge/` on the
TypeScript side.

Usage
-----

    echo '<envelope>' | python3 bridge.py          # dispatch one request
    python3 bridge.py --describe                   # protocol + operation table

Exactly one JSON object is written to stdout. Everything else — the human
notes the dispatchers print, tracebacks, anything a library decides to log —
goes to stderr, and the notes are folded into the result's `warnings`.
The process exits 0 whenever it produced a result envelope, including a
failing one: an `ok: false` envelope is a successful conversation about a
failed operation. Exit 1 means no envelope could be produced at all.

Design rules
------------

**Nothing is inferred.** The request carries `context.repoRoot` and
`context.sessionId`; the bridge chdirs to the former and refuses to guess.
The dispatchers resolve config by walking up from the cwd, so setting the cwd
from an explicit field is precisely how the inference is removed — an MCP
server serving three repos cannot rely on its own process cwd.

**Existing implementations are called, not re-implemented.** Every operation
builds the `argparse.Namespace` the dispatcher's own `cmd_*` function already
expects and calls it. There is no second copy of the transition rules, the
acceptance guards or the wrap validation, and a fix in `issues.py` is a fix
here for free. The cost is that stdout has to be captured and re-parsed, which
is cheap and honest.

**Errors keep their codes.** `fail()` renders a lifecycle code as a sentence
and exits; the bridge reads the structured error the relay recorded
(`flydocs_api.take_last_relay_error`) and returns `REVISION_MISMATCH` as a
code. Bridge-local failures get their own codes, listed in `ERROR_CODES`.

Protocol versioning
-------------------

`protocolVersion` is `MAJOR.MINOR`.

* **MINOR** covers additive change only: a new operation, a new *optional*
  request parameter, a new result field, a new error code. A client on 1.0
  keeps working against a 1.4 bridge, because everything 1.4 added is
  optional and every result reader tolerates fields it does not know.
* **MAJOR** covers everything else: removing or renaming a parameter,
  changing a type, making an optional parameter required, changing the
  meaning of a result field.
* The bridge accepts a request whose major matches and whose minor is **at or
  below** its own. A higher minor is rejected with `PROTOCOL_MISMATCH` rather
  than best-guessed, because request validation is strict — a newer client's
  new optional parameter would otherwise be reported as an unknown field,
  which sends the reader looking for a typo instead of a version gap.
* Requests reject unknown fields. Results tolerate them. That asymmetry is
  what makes minor bumps safe in both directions: an old client ignores a new
  result field, and a new client's new request field is refused loudly by an
  old bridge instead of being silently dropped.
* `--describe` is the machine-readable form of everything above and is what a
  future TypeScript client (FLY-921) checks itself against.
"""

import argparse
import contextlib
import io
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional

# 1.1 (FLY-1294) — additive: `graph.change_context` and the `NOT_AVAILABLE`
# code. A client on 1.0 keeps working unchanged; a client on 1.1 talking to a
# 1.0 bridge is refused with PROTOCOL_MISMATCH rather than told that an
# operation it can see in its own roster does not exist.
PROTOCOL_VERSION = "1.1"

# Python floor, declared and enforced (agent-platform-v2-spec.md §4.2).
#
# The check runs *before* the dispatchers are imported, and that ordering is
# the whole point: `flydocs_api` annotates with `Path | None`, which a 3.9
# interpreter rejects at import time. Checking after the import would mean a
# 3.9 workspace never reaches this message — it would get a TypeError from
# inside a module it has no reason to have heard of. The envelope is written
# out by hand here because none of the helpers exist yet.
MIN_PYTHON = (3, 10)

if sys.version_info < MIN_PYTHON:
    _floor = ".".join(str(part) for part in MIN_PYTHON)
    _running = ".".join(str(part) for part in sys.version_info[:3])
    print(json.dumps({
        "protocolVersion": PROTOCOL_VERSION,
        "operation": None,
        "operationId": None,
        "ok": False,
        "error": {
            "code": "PROTOCOL_MISMATCH",
            "message": f"FlyDocs requires Python {_floor}+, this is {_running}.",
            "retryable": False,
            "details": {"minPython": _floor, "running": _running},
        },
        "warnings": [],
    }))
    raise SystemExit(1)

sys.path.insert(0, str(Path(__file__).parent))

import change_context as change_context_lib  # noqa: E402
import issues as issues_cmd  # noqa: E402
import session as session_cmd  # noqa: E402
from flydocs_api import (  # noqa: E402
    DEFAULT_LIST_LIMIT,
    RelayError,
    set_operation_seed,
    take_last_relay_error,
)


# ---------------------------------------------------------------------------
# Error codes
# ---------------------------------------------------------------------------
#
# Two families, one namespace. Relay lifecycle codes pass through untouched
# (relay-lifecycle-authority-spec.md §11) so a caller matching on
# `REVISION_MISMATCH` matches the same string whether it came through the
# bridge or a direct HTTP call. Bridge-local codes name failures that happen
# before, around or instead of a relay call, and cannot collide: no relay code
# describes a subprocess.

RELAY_ERROR_CODES = (
    "TRANSITION_ILLEGAL",
    "ASSIGNMENT_REQUIRED",
    "ACCEPTANCE_INCOMPLETE",
    "REVISION_MISMATCH",
    "REVISION_UNAVAILABLE",
    "CRITERION_MISMATCH",
    "OPERATION_IN_FLIGHT",
    "OPERATION_ID_REUSED",
    "OPERATION_ID_REQUIRED",
    "POLICY_OVERRIDE_DENIED",
    "VALIDATION_ERROR",
    "INVALID_STATUS",
    "STATUS_NOT_MAPPED",
    "STATUS_NOT_REACHABLE",
    "STATUS_MAPPING_ERROR",
    "UNMAPPED_CURRENT_STATE",
    "PROVIDER_AUTH_FAILED",
    "PROVIDER_TOKEN_REFRESH_FAILED",
    "NETWORK_ERROR",
    "UNKNOWN",
)

# Bridge-local codes. `TIMEOUT`, `CANCELED`, `INTERPRETER_NOT_FOUND`,
# `SPAWN_FAILED` and `OUTPUT_LIMIT_EXCEEDED` are produced by the TypeScript
# spawn layer rather than here — the process that fails to start, hangs or
# floods cannot report on itself — but they are declared in one place so the
# code table has a single home.
BRIDGE_ERROR_CODES = (
    "PROTOCOL_MISMATCH",     # unsupported protocolVersion
    "INVALID_REQUEST",       # envelope failed schema validation
    "UNKNOWN_OPERATION",     # operation not in the v1 set
    "NOT_AVAILABLE",         # the operation is real; this mode has no data yet
    "CONTEXT_INVALID",       # repoRoot missing, not a directory, or not FlyDocs
    "OPERATION_FAILED",      # the dispatcher exited non-zero with no relay code
    "INVALID_RESULT",        # the dispatcher printed something unparseable
    "BRIDGE_CRASHED",        # unhandled exception inside the bridge
    "TIMEOUT",               # spawn layer: wall-clock budget exhausted
    "CANCELED",              # spawn layer: caller aborted the call
    "INTERPRETER_NOT_FOUND",  # spawn layer: no usable Python
    "SPAWN_FAILED",          # spawn layer: exec failed for another reason
    "OUTPUT_LIMIT_EXCEEDED",  # spawn layer: output cap hit, result unusable
)

ERROR_CODES = RELAY_ERROR_CODES + BRIDGE_ERROR_CODES

# Codes worth retrying with the same envelope. The operation id makes that
# safe: a replay of an already-applied mutation returns the stored outcome
# instead of writing twice (FLY-1263 §7.2).
RETRYABLE_CODES = frozenset({
    "OPERATION_IN_FLIGHT",
    "NETWORK_ERROR",
    "TIMEOUT",
})


class BridgeError(Exception):
    """A failure with a code, on its way to an `ok: false` envelope."""

    def __init__(self, code: str, message: str,
                 details: Optional[dict] = None):
        super().__init__(message)
        self.code = code
        self.message = message
        self.details = details or {}


# ---------------------------------------------------------------------------
# Parameter specs
# ---------------------------------------------------------------------------
#
# The strict schemas live on the TypeScript side (zod). This table is the
# Python half of the same contract — deliberately duplicated, because a bridge
# that trusts its caller is a bridge with no contract, and the golden fixtures
# in the CLI repo's `tests/bridge/fixtures/` are what prove the two halves
# still agree.

class Param:
    """One request parameter: type, requiredness, default, allowed values."""

    def __init__(self, kind: str, *, required: bool = False,
                 default: Any = None, choices: Optional[tuple] = None,
                 minimum: Optional[int] = None, doc: str = ""):
        self.kind = kind
        self.required = required
        self.default = default
        self.choices = choices
        self.minimum = minimum
        self.doc = doc

    def describe(self) -> dict:
        spec: dict = {"type": self.kind, "required": self.required}
        if self.default is not None:
            spec["default"] = self.default
        if self.choices:
            spec["choices"] = list(self.choices)
        if self.minimum is not None:
            spec["minimum"] = self.minimum
        if self.doc:
            spec["doc"] = self.doc
        return spec


_TYPE_CHECKS: dict[str, Callable[[Any], bool]] = {
    "string": lambda v: isinstance(v, str),
    "integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
    "boolean": lambda v: isinstance(v, bool),
    "string[]": lambda v: isinstance(v, list) and all(isinstance(x, str) for x in v),
    "object[]": lambda v: isinstance(v, list) and all(isinstance(x, dict) for x in v),
}


class Operation:
    """One bridge operation and how it reaches the dispatcher behind it."""

    def __init__(self, name: str, *, mutating: bool, params: dict[str, Param],
                 handler: Callable[[dict], Any], summary: str):
        self.name = name
        self.mutating = mutating
        self.params = params
        self.handler = handler
        self.summary = summary

    def describe(self) -> dict:
        return {
            "operation": self.name,
            "mutating": self.mutating,
            "summary": self.summary,
            "params": {k: v.describe() for k, v in self.params.items()},
        }

    def validate(self, raw: dict) -> dict:
        """Check `raw` against the spec and fill defaults.

        Unknown parameters are an error, not noise: the one thing a caller can
        do that silently does nothing is misspell a parameter, and that is
        exactly the failure a strict request schema exists to prevent.
        """
        unknown = sorted(set(raw) - set(self.params))
        if unknown:
            raise BridgeError(
                "INVALID_REQUEST",
                f"{self.name}: unknown parameter(s) {', '.join(unknown)}. "
                f"Accepted: {', '.join(sorted(self.params))}.",
                {"operation": self.name, "unknown": unknown},
            )

        resolved: dict = {}
        for name, spec in self.params.items():
            if name not in raw or raw[name] is None:
                if spec.required:
                    raise BridgeError(
                        "INVALID_REQUEST",
                        f"{self.name}: '{name}' is required.",
                        {"operation": self.name, "param": name},
                    )
                resolved[name] = spec.default
                continue

            value = raw[name]
            check = _TYPE_CHECKS[spec.kind]
            if not check(value):
                raise BridgeError(
                    "INVALID_REQUEST",
                    f"{self.name}: '{name}' must be {spec.kind}, got "
                    f"{type(value).__name__}.",
                    {"operation": self.name, "param": name},
                )
            if spec.required and spec.kind == "string" and not value.strip():
                # Mirrors `z.string().min(1)` on the TypeScript side. Catching
                # it here means the caller is told which parameter was empty,
                # instead of reading a dispatcher message about piping to
                # stdin — advice that means nothing to a tool call.
                raise BridgeError(
                    "INVALID_REQUEST",
                    f"{self.name}: '{name}' must not be empty."
                    + (f" {spec.doc}." if spec.doc else ""),
                    {"operation": self.name, "param": name},
                )
            if spec.choices is not None and value not in spec.choices:
                raise BridgeError(
                    "INVALID_REQUEST",
                    f"{self.name}: '{name}' must be one of "
                    f"{', '.join(str(c) for c in spec.choices)}, got {value!r}.",
                    {"operation": self.name, "param": name},
                )
            if spec.minimum is not None and value < spec.minimum:
                # Mirrors `z.number().int().min(n)` on the TypeScript side.
                raise BridgeError(
                    "INVALID_REQUEST",
                    f"{self.name}: '{name}' must be at least {spec.minimum}, "
                    f"got {value!r}.",
                    {"operation": self.name, "param": name},
                )
            resolved[name] = value

        return resolved


# ---------------------------------------------------------------------------
# Dispatcher invocation
# ---------------------------------------------------------------------------

def _warnings_from(stderr_text: str) -> list[str]:
    """The dispatchers' stderr notes, as a list a tool caller can surface.

    These are the "Note: returned 250 of 613 issues" lines — advice the human
    path prints and the subprocess path has been throwing away.
    """
    return [line.strip() for line in stderr_text.splitlines() if line.strip()]


def _parse_output(stdout_text: str) -> Any:
    """Read the dispatcher's JSON result off captured stdout.

    `output_json` prints exactly one line, but a handler is free to print
    advice above it, so the last JSON-parseable line wins rather than the
    whole buffer.
    """
    lines = [ln for ln in stdout_text.splitlines() if ln.strip()]
    for line in reversed(lines):
        try:
            return json.loads(line)
        except json.JSONDecodeError:
            continue
    raise BridgeError(
        "INVALID_RESULT",
        "The dispatcher produced no JSON result.",
        {"stdout": stdout_text[-2000:]},
    )


def _invoke(handler: Callable[[argparse.Namespace], None],
            namespace: argparse.Namespace) -> tuple[Any, list[str]]:
    """Run one `cmd_*` function and turn its console output into a value.

    Three exits are possible and all three are handled here:
    a normal return (parse stdout), a `fail()` (SystemExit — recover the relay
    code if there was one, otherwise report the message it printed), and an
    unhandled exception (surfaced as `BRIDGE_CRASHED` by the caller).
    """
    take_last_relay_error()  # discard anything a previous call left behind
    out, err = io.StringIO(), io.StringIO()

    try:
        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
            handler(namespace)
    except SystemExit as exc:
        code = exc.code if isinstance(exc.code, int) else 1
        warnings = _warnings_from(err.getvalue())
        if code == 0:
            return _parse_output(out.getvalue()), warnings
        relay_error = take_last_relay_error()
        message = err.getvalue().strip() or f"exited with status {code}"
        if relay_error is not None:
            raise BridgeError(
                relay_error.code,
                relay_error.message or message,
                {
                    "status": relay_error.status,
                    "body": relay_error.body,
                    "rendered": message,
                    "warnings": warnings,
                },
            ) from exc
        # The message *is* the stderr on this path, so anything already inside
        # it would otherwise be reported twice — once as the failure and once
        # as advice. Notes printed before the failure survive.
        raise BridgeError(
            "OPERATION_FAILED", message,
            {"warnings": [w for w in warnings if w not in message]},
        ) from exc
    except RelayError as exc:
        raise BridgeError(
            exc.code, exc.message,
            {
                "status": exc.status,
                "body": exc.body,
                "warnings": _warnings_from(err.getvalue()),
            },
        ) from exc

    return _parse_output(out.getvalue()), _warnings_from(err.getvalue())


def _ns(**fields: Any) -> argparse.Namespace:
    """An argparse namespace with every attribute the handler will read.

    Written out per operation rather than generated: the dispatchers use
    `args.x` freely, and a missing attribute is an AttributeError at call
    time, so the full set belongs where a reader can check it against the
    parser definition.
    """
    return argparse.Namespace(**fields)


# ---------------------------------------------------------------------------
# Operation handlers
# ---------------------------------------------------------------------------

def _op_issue_get(p: dict) -> dict:
    data, warnings = _invoke(
        issues_cmd.cmd_get, _ns(ref=p["ref"], fields=p["fields"]),
    )
    return {"data": data, "warnings": warnings}


def _op_issue_list(p: dict) -> dict:
    data, warnings = _invoke(issues_cmd.cmd_list, _ns(
        status=p["status"],
        active=p["active"],
        project=p["project"],
        assignee=p["assignee"],
        milestone=p["milestone"],
        mine=p["mine"],
        show_all=p["all"],
        limit=p["limit"],
        sprint=p["sprint"],
        board=p["board"],
        focused=p["focused"],
    ))
    return {"data": {"issues": data, "count": len(data) if isinstance(data, list) else None},
            "warnings": warnings}


def _op_issue_create(p: dict) -> dict:
    data, warnings = _invoke(issues_cmd.cmd_create, _ns(
        title=p["title"],
        type=p["type"],
        description=p["description"],
        # Deliberately not exposed: --description-file and --template read
        # paths relative to the dispatcher's cwd. A tool caller that can name
        # a file can read it, and passing text keeps the boundary free of
        # filesystem semantics it would have to sandbox.
        description_file=None,
        template=False,
        priority=p["priority"],
        estimate=p["estimate"],
        assignee=p["assignee"],
        project=p["project"],
        milestone=p["milestone"],
        triage=p["triage"],
    ))
    return {"data": data, "warnings": warnings}


def _op_issue_transition(p: dict) -> dict:
    data, warnings = _invoke(issues_cmd.cmd_transition, _ns(
        ref=p["ref"], status=p["status"], comment=p["comment"], force=p["force"],
    ))
    return {"data": data, "warnings": warnings}


def _op_issue_assign(p: dict) -> tuple[Any, list[str]]:
    return _invoke(issues_cmd.cmd_assign, _ns(
        ref=p["ref"], assignee=p["assignee"], unassign=False,
    ))


def _op_issue_activate(p: dict) -> dict:
    """Assign, transition and land session state as one outcome.

    The roster's flagship flow could not be completed by any single tool
    (agent-platform-v2-spec.md §4.2): assignment had no tool at all, and the
    relay gates IMPLEMENTING on it. Composing here rather than adding a new
    dispatcher subcommand keeps the composition in the layer that owns
    multi-step outcomes, and each leg keeps its own operation id because the
    ids are derived per relay call.

    Not a transaction. A failure after assignment leaves the issue assigned
    and unmoved, which is the recoverable half — re-running the same envelope
    replays the assignment and retries the transition.
    """
    warnings: list[str] = []
    assigned, assign_warnings = _op_issue_assign(p)
    warnings.extend(assign_warnings)

    transitioned, transition_warnings = _invoke(issues_cmd.cmd_transition, _ns(
        ref=p["ref"], status=p["status"], comment=p["comment"], force=None,
    ))
    warnings.extend(transition_warnings)

    session_dir = issues_cmd._resolve_session_dir()
    return {
        "data": {
            "issue": p["ref"].upper(),
            "assigned": assigned,
            "transitioned": transitioned,
            "sessionState": {
                "dir": str(session_dir),
                "focus": (session_dir / "focus.md").exists(),
                "status": _read_if_present(session_dir / "status"),
            },
        },
        "warnings": warnings,
    }


def _read_if_present(path: Path) -> Optional[str]:
    try:
        return path.read_text().strip()
    except OSError:
        return None


def _op_issue_comment(p: dict) -> dict:
    data, warnings = _invoke(issues_cmd.cmd_comment, _ns(
        ref=p["ref"], body=p["body"],
    ))
    return {"data": data, "warnings": warnings}


_ACCEPTANCE_STATUSES = ("checked", "unchecked", "deferred", "note")


def _op_issue_acceptance_update(p: dict) -> dict:
    """Semantic acceptance edits, expressed as the flags `cmd_acceptance` parses.

    The tool surface is `changes: [{criterionId, status, …}]` (§4.2 rev 2),
    which is the shape the relay route already takes. `cmd_acceptance` owns
    the read, the text guards, the revision check and the one-shot retry, so
    the changes are rendered back into its flag vocabulary rather than
    duplicating any of that here.
    """
    check: list[str] = []
    uncheck: list[str] = []
    defer: list[str] = []
    note: list[str] = []

    for index, change in enumerate(p["changes"]):
        unknown = sorted(set(change) - {"criterionId", "status", "deferredTo", "note"})
        if unknown:
            raise BridgeError(
                "INVALID_REQUEST",
                f"changes[{index}]: unknown field(s) {', '.join(unknown)}.",
                {"index": index, "unknown": unknown},
            )
        criterion = change.get("criterionId")
        status = change.get("status")
        if not isinstance(criterion, int) or isinstance(criterion, bool) or criterion < 1:
            raise BridgeError(
                "INVALID_REQUEST",
                f"changes[{index}]: 'criterionId' must be a positive integer "
                "(the number `issue.get` returns under `acceptance`).",
                {"index": index},
            )
        if status not in _ACCEPTANCE_STATUSES:
            raise BridgeError(
                "INVALID_REQUEST",
                f"changes[{index}]: 'status' must be one of "
                f"{', '.join(_ACCEPTANCE_STATUSES)}.",
                {"index": index},
            )

        if status == "checked":
            check.append(str(criterion))
        elif status == "unchecked":
            uncheck.append(str(criterion))
        elif status == "deferred":
            destination = change.get("deferredTo")
            if not isinstance(destination, str) or not destination.strip():
                raise BridgeError(
                    "INVALID_REQUEST",
                    f"changes[{index}]: 'deferredTo' is required for a deferral "
                    "— a deferral with no destination is an unfinished criterion "
                    "with better manners (FLY-1087).",
                    {"index": index},
                )
            defer.append(f"{criterion}:{destination.strip()}")
        else:
            text = change.get("note")
            if not isinstance(text, str) or not text.strip():
                raise BridgeError(
                    "INVALID_REQUEST",
                    f"changes[{index}]: 'note' text is required for a note change.",
                    {"index": index},
                )
            note.append(f"{criterion}:{text.strip()}")

    data, warnings = _invoke(issues_cmd.cmd_acceptance, _ns(
        ref=p["ref"],
        check=check or None,
        uncheck=uncheck or None,
        defer=defer or None,
        note=note or None,
    ))
    return {"data": data, "warnings": warnings}


def _op_session_start(p: dict) -> dict:
    data, warnings = _invoke(session_cmd.cmd_start_context, _ns())
    return {"data": data, "warnings": warnings}


def _op_session_wrap(p: dict) -> dict:
    # FLY-1498: `summary` is the spelling the tool surface uses, `notes` the
    # one the CLI has always had, and they are the same field — the record's
    # `summary`. `summary` wins when a caller sends both, because it is the
    # one the current schema describes.
    data, warnings = _invoke(session_cmd.cmd_wrap, _ns(
        issues=p["issues"],
        health=p["health"],
        notes=p["summary"] or p["notes"],
        title=p["title"],
        pending=p["pending"],
        blockers=p["blockers"],
        body=p["body"],
        body_file=None,
        project=p["project"],
    ))
    return {"data": data, "warnings": warnings}


def _op_project_update(p: dict) -> dict:
    data, warnings = _invoke(session_cmd.cmd_project_update, _ns(
        health=p["health"], body=p["body"], body_file=None, project=p["project"],
    ))
    return {"data": data, "warnings": warnings}


# The one dependency `impact` is waiting on, named in one place so the two
# refusals below cannot drift apart.
_GRAPH_GATE = "Phase 8 — Graph Pilot & Review Lane (graph correctness)"


def _now() -> str:
    """The observation time a slice is stamped with, to the second, in UTC."""
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
        "+00:00", "Z",
    )


def _op_graph_change_context(p: dict) -> dict:
    """The shipped-mode join: issue → PRs → related issues → decisions.

    Reads through `issues.py cmd_get` — the same function `issue.get` calls —
    so there is one issue read path in this process, not two. Everything after
    the read is `change_context.py`, which is pure and holds the §5.5
    provenance rules.

    `impact` and non-issue targets both refuse rather than approximate. The
    honest answer to "what would this change break" is a graph read, the graph's
    correctness fix is Phase 8 work, and a blast radius assembled from a stale
    local graph would be a set of claims whose sources do not support them —
    which is the one thing §5.5 forbids outright.
    """
    target = p["target"].strip()
    mode = p["mode"]
    limit = p["limit"]

    problem = change_context_lib.check_limit(limit)
    if problem:
        raise BridgeError(
            "INVALID_REQUEST", f"graph.change_context: {problem}",
            {"operation": "graph.change_context", "param": "limit"},
        )

    if mode == "impact":
        raise BridgeError(
            "NOT_AVAILABLE",
            "change_context does not serve 'impact' yet. A blast radius is a "
            "graph read, and the graph's correctness fix is "
            f"{_GRAPH_GATE}; answering it from today's local graph would "
            "produce claims their sources do not support. Use mode 'shipped' "
            "for what an issue actually records.",
            {"mode": mode, "target": target, "dependency": _GRAPH_GATE,
             "availableModes": ["shipped"]},
        )

    if not change_context_lib.is_issue_ref(target):
        raise BridgeError(
            "NOT_AVAILABLE",
            f"'{target}' is not an issue reference (e.g. FLY-123), and "
            "resolving a module or topic to the work that touched it is the "
            f"same graph read that {_GRAPH_GATE} is fixing. Pass an issue "
            "reference, or find one with issue.list first.",
            {"mode": mode, "target": target, "dependency": _GRAPH_GATE,
             "resolves": ["issueRef"]},
        )

    issue, warnings = _invoke(issues_cmd.cmd_get, _ns(
        ref=change_context_lib.normalize_ref(target), fields="full",
    ))
    slice_, slice_warnings = change_context_lib.build_shipped_slice(
        issue, target=target, observed_at=_now(), limit=limit,
    )
    return {"data": slice_, "warnings": warnings + slice_warnings}


# ---------------------------------------------------------------------------
# The v1 operation set
# ---------------------------------------------------------------------------

_HEALTH = ("onTrack", "atRisk", "offTrack")

OPERATIONS: dict[str, Operation] = {
    op.name: op for op in (
        Operation(
            "issue.get", mutating=False, handler=_op_issue_get,
            summary="Read one issue, including its parsed acceptance criteria.",
            params={
                "ref": Param("string", required=True, doc="Issue reference, e.g. FLY-123"),
                "fields": Param("string", default="full", choices=("basic", "full")),
            },
        ),
        Operation(
            "issue.list", mutating=False, handler=_op_issue_list,
            summary="List issues with focus-aware filters.",
            params={
                "status": Param("string"),
                "active": Param("boolean", default=False),
                "project": Param("string"),
                "assignee": Param("string"),
                "milestone": Param("string"),
                "mine": Param("boolean", default=False),
                "all": Param("boolean", default=False,
                             doc="Bypass the product scope cascade"),
                "limit": Param("integer", default=DEFAULT_LIST_LIMIT),
                "sprint": Param("string", doc="Sprint id, or 'active'"),
                "board": Param("string", doc="Board id, or 'active'"),
                "focused": Param("boolean", default=False),
            },
        ),
        Operation(
            "issue.create", mutating=True, handler=_op_issue_create,
            summary="Create an issue from typed template fields.",
            params={
                "title": Param("string", required=True),
                "type": Param("string", required=True,
                              choices=("feature", "bug", "chore", "idea")),
                "description": Param("string",
                                     doc="Required unless triage is true"),
                "priority": Param("integer", choices=(0, 1, 2, 3, 4)),
                "estimate": Param(
                    "integer", minimum=0,
                    doc="Points on the provider scale — "
                        "workspace.get-estimate-scale reports it",
                ),
                "assignee": Param("string", doc="Provider id, or 'me'"),
                "project": Param("string"),
                "milestone": Param("string"),
                "triage": Param("boolean", default=False),
            },
        ),
        Operation(
            "issue.activate", mutating=True, handler=_op_issue_activate,
            summary="Assign, transition and record session state as one outcome.",
            params={
                "ref": Param("string", required=True),
                "assignee": Param("string", required=True, doc="Provider id, or 'me'"),
                "comment": Param("string", required=True,
                                 doc="Why the work is starting — required by the transition"),
                "status": Param("string", default="IMPLEMENTING",
                                doc="Target status; the default is the activation flow"),
            },
        ),
        Operation(
            "issue.transition", mutating=True, handler=_op_issue_transition,
            summary="Move an issue to a canonical status, with the mandatory comment.",
            params={
                "ref": Param("string", required=True),
                "status": Param("string", required=True,
                                doc="Canonical FlyDocs status, e.g. IMPLEMENTING"),
                "comment": Param("string", required=True,
                                 doc="Never optional — no status moves silently"),
                "force": Param("string",
                               doc="Provider-native override for STATUS_NOT_REACHABLE"),
            },
        ),
        Operation(
            "issue.comment", mutating=True, handler=_op_issue_comment,
            summary="Post a comment on an issue.",
            params={
                "ref": Param("string", required=True),
                "body": Param("string", required=True),
            },
        ),
        Operation(
            "issue.acceptance_update", mutating=True,
            handler=_op_issue_acceptance_update,
            summary="Check, uncheck, defer or annotate acceptance criteria by number.",
            params={
                "ref": Param("string", required=True),
                "changes": Param(
                    "object[]", required=True,
                    doc="[{criterionId, status: checked|unchecked|deferred|note, "
                        "deferredTo?, note?}]",
                ),
            },
        ),
        Operation(
            "session.start", mutating=False, handler=_op_session_start,
            summary="Gather identity, focus, config and session state in one read.",
            params={},
        ),
        Operation(
            "session.wrap", mutating=True, handler=_op_session_wrap,
            summary="Write the handoff summary, post the project update, clear state.",
            params={
                "issues": Param("string[]", default=[]),
                "health": Param("string", choices=_HEALTH,
                                doc="Omit to wrap without posting a project update"),
                "notes": Param("string", default="",
                               doc="Legacy spelling of `summary`; `summary` "
                                   "wins when both are sent"),
                "title": Param("string",
                               doc="One line for a teammate naming what the "
                                   "session did, max 120 characters"),
                "summary": Param("string",
                                 doc="2-3 plain sentences for a teammate: "
                                     "what changed and what it means, max "
                                     "600 characters"),
                "pending": Param("string[]", default=[]),
                "blockers": Param("string[]", default=[]),
                "body": Param("string",
                              doc="Filled session-wrap template; required sections are enforced"),
                "project": Param("string"),
            },
        ),
        Operation(
            "graph.change_context", mutating=False,
            handler=_op_graph_change_context,
            summary="The provenanced shipped-mode slice: issue, its pull "
                    "requests, related issues and decisions.",
            params={
                "target": Param("string", required=True,
                                doc="Issue reference, e.g. FLY-123"),
                "mode": Param("string", default="shipped",
                              choices=change_context_lib.MODES,
                              doc="shipped: what an issue records. "
                                  "impact: blast radius (not available yet)"),
                "limit": Param("integer",
                               default=change_context_lib.DEFAULT_LIMIT,
                               doc="Per-section cap: pull requests, related "
                                   "issues, decisions"),
            },
        ),
        Operation(
            "project.update", mutating=True, handler=_op_project_update,
            summary="Post a standalone mid-session project update.",
            params={
                "health": Param("string", required=True, choices=_HEALTH),
                "body": Param("string", required=True),
                "project": Param("string"),
            },
        ),
    )
}


# ---------------------------------------------------------------------------
# Envelope handling
# ---------------------------------------------------------------------------

_ENVELOPE_FIELDS = {"protocolVersion", "operation", "operationId", "context", "params"}
_CONTEXT_FIELDS = {"repoRoot", "workspaceRoot", "sessionId"}


def _parse_version(raw: str) -> tuple[int, int]:
    major, _, minor = raw.partition(".")
    return int(major), int(minor)


def _check_protocol(raw: Any) -> None:
    if not isinstance(raw, str):
        raise BridgeError(
            "PROTOCOL_MISMATCH",
            f"protocolVersion must be a MAJOR.MINOR string; this bridge speaks "
            f"{PROTOCOL_VERSION}.",
            {"supported": PROTOCOL_VERSION},
        )
    try:
        major, minor = _parse_version(raw)
    except ValueError:
        raise BridgeError(
            "PROTOCOL_MISMATCH",
            f"protocolVersion '{raw}' is not MAJOR.MINOR; this bridge speaks "
            f"{PROTOCOL_VERSION}.",
            {"supported": PROTOCOL_VERSION, "requested": raw},
        ) from None

    own_major, own_minor = _parse_version(PROTOCOL_VERSION)
    if major != own_major:
        raise BridgeError(
            "PROTOCOL_MISMATCH",
            f"Protocol major {major} is not supported; this bridge speaks "
            f"{PROTOCOL_VERSION}. Update the CLI and the workspace template "
            "together — a major bump is a breaking change on both sides.",
            {"supported": PROTOCOL_VERSION, "requested": raw},
        )
    if minor > own_minor:
        raise BridgeError(
            "PROTOCOL_MISMATCH",
            f"Client speaks {raw}, this bridge speaks {PROTOCOL_VERSION}. The "
            "extra minor version may carry request fields this bridge would "
            "reject as unknown. Run `flydocs update` to reseed the workspace "
            "scripts.",
            {"supported": PROTOCOL_VERSION, "requested": raw},
        )


def _check_context(raw: Any) -> dict:
    """Validate the explicit repo/session context and enter the repo.

    Every dispatcher resolves config by walking up from the cwd. Pointing the
    cwd at `repoRoot` is what turns that walk from an inference into an
    instruction — and it is why `repoRoot` is required rather than optional
    with a cwd fallback.
    """
    if not isinstance(raw, dict):
        raise BridgeError("INVALID_REQUEST", "context must be an object.")

    unknown = sorted(set(raw) - _CONTEXT_FIELDS)
    if unknown:
        raise BridgeError(
            "INVALID_REQUEST",
            f"context: unknown field(s) {', '.join(unknown)}. Accepted: "
            f"{', '.join(sorted(_CONTEXT_FIELDS))}.",
            {"unknown": unknown},
        )

    repo_root = raw.get("repoRoot")
    session_id = raw.get("sessionId")
    for name, value in (("repoRoot", repo_root), ("sessionId", session_id)):
        if not isinstance(value, str) or not value.strip():
            raise BridgeError(
                "INVALID_REQUEST",
                f"context.{name} is required — the bridge infers neither the "
                "repo nor the session from its environment.",
                {"param": name},
            )

    workspace_root = raw.get("workspaceRoot")
    if workspace_root is not None and not isinstance(workspace_root, str):
        raise BridgeError("INVALID_REQUEST", "context.workspaceRoot must be a string.")

    path = Path(repo_root)
    if not path.is_dir():
        raise BridgeError(
            "CONTEXT_INVALID",
            f"context.repoRoot is not a directory: {repo_root}",
            {"repoRoot": repo_root},
        )
    if not any((parent / ".flydocs").is_dir() for parent in (path, *path.parents)):
        raise BridgeError(
            "CONTEXT_INVALID",
            f"No .flydocs directory at or above {repo_root} — this is not a "
            "FlyDocs repo. Run `flydocs init` there first.",
            {"repoRoot": repo_root},
        )

    os.chdir(path)
    return {
        "repoRoot": str(path),
        "workspaceRoot": workspace_root,
        "sessionId": session_id,
    }


def handle(envelope: Any) -> dict:
    """Validate one request envelope, run it, and return the result envelope.

    Never raises for an operational failure — every path produces an envelope,
    because the caller reads stdout, not exceptions.
    """
    operation_name = None
    operation_id = None
    try:
        if not isinstance(envelope, dict):
            raise BridgeError(
                "INVALID_REQUEST",
                f"Request must be a JSON object, got {type(envelope).__name__}.",
            )

        unknown = sorted(set(envelope) - _ENVELOPE_FIELDS)
        if unknown:
            raise BridgeError(
                "INVALID_REQUEST",
                f"Envelope: unknown field(s) {', '.join(unknown)}. Accepted: "
                f"{', '.join(sorted(_ENVELOPE_FIELDS))}.",
                {"unknown": unknown},
            )

        _check_protocol(envelope.get("protocolVersion"))

        operation_name = envelope.get("operation")
        if not isinstance(operation_name, str) or not operation_name:
            raise BridgeError("INVALID_REQUEST", "operation is required.")
        operation = OPERATIONS.get(operation_name)
        if operation is None:
            raise BridgeError(
                "UNKNOWN_OPERATION",
                f"'{operation_name}' is not a v1 operation. Supported: "
                f"{', '.join(sorted(OPERATIONS))}.",
                {"supported": sorted(OPERATIONS)},
            )

        operation_id = envelope.get("operationId")
        if operation_id is not None and not isinstance(operation_id, str):
            raise BridgeError("INVALID_REQUEST", "operationId must be a string.")
        if operation.mutating and not (operation_id or "").strip():
            raise BridgeError(
                "INVALID_REQUEST",
                f"{operation_name} is a mutating operation and requires "
                "operationId — it is what lets a retry replay the first "
                "outcome instead of writing twice.",
                {"operation": operation_name},
            )

        context = _check_context(envelope.get("context"))

        raw_params = envelope.get("params", {})
        if raw_params is None:
            raw_params = {}
        if not isinstance(raw_params, dict):
            raise BridgeError("INVALID_REQUEST", "params must be an object.")
        params = operation.validate(raw_params)

        # Reads carry no seed: keying a read would create operation records for
        # traffic that changes nothing (FLY-1265). An operationId sent on a
        # read is still echoed, so a caller can correlate a read with the
        # write it preceded.
        set_operation_seed(operation_id if operation.mutating else None)
        try:
            outcome = operation.handler(params)
        finally:
            set_operation_seed(None)

        return _ok(operation_name, operation_id, context,
                   outcome["data"], outcome.get("warnings") or [])

    except BridgeError as err:
        return _err(operation_name, operation_id, err.code, err.message,
                    err.details)
    except Exception as err:  # noqa: BLE001 — the boundary must not leak a traceback
        import traceback
        return _err(
            operation_name, operation_id, "BRIDGE_CRASHED",
            f"{type(err).__name__}: {err}",
            {"traceback": traceback.format_exc()[-4000:]},
        )


def _base(operation: Optional[str], operation_id: Optional[str]) -> dict:
    return {
        "protocolVersion": PROTOCOL_VERSION,
        "operation": operation,
        "operationId": operation_id,
    }


def _ok(operation: Optional[str], operation_id: Optional[str], context: dict,
        data: Any, warnings: list[str]) -> dict:
    envelope = _base(operation, operation_id)
    envelope.update({
        "ok": True,
        "context": context,
        "data": data,
        "warnings": warnings,
    })
    return envelope


def _err(operation: Optional[str], operation_id: Optional[str], code: str,
         message: str, details: Optional[dict] = None) -> dict:
    details = dict(details or {})
    warnings = details.pop("warnings", []) or []
    envelope = _base(operation, operation_id)
    envelope.update({
        "ok": False,
        "error": {
            "code": code,
            "message": message,
            "retryable": code in RETRYABLE_CODES,
            "details": details,
        },
        "warnings": warnings,
    })
    return envelope


def describe() -> dict:
    """The protocol, machine-readable — the parity target for a TS client."""
    return {
        "protocolVersion": PROTOCOL_VERSION,
        "minPython": ".".join(str(part) for part in MIN_PYTHON),
        "errorCodes": {
            "relay": list(RELAY_ERROR_CODES),
            "bridge": list(BRIDGE_ERROR_CODES),
            "retryable": sorted(RETRYABLE_CODES),
        },
        "operations": [OPERATIONS[name].describe() for name in sorted(OPERATIONS)],
    }


def main(argv: Optional[list[str]] = None) -> int:
    argv = list(sys.argv[1:] if argv is None else argv)

    # The version floor is enforced at import time — see MIN_PYTHON above.

    if "--describe" in argv:
        print(json.dumps(describe()))
        return 0

    if argv:
        print(json.dumps(_err(None, None, "INVALID_REQUEST",
                              f"Unexpected argument(s): {' '.join(argv)}. The "
                              "bridge reads one envelope on stdin.")))
        return 1

    raw = sys.stdin.read()
    # The dispatchers read stdin themselves for piped bodies. The envelope has
    # already consumed it, so hand them an empty one rather than a closed file
    # they would block on or crash reading.
    sys.stdin = io.StringIO("")

    if not raw.strip():
        print(json.dumps(_err(None, None, "INVALID_REQUEST",
                              "No request envelope on stdin.")))
        return 0

    try:
        envelope = json.loads(raw)
    except json.JSONDecodeError as exc:
        print(json.dumps(_err(None, None, "INVALID_REQUEST",
                              f"Request is not valid JSON: {exc}")))
        return 0

    print(json.dumps(handle(envelope)))
    return 0


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