"""FlyDocs Unified API Client — tier-aware routing. Reads tier from .flydocs/config.json and routes operations to either the relay REST API (cloud tier) or local filesystem (_local/file_store.py). Usage: from flydocs_api import get_client, output_json, fail client = get_client() result = client.create_issue(title="New issue", issue_type="feature") """ import hashlib import json import os import re import subprocess import sys import time import uuid from pathlib import Path from typing import Optional # FLY-1105 / FLY-1115: default page size for issue listing. # # Was 50, and a saturated result was reported as if it were the whole set — a # phase rollup read 50 issues / 131 points when the real figures were 57 / 144, # and made it look like issues had been lost. Defined here rather than in # issues.py so the cloud path, the local path and the CLI cannot drift apart; # they were three separate literals before. DEFAULT_LIST_LIMIT = 250 # FLY-1263 (relay-lifecycle-authority-spec.md §7.2 / §12.1): mutation idempotency # wire contract with the relay lifecycle service (RLA-4). # # A stable operation id per logical mutation lets the server (RLA-4) replay the # first outcome instead of writing twice when a retry fires. The server reads the # header case-insensitively; we send the canonical casing. OPERATION_ID_HEADER = "X-Operation-Id" # Set by the server on a replayed mutation response — a retry that lands on it is # a success, not a duplicate. Surfaced through `last_response_headers` (lowercased) # and, because a replay is a 2xx carrying the stored body, handled by the normal # success path with no special casing. IDEMPOTENT_REPLAY_HEADER = "X-Idempotent-Replay" # --------------------------------------------------------------------------- # Description content guard (FLY-1470 — relay-lifecycle-authority-spec.md §8) # --------------------------------------------------------------------------- # # `expectedRevision` is a last-modified timestamp — Linear's `Issue.updatedAt`, # Jira's `fields.updated` — so it answers "did anything about this issue # change", which is not the question a whole-document description rewrite is # asking. Measured on 2026-08-29: a comment by another actor did NOT move the # token on the current mapping, while a status change or a field write DID. So # a correctly-read token 409s on activity that never touched a word of the # prose, which is precisely what an agent produces between reading an issue and # rewriting its description. # # `expectedDescriptionHash` guards the document the write actually replaces. # The normalization below is the Python half of a two-language contract: the # authority is `src/lib/relay/lifecycle/description-hash.ts` in flydocs-app, # and the fixtures in `test_enforcement.py` are copied verbatim from that # module's test so a drift in either implementation goes red in both suites # rather than 409-ing in production. # # In order: # # 1. CRLF -> LF (a lone CR too). Line endings mean nothing here. # 2. Trailing spaces and tabs stripped per line. Exactly `[ \t]+$` — NOT a # `\s`-class strip, because Python and JavaScript disagree about which # Unicode code points `\s` covers and this has to be byte-identical. # 3. Leading FlyDocs attribution blocks removed, while they match. FLY-560 # prepends `**@handle** (via FlyDocs)` to descriptions written with # workspace credentials, so what a client composed and what the provider # stores differ by a header the client never wrote. Removed after step 2 # so trailing whitespace on that line cannot defeat the match. # 4. Trailing whitespace at the end of the document removed. # # Interior blank lines and case are preserved — they are prose. # Mirrors `ATTRIBUTION_BLOCK` in flydocs-app `src/lib/relay/attribution.ts`, # after step 1 has already collapsed CRLF. _ATTRIBUTION_BLOCK_RE = re.compile(r"^\*\*@[^*]+\*\* \(via FlyDocs\)\n{1,2}") _LINE_TRAILING_RE = re.compile(r"[ \t]+$") _DOCUMENT_TRAILING_RE = re.compile(r"[ \t\n]+$") def normalize_description_for_hash(text: str) -> str: """Reduce a description to the bytes `description_hash` digests.""" unix_newlines = text.replace("\r\n", "\n").replace("\r", "\n") line_trimmed = "\n".join( _LINE_TRAILING_RE.sub("", line) for line in unix_newlines.split("\n") ) without_attribution = line_trimmed while _ATTRIBUTION_BLOCK_RE.match(without_attribution): without_attribution = _ATTRIBUTION_BLOCK_RE.sub("", without_attribution, count=1) return _DOCUMENT_TRAILING_RE.sub("", without_attribution) # Unpaired surrogates survive `json.loads` in Python (a relay response really # can carry one) and then raise on `.encode("utf-8")`. Node's `Buffer.from(s, # "utf8")` substitutes U+FFFD instead, so the two languages only agree if # Python substitutes the same code point. `errors="replace"` is NOT that: it # writes `?` (0x3f) and would make every such digest diverge. _LONE_SURROGATE_RE = re.compile("[\ud800-\udfff]") def description_hash(text: str) -> str: """sha256 (lowercase hex) of the normalized description. The surrogate scrub is part of the cross-language contract, not defensive padding: without it a description carrying an unpaired surrogate crashes the CLI, and with the wrong substitution it silently 409s forever. """ normalized = _LONE_SURROGATE_RE.sub("\ufffd", normalize_description_for_hash(text)) return hashlib.sha256(normalized.encode("utf-8")).hexdigest() # --------------------------------------------------------------------------- # Project root discovery # --------------------------------------------------------------------------- def find_project_root() -> Path: """Resolve the authoritative project root for FlyDocs operations. In multi-repo workspaces (marked by ``.flydocs-workspace.json`` at an ancestor of ``cwd``), the authoritative config lives in each child repo's ``.flydocs/config.json``. The workspace-root ``.flydocs/`` may contain a stale single-repo config left over from pre-multi-repo onboarding (FLY-723) and must NOT be treated as authoritative. Resolution order: 1. If ``.flydocs-workspace.json`` is found at ``cwd`` or any ancestor, the workspace is multi-repo. Resolve to the active child repo via: a. cwd is inside a known child repo → return that repo b. ``.flydocs/session/active-repo`` pointer at workspace root → use it c. First child repo with a ``.flydocs/`` directory (stable fallback) 2. Otherwise (single-repo) walk up from cwd to the first ``.flydocs/`` directory — legacy behavior. """ cwd = Path.cwd() workspace_root = _find_workspace_root(cwd) if workspace_root is not None: child = _resolve_child_repo(cwd, workspace_root) if child is not None: return child # Degenerate multi-repo: workspace file exists but no children are # resolvable. Fall through to the legacy walk as a last resort so # we don't hard-fail — but callers should expect stale config. current = cwd while current != current.parent: if (current / ".flydocs").is_dir(): return current current = current.parent return cwd def _find_workspace_root(start: Path) -> Path | None: """Walk up from ``start`` to find ``.flydocs-workspace.json``.""" current = start while current != current.parent: if (current / ".flydocs-workspace.json").is_file(): return current current = current.parent return None def _resolve_child_repo(cwd: Path, workspace_root: Path) -> Path | None: """Resolve the active child repo in a multi-repo workspace. Returns the child repo directory, or ``None`` if no child can be resolved (misconfigured workspace). """ try: ws_data = json.loads( (workspace_root / ".flydocs-workspace.json").read_text() ) except (OSError, json.JSONDecodeError): return None repos = ws_data.get("repos", {}) if not repos: return None cwd_real = cwd.resolve() # 1. cwd is inside a known child repo for _name, entry in repos.items(): path = entry.get("path", "") repo_dir = (workspace_root / path).resolve() try: cwd_real.relative_to(repo_dir) except ValueError: continue if (repo_dir / ".flydocs").is_dir(): return repo_dir # 2. active-repo pointer at workspace root pointer = workspace_root / ".flydocs" / "session" / "active-repo" if pointer.exists(): try: name = pointer.read_text().strip() except (OSError, IOError): name = "" entry = repos.get(name) if entry: repo_dir = (workspace_root / entry.get("path", "")).resolve() if (repo_dir / ".flydocs").is_dir(): return repo_dir # 3. First child repo with a .flydocs/ directory (stable fallback) for _name, entry in repos.items(): repo_dir = (workspace_root / entry.get("path", "")).resolve() if (repo_dir / ".flydocs").is_dir(): return repo_dir return None # --------------------------------------------------------------------------- # Structured relay errors # --------------------------------------------------------------------------- class RelayError(Exception): """A structured relay rejection, raised instead of exiting (FLY-1265). `_request` normally renders an API error and calls `fail()`, which is right for a caller with nothing to do about it. The lifecycle codes (§11) are the exception: `REVISION_MISMATCH` carries the fresh state to retry against and `CRITERION_MISMATCH` carries the list the caller addressed the wrong thing in. Those are recoveries, not exits, so callers opt in with `raise_on_error=True` and read `code` / `body`. """ def __init__(self, status: int, code: str, message: str, body: dict): super().__init__(message) self.status = status self.code = code or "UNKNOWN" self.message = message self.body = body # FLY-1293: the last structured rejection the relay produced, whether it was # raised or rendered-and-exited. # # `fail()` turns a lifecycle code into prose and exits, which is the right # behavior for a human at a terminal and a total loss of structure for the # bridge — it would have to regex `Relay API error (CODE): …` back out of # stderr to answer "which code was this?". Recording the error here costs one # assignment on the failure path, changes no existing behavior, and lets # `bridge.py` return `REVISION_MISMATCH` as a code rather than as a sentence. _LAST_RELAY_ERROR: Optional[RelayError] = None def _record_relay_error(err: RelayError) -> None: """Remember a structured rejection for a caller that wants the code.""" global _LAST_RELAY_ERROR _LAST_RELAY_ERROR = err def take_last_relay_error() -> Optional[RelayError]: """Return and clear the last structured relay rejection (FLY-1293). Consuming read: an error belongs to exactly one failure, and a stale one left in place would be attributed to the next operation in the process. """ global _LAST_RELAY_ERROR err = _LAST_RELAY_ERROR _LAST_RELAY_ERROR = None return err # --------------------------------------------------------------------------- # Caller-supplied operation identity (FLY-1293) # --------------------------------------------------------------------------- # # A mutation's operation id is normally invented here (uuid4), which is correct # for a human running one command: the intent begins and ends inside the # process. The bridge inverts that — the intent is named by the caller (an MCP # tool call), survives a process that was killed mid-flight, and is retried by # spawning the bridge again. A fresh uuid4 on the second spawn would present # the same write to the relay as a different intent and write it twice, which # is the exact failure `X-Operation-Id` exists to prevent. # # So the ambient id below is a *seed*, not the header value. Each mutation # derives its own id from (seed, method, path, body) via uuid5: # # * same envelope replayed after a kill -> same derived id -> the relay # replays the stored outcome (`X-Idempotent-Replay`) instead of writing; # * the three mutations inside one `issue.activate` (assign, transition, # comment) get three distinct ids, so none of them trips # `OPERATION_ID_REUSED` — which is what sending the seed verbatim would do, # since the relay keys on (id, input hash). # # Unset, everything below is inert and mutations key themselves exactly as # before. _OPERATION_SEED: Optional[str] = None # Fixed namespace so a derived id is reproducible across processes, machines # and CLI versions. Any UUID works as long as it never changes. _OPERATION_NAMESPACE = uuid.UUID("6f1e5a2c-4c0d-5b7e-9a3f-1d2c3b4a5e6f") def set_operation_seed(seed: Optional[str]) -> None: """Seed derived operation ids for every mutation in this process. Called by the bridge with the envelope's `operationId`. Passing ``None`` restores uuid4-per-mutation. """ global _OPERATION_SEED _OPERATION_SEED = seed or None def get_operation_seed() -> Optional[str]: """The active operation seed, or ``None`` when ids are per-process.""" return _OPERATION_SEED def derive_operation_id(seed: str, method: str, path: str, body: Optional[dict]) -> str: """Derive a stable operation id for one mutation under ``seed``. The body participates so that a retry carrying *different* content is a different intent — the same rule `_retry_acceptance_once` relies on when it re-issues an acceptance batch against a fresh revision. """ payload = json.dumps(body, sort_keys=True, separators=(",", ":")) if body else "" return str(uuid.uuid5(_OPERATION_NAMESPACE, f"{seed}\n{method} {path}\n{payload}")) # --------------------------------------------------------------------------- # Relay backend (cloud tier) # --------------------------------------------------------------------------- class RelayBackend: """REST client for the FlyDocs Relay API.""" DEFAULT_BASE_URL = "https://app.flydocs.ai/api/relay" LOCAL_BASE_URL = "http://localhost:3000/api/relay" MAX_RETRIES = 3 RETRY_DELAY = 2 # FLY-1263 (§12.1): mutations carry an operation id, and a keyed request is # the only one safe to retry after an ambiguous failure. # # FLY-1265: **every** mutating relay call is keyed, not just `/issues*` and # `/operations*`. The narrower rule left context push, workspace rules, # scan and usage writes unkeyed, which meant they could not be retried # after a 5xx — the exact failure mode the id exists to make survivable — # and it made "is this call idempotent?" a per-route question every future # caller would have to re-answer. The server ignores an id on a route it # does not protect, so keying everything costs a header and removes the # question. MUTATING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"}) # Cap on a server-supplied Retry-After sleep — a runaway header must not hang # the CLI (§12.1). RETRY_AFTER_CAP = 30 # FLY-1556: how long a keyed mutation keeps waiting on 409 # OPERATION_IN_FLIGHT before giving up. An in-flight answer is the relay # finishing our own earlier attempt, not a failure, so it must not spend # the transient-error attempts: it did, and a Jira create that took 11s on # the server got a 502 then two 409s five seconds apart, the loop gave up # ~6s after the first answer while the relay still held the claim, and the # agent created again under a new id. The relay's request budget is 30s; # twice that covers the original attempt and its release. IN_FLIGHT_WAIT_BUDGET = 60 def __init__(self, project_root: Path, config: dict): self.project_root = project_root self.config = config self.log_path = project_root / ".flydocs" / "logs" / "relay-ops.jsonl" self.api_key = self._load_api_key() if not self.api_key: # Build a detailed list of every location we actually checked so # the user can see which file is missing or malformed. checked_paths: list[str] = ["FLYDOCS_API_KEY env var"] for p in self._candidate_credential_paths(): exists = "exists" if p.exists() else "not found" checked_paths.append(f"{p} ({exists})") global_cred = Path.home() / ".flydocs" / "credentials" checked_paths.append( f"{global_cred} ({'exists' if global_cred.exists() else 'not found'})" ) checked_paths.append(".env.local / .env (legacy)") fail( "FlyDocs API key not found.\n" "Project root resolved to: " + str(project_root) + "\n" "Checked (in order):\n - " + "\n - ".join(checked_paths) + "\n\nTo set up your API key:\n" " 1. Run: flydocs init (interactive, writes global)\n" " 2. Or: flydocs init --key KEY --project-key (writes project-scoped)\n" " 3. Or: flydocs auth (set global key directly)\n" " 4. CI/CD: set FLYDOCS_API_KEY environment variable" ) self.workspace_id = self._resolve_workspace_id(config) if not self.workspace_id: fail("workspaceId not found. Run 'flydocs init' to set up your workspace.") self.repo_slug = self._detect_repo_slug() self.base_url = self._resolve_base_url() # FLY-699: Normalize config paths across v1/v2/v3 formats. # v1 nests under workspace/provider; v2+ uses top-level keys. self._normalize_config() # Workspace and label helpers (for create_issue label resolution) self.workspace = self.config.get("workspace", {}) def _normalize_config(self) -> None: """Ensure operational fields are accessible at expected paths. ADR-011: v3 configs store all fields at top level. This method populates the 'workspace' dict from top-level fields so scripts that read via relay.workspace.get() continue working. Also handles activeProjects -> activeProjectId migration. """ cfg = self.config fmt = cfg.get("configFormat", 1) if fmt < 2: return # ADR-011: Migrate activeProjects array to activeProjectId string if "activeProjectId" not in cfg and "activeProjects" in cfg: ap = cfg["activeProjects"] if isinstance(ap, list) and ap: cfg["activeProjectId"] = ap[0] # Build workspace dict from top-level fields ws = cfg.get("workspace") or {} for key in ("activeProjectId", "activeContexts", "activeSprintId", "repoSlug", "statusMapping"): if key not in ws and key in cfg: ws[key] = cfg[key] cfg["workspace"] = ws # Ensure issueLabels is at top level if "issueLabels" not in cfg: cfg["issueLabels"] = {} # Provider — keep provider.type for provider-specific logic if "provider" not in cfg: cfg["provider"] = {} def _candidate_credential_paths(self) -> list[Path]: """ FLY-648: Return ordered list of credential file locations to check. In multi-repo workspaces, credentials.json lives at the workspace root (sibling of .flydocs-workspace.json), one level up from each child repo. Check the project root first, then the workspace root if it exists. """ paths = [self.project_root / ".flydocs" / "credentials.json"] # Walk up: if parent has .flydocs-workspace.json, check its credentials too parent = self.project_root.parent if (parent / ".flydocs-workspace.json").exists(): paths.append(parent / ".flydocs" / "credentials.json") return paths def _load_api_key(self) -> Optional[str]: # 1. Environment variable (CI/CD override) if os.environ.get("FLYDOCS_API_KEY"): return os.environ["FLYDOCS_API_KEY"] # 2. FLY-644 + FLY-648: Project-scoped or workspace-scoped credentials for cred_path in self._candidate_credential_paths(): if cred_path.exists(): try: cred_data = json.loads(cred_path.read_text()) key = cred_data.get("apiKey") or cred_data.get("api_key") if key: return key # FLY-704: File exists, valid JSON, but missing apiKey print(f"Warning: Credential at {cred_path} is missing apiKey — falling back.", file=sys.stderr) except (json.JSONDecodeError, OSError): # FLY-704: File exists but can't be parsed print(f"Warning: Malformed credential at {cred_path} — falling back. Fix or delete the file.", file=sys.stderr) # 3. Global credential file (v2 — written by flydocs auth/init) cred_file = Path.home() / ".flydocs" / "credentials" if cred_file.exists(): try: cred_data = json.loads(cred_file.read_text()) key = cred_data.get("apiKey") or cred_data.get("api_key") if key: return key print(f"Warning: Global credential at {cred_file} is missing apiKey.", file=sys.stderr) except (json.JSONDecodeError, OSError): print(f"Warning: Malformed global credential at {cred_file} — fix or delete the file.", file=sys.stderr) # 4. Legacy per-project env files (v1 fallback) for name in [".env.local", ".env"]: env_file = self.project_root / name if env_file.exists(): key = self._parse_env_file(env_file, "FLYDOCS_API_KEY") if key: return key return None def _resolve_workspace_id(self, config: dict) -> Optional[str]: # 1. Config file (v2 — written by flydocs init/sync) ws_id = config.get("workspaceId") if ws_id: return ws_id # 2. FLY-644 + FLY-648: Project-scoped or workspace-scoped credentials for cred_path in self._candidate_credential_paths(): if cred_path.exists(): try: cred_data = json.loads(cred_path.read_text()) ws_id = cred_data.get("workspaceId") if ws_id: return ws_id except (json.JSONDecodeError, OSError): print(f"Warning: Malformed credential at {cred_path} — skipping.", file=sys.stderr) # 3. Global credential file (v2 fallback) cred_file = Path.home() / ".flydocs" / "credentials" if cred_file.exists(): try: cred_data = json.loads(cred_file.read_text()) ws_id = cred_data.get("workspaceId") if ws_id: return ws_id except (json.JSONDecodeError, OSError): print(f"Warning: Malformed global credential at {cred_file} — skipping.", file=sys.stderr) return None def _parse_env_file(self, path: Path, key: str) -> Optional[str]: with open(path, "r") as f: for line in f: line = line.strip() if line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") if k.strip() == key: v = v.strip().strip("\"'") return v if v else None return None def _detect_repo_slug(self) -> Optional[str]: try: url = subprocess.check_output( ["git", "remote", "get-url", "origin"], stderr=subprocess.DEVNULL, timeout=5, ).decode().strip() if url.endswith(".git"): url = url[:-4] if ":" in url and "@" in url: return url.split(":")[-1] return "/".join(url.split("/")[-2:]) except Exception: return None def _resolve_base_url(self) -> str: env_url = os.environ.get("FLYDOCS_RELAY_URL") if env_url: return env_url.rstrip("/") config_url = self.config.get("relay", {}).get("url") if config_url: return config_url.rstrip("/") return self.DEFAULT_BASE_URL # FLY-1115: headers from the most recent response. Reset per request. last_response_headers: dict = {} def _retry_after_seconds(self, err, attempt: int) -> int: """Delay for a retry: the server's `Retry-After` (delta-seconds, capped), else exponential backoff (FLY-1263 §12.1). Only the numeric form is honored — the relay sends seconds (§7.2); an HTTP-date value falls back to backoff rather than being mis-parsed into a huge or negative sleep. """ raw = None try: raw = err.headers.get("Retry-After") if err.headers else None except Exception: raw = None if raw is not None: try: seconds = int(str(raw).strip()) if seconds >= 0: return min(seconds, self.RETRY_AFTER_CAP) except (TypeError, ValueError): pass return self.RETRY_DELAY * (2 ** attempt) def _operation_id_for(self, method: str, path: str, body: Optional[dict]) -> str: """The `X-Operation-Id` for one mutation (FLY-1263, FLY-1293). A uuid4 per call unless a caller seeded the process (`set_operation_seed`), in which case the id is derived from the seed and the request — see the module note above for why the seed is not sent verbatim. """ seed = get_operation_seed() if seed: return derive_operation_id(seed, method, path, body) return str(uuid.uuid4()) def _request(self, method: str, path: str, body: Optional[dict] = None, params: Optional[dict] = None, best_effort: bool = False, raise_on_error: bool = False, id_payload: Optional[dict] = None) -> dict: import urllib.request import urllib.error import urllib.parse url = f"{self.base_url}{path}" if params: filtered = {k: v for k, v in params.items() if v is not None and v != ""} if filtered: url += "?" + urllib.parse.urlencode(filtered, doseq=True) headers = { "Authorization": f"Bearer {self.api_key}", "X-Workspace": self.workspace_id, "Content-Type": "application/json", "Accept": "application/json", } if self.repo_slug: headers["X-Repo"] = self.repo_slug # FLY-1263 / FLY-1265 (§7.2 / §12.1): key every mutation with one # operation id. Generated once, held across the retry loop below — the # id identifies the intent, not the attempt, so a retry replays the # first outcome instead of writing twice. is_mutation = method in self.MUTATING_METHODS operation_id = None if is_mutation: # FLY-1411: `id_payload` is the projection of the body that carries # the *intent*, for a request whose body also carries something # that changes on its own. A session record embeds # `window.endedAt = now()`, so hashing the whole body would give a # respawned wrap a fresh id — and a fresh id is a second record and # a second project update, which is precisely what the id exists to # prevent. Defaults to the body, so every other route is unchanged. operation_id = self._operation_id_for( method, path, body if id_payload is None else id_payload) headers[OPERATION_ID_HEADER] = operation_id # A mutation is only safe to retry after an ambiguous failure (5xx or a # network drop) when it is keyed — otherwise the outcome is unknown and a # retry could double-write (§12.1). Reads carry no such risk. A 429 or a # 409 OPERATION_IN_FLIGHT is retried regardless: neither executed the # write, so replaying them is always safe. # # Since FLY-1265 every mutation is keyed, so this is now always true — # kept as a guard rather than deleted because it is the rule ("no key, # no ambiguous retry") and deleting it would make a future unkeyed path # silently retryable. can_retry_ambiguous = (not is_mutation) or operation_id is not None data = json.dumps(body).encode("utf-8") if body else None # FLY-861: 60s timeout (was 15s) gives international customers # headroom for slow international links + heavy responses without # making genuine network failures hang for too long. Server-side # FLY-858 perf work shrinks typical responses; this is the # belt-and-suspenders ceiling for the long tail. # FLY-1556: `attempt` counts transient-error retries; in-flight waits # are budgeted in seconds and do not advance it. attempt = 0 in_flight_waited = 0.0 while attempt < self.MAX_RETRIES: try: req = urllib.request.Request(url, data=data, headers=headers, method=method) with urllib.request.urlopen(req, timeout=60) as resp: response_body = resp.read().decode("utf-8") result = json.loads(response_body) if response_body else {} # FLY-1115: retain response headers so callers can read # pagination context. Kept off the return value because # every existing caller consumes the parsed body directly. # FLY-1263: this is also where a replayed mutation surfaces — # `x-idempotent-replay` rides here; a replay is a 2xx carrying # the stored body, so it is treated as success with no special # casing, which is the point of the stable operation id. self.last_response_headers = { k.lower(): v for k, v in resp.headers.items() } self._log_operation(method, path, resp.status, result) return result except urllib.error.HTTPError as e: error_body = e.read().decode("utf-8") if e.fp else "" try: error_data = json.loads(error_body) if error_body else {} except json.JSONDecodeError: error_data = {"error": error_body} self._log_operation(method, path, e.code, error_data) error_code = error_data.get("code", "") # 429 throttle — honor Retry-After when present (§12.1). The # request never executed, so retrying is safe even unkeyed. if e.code == 429 and attempt < self.MAX_RETRIES - 1: delay = self._retry_after_seconds(e, attempt) print(f"Rate limited, retrying in {delay}s...", file=sys.stderr) time.sleep(delay) attempt += 1 continue # 409 OPERATION_IN_FLIGHT — our own earlier keyed attempt is still # finishing; wait the Retry-After and let the replay land (§7.2). # FLY-1556: on a time budget, not the attempt counter — waiting # has no side effect and giving up here is what duplicates. if (e.code == 409 and error_code == "OPERATION_IN_FLIGHT" and in_flight_waited < self.IN_FLIGHT_WAIT_BUDGET): delay = self._retry_after_seconds(e, 0) in_flight_waited += delay print(f"Operation in flight, retrying in {delay}s...", file=sys.stderr) time.sleep(delay) continue # 5xx transient — retry only when the request is safe to repeat: # a read, or a mutation that carried an operation id (§12.1). if (e.code >= 500 and attempt < self.MAX_RETRIES - 1 and can_retry_ambiguous): # FLY-634: Don't retry on provider auth failures — credentials are expired if error_code in ("PROVIDER_AUTH_FAILED", "PROVIDER_TOKEN_REFRESH_FAILED"): auth_err = RelayError( e.code, error_code, error_data.get("error", f"HTTP {e.code}"), error_data, ) _record_relay_error(auth_err) # FLY-1293 if raise_on_error: raise auth_err if best_effort: return {} fail( f"Provider credentials expired ({error_code}). " "Reconnect your provider in the FlyDocs dashboard at app.flydocs.ai" ) delay = self.RETRY_DELAY * (2 ** attempt) print(f"Server error ({e.code}), retrying in {delay}s...", file=sys.stderr) time.sleep(delay) attempt += 1 continue # Terminal — includes 422 OPERATION_ID_REUSED and 400 # OPERATION_ID_REQUIRED, which are client-bug states that a retry # would only loop on (§7.2). error_msg = error_data.get("error", f"HTTP {e.code}") error_code = error_code or "UNKNOWN" provider = error_data.get("provider_error", "") # FLY-1293: record before the render-and-exit paths below turn # the code into a sentence. terminal_err = RelayError(e.code, error_code, error_msg, error_data) _record_relay_error(terminal_err) # FLY-1265: hand the structured body back to a caller that has a # recovery for it, before any of the render-and-exit paths. if raise_on_error: raise terminal_err if best_effort: return {} # FLY-653: Handle structured transition errors with a recovery prompt. # Supports both the structured shape (FLY-651 landing) AND today's # unstructured STATUS_MAPPING_ERROR for backward compatibility. if error_code in ("STATUS_NOT_REACHABLE", "STATUS_MAPPING_ERROR"): fail(_format_transition_recovery_prompt(error_data, error_msg)) msg = f"Relay API error ({error_code}): {error_msg}" if provider: msg += f" — provider: {provider}" fail(msg) except (urllib.error.URLError, TimeoutError) as e: # FLY-861: include the underlying exception class on retry # messages so future incidents have specific signal — # `gaierror` (DNS), `ConnectionRefusedError`, `TimeoutError`, # `ssl.SSLError` etc. all funnel through this catch. exc_label = type(e).__name__ # FLY-1263: a network drop leaves the outcome ambiguous, so an # unkeyed mutation must not retry it — same rule as 5xx. if attempt < self.MAX_RETRIES - 1 and can_retry_ambiguous: delay = self.RETRY_DELAY * (2 ** attempt) print( f"Network error ({exc_label}), retrying in {delay}s...", file=sys.stderr, ) time.sleep(delay) attempt += 1 continue if best_effort: return {} # FLY-1293: a transport failure is structured too — the bridge # reports it as RELAY_UNREACHABLE rather than as free text. network_err = RelayError( 0, "NETWORK_ERROR", f"unable to reach relay API ({exc_label})", {}, ) _record_relay_error(network_err) # FLY-1411: and `raise_on_error` means it, on this branch too. # It used to mean "raise the rejections the server sent and exit # on the ones it never received", which made a caller holding a # recovery for an unreachable relay — the session wrap, whose # whole point is that a network failure must not lose the # record — unable to run it: `fail()` raises SystemExit, and no # `except Exception` catches that. if raise_on_error: raise network_err # FLY-861: actionable recovery hint when the request was # for /issues/{ref} on the default fields=full path. Tells # the agent (or human) to retry with --fields basic, # turning `basic` into a real fallback valve instead of # a dead-end parameter. params_str = str(params or {}) is_issue_get = ( method == "GET" and "/issues/" in path and "fields=basic" not in params_str ) if is_issue_get and isinstance(e, TimeoutError): fail( f"Timed out reaching relay API after 60s ({exc_label}).\n" "Hint: this issue may be large or your link is slow. Try:\n" " python3 .claude/skills/flydocs-workflow/scripts/issues.py " "get --fields basic\n" "to fetch a lighter response. Re-run with --fields full " "if you need comments/relations." ) fail(f"Network error: unable to reach relay API ({exc_label})") if best_effort: return {} # The same rule as the branch above: the retries are exhausted, the # request never landed, and a caller that asked for the failure gets it # as a value (FLY-1411). exhausted = RelayError( 0, "NETWORK_ERROR", "max retries exceeded reaching relay API", {}, ) _record_relay_error(exhausted) if raise_on_error: raise exhausted fail("Max retries exceeded") return {} # unreachable def get(self, path: str, params: Optional[dict] = None, best_effort: bool = False, raise_on_error: bool = False) -> dict | list: """A read. `best_effort` returns `{}` instead of exiting on failure, and `raise_on_error` raises `RelayError` instead — both for a read whose failure the caller has a plan for (FLY-1592: the context push reads the stored rules sections through a route newer than itself, and needs the reason to report when it cannot).""" return self._request("GET", path, params=params, best_effort=best_effort, raise_on_error=raise_on_error) def post(self, path: str, body: Optional[dict] = None, raise_on_error: bool = False, id_payload: Optional[dict] = None) -> dict: return self._request("POST", path, body=body, raise_on_error=raise_on_error, id_payload=id_payload) def put(self, path: str, body: Optional[dict] = None, raise_on_error: bool = False) -> dict: return self._request("PUT", path, body=body, raise_on_error=raise_on_error) def patch(self, path: str, body: Optional[dict] = None) -> dict: return self._request("PATCH", path, body=body) def delete(self, path: str) -> dict: return self._request("DELETE", path) def repair_operation(self, operation_id: Optional[str]) -> None: """Fire one best-effort reconciliation repair (§9.4 / §12.1). When a transition landed but its audit comment did not, the server returns `reconciliation: transitioned_comment_pending`; this posts the pending comment via `POST /operations/{id}/repair`. Best-effort by contract: it never raises and never exits — a failed repair leaves the record durably pending and repairable on the next touch, so surfacing an error here would only trigger the retry loop the reconciliation design exists to avoid (§9.5). NOTE: the repair route (RLA-6) is not deployed yet. This is the client side built against the frozen spec contract; until RLA-6 lands the call 404s and is swallowed exactly as a transient failure would be. It is isolated behind `best_effort` so it cannot fail the transition. """ if not operation_id: return self._request( "POST", f"/operations/{operation_id}/repair", body={}, best_effort=True ) def _log_operation(self, method: str, path: str, status: int, result: dict | list) -> None: try: from datetime import datetime, timezone self.log_path.parent.mkdir(parents=True, exist_ok=True) entry = { "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "method": method, "path": path, "status": status, "success": 200 <= status < 300, } with open(self.log_path, "a") as f: f.write(json.dumps(entry) + "\n") except Exception: pass def get_category_label_id(self, issue_type: str) -> Optional[str]: labels = self.config.get("issueLabels", {}).get("category", {}) return labels.get(issue_type) def get_other_label_id(self, label_name: str) -> Optional[str]: labels = self.config.get("issueLabels", {}).get("other", {}) return labels.get(label_name) def resolve_user_id(self, name_or_id: str) -> tuple[Optional[str], Optional[str]]: """Resolve a user name or ID. Returns (id, displayName).""" # If it looks like a UUID, return as-is if len(name_or_id) == 36 and "-" in name_or_id: return name_or_id, None # Try identity file me_path = self.project_root / ".flydocs" / "me.json" if me_path.exists(): try: me = json.loads(me_path.read_text()) if me.get("displayName", "").lower() == name_or_id.lower(): return me.get("id"), me.get("displayName") except (json.JSONDecodeError, OSError): pass return name_or_id, None # --------------------------------------------------------------------------- # Local backend # --------------------------------------------------------------------------- class LocalBackend: """Filesystem-based backend using _local/file_store.py.""" def __init__(self, project_root: Path, config: dict): self.project_root = project_root self.config = config # --------------------------------------------------------------------------- # Unified client # --------------------------------------------------------------------------- class FlyDocsClient: """Tier-aware client. Routes to relay (cloud) or filesystem (local).""" def __init__(self): self.project_root = find_project_root() self.config_path = self.project_root / ".flydocs" / "config.json" self.config = self._load_config() self.tier = self.config.get("tier", "local") if self.tier == "cloud": self._relay = RelayBackend(self.project_root, self.config) else: self._relay = None self._local_backend = LocalBackend(self.project_root, self.config) def _load_config(self) -> dict: if self.config_path.exists(): with open(self.config_path, "r") as f: return json.load(f) return {} @property def is_cloud(self) -> bool: return self.tier == "cloud" def require_cloud(self, operation: str) -> None: """Fail with clear message if operation requires cloud tier.""" if not self.is_cloud: fail(f"{operation} requires cloud tier. Current tier: {self.tier}") def _resolve_local_identity(self) -> tuple[str | None, str | None]: """Resolve the local user's provider ID and display name from me.json. FLY-717: Used as fallback when server-side 'mine' filter resolves to the wrong user (e.g., API key owner instead of current user). Returns (provider_id, display_name). Either may be None. """ provider_type = self.config.get("provider", {}).get("type", "") for me_path in ( self.project_root / ".flydocs" / "me.json", Path.home() / ".flydocs" / "me.json", ): if me_path.exists(): try: me = json.loads(me_path.read_text()) name = me.get("displayName") identities = me.get("providerIdentities", {}) provider_id = identities.get(provider_type) if provider_type else None if name or provider_id: return provider_id, name except (json.JSONDecodeError, OSError): continue return None, None # --- Relay passthrough (for cloud scripts that need raw HTTP) --- @property def relay(self) -> RelayBackend: if self._relay is None: fail("Relay not available on local tier") return self._relay # --- Issue operations (tier-aware) --- def create_issue(self, **kwargs: object) -> dict: if self.is_cloud: return self._cloud_create_issue(**kwargs) from _local.file_store import create_issue return create_issue(self.project_root, **kwargs) def _cloud_create_issue(self, **kwargs: object) -> dict: relay = self.relay auto_resolved: dict[str, str] = {} issue_input: dict = { "title": kwargs.get("title", ""), "description": kwargs.get("description", ""), } # FLY-650: Only send priority when explicitly set. Jira rejects Linear's # 0-4 scale with a 400 error, so defaulting to 3 breaks all Jira creates. # Full provider-agnostic priority translation is tracked in FLY-659. priority = kwargs.get("priority") if priority is not None: issue_input["priority"] = priority estimate = kwargs.get("estimate") if estimate: issue_input["estimate"] = estimate # FLY-697: Send issue type so relay can apply workspace type mapping issue_type = kwargs.get("issue_type", "") if issue_type: issue_input["type"] = issue_type # Labels — category + triage + repo (auto-resolved) # FLY-699: For Jira, category is applied via issueTypeMapping in the # relay (using the `type` field sent above). Don't add category label # IDs to labelIds — those values are Jira issue type IDs, not label # IDs, and sending them as labels causes Jira to reject or misapply. provider_type = relay.config.get("provider", {}).get("type", "") label_ids: list[str] = [] if isinstance(issue_type, str) and provider_type != "jira": cat_id = relay.get_category_label_id(issue_type) if cat_id: label_ids.append(cat_id) auto_resolved["categoryLabel"] = issue_type if kwargs.get("triage"): triage_id = relay.get_other_label_id("triage") if triage_id: label_ids.append(triage_id) # ADR-011: repoDefaults — per-repo labels and component repo_defaults = relay.config.get("repoDefaults", {}) if repo_defaults.get("labels"): for label_name in repo_defaults["labels"]: # Resolve label name to ID via the relay's label cache label_id = relay.get_other_label_id(label_name) if label_id and label_id not in label_ids: label_ids.append(label_id) if repo_defaults["labels"]: auto_resolved["repoLabels"] = repo_defaults["labels"] # Fallback: legacy issueLabels.repo (deprecated, still read) if not repo_defaults.get("labels"): repo_labels = relay.config.get("issueLabels", {}).get("repo", {}) topology = relay.config.get("topology", {}) topo_type = topology.get("type", 1) if topo_type in (3, 4) and repo_labels: slug = relay.repo_slug or "" repo_name = slug.split("/")[-1] if "/" in slug else slug for name, label_id in repo_labels.items(): if label_id and name.lower() in repo_name.lower(): if label_id not in label_ids: label_ids.append(label_id) auto_resolved["repoLabel"] = name break if label_ids: issue_input["labelIds"] = label_ids # ADR-011: repoDefaults.component for Jira if repo_defaults.get("component") and provider_type == "jira": issue_input["component"] = repo_defaults["component"] auto_resolved["component"] = repo_defaults["component"] # Project — ADR-011: activeProjectId (singular) project_id = kwargs.get("project") if not project_id: project_id = relay.workspace.get("activeProjectId") if project_id: auto_resolved["project"] = "activeProjectId" if project_id: issue_input["projectId"] = project_id # Milestone — explicit only (ADR-011: defaultMilestoneId removed) milestone_id = kwargs.get("milestone") if milestone_id: # FLY-1103: the relay reads `milestoneId`. This sent # `projectMilestoneId`, which matched nothing, so the milestone was # silently dropped on every create while the call still succeeded. issue_input["milestoneId"] = milestone_id # Assignee assignee = kwargs.get("assignee") if assignee and isinstance(assignee, str): user_id, _ = relay.resolve_user_id(assignee) if user_id: issue_input["assigneeId"] = user_id result = relay.post("/issues", issue_input) issue = result.get("issue", result) response: dict = { "id": issue.get("id", ""), "identifier": issue.get("identifier", ""), "title": issue.get("title", kwargs.get("title", "")), "url": issue.get("url", ""), } if auto_resolved: response["autoResolved"] = auto_resolved return response def transition(self, ref: str, status: str, comment: str, force: str | None = None) -> dict: if self.is_cloud: payload: dict = { "status": status.upper(), "comment": comment, } # FLY-689: Force override — pass provider-native target directly if force: payload["force"] = force result = self.relay.post(f"/issues/{ref}/transition", payload) # FLY-1356: neither field is defaulted to something the relay did # not say. `success` defaulted to True, so a reply that omitted it # was reported as a success; `newStatus` defaulted to the requested # status, so a reply that omitted it echoed the request back as if # the provider had confirmed it. Both defaults manufactured # confirmation out of silence — the caller decides what to do with # the absence (issues.py refuses to write session state for it). response = { "success": result.get("success", False), "issue": result.get("issue", ref), "previousStatus": result.get("previousStatus", ""), "newStatus": result.get("newStatus", ""), } # FLY-653: Surface fallback info when the relay used a fallback target if "fallbackUsed" in result: response["fallbackUsed"] = result["fallbackUsed"] if "actualStatus" in result: response["actualStatus"] = result["actualStatus"] if "mappedFromFlydocsStatus" in result: response["mappedFromFlydocsStatus"] = result["mappedFromFlydocsStatus"] # FLY-689: Surface force override info if "forceUsed" in result: response["forceUsed"] = result["forceUsed"] if "forceTarget" in result: response["forceTarget"] = result["forceTarget"] # FLY-1263 (§9 / §12.1): the transition succeeded but its audit # comment is pending. Fire one best-effort repair; the record is # durable, so a failure here is recoverable rather than fatal. if result.get("reconciliation") == "transitioned_comment_pending": print( "transition recorded; audit comment pending — repairing", file=sys.stderr, ) self.relay.repair_operation(result.get("operationId")) response["reconciliation"] = result["reconciliation"] return response from _local.file_store import transition return transition(self.project_root, ref, status, comment) def comment(self, ref: str, body: str) -> dict: if self.is_cloud: result = self.relay.post(f"/issues/{ref}/comment", {"body": body}) return { "success": result.get("success", True), "commentId": result.get("commentId", ""), } from _local.file_store import add_comment return add_comment(self.project_root, ref, body) def list_issues(self, **kwargs: object) -> list[dict]: if self.is_cloud: params: dict = {} for key in ("status", "assignee", "project", "milestone", "limit"): val = kwargs.get(key) if val is not None and val != "": params[key] = str(val).upper() if key == "status" else str(val) if kwargs.get("active"): params["active"] = "true" if kwargs.get("mine"): params["mine"] = "true" # FLY-692: Sprint and board filters sprint = kwargs.get("sprint") if sprint: params["sprint"] = str(sprint) board = kwargs.get("board") if board: params["board"] = str(board) # Product scope cascade (bypassed by explicit --project or --all) if "project" not in params and not kwargs.get("show_all"): active_project = self.relay.workspace.get("activeProjectId") if active_project: params["project"] = active_project else: product_labels = self.relay.workspace.get("product", {}).get("labelIds", []) if product_labels: params["label"] = product_labels[0] result = self.relay.get("/issues", params=params) issues = result if isinstance(result, list) else [] # FLY-717: mine fallback — if server-side mine resolved to a # different user (e.g., API key owner instead of current user), # retry with explicit assignee from local me.json identity. if kwargs.get("mine") and issues: _, local_name = self._resolve_local_identity() if local_name and not any( (i.get("assignee") or "").lower() == local_name.lower() for i in issues ): local_id, _ = self._resolve_local_identity() if local_id: print( f"Note: --mine returned issues for a different user. " f"Retrying with local identity ({local_name}).", file=sys.stderr, ) params.pop("mine", None) params["assignee"] = local_id retry = self.relay.get("/issues", params=params) issues = retry if isinstance(retry, list) else [] return issues from _local.file_store import list_issues # FLY-1115: was a bare 50 while the cloud path defaulted to 250 — one of # the sibling caps the FLY-1105 audit recorded. Aligned so local and # cloud tiers truncate at the same point. return list_issues( self.project_root, status=str(kwargs.get("status", "")), assignee=str(kwargs.get("assignee", "")), limit=int(kwargs.get("limit", DEFAULT_LIST_LIMIT)), ) def last_list_pagination(self) -> dict: """Pagination context from the most recent /issues list (FLY-1115). Returns `{"has_more": bool, "total": int | None, "returned": int | None}`. `total` is None when the provider could not supply one — Linear exposes `hasNextPage` but no cheap exact count, so "more exists, quantity unknown" is a real state that callers must render rather than guess at. """ h = getattr(self.relay, "last_response_headers", {}) or {} total = h.get("x-total-count") returned = h.get("x-returned-count") return { "has_more": h.get("x-has-more") == "true", "total": int(total) if total and total.isdigit() else None, "returned": int(returned) if returned and returned.isdigit() else None, } def get_issue(self, ref: str, **kwargs: object) -> dict: if self.is_cloud: params: dict = {} fields = kwargs.get("fields") if fields: params["fields"] = str(fields) return self.relay.get(f"/issues/{ref}", params=params) from _local.file_store import get_issue return get_issue(self.project_root, ref) def assign(self, ref: str, assignee: str | None) -> dict: if self.is_cloud: result = self.relay.post(f"/issues/{ref}/assign", {"assignee": assignee}) return { "success": result.get("success", True), "issue": result.get("issue", ref), "assignee": result.get("assignee", assignee), } from _local.file_store import assign_issue return assign_issue(self.project_root, ref, assignee) def update_description(self, ref: str, text: str, expected_revision: str | None = None, expected_description_hash: str | None = None) -> dict: """Replace an issue's description under the §8 concurrency guards. The token travels in the request **body** as `expectedRevision` — the same field the acceptance route takes. `issues/[ref]/description/ route.ts` reads `parsed.body.expectedRevision` and validates it with `validateStringField(..., { required: false })`; there is no If-Match header on this route, and the schema note's request shape (`{ "text": ... }`) predates RLA-9. Omitted, not empty, when the caller has no token: `evaluateRevision` treats an absent field as "did not opt in" (pass-open while `requireRevision` is off) and an empty string as a value that can never match. FLY-1292 found this writer sending nothing at all — every `REVISION_REQUIRED` would-block verdict in the warn window was `operation: issue.description`, so under `enforce` every rewrite here would have been a 400. `expectedDescriptionHash` is the FLY-1470 content guard, and travels in the body beside the token. Either one satisfies the route; sending both is the strongest shape, because a matched hash overrides a stale revision (the prose is provably unchanged) while a stale revision with no hash is still a 409. Omitted, not empty, for the same reason the token is: a digest of nothing matches no real document. Rejections raise `RelayError` rather than exiting: a 409 on a whole-document write is something `cmd_description` has to explain, not a bare failure. """ if self.is_cloud: body: dict = {"text": text} if expected_revision: body["expectedRevision"] = expected_revision if expected_description_hash: body["expectedDescriptionHash"] = expected_description_hash result = self.relay.put( f"/issues/{ref}/description", body, raise_on_error=True ) return { "success": result.get("success", True), "issue": result.get("issue", ref), } from _local.file_store import update_description return update_description(self.project_root, ref, text) def acceptance(self, ref: str, changes: list[dict], expected_revision: str) -> dict: """Apply criterion-addressed acceptance edits (§10, FLY-1265). Cloud only. The merge happens server-side under one `expectedRevision` check, which is the point: read, merge and write all take place inside one operation, so the read-modify-write race that the description route has by construction cannot open. Rejections come back as `RelayError` so the caller can re-check its intent against the fresh state instead of dying on a 409. """ self.require_cloud("issues.py acceptance") payload = {"expectedRevision": expected_revision, "changes": changes} return self.relay.post( f"/issues/{ref}/acceptance", payload, raise_on_error=True ) def update_issue(self, ref: str, **fields: object) -> dict: """Update issue fields through `PATCH /issues/:ref`. `description` is still forwarded for callers outside this repo, but no dispatcher sends one any more: FLY-1469 retired `issues.py update --description` in favour of `update_description`, which carries the §8 `expectedRevision`. The relay now runs the same check on this route, so a description sent here without a token records a `REVISION_REQUIRED` would-block verdict (and is refused once the workspace requires one). """ if self.is_cloud: body: dict = {} updated: list[str] = [] for key in ("title", "priority", "estimate", "assignee", "description", "comment"): val = fields.get(key) if val is not None: body[key] = val updated.append(key) # FLY-663: dueDate — null is a valid value (clears the due date), # so we check for presence in fields rather than non-None. if "dueDate" in fields: body["dueDate"] = fields["dueDate"] updated.append("dueDate") state = fields.get("state") if state and isinstance(state, str): body["status"] = state.upper() updated.append("state") # FLY-661: labels accepts a list (preferred) or comma-separated string (legacy) labels = fields.get("labels") if labels: if isinstance(labels, list): body["labels"] = [str(l).strip() for l in labels if str(l).strip()] updated.append("labels") elif isinstance(labels, str): body["labels"] = [l.strip() for l in labels.split(",") if l.strip()] updated.append("labels") # Milestone resolution milestone = fields.get("milestone") if milestone and isinstance(milestone, str): milestone_id = milestone if len(milestone_id) != 36 or "-" not in milestone_id: milestones = self.relay.get("/milestones") match = next((m for m in milestones if m["name"].lower() == milestone_id.lower()), None) if not match: fail(f"Milestone not found: {milestone_id}") milestone_id = match["id"] body["milestoneId"] = milestone_id updated.append("milestone") # FLY-751: Project reassignment project_id = fields.get("projectId") if project_id and isinstance(project_id, str): body["projectId"] = project_id updated.append("project") if not body: fail("No fields to update") result = self.relay.patch(f"/issues/{ref}", body) return { "success": result.get("success", True), "issue": result.get("issue", ref), "updated": updated, } from _local.file_store import update_issue return update_issue(self.project_root, ref, **fields) def estimate(self, ref: str, points: int) -> dict: if self.is_cloud: result = self.relay.put(f"/issues/{ref}/estimate", {"estimate": points}) return { "success": result.get("success", True), "issue": result.get("issue", ref), "estimate": result.get("estimate", points), } from _local.file_store import estimate_issue return estimate_issue(self.project_root, ref, points) def priority(self, ref: str, level: int) -> dict: if self.is_cloud: result = self.relay.put(f"/issues/{ref}/priority", {"priority": level}) return { "success": result.get("success", True), "issue": result.get("issue", ref), "priority": result.get("priority", level), } from _local.file_store import priority_issue return priority_issue(self.project_root, ref, level) def link(self, ref: str, related_ref: str, link_type: str) -> dict: if self.is_cloud: result = self.relay.post(f"/issues/{ref}/link", { "relatedRef": related_ref, "type": link_type, }) return { "success": result.get("success", True), "type": result.get("type", link_type), } from _local.file_store import link_issues return link_issues(self.project_root, ref, related_ref, link_type) def assign_milestone(self, ref: str, milestone_id: str) -> dict: self.require_cloud("assign_milestone") result = self.relay.put(f"/issues/{ref}/milestone", {"milestoneId": milestone_id}) return { "success": result.get("success", True), "issue": result.get("issue", ref), "milestone": result.get("milestone", milestone_id), } def assign_cycle(self, ref: str, cycle_id: str | None = None) -> dict: self.require_cloud("assign_cycle") body: dict = {"cycleId": cycle_id} if cycle_id else {} result = self.relay.put(f"/issues/{ref}/cycle", body) return { "success": result.get("success", True), "issue": result.get("issue", ref), "cycle": result.get("cycle", cycle_id), } # --- Project operations (cloud only) --- def list_projects(self, **kwargs: object) -> list[dict]: self.require_cloud("list_projects") params: dict = {} if kwargs.get("active"): params["active"] = "true" if kwargs.get("show_all"): params["all"] = "true" result = self.relay.get("/projects", params=params) return result if isinstance(result, list) else [] def create_project(self, name: str, description: str | None = None) -> dict: self.require_cloud("create_project") body: dict = {"name": name} if description: body["description"] = description result = self.relay.post("/projects", body) return {"id": result["id"], "name": result["name"], "url": result.get("url", "")} def update_project(self, ref: str, name: str | None = None, description: str | None = None, state: str | None = None) -> dict: """Rename a project or edit its description (FLY-1073). `ref` accepts a name or an id, matching how `issues.py` resolves references. PATCH rather than POST /projects/update — that path already means "post a project status update". """ self.require_cloud("update_project") body: dict = {"ref": ref} if name is not None: body["name"] = name if description is not None: body["description"] = description if state is not None: body["state"] = state result = self.relay.patch("/projects", body) return { "success": result.get("success", True), "id": result.get("id", ref), "name": result.get("name", ""), "updated": result.get("updated", []), } def archive_project(self, ref: str) -> dict: """Archive a project where the provider supports it (FLY-1073). `mappedTo` reports what the provider actually did — Linear archives, Jira closes the Epic — because "archive" is not symmetric and a bare success would hide the difference. """ self.require_cloud("archive_project") result = self.relay.post("/projects/archive", {"ref": ref}) return { "success": result.get("success", True), "id": result.get("id", ref), "mappedTo": result.get("mappedTo", ""), } def list_milestones(self, **kwargs: object) -> list[dict]: self.require_cloud("list_milestones") params: dict = {} if kwargs.get("show_all"): params["all"] = "true" result = self.relay.get("/milestones", params=params) return result if isinstance(result, list) else [] def create_milestone(self, name: str, project: str | None = None, target_date: str | None = None) -> dict: self.require_cloud("create_milestone") body: dict = {"name": name} if project: body["projectId"] = project if target_date: body["targetDate"] = target_date return self.relay.post("/milestones", body) def update_milestone(self, milestone_id: str, **fields: object) -> dict: self.require_cloud("update_milestone") body: dict = {} for key in ("name", "targetDate", "description"): val = fields.get(key) if val is not None: body[key] = val result = self.relay.patch(f"/milestones/{milestone_id}", body) return {"success": result.get("success", True), "id": milestone_id, "name": result.get("name", "")} def delete_milestone(self, milestone_id: str) -> dict: self.require_cloud("delete_milestone") self.relay.delete(f"/milestones/{milestone_id}") return {"success": True, "id": milestone_id} def list_cycles(self, active: bool = False) -> list[dict]: """@deprecated FLY-655: use list_sprints().""" return self.list_sprints(active=active, include_all=not active) def list_sprints( self, active: bool = False, future: bool = False, closed: bool = False, current: bool = False, include_all: bool = False, ) -> list[dict]: """ FLY-656: List sprints across providers. Hits the canonical /sprints endpoint (FLY-657). Falls back to /cycles if /sprints is unavailable (older relay deploys during rollout window). Defaults to active+future (hides closed unless explicitly requested). """ self.require_cloud("list_sprints") params: dict = {} if active: params["active"] = "true" if future: params["future"] = "true" if closed: params["closed"] = "true" if current: params["current"] = "true" if include_all: params["all"] = "true" # FLY-657: prefer canonical /sprints; /cycles is retained as legacy alias result = self.relay.get("/sprints", params=params) # FLY-657 normalized shape may be { sprints: [...], meta: {...} } or a list if isinstance(result, dict) and "sprints" in result: return result.get("sprints", []) if isinstance(result, dict) and "cycles" in result: return result.get("cycles", []) return result if isinstance(result, list) else [] def assign_sprint(self, ref: str, sprint_id: str | None = None) -> dict: """FLY-656: Assign an issue to a sprint. Alias for assign_cycle.""" return self.assign_cycle(ref, sprint_id) def project_update(self, health: str, body: str, project_id: str | None = None, **_kwargs: object) -> dict: if self.is_cloud: payload: dict = {"health": health, "body": body} if project_id: payload["projectId"] = project_id result = self.relay.post("/projects/update", payload) return {"success": result.get("success", True), "id": result.get("id", "")} from _local.file_store import project_update return project_update(self.project_root, health, body) def session_update_create(self, envelope: dict, project_id: str | None = None) -> dict: """Store a SessionUpdate v1 record, with the provider update as a destination of it (FLY-1411, phase 10 spec §4.2). This is the governed replacement for posting a wrap straight to `/projects/update`: the record is the trunk, and `destinations.providerProjectUpdate` asks the relay to render it into the provider's project feed afterwards. A provider with no project-update concept still gets a record and reports `providerUpdate.posted: false` — which is how Jira and GitHub Issues teams stop being the tier that gets nothing. `raise_on_error` so the caller can read `WRAP_VALIDATION_FAILED` as a code and reproduce the local validator's refusal, rather than reading it back out of a rendered sentence. """ self.require_cloud("session_update_create") payload = dict(envelope) destination: dict = {} if project_id: destination["projectId"] = project_id payload["destinations"] = {"providerProjectUpdate": destination} # `window` and `sessionId` are the two fields that differ between two # attempts at the *same* wrap: the second is composed a few seconds # later, and its sequence number has moved on if the first attempt got # as far as appending. Both are therefore out of the seed. Everything # that identifies the intent — repo, health, issues, narrative, # destinations — stays in, so two genuinely different wraps under one # operation seed still key differently, and a retry of one wrap replays # the record instead of storing a second. # # **This projection is shared, not local.** `sessionUpdate.create` on # the relay (FLY-1410) hashes the same one — body minus `window` and # `sessionId`, `destinations` included — for its own idempotency key. # A client that quietly changed the shape here would not fail a test; # it would file a second record under a key the server thought was new. # Change one side and you are changing both. id_payload = { k: v for k, v in payload.items() if k not in ("window", "sessionId") } return self.relay.post("/session-updates", payload, raise_on_error=True, id_payload=id_payload) def status_summary(self) -> dict: if self.is_cloud: # Cloud could use relay, but local summary is always available pass from _local.file_store import status_summary return status_summary(self.project_root) # --- Workspace operations (cloud only) --- def validate_setup(self) -> dict: self.require_cloud("validate_setup") return self.relay.get("/auth/config") def list_labels(self) -> list[dict]: self.require_cloud("list_labels") result = self.relay.get("/labels") return result if isinstance(result, list) else [] def list_statuses(self) -> list[dict]: self.require_cloud("list_statuses") result = self.relay.get("/auth/statuses") return result if isinstance(result, list) else [] def list_providers(self) -> list[dict]: self.require_cloud("list_providers") result = self.relay.get("/providers") return result if isinstance(result, list) else [] def list_teams(self) -> list[dict]: self.require_cloud("list_teams") result = self.relay.get("/teams") return result if isinstance(result, list) else [] def create_team(self, name: str, key: str | None = None, description: str | None = None, parent: str | None = None) -> dict: self.require_cloud("create_team") body: dict = {"name": name} if key: body["key"] = key if description: body["description"] = description if parent: body["parentId"] = parent result = self.relay.post("/teams", body) return {"id": result["id"], "name": result["name"], "key": result.get("key", "")} # --------------------------------------------------------------------------- # Module-level helpers # --------------------------------------------------------------------------- _client: Optional[FlyDocsClient] = None def get_client() -> FlyDocsClient: """Get or create singleton client.""" global _client if _client is None: _client = FlyDocsClient() return _client def output_json(data: dict | list) -> None: """Print JSON to stdout — standard contract output.""" print(json.dumps(data)) def fail(message: str) -> None: """Print error to stderr and exit 1.""" print(message, file=sys.stderr) sys.exit(1) def _format_transition_recovery_prompt(error_data: dict, error_msg: str) -> str: """ FLY-653: Format a structured transition error into an agent-friendly recovery prompt. Handles two shapes: - Structured (FLY-651): { code: STATUS_NOT_REACHABLE, flydocsStatus, mappedTarget, currentState, availableTransitions: [{flydocsStatus?, jiraTarget}, ...] } - Legacy (today): { code: STATUS_MAPPING_ERROR, error: "No transition found for status X. Available: A → B, C → D, ..." } Output is multi-line plain text that the agent can read and act on. Includes a dashboard link so the user can fix mappings if the problem is configuration. """ flydocs_status = error_data.get("flydocsStatus") mapped_target = error_data.get("mappedTarget") current_state = error_data.get("currentState") available = error_data.get("availableTransitions") dashboard_url = "https://app.flydocs.ai/workspace/settings/status-mapping" lines = ["Transition blocked by provider workflow:"] if flydocs_status and mapped_target: lines.append( f" You asked: {flydocs_status} (mapped to '{mapped_target}' in this workspace)" ) elif flydocs_status: lines.append(f" You asked: {flydocs_status}") if current_state: lines.append(f" Currently: {current_state}") if flydocs_status and mapped_target and current_state: lines.append( f" Problem: No direct path from '{current_state}' to '{mapped_target}' in your workflow" ) elif not (flydocs_status or mapped_target or current_state): # No structured fields — fall through with the raw error message lines.append(f" Error: {error_msg}") # Available transitions (structured shape from FLY-651) if isinstance(available, list) and available: lines.append("") lines.append(" Available from here:") for t in available: if not isinstance(t, dict): continue flydocs = t.get("flydocsStatus") target = t.get("jiraTarget") or t.get("target") or t.get("name") if flydocs and target: lines.append(f" → {flydocs:<12} ({target})") elif target: lines.append(f" → {target} (no FlyDocs equivalent)") elif not available and "Available:" in error_msg: # Legacy unstructured format — extract the list from the error string try: avail_part = error_msg.split("Available:", 1)[1].strip() lines.append("") lines.append(" Available transitions:") for item in avail_part.split(","): lines.append(f" → {item.strip()}") except (IndexError, AttributeError): pass lines.append("") lines.append( " Choose an available status and re-run the transition, or update your" ) lines.append(f" status mapping at {dashboard_url}") return "\n".join(lines) def resolve_text_input(text_arg: str | None = None, file_arg: str | None = None) -> str | None: """Resolve text from --file > --text > stdin. Shared helper for dispatchers. FLY-699: Priority changed to file > text > stdin (was file > text > stdin). When called from subprocess harnesses (Claude Code, CI, etc.), stdin is often a pipe that's open but empty and never closed. isatty() returns False but sys.stdin.read() blocks forever waiting for EOF. Use stdin_has_data() to check non-blockingly before reading. """ if file_arg: path = Path(file_arg) if not path.exists(): fail(f"File not found: {file_arg}") return path.read_text() if text_arg is not None: return text_arg if stdin_has_data(): return sys.stdin.read().strip() return None def stdin_has_data() -> bool: """Non-blocking check: does stdin have data ready to read? Returns False if stdin is a TTY (interactive), an empty/open pipe, or otherwise not ready. Returns True only if data is actually available, preventing hangs when scripts are invoked from harnesses that don't close stdin. """ if sys.stdin.isatty(): return False try: import select # 0 timeout — check without blocking ready, _, _ = select.select([sys.stdin], [], [], 0) return bool(ready) except (ValueError, OSError): # select() can fail on some platforms / stdin states — play it safe return False