linear

#!/usr/bin/env python3
"""linear - Task management CLI for AI agent teams.

General-purpose Linear CLI designed for orchestrating work across teams of
AI agents. Each agent can query its work queue, pick up tasks, report progress,
and see the full team board.

Commands:
    setup               Configure Linear connection
    tasks               List/view/board tasks (paginated; --cycle
                        active|next|all|none|<name|id>, --since, --assignee)
    update              Modify one or many issues (status, comment, relations);
                        bulk via multiple ids or --stdin
    create              Create a new issue
    cycles              List / create / update cycles
    projects            List / show / create / update / archive / delete projects
    milestones          List / create / move / retarget / delete milestones
    initiatives         List / show / create / update / link / unlink / archive
    goals               Company, week and day goals; `set` and `todo` write yours
    labels              List / create / update / delete labels
    users               List assignable users
    agents              List agent members you can --delegate to (auto-detected)
    migrate-agent-labels
                        One-time: move legacy agent:<name> labels onto delegates
    states              List the team's workflow states
    queue               Durable local queue for closes that hit rate limits
                        (`queue drain` applies pending closes)

Config: ~/.linear-cli/config.json
Cache (team board for --similar, optional embedding vectors): ~/.linear-cli/cache.sqlite
"""

from __future__ import annotations

import argparse
import array
import contextlib
import difflib
import errno
import hashlib
import json
import math
import mimetypes
import os
import re
import sqlite3
import subprocess
import sys
from datetime import date, datetime, timedelta, timezone, tzinfo
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import URLError

# File locking is platform-split: fcntl on POSIX, msvcrt on Windows. Importing
# fcntl unconditionally at module scope took the whole CLI down on Windows —
# every command, not just `queue` — so both are optional and the lock helpers
# below dispatch on whichever landed.
try:
    import fcntl
except ImportError:  # Windows
    fcntl = None
try:
    import msvcrt
except ImportError:  # POSIX
    msvcrt = None

# msvcrt.locking reports contention as EDEADLOCK (Windows spells it that way;
# POSIX errno only has EDEADLK) or EACCES. Resolved once, defensively, so the
# lookup can't AttributeError on whichever platform is running.
_LOCK_CONTENDED_ERRNOS = tuple(
    e for e in (getattr(errno, "EDEADLOCK", None), getattr(errno, "EDEADLK", None),
                getattr(errno, "EACCES", None))
    if e is not None
)

__version__ = "0.31.0"

# Sentinel for "flag not supplied" — distinct from None, which means an explicit
# clear (e.g. `--project none`). Lets update pre-resolve a field once and pass
# the result down without re-deriving whether the flag was present.
_UNSET = object()

CONFIG_PATH = Path.home() / ".linear-cli" / "config.json"
LEGACY_CONFIG_PATH = Path.home() / ".agents" / "linear.json"
API_URL = "https://api.linear.app/graphql"
API_TIMEOUT = 20

# Config holds the Linear API key — keep it private on multi-user boxes.
CONFIG_DIR_MODE = 0o700
CONFIG_FILE_MODE = 0o600

# How long a cached agent roster stays fresh before an auto-refresh. Agent apps
# (Claude, Codex, Kimi, …) are installed/removed rarely, so a periodic refresh
# keeps `--delegate <name>` resolving without a per-call lookup.
AGENTS_TTL_SECONDS = 6 * 3600

# Durable local queue for closes that could not be written immediately (e.g.
# Linear rate limit). One JSON intent per issue identifier; the directory is
# kept private because it carries proof URLs/text from the user's workflow.
QUEUE_DIR = Path.home() / ".linear-cli" / "queue"
MAX_QUEUE_SIZE = 1000
MAX_QUEUE_ATTEMPTS = 10
QUEUE_BACKOFF_BASE_SECONDS = 2
QUEUE_BACKOFF_MAX_SECONDS = 300  # 5 minutes


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

def _harden_config_perms() -> None:
    """Best-effort: force 0700/0600 on an existing config tree.

    Upgrades written before this release (or created with a loose umask) stay
    world-readable until the next `save_config`. Touching modes on every load
    closes that window without requiring a re-setup.
    """
    try:
        if CONFIG_PATH.parent.exists():
            CONFIG_PATH.parent.chmod(CONFIG_DIR_MODE)
        if CONFIG_PATH.exists():
            CONFIG_PATH.chmod(CONFIG_FILE_MODE)
    except OSError:
        pass


def write_config_file(cfg: dict) -> None:
    """Write config.json with private dir/file modes (API key lives here)."""
    CONFIG_PATH.parent.mkdir(mode=CONFIG_DIR_MODE, parents=True, exist_ok=True)
    # mkdir(mode=…) is ignored when the dir already exists — always re-assert.
    CONFIG_PATH.parent.chmod(CONFIG_DIR_MODE)
    data = (json.dumps(cfg, indent=2) + "\n").encode()
    # O_CREAT with mode is masked by umask; chmod after write is the real gate.
    fd = os.open(CONFIG_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, CONFIG_FILE_MODE)
    try:
        os.write(fd, data)
    finally:
        os.close(fd)
    CONFIG_PATH.chmod(CONFIG_FILE_MODE)


def load_config() -> dict:
    if CONFIG_PATH.exists():
        _harden_config_perms()
        return json.loads(CONFIG_PATH.read_text())
    if LEGACY_CONFIG_PATH.exists():
        cfg = json.loads(LEGACY_CONFIG_PATH.read_text())
        write_config_file(cfg)
        print(f"Migrated config: {LEGACY_CONFIG_PATH} -> {CONFIG_PATH}", file=sys.stderr)
        return cfg
    return {}


def save_config(cfg: dict) -> bool:
    """Persist config. Returns False when the file could not be written.

    Most writes are cache refreshes (the agent roster, the states map) on a
    read path. A read-only or unwritable config dir must not take down the
    command that triggered the refresh — the in-memory values are still good for
    this run. `setup` checks the return value, because there persistence IS the
    operation.

    Always writes dir `0700` and file `0600` so the API key is not world-readable.
    """
    # Volatile cfg (e.g. --team override) opts out of persistence.
    if cfg.get("__volatile__"):
        return True
    try:
        write_config_file(cfg)
        return True
    except OSError as e:
        print(f"Warning: could not write {CONFIG_PATH}: {e}", file=sys.stderr)
        return False


def get_api_key(cfg: dict) -> str:
    """Resolve API key: config > env > Keychain."""
    key = cfg.get("apiKey") or os.environ.get("LINEAR_API_KEY")
    if key:
        return key
    try:
        result = subprocess.run(
            ["security", "find-generic-password", "-s", "linear-api-key", "-w"],
            capture_output=True, text=True,
        )
        if result.returncode == 0 and result.stdout.strip():
            return result.stdout.strip()
    except FileNotFoundError:
        pass
    return ""


def get_team_id(cfg: dict) -> str:
    """Resolve team ID: config > env > Keychain."""
    tid = cfg.get("teamId") or os.environ.get("LINEAR_TEAM_ID")
    if tid:
        return tid
    try:
        result = subprocess.run(
            ["security", "find-generic-password", "-s", "linear-team-id", "-w"],
            capture_output=True, text=True,
        )
        if result.returncode == 0 and result.stdout.strip():
            return result.stdout.strip()
    except FileNotFoundError:
        pass
    return ""


# ---------------------------------------------------------------------------
# GraphQL client
# ---------------------------------------------------------------------------

def _header_retry_after(err: BaseException) -> int | None:
    """Parse Retry-After (seconds) from an HTTPError-like response, if present."""
    headers = getattr(err, "headers", None)
    if headers is None:
        return None
    raw = None
    try:
        raw = headers.get("Retry-After") or headers.get("retry-after")
    except Exception:
        return None
    if raw is None:
        return None
    try:
        seconds = int(str(raw).strip())
    except (TypeError, ValueError):
        return None
    if seconds < 0:
        return None
    return seconds


def _annotate_error_extensions(err: dict, status: int | None,
                               retry_after: int | None) -> None:
    """Attach HTTP status / Retry-After onto a GraphQL error's extensions."""
    if not isinstance(err, dict):
        return
    if status is None and retry_after is None:
        return
    ext = err.setdefault("extensions", {})
    if status is not None:
        ext.setdefault("status", status)
    if retry_after is not None:
        ext.setdefault("retry_after", retry_after)


# Linear's "duplicate" state type is closed work, like completed and canceled.
OPEN_STATE_FILTER = 'state: { type: { nin: ["completed", "canceled", "duplicate"] } }'


def gql(api_key: str, query: str, variables: dict | None = None) -> dict:
    payload: dict = {"query": query}
    if variables:
        payload["variables"] = variables
    body = json.dumps(payload).encode()

    req = Request(API_URL, data=body, method="POST")
    req.add_header("Content-Type", "application/json")
    req.add_header("Authorization", api_key)

    try:
        with urlopen(req, timeout=API_TIMEOUT) as resp:
            return json.loads(resp.read())
    except (URLError, TimeoutError) as e:
        # HTTPError has a response body with actual GraphQL error details
        body = ""
        status = getattr(e, "code", None)
        retry_after = _header_retry_after(e)
        if hasattr(e, "read"):
            try:
                body = e.read().decode("utf-8", errors="replace")
                parsed = json.loads(body)
                if "errors" in parsed:
                    # Surface the HTTP status / Retry-After so callers can
                    # distinguish rate-limit (429) and honor server backoff.
                    for err in parsed.get("errors", []):
                        _annotate_error_extensions(err, status, retry_after)
                    return parsed
            except (json.JSONDecodeError, Exception):
                pass
        timed_out = isinstance(e, TimeoutError) or isinstance(getattr(e, "reason", None), TimeoutError)
        detail = (f"Linear API timeout after {API_TIMEOUT} seconds"
                  if timed_out else body if body else str(e))
        err: dict = {"message": detail}
        _annotate_error_extensions(err, status, retry_after)
        return {"errors": [err]}


def check_errors(data: dict) -> bool:
    errors = data.get("errors")
    if errors:
        print(f"Error: {errors[0].get('message', 'Unknown error')}", file=sys.stderr)
        return True
    return False


def _error_status(err: dict) -> int | None:
    """HTTP status carried by a GraphQL error extension, if any."""
    ext = err.get("extensions") or {}
    status = ext.get("status")
    return int(status) if isinstance(status, int) or (isinstance(status, str) and status.isdigit()) else None


def _error_message(err: dict) -> str:
    return str(err.get("message", "")).lower()


def _error_retry_after(err: dict) -> int | None:
    """Retry-After seconds from a GraphQL error extension, if any."""
    ext = err.get("extensions") or {}
    raw = ext.get("retry_after")
    if raw is None:
        return None
    try:
        seconds = int(raw)
    except (TypeError, ValueError):
        return None
    return seconds if seconds >= 0 else None


def is_rate_limited(data: dict) -> bool:
    """True when Linear answered 429 or the error text names a rate limit."""
    for err in data.get("errors", []):
        if not isinstance(err, dict):
            continue
        if _error_status(err) == 429:
            return True
        msg = _error_message(err)
        if "rate limit" in msg or "too many requests" in msg:
            return True
    return False


def is_transient_error(data: dict) -> bool:
    """True for errors where a retry may reasonably succeed: 429/502/503/504,
    timeouts, and network blips. Permanent GraphQL validation errors are not."""
    if is_rate_limited(data):
        return True
    transient_statuses = {429, 502, 503, 504}
    transient_phrases = ("timeout", "temporary", "unreachable", "connection",
                         "name or service not known", "nodename nor servname")
    for err in data.get("errors", []):
        if not isinstance(err, dict):
            continue
        status = _error_status(err)
        if status in transient_statuses:
            return True
        msg = _error_message(err)
        if any(p in msg for p in transient_phrases):
            return True
    return False


def response_retry_after_seconds(data: dict) -> int | None:
    """Best Retry-After (seconds) from a GraphQL error response, if any."""
    best: int | None = None
    for err in data.get("errors", []):
        if not isinstance(err, dict):
            continue
        seconds = _error_retry_after(err)
        if seconds is None:
            continue
        if best is None or seconds > best:
            best = seconds
    return best


# ---------------------------------------------------------------------------
# Durable close queue
# ---------------------------------------------------------------------------

def queue_dir() -> Path:
    """Return the directory that holds queued close intents."""
    return QUEUE_DIR


def _intent_path(identifier: str) -> Path:
    """Safe filename for an intent: one JSON file per identifier."""
    safe = identifier.replace("/", "_").replace("\\", "_")
    return queue_dir() / f"{safe}.json"


def _ensure_queue_dir() -> None:
    """Create the queue directory with the same private mode as config."""
    qd = queue_dir()
    qd.mkdir(mode=CONFIG_DIR_MODE, parents=True, exist_ok=True)
    try:
        qd.chmod(CONFIG_DIR_MODE)
    except OSError:
        pass


def list_queue_intents() -> list[dict]:
    """Return all queued intents, oldest first (by filename for stability)."""
    qd = queue_dir()
    if not qd.exists():
        return []
    intents = []
    for path in sorted(qd.glob("*.json")):
        try:
            intents.append(json.loads(path.read_text()))
        except (json.JSONDecodeError, OSError):
            # Corrupt file: leave it; a human can inspect, and we won't lose it.
            pass
    return intents


def save_queue_intent(intent: dict) -> bool:
    """Persist an intent to disk. Overwrites any existing intent for the same
    identifier so duplicate closes collapse to the latest desired state.
    Returns False only when the queue is at MAX_QUEUE_SIZE and this would add
    a new file (existing intents can still be updated).
    """
    _ensure_queue_dir()
    identifier = intent.get("identifier", "")
    if not identifier:
        return False
    path = _intent_path(identifier)
    qd = queue_dir()
    existing = set(qd.glob("*.json"))
    is_new = path not in existing
    if is_new and len(existing) >= MAX_QUEUE_SIZE:
        return False
    tmp = path.with_suffix(".tmp")
    try:
        tmp.write_text(json.dumps(intent, indent=2) + "\n")
        tmp.replace(path)
        return True
    except OSError:
        return False


def remove_queue_intent(identifier: str) -> None:
    """Delete an intent once it has been successfully applied."""
    path = _intent_path(identifier)
    if path.exists():
        try:
            path.unlink()
        except OSError:
            pass


def queue_backoff_delay(attempts: int) -> int:
    """Exponential backoff, bounded by QUEUE_BACKOFF_MAX_SECONDS."""
    return min(
        QUEUE_BACKOFF_BASE_SECONDS * (2 ** max(0, attempts - 1)),
        QUEUE_BACKOFF_MAX_SECONDS,
    )


def resolve_backoff_delay(attempts: int, retry_after: int | None = None) -> int:
    """Prefer server Retry-After when present; otherwise exponential backoff.

    Both paths are capped at QUEUE_BACKOFF_MAX_SECONDS so a pathological
    Retry-After cannot stall the queue indefinitely.
    """
    if retry_after is not None:
        try:
            seconds = int(retry_after)
        except (TypeError, ValueError):
            seconds = -1
        if seconds >= 0:
            return min(max(seconds, 0), QUEUE_BACKOFF_MAX_SECONDS)
    return queue_backoff_delay(attempts)


def _now_utc() -> datetime:
    return datetime.now(timezone.utc)


def _lock_fd(fd: int, blocking: bool) -> bool:
    """Take an exclusive OS lock on fd. True when held, False when another
    process holds it. Both backends release on process death, so a killed drain
    cannot wedge the queue behind a stale lock."""
    if fcntl is not None:
        flags = fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB
        try:
            fcntl.flock(fd, flags)
            return True
        except (BlockingIOError, OSError):
            return False
    if msvcrt is not None:
        # Byte-range lock on the first byte, from the current file position —
        # hence the seek before both lock and unlock.
        #
        # LK_LOCK is NOT flock's LOCK_EX: it retries for ~10s and then raises,
        # where LOCK_EX waits forever. Loop it so blocking=True means blocking
        # on both platforms instead of quietly returning "not held" after 10s.
        while True:
            os.lseek(fd, 0, os.SEEK_SET)
            try:
                msvcrt.locking(fd, msvcrt.LK_LOCK if blocking else msvcrt.LK_NBLCK, 1)
                return True
            except OSError as exc:
                if not blocking:
                    return False
                # Contention is EDEADLOCK (LK_LOCK gave up after its own ~10s of
                # retries) or EACCES. Keep waiting on those, the way LOCK_EX
                # would. Anything else is a real I/O failure — surface it rather
                # than retry a broken fd forever.
                if exc.errno not in _LOCK_CONTENDED_ERRNOS:
                    raise
                # Ctrl-C during the wait propagates, same as on POSIX.
    # Neither backend exists (not a platform Python actually ships today: POSIX
    # has fcntl, Windows has msvcrt). Nothing serializes concurrent drains here,
    # so warn rather than let a caller believe it holds an exclusive lock.
    print("Warning: no file-locking primitive on this platform; concurrent "
          "`queue drain` runs are not serialized.", file=sys.stderr)
    return True


def _unlock_fd(fd: int) -> None:
    try:
        if fcntl is not None:
            fcntl.flock(fd, fcntl.LOCK_UN)
        elif msvcrt is not None:
            os.lseek(fd, 0, os.SEEK_SET)
            msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
    except OSError:
        pass


@contextlib.contextmanager
def queue_drain_lock(blocking: bool = False):
    """Exclusive lock around drain so concurrent drains on one machine are safe.

    Yields True when the lock is held. Yields False when non-blocking and another
    drain already holds the lock (caller should skip, not corrupt intents).
    """
    _ensure_queue_dir()
    lock_path = queue_dir() / ".drain.lock"
    fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o600)
    held = False
    try:
        held = _lock_fd(fd, blocking)
        yield held
    finally:
        if held:
            _unlock_fd(fd)
        try:
            os.close(fd)
        except OSError:
            pass


def build_close_intent(identifier: str, issue_id: str, state: str,
                       proof: list[str] | None = None,
                       comment: str | None = None) -> dict:
    """Construct a durable intent for closing an issue with proof.

    The intent never stores the API key or any credential; only the public
    issue identifier, the resolved issue id, the desired state name, proof
    items, and an optional comment.
    """
    return {
        "identifier": identifier,
        "issue_id": issue_id,
        "state": state,
        "proof": list(proof or []),
        "comment": comment,
        "proof_posted": False,
        "created_at": _now_utc().isoformat(),
        "attempts": 0,
        "next_attempt": None,
    }


def apply_close_intent(api_key: str, team_id: str, cfg: dict,
                       intent: dict) -> tuple[bool, bool]:
    """Apply one queued close intent. Returns (success, transient).

    * success  = the issue is now in the desired state (or already was).
    * transient = the failure should be retried later (rate limit / network).

    Idempotent: if the issue already has the desired state, the intent is
    considered successful and no duplicate proof is posted.

    On a transient failure, any server Retry-After is stashed on the intent as
    ``last_retry_after`` so drain can honor it.
    """
    identifier = intent.get("identifier", "")
    issue_id = intent.get("issue_id")
    desired_state = intent.get("state", "")
    intent.pop("last_retry_after", None)

    def _transient(data: dict | None = None) -> tuple[bool, bool]:
        if data is not None:
            ra = response_retry_after_seconds(data)
            if ra is not None:
                intent["last_retry_after"] = ra
        return False, True

    # Resolve fresh state each attempt so a close that already landed is a no-op.
    issue = resolve_issue(api_key, team_id, identifier)
    if not issue:
        # resolve_issue already printed the error. Treat as transient because
        # "not found" after a rate limit is often a misleading symptom.
        return _transient()

    current_state = (issue.get("state") or {}).get("name", "")
    if current_state == desired_state:
        return True, False

    # Issue id can change if an issue is moved teams; prefer the fresh one.
    issue_id = issue.get("id") or issue_id

    states = get_states(api_key, team_id, cfg)
    state_id = resolve_state_id(states, desired_state)
    if not state_id:
        print(f"State '{desired_state}' not found. Available: {', '.join(states.keys())}",
              file=sys.stderr)
        return False, False

    proof = intent.get("proof") or []
    comment = intent.get("comment")
    proof_posted = intent.get("proof_posted", False)
    if proof and not proof_posted:
        proof_body = build_proof_comment(api_key, proof)
        if comment:
            proof_body = f"{comment}\n\n{proof_body}"
        data = gql(api_key, """
            mutation($input: CommentCreateInput!) {
                commentCreate(input: $input) { success }
            }
        """, {"input": {"issueId": issue_id, "body": proof_body}})
        if is_transient_error(data):
            return _transient(data)
        if check_errors(data):
            # A non-transient proof failure (e.g. missing file) should not
            # block forever; the caller will count attempts and eventually drop.
            return False, False
        if not data.get("data", {}).get("commentCreate", {}).get("success"):
            return False, False
        intent["proof_posted"] = True
        save_queue_intent(intent)

    data = gql(api_key, """
        mutation($id: String!, $stateId: String!) {
            issueUpdate(id: $id, input: { stateId: $stateId }) {
                success
                issue { identifier title state { name } }
            }
        }
    """, {"id": issue_id, "stateId": state_id})

    if is_transient_error(data):
        return _transient(data)
    if check_errors(data):
        return False, False

    result = data.get("data", {}).get("issueUpdate") or {}
    if result.get("success"):
        return True, False
    return False, False


def drain_queue(api_key: str, team_id: str, cfg: dict,
                quiet: bool = False, once: bool = False,
                acquire_lock: bool = True) -> tuple[int, int]:
    """Apply queued intents that are due. Returns (applied, remaining).

    Each intent is retried with exponential backoff (or server Retry-After)
    up to MAX_QUEUE_ATTEMPTS. Intents that exhaust attempts are dropped so the
    queue cannot grow forever on permanently stuck items.

    Concurrent drains on one machine serialize via ``queue_drain_lock``. When
    another drain holds the lock, this call is a no-op (returns 0, remaining).
    Pass ``once=True`` to process at most one due intent.
    """
    if acquire_lock:
        with queue_drain_lock(blocking=False) as held:
            if not held:
                if not quiet:
                    print("Another drain holds the queue lock; skipping.",
                          file=sys.stderr)
                return 0, len(list_queue_intents())
            return drain_queue(api_key, team_id, cfg, quiet=quiet, once=once,
                               acquire_lock=False)

    intents = list_queue_intents()
    if not intents:
        if not quiet:
            print("Queue empty.")
        return 0, 0

    applied = 0
    remaining = 0
    processed_due = 0
    for intent in intents:
        identifier = intent.get("identifier", "")
        attempts = intent.get("attempts", 0) or 0
        next_attempt = intent.get("next_attempt")
        if next_attempt and _now_utc().isoformat() < next_attempt:
            remaining += 1
            continue

        if once and processed_due >= 1:
            remaining += 1
            continue

        processed_due += 1
        ok, transient = apply_close_intent(api_key, team_id, cfg, intent)
        if ok:
            remove_queue_intent(identifier)
            applied += 1
            if not quiet:
                print(f"Applied {identifier} -> {intent.get('state', 'Done')}")
            continue

        if transient:
            attempts += 1
            if attempts >= MAX_QUEUE_ATTEMPTS:
                remove_queue_intent(identifier)
                if not quiet:
                    print(f"Gave up on {identifier} after {attempts} attempts.",
                          file=sys.stderr)
            else:
                delay = resolve_backoff_delay(
                    attempts, intent.get("last_retry_after"),
                )
                intent.pop("last_retry_after", None)
                intent["attempts"] = attempts
                intent["next_attempt"] = (
                    _now_utc() + timedelta(seconds=delay)
                ).isoformat()
                save_queue_intent(intent)
                remaining += 1
                if not quiet:
                    print(f"Retained {identifier} in queue (retry {attempts} in {delay}s)")
        else:
            remove_queue_intent(identifier)
            if not quiet:
                print(f"Dropped {identifier}: permanent failure.", file=sys.stderr)

    return applied, remaining


def queue_close_and_try(api_key: str, team_id: str, cfg: dict,
                        issue: dict, state: str,
                        proof: list[str] | None = None,
                        comment: str | None = None,
                        quiet: bool = False) -> bool:
    """Persist a close intent, attempt it immediately, and keep it on transient
    failure. Returns True when the issue is applied or durably queued; False on
    a permanent failure (e.g. unknown state or queue full).
    """
    identifier = issue.get("identifier", "")
    intent = build_close_intent(identifier, issue.get("id"), state,
                                proof=proof, comment=comment)
    if not save_queue_intent(intent):
        print(f"Queue full; could not queue close for {identifier}.",
              file=sys.stderr)
        return False

    ok, transient = apply_close_intent(api_key, team_id, cfg, intent)
    if ok:
        remove_queue_intent(identifier)
        if not quiet:
            print(f"{identifier} -> {state}")
        return True
    if transient:
        if not quiet:
            print(f"{identifier} -> queued for retry (rate limited / transient)")
        return True
    remove_queue_intent(identifier)
    return False


# ---------------------------------------------------------------------------
# State resolution (dynamic, not hardcoded)
# ---------------------------------------------------------------------------

def get_states(api_key: str, team_id: str, cfg: dict) -> dict:
    cached = cfg.get("states")
    if cached:
        return cached

    data = gql(api_key, """
        query($teamId: ID!) {
            workflowStates(filter: { team: { id: { eq: $teamId } } }) {
                nodes { id name type }
            }
        }
    """, {"teamId": team_id})

    if check_errors(data):
        return {}

    states = {}
    for node in data["data"]["workflowStates"]["nodes"]:
        states[node["name"]] = {"id": node["id"], "type": node["type"]}

    cfg["states"] = states
    save_config(cfg)
    return states


def resolve_state_id(states: dict, name: str) -> str | None:
    if name in states:
        return states[name]["id"]
    for k, v in states.items():
        if k.lower() == name.lower():
            return v["id"]
    aliases = {
        "progress": "In Progress",
        "in-progress": "In Progress",
        "wip": "In Progress",
    }
    mapped = aliases.get(name.lower(), "")
    if mapped in states:
        return states[mapped]["id"]
    return None


# ---------------------------------------------------------------------------
# Viewer + user resolution (for auto-assign on create)
# ---------------------------------------------------------------------------

def get_viewer_id(api_key: str, cfg: dict) -> str | None:
    """Return the API-key owner's user ID. Cached in config after first lookup."""
    cached = cfg.get("viewerId")
    if cached:
        return cached

    data = gql(api_key, "{ viewer { id name email } }")
    if check_errors(data):
        return None

    viewer = data.get("data", {}).get("viewer") or {}
    vid = viewer.get("id")
    if vid:
        cfg["viewerId"] = vid
        # Stash email too for setup output; not load-bearing.
        if viewer.get("email"):
            cfg["viewerEmail"] = viewer["email"]
        save_config(cfg)
    return vid


def resolve_team_id_by_key(api_key: str, value: str) -> str | None:
    """Resolve --team to a team ID. Accepts team key (e.g. 'ANT') or UUID."""
    if not value:
        return None
    if len(value) == 36 and value.count("-") == 4:
        return value
    data = gql(api_key, """
        query($key: String!) {
            teams(filter: { key: { eqIgnoreCase: $key } }) {
                nodes { id name key }
            }
        }
    """, {"key": value})
    if check_errors(data):
        return None
    nodes = (data.get("data") or {}).get("teams", {}).get("nodes", [])
    if not nodes:
        print(f"Warning: team '{value}' not found.", file=sys.stderr)
        return None
    return nodes[0]["id"]


def resolve_user_id_by_email(api_key: str, email: str) -> str | None:
    """Look up a user by email address. Returns user ID or None."""
    data = gql(api_key, """
        query($email: String!) {
            users(filter: { email: { eq: $email } }) {
                nodes { id name email }
            }
        }
    """, {"email": email})
    if check_errors(data):
        return None
    nodes = data.get("data", {}).get("users", {}).get("nodes", [])
    if not nodes:
        return None
    return nodes[0]["id"]


def list_workspace_users(api_key: str) -> list[dict]:
    """All active workspace members, carrying the `app` flag so callers can split
    humans (assignable) from agents (delegatable). Fully paginated."""
    query = """query($cursor: String) {
        users(filter: { active: { eq: true } }, first: 100, after: $cursor) {
            pageInfo { hasNextPage endCursor }
            nodes { id name email displayName app guest avatarUrl }
        }
    }"""
    return paginate_connection(api_key, query, ["users"]) or []


def resolve_assignee_id(api_key: str, value: str) -> str | None:
    """Resolve --assign to a user ID. Accepts an email, or a human's name /
    displayName (case-insensitive) so `--assign bisma` works like `--delegate
    claude`. Agents (app users) are excluded on purpose — delegate those instead.
    Exact match wins over substring; ambiguous names note the pick to stderr."""
    v = (value or "").strip()
    if not v:
        return None
    if "@" in v:
        return resolve_user_id_by_email(api_key, v)
    humans = [u for u in list_workspace_users(api_key) if not u.get("app")]
    lv = v.lower()
    matches = [u for u in humans
               if lv in (u["name"].lower(), (u.get("displayName") or "").lower())]
    if not matches:
        matches = [u for u in humans
                   if lv in u["name"].lower() or lv in (u.get("displayName") or "").lower()]
    if not matches:
        return None
    if len(matches) > 1:
        print(f"Note: '{value}' matched {len(matches)} people; picked '{matches[0]['name']}'.",
              file=sys.stderr)
    return matches[0]["id"]


# ---------------------------------------------------------------------------
# Agent detection (Linear app users you can delegate to)
# ---------------------------------------------------------------------------

def get_agents(api_key: str, cfg: dict, force: bool = False) -> list[dict]:
    """Return the workspace's agent members — Linear app users (Claude, Codex,
    Kimi, Antigravity, Grok, Droid, …). Detected via the `app` user flag, cached
    in config, and auto-refreshed every AGENTS_TTL_SECONDS so the roster tracks
    apps installed/removed without a manual step. Pass force=True to refresh now.
    """
    cached = cfg.get("agents")
    fetched = cfg.get("agentsFetchedAt")
    if cached is not None and not force and fetched:
        try:
            age = (datetime.now(timezone.utc) - datetime.fromisoformat(fetched)).total_seconds()
            if age < AGENTS_TTL_SECONDS:
                return cached
        except ValueError:
            pass  # unparseable timestamp — fall through and refresh

    data = gql(api_key, """
        query {
            users(filter: { app: { eq: true }, active: { eq: true } }, first: 100) {
                nodes { id name }
            }
        }
    """)
    if check_errors(data):
        return cached or []  # keep serving the stale roster on a transient error

    nodes = (data.get("data") or {}).get("users", {}).get("nodes", []) or []
    agents = sorted(
        [{"id": n["id"], "name": n["name"]} for n in nodes],
        key=lambda a: a["name"].lower(),
    )
    cfg["agents"] = agents
    cfg["agentsFetchedAt"] = datetime.now(timezone.utc).isoformat()
    save_config(cfg)
    return agents


def resolve_agent_id(api_key: str, cfg: dict, name: str) -> str | None:
    """Resolve an agent name (e.g. 'claude') to its Linear user id, case- and
    space-insensitive. On a cache miss, refreshes once — a freshly-installed
    agent shouldn't have to wait out the TTL to be delegatable."""
    want = name.strip().lower()
    for force in (False, True):
        for agent in get_agents(api_key, cfg, force=force):
            if agent["name"].lower() == want:
                return agent["id"]
        # only bother with the forced refresh if the first pass missed
    return None


def resolve_agent_name(api_key: str, cfg: dict, name: str) -> str | None:
    """Resolve an agent name to the roster's canonical spelling ('claude' ->
    'Claude'), case- and space-insensitive. Ownership is Linear's native
    `delegate` field and delegate names come back in roster spelling, so a queue
    filter has to compare against the canonical name, not the raw argument."""
    want = name.strip().lower()
    for force in (False, True):
        for agent in get_agents(api_key, cfg, force=force):
            if agent["name"].lower() == want:
                return agent["name"]
    return None


def delegate_of(node: dict) -> str | None:
    """The agent an issue is delegated to, or None when it is unowned.
    Unowned means exactly one thing: the native `delegate` field is null."""
    return (node.get("delegate") or {}).get("name") or None


def delegated_to(node: dict, agent_name: str) -> bool:
    """True when this issue's native delegate is `agent_name` (case-insensitive)."""
    dn = delegate_of(node)
    return dn is not None and dn.lower() == agent_name.lower()


# ---------------------------------------------------------------------------
# Cycle resolution (for auto-attach on create, move on update, listing)
# ---------------------------------------------------------------------------

def cycle_label(node: dict | None) -> str:
    """Human label for a cycle node. Linear numbered cycles have name=None, so
    fall back to `Cycle {number}`, then to 'no cycle'. Selections feeding this
    must request `number` alongside `name`."""
    if not node:
        return "no cycle"
    name = node.get("name")
    if name:
        return name
    number = node.get("number")
    if number is not None:
        return f"Cycle {number}"
    return "no cycle"


def resolve_cycle_meta(api_key: str, team_id: str, which: str) -> dict | None:
    """Return active or next cycle metadata for a team. Never cached (cycles rotate)."""
    if which not in ("active", "next"):
        return None
    if which == "active":
        data = gql(api_key, f"""{{
            team(id: "{team_id}") {{ activeCycle {{ id number name startsAt endsAt }} }}
        }}""")
        if check_errors(data):
            return None
        cycle = (data.get("data") or {}).get("team", {}).get("activeCycle")
        return cycle or None

    # "next" — Linear has no team.nextCycle field. Find the first cycle whose
    # startsAt is after the active cycle's endsAt, ordered ascending.
    active_query = gql(api_key, f"""{{
        team(id: "{team_id}") {{
            activeCycle {{ endsAt }}
        }}
    }}""")
    if check_errors(active_query):
        return None
    active_ends = (
        ((active_query.get("data") or {}).get("team", {}) or {})
        .get("activeCycle", {}) or {}
    ).get("endsAt")
    if not active_ends:
        return None

    data = gql(api_key, """
        query($teamId: ID!, $after: DateTimeOrDuration!) {
            cycles(
                filter: {
                    team: { id: { eq: $teamId } }
                    startsAt: { gte: $after }
                }
                orderBy: updatedAt
                first: 20
            ) {
                nodes { id number name startsAt endsAt }
            }
        }
    """, {"teamId": team_id, "after": active_ends})
    if check_errors(data):
        return None

    nodes = (data.get("data") or {}).get("cycles", {}).get("nodes", []) or []
    nodes = [n for n in nodes if n.get("startsAt") and n["startsAt"] >= active_ends]
    nodes.sort(key=lambda n: n["startsAt"])
    return nodes[0] if nodes else None


def get_cycle_id(api_key: str, team_id: str, which: str) -> str | None:
    cycle = resolve_cycle_meta(api_key, team_id, which)
    return cycle.get("id") if cycle else None


# Every non-archived project a team can see (Linear's `team.projects` leaves
# archived ones out unless asked). Shared by the project list and the overview
# so both read the same rows.
_TEAM_PROJECTS_QUERY = """query($teamId: String!, $cursor: String) {
    team(id: $teamId) {
        projects(first: 100, after: $cursor) {
            pageInfo { hasNextPage endCursor }
            nodes {
                id name state status { name type } progress scope priority
                startDate targetDate updatedAt
                lead { name }
            }
        }
    }
}"""


def list_team_projects(api_key: str, team_id: str) -> list[dict]:
    """All projects accessible to a team. Sorted most-recently-updated first.
    Fully paginated — no silent truncation at Linear's default page size."""
    projects = paginate_connection(api_key, _TEAM_PROJECTS_QUERY, ["team", "projects"],
                                   {"teamId": team_id})
    if projects is None:
        return []
    for p in projects:
        p["state"] = project_state(p)
    projects.sort(key=lambda p: p.get("updatedAt") or "", reverse=True)
    return projects


def project_state(project: dict) -> str | None:
    """The project's status type, read from `status` (Linear deprecated `state`).
    Linear's settings UI files a Paused status under the Planned category, so
    its type is `planned`; a status named Paused is read as paused."""
    status = project.get("status") or {}
    if (status.get("name") or "").strip().lower() == "paused":
        return "paused"
    return status.get("type") or project.get("state")


def _suggest(value: str, names: list[str]) -> str:
    """A ' Did you mean: a, b?' hint for a name that didn't resolve — fuzzy
    close-matches first, then substring, so a typo or partial name is actionable
    instead of a dead end."""
    close = difflib.get_close_matches(value, names, n=3, cutoff=0.5)
    if not close:
        close = [n for n in names if value.lower() in n.lower()][:3]
    return f" Did you mean: {', '.join(close)}?" if close else ""


def _unknown_label_message(label: str, names: list[str]) -> str:
    """Why a label was skipped and what to use instead. Labels are the owner's
    taxonomy, so the fix is picking an existing one, never creating it — the
    message says so at the moment an agent would otherwise reach for
    `linear labels create`."""
    if not names:
        return (f"label '{label}' not found, skipping. The team has no labels yet; "
                "labels are owner-managed, so ask the owner rather than creating one.")
    names = sorted(names, key=str.lower)  # stable, scannable; the API order is creation order
    shown = ", ".join(names[:12]) + (", …" if len(names) > 12 else "")
    return (f"label '{label}' not found, skipping.{_suggest(label, names)} "
            f"Labels are owner-managed: pick one of the team's {len(names)} labels "
            f"({shown}); full list: linear labels.")


def _sort_by_recent(nodes: list[dict]) -> list[dict]:
    return sorted(
        nodes,
        key=lambda n: n.get("updatedAt") or n.get("startsAt") or "",
        reverse=True,
    )


def _select_named_node(nodes: list[dict], value: str, kind: str,
                       strict: bool = False, suggest: bool = False,
                       skip_word: bool = False) -> dict | None:
    """Select a node from already-fetched data: exact name, then substring,
    newest first on ambiguity. This keeps API wrappers thin and testable."""
    matches = [n for n in nodes if (n.get("name") or "").lower() == value.lower()]
    if not matches:
        matches = [n for n in nodes if value.lower() in (n.get("name") or "").lower()]
    if not matches:
        msg = f"{kind} '{value}' not found."
        if suggest:
            msg += _suggest(value, [n.get("name") or "" for n in nodes])
        if strict:
            raise LookupError(msg)
        suffix = " Skipping." if skip_word else ""
        print(f"Warning: {msg}{suffix}", file=sys.stderr)
        return None
    matches = _sort_by_recent(matches)
    if len(matches) > 1:
        suffix = ""
        if kind == "milestone":
            project_name = matches[0].get("_projectName", "")
            suffix = f" in project '{project_name}'" if project_name else ""
        print(f"Note: '{value}' matched {len(matches)} {kind}s; "
              f"picked '{matches[0].get('name')}'{suffix} (most recent).", file=sys.stderr)
    return matches[0]


def resolve_project_id(api_key: str, team_id: str, value: str,
                       strict: bool = False) -> str | None:
    """Resolve --project to a projectId. Accepts UUID or name (exact match
    preferred, then substring). On ambiguity, picks the most-recently-active
    match and notes it to stderr.

    On no match: when strict, raise LookupError with close-match suggestions so
    the caller aborts instead of silently proceeding — a mistyped or not-yet-
    created project must never no-op an issue create/update (RUSH-1496). When
    not strict, warn to stderr and return None."""
    if not value:
        return None
    # Looks like a UUID? Pass through.
    if len(value) == 36 and value.count("-") == 4:
        return value
    projects = list_team_projects(api_key, team_id)
    match = _select_named_node(projects, value, "project", strict=strict,
                               suggest=True, skip_word=True)
    if not match:
        return None
    return match["id"]


def resolve_cwd_project(timeout: float = 3.0) -> tuple[str | None, str | None]:
    """Best-effort: which Linear project owns the current working directory?

    The directory -> project mapping is owned by the `agents` CLI: `agents
    projects` binds a repo root and its monorepo subpaths to a Linear project
    (`linear.projectId` / `linear.name`). We ask it rather than re-deriving the
    mapping, because `projects view <dir>` does a longest-match over every bound
    root and subpath — a worktree, a subdir, and two projects sharing one
    monorepo root all resolve correctly, none of which a basename comparison
    could do.

    Returns (projectId, projectName), or (None, None) when nothing claims this
    directory. Fail-open by design: no `agents` on PATH, no def for this cwd, a
    def with no Linear binding, or a slow/broken call all yield (None, None) and
    the caller falls back to the unscoped team view. Never raises — this runs on
    the hot path of the most-used command and must not brick it.

    One bounded subprocess call (~0.3s on an idle box): `agents projects view .
    --json` longest-matches the cwd and returns `{name, linear:{name,
    projectId}, root}` in a single shot (an unmatched cwd returns those fields
    as null with exit 0).
    """
    try:
        out = subprocess.run(["agents", "projects", "view", ".", "--json"],
                             capture_output=True, text=True, timeout=timeout)
    except (OSError, ValueError, subprocess.SubprocessError):
        return (None, None)
    if out.returncode != 0 or not out.stdout.strip():
        return (None, None)
    try:
        data = json.loads(out.stdout)
    except (ValueError, TypeError):
        return (None, None)
    if not isinstance(data, dict):
        return (None, None)
    lin = data.get("linear") or {}
    return (lin.get("projectId"), lin.get("name"))


def resolve_auto_project_id(api_key: str, team_id: str, value: str) -> str | None:
    """Validate an auto-detected (cwd) project id/name against the team's live
    projects. Returns the matching projectId, or None when it doesn't match.

    NEVER raises and never exits — the key difference from
    resolve_project_id(strict=True). An explicit --project that doesn't resolve
    is a user typo worth aborting on (RUSH-1496); an auto-detected value that
    doesn't resolve just means the `agents projects` binding drifted (a project
    renamed/recreated, or a stale recorded id) — a condition outside this CLI's
    control that must fall back to the unscoped view, not brick the most-common
    command. Membership is checked against the live project list, so it also
    catches a stale-but-UUID-shaped id that resolve_project_id's shape-only UUID
    passthrough would wave through into a filter that silently matches nothing.
    """
    if not value:
        return None
    want = value.lower()
    for p in list_team_projects(api_key, team_id):
        if p.get("id") == value or (p.get("name") or "").lower() == want:
            return p.get("id")
    return None


def list_project_statuses(api_key: str) -> list[dict]:
    """Workspace project statuses (Backlog / Planned / In Progress / ...)."""
    data = gql(api_key, """
        query {
            projectStatuses {
                nodes { id name type }
            }
        }
    """)
    if check_errors(data):
        return []
    return ((data.get("data") or {}).get("projectStatuses") or {}).get("nodes") or []


def resolve_project_status_id(api_key: str, value: str) -> str:
    """Resolve --state to a ProjectStatus id.

    Accepts a status *name* (exact), a status *type* (backlog | planned |
    started | paused | completed | canceled), or a unique name substring. Raises LookupError on
    unknown values so callers fail loud instead of no-opping the update.
    """
    statuses = list_project_statuses(api_key)
    if not statuses:
        raise LookupError("no project statuses available from Linear")
    v = (value or "").strip().lower()
    if not v:
        raise LookupError("project status is required")
    # Name first: a Paused status has type `planned`, so `planned` must not
    # resolve to it by type when a status is named Planned.
    by_name = [s for s in statuses if (s.get("name") or "").lower() == v]
    if by_name:
        return by_name[0]["id"]
    by_type = [s for s in statuses if (s.get("type") or "").lower() == v]
    if by_type:
        return by_type[0]["id"]
    sub = [s for s in statuses if v in (s.get("name") or "").lower()]
    if len(sub) == 1:
        return sub[0]["id"]
    choices = sorted({
        *(s.get("type") or "" for s in statuses),
        *(s.get("name") or "" for s in statuses),
    } - {""})
    raise LookupError(
        f"project status '{value}' not found." + _suggest(value, choices)
    )


def list_initiatives(api_key: str) -> list[dict]:
    """All workspace initiatives, most-recently-updated first. Fully paginated."""
    query = """query($cursor: String) {
        initiatives(first: 100, after: $cursor) {
            pageInfo { hasNextPage endCursor }
            nodes {
                id name description status targetDate updatedAt
                owner { name email }
            }
        }
    }"""
    nodes = paginate_connection(api_key, query, ["initiatives"]) or []
    nodes.sort(key=lambda n: n.get("updatedAt") or "", reverse=True)
    return nodes


def resolve_initiative_id(api_key: str, value: str,
                          strict: bool = False) -> str | None:
    """Resolve an initiative by UUID or name (exact, then substring).

    On no match: strict raises LookupError with suggestions; otherwise warns
    and returns None — same contract as resolve_project_id.
    """
    if not value:
        return None
    if len(value) == 36 and value.count("-") == 4:
        return value
    initiatives = list_initiatives(api_key)
    match = _select_named_node(initiatives, value, "initiative", strict=strict,
                               suggest=True, skip_word=True)
    if not match:
        return None
    return match["id"]


def find_initiative_to_project_id(api_key: str, initiative_id: str,
                                  project_id: str) -> str | None:
    """Return the initiativeToProject join-row id for (initiative, project), or
    None if not linked. Fully paginated — a workspace can outgrow one page."""
    query = """query($cursor: String) {
        initiativeToProjects(first: 100, after: $cursor) {
            pageInfo { hasNextPage endCursor }
            nodes {
                id
                initiative { id }
                project { id }
            }
        }
    }"""
    nodes = paginate_connection(api_key, query, ["initiativeToProjects"]) or []
    for n in nodes:
        if ((n.get("initiative") or {}).get("id") == initiative_id
                and (n.get("project") or {}).get("id") == project_id):
            return n.get("id")
    return None


def resolve_milestone_id(api_key: str, team_id: str, name: str,
                         project_id: str | None,
                         strict: bool = False) -> str | None:
    """Resolve --milestone to a projectMilestoneId. If project_id is given,
    look only there. Else search all team projects and pick the
    most-recently-updated match.

    On no match: when strict, raise LookupError with suggestions (so a batch
    aborts rather than silently dropping the milestone); else warn and return
    None."""
    if not name:
        return None
    if len(name) == 36 and name.count("-") == 4:
        return name

    if project_id:
        data = gql(api_key, """
            query($id: String!) {
                project(id: $id) {
                    projectMilestones(first: 100) {
                        nodes { id name updatedAt }
                    }
                }
            }
        """, {"id": project_id})
        if check_errors(data):
            return None
        nodes = (
            ((data.get("data") or {}).get("project") or {})
            .get("projectMilestones", {}).get("nodes", [])
        )
    else:
        projects = list_team_projects(api_key, team_id)
        nodes = []
        for p in projects:
            data = gql(api_key, """
                query($id: String!) {
                    project(id: $id) {
                        projectMilestones(first: 100) {
                            nodes { id name updatedAt }
                        }
                    }
                }
            """, {"id": p["id"]})
            if check_errors(data):
                continue
            for m in (((data.get("data") or {}).get("project") or {})
                      .get("projectMilestones", {}).get("nodes", [])):
                m["_projectName"] = p["name"]
                nodes.append(m)

    match = _select_named_node(nodes, name, "milestone", strict=strict,
                               suggest=True, skip_word=True)
    if not match:
        return None
    return match["id"]


# Cap the create-guard's printed catalog so a team with dozens of milestones
# still yields a readable refusal. The error then points at
# `linear milestones list` / `linear projects` for the rest.
_MILESTONE_CATALOG_CAP = 20

_MILESTONE_CATALOG_QUERY = """query($ids: [ID!], $cursor: String) {
    projectMilestones(first: 100, after: $cursor,
                      filter: { project: { id: { in: $ids } } }) {
        pageInfo { hasNextPage endCursor }
        nodes { id name targetDate progress sortOrder project { id name } }
    }
}"""


def list_milestone_catalog(api_key: str, team_id: str,
                           project_id: str | None = None) -> list[dict]:
    """Milestones the create-guard can print so an agent can pick one.

    When `project_id` is set, only that project's; otherwise every
    non-archived project on the team. Mock this in tests so they never
    hit the network.
    """
    if project_id:
        ids = [project_id]
    else:
        ids = [p["id"] for p in list_team_projects(api_key, team_id)]
    if not ids:
        return []
    nodes = paginate_connection(api_key, _MILESTONE_CATALOG_QUERY,
                                ["projectMilestones"], {"ids": ids})
    return nodes or []


def list_team_cycles(api_key: str, team_id: str) -> list[dict]:
    """Return all cycles for a team, fully paginated (no 50-cycle truncation)."""
    query = """query($teamId: ID!, $cursor: String) {
        cycles(
            filter: { team: { id: { eq: $teamId } } }
            orderBy: updatedAt
            first: 100
            after: $cursor
        ) {
            pageInfo { hasNextPage endCursor }
            nodes {
                id number name startsAt endsAt completedAt
                issueCountHistory updatedAt
            }
        }
    }"""
    return paginate_connection(api_key, query, ["cycles"], {"teamId": team_id}) or []


# ---------------------------------------------------------------------------
# File upload
# ---------------------------------------------------------------------------

def upload_file(api_key: str, filepath: str) -> str | None:
    """Upload a file to Linear and return the asset URL."""
    path = Path(filepath)
    if not path.exists():
        print(f"File not found: {filepath}", file=sys.stderr)
        return None

    size = path.stat().st_size
    content_type = mimetypes.guess_type(filepath)[0] or "application/octet-stream"

    data = gql(api_key, """
        mutation($filename: String!, $contentType: String!, $size: Int!) {
            fileUpload(filename: $filename, contentType: $contentType, size: $size) {
                success
                uploadFile { uploadUrl assetUrl headers { key value } }
            }
        }
    """, {"filename": path.name, "contentType": content_type, "size": size})

    if check_errors(data):
        return None

    upload = data["data"]["fileUpload"]
    if not upload["success"]:
        print("Failed to get upload URL.", file=sys.stderr)
        return None

    uf = upload["uploadFile"]
    file_bytes = path.read_bytes()

    req = Request(uf["uploadUrl"], data=file_bytes, method="PUT")
    req.add_header("Content-Type", content_type)
    for h in uf.get("headers") or []:
        req.add_header(h["key"], h["value"])

    try:
        with urlopen(req) as resp:
            if resp.status not in (200, 201):
                print(f"Upload failed: HTTP {resp.status}", file=sys.stderr)
                return None
    except URLError as e:
        print(f"Upload failed: {e}", file=sys.stderr)
        return None

    return uf["assetUrl"]


def build_proof_comment(api_key: str, proofs: list[str]) -> str:
    """Build a markdown comment body from proof items.

    Each item is auto-detected:
      - File path (exists on disk) -> upload and embed as image/link
      - URL (starts with http) -> embed as link
      - Plain text -> inline as-is
    """
    parts = []
    for i, proof in enumerate(proofs, 1):
        label = f"**Proof {i}:**" if len(proofs) > 1 else "**Proof:**"
        path = Path(proof)

        if path.exists() and path.is_file():
            asset_url = upload_file(api_key, proof)
            if not asset_url:
                parts.append(f"{label} (upload failed: {path.name})")
                continue
            ct = mimetypes.guess_type(proof)[0] or ""
            if ct.startswith("image/"):
                parts.append(f"{label}\n![{path.name}]({asset_url})")
            else:
                parts.append(f"{label} [{path.name}]({asset_url})")
        elif proof.startswith("http://") or proof.startswith("https://"):
            parts.append(f"{label} {proof}")
        else:
            parts.append(f"{label} {proof}")

    return "\n\n".join(parts)


# ---------------------------------------------------------------------------
# Issue resolution
# ---------------------------------------------------------------------------

def resolve_issue(api_key: str, team_id: str, identifier: str) -> dict | None:
    number = identifier.split("-")[-1]
    try:
        number = int(number)
    except ValueError:
        print(f"Invalid identifier: {identifier}", file=sys.stderr)
        return None

    data = gql(api_key, f"""{{
        issues(filter: {{
            team: {{ id: {{ eq: "{team_id}" }} }}
            number: {{ eq: {number} }}
        }}) {{
            nodes {{ id identifier title }}
        }}
    }}""")

    if check_errors(data):
        return None

    nodes = data.get("data", {}).get("issues", {}).get("nodes", [])
    return nodes[0] if nodes else None


def _warn_sub_issue(child: str, parent_ident: str):
    """Non-blocking nudge when an issue is nested under a parent. Sub-issues add
    hierarchy that's easy to lose track of; a flat issue under a project/milestone
    is usually clearer. A stderr tip (not a prompt) keeps bulk/agent runs
    unblocked while still discouraging casual nesting."""
    print(f"Tip: making '{child}' a sub-issue of {parent_ident}. "
          f"Sub-issues add nesting — prefer a top-level issue under a "
          f"project/milestone unless you truly need it.", file=sys.stderr)


# ---------------------------------------------------------------------------
# Formatters
# ---------------------------------------------------------------------------

PRIORITY_MAP = {0: "-", 1: "Urgent", 2: "High", 3: "Medium", 4: "Low"}

PRIORITY_NAMES = {
    "urgent": 1, "high": 2, "medium": 3, "med": 3, "low": 4,
    "none": 0, "no": 0, "no-priority": 0,
}


def parse_priority(value: str) -> int:
    """Accept named priorities: urgent, high, medium, low, none."""
    v = str(value).strip().lower()
    if v in PRIORITY_NAMES:
        return PRIORITY_NAMES[v]
    raise SystemExit(
        f"Invalid --priority '{value}'. Use: urgent, high, medium, low, or none."
    )


def derive_title(description: str) -> str:
    """Pull a sensible title out of a description body. First sentence or first
    line, stripped of markdown bullets/headers, capped at 80 chars."""
    text = description.strip()
    # First newline-bounded line
    first_line = text.split("\n", 1)[0].strip()
    # Strip leading markdown noise: headers, bullets, blockquotes
    while first_line and first_line[0] in "#*->":
        first_line = first_line[1:].lstrip()
    # First sentence within that line
    for sep in (". ", "! ", "? "):
        idx = first_line.find(sep)
        if 0 < idx < 80:
            first_line = first_line[: idx + 1]
            break
    if len(first_line) > 80:
        first_line = first_line[:79].rstrip() + "…"
    return first_line or "Untitled"


def read_description(inline: str | None, path: str | None) -> str | None:
    """Resolve description from --description (inline), --description-file PATH,
    or stdin via --description-file -. Inline wins if both are given."""
    if inline is not None:
        return inline
    if path is None:
        return None
    if path == "-":
        return sys.stdin.read()
    p = Path(path)
    if not p.exists():
        raise SystemExit(f"--description-file: {path} not found")
    return p.read_text()


def parse_due_date(value: str) -> str | None:
    """Accept YYYY-MM-DD, or 'none' to clear. Returns the validated string,
    or None when clearing. Linear's dueDate is a TimelessDate (date-only)."""
    v = str(value).strip()
    if v.lower() in ("none", ""):
        return None
    try:
        date.fromisoformat(v)
    except ValueError:
        raise SystemExit(
            f"Invalid --due-date '{value}'. Use YYYY-MM-DD (e.g. 2026-04-28) or 'none'."
        )
    return v

def parse_remind_at(value: str) -> str:
    """An exact time for Linear's issue reminder ("Remind me"), the time a
    TimelessDate dueDate cannot hold. Accepts YYYY-MM-DDTHH:MM (or a space for
    the T), with seconds and a zone optional; no zone means local time.
    Returns UTC ISO 8601."""
    raw = str(value).strip()
    v = raw.replace(" ", "T", 1)
    try:
        at = datetime.fromisoformat(v.replace("Z", "+00:00")) if "T" in v else None
    except ValueError:
        at = None
    if at is None:
        raise SystemExit(
            f"Invalid --remind-at '{raw}'. Use YYYY-MM-DDTHH:MM in local time "
            "(e.g. 2026-10-08T09:10) or with a zone (2026-10-08T09:10-07:00)."
        )
    if at.tzinfo is None:
        at = at.astimezone()
    return at.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")


def set_issue_reminder(api_key: str, issue_id: str, at: str) -> bool:
    data = gql(api_key, """
        mutation($id: String!, $at: DateTime!) {
            issueReminder(id: $id, reminderAt: $at) { success }
        }
    """, {"id": issue_id, "at": at})
    return not check_errors(data) and bool(((data.get("data") or {}).get("issueReminder") or {}).get("success"))


# `id` and `updatedAt` are what the embeddings cache keys on — the vector for an
# issue is reused until its updatedAt moves.
ISSUE_FIELDS = """
    id updatedAt
    identifier title description state { name type }
    priority labels { nodes { name } }
    assignee { name }
    delegate { name }
    project { name id }
    projectMilestone { name }
    cycle { number name }
    dueDate createdAt url
"""


# Sort sentinel for missing due dates — push them to the end of their tier.
_NO_DUE = "9999-99-99"


def issue_sort_key(node: dict) -> tuple:
    """Primary: priority (Urgent=1 first, No-priority=4 last).
    Secondary: due date ascending (earliest first, nulls last).
    Tertiary: identifier so ties are deterministic."""
    priority = node.get("priority") or 4
    due = node.get("dueDate") or _NO_DUE
    ident = node.get("identifier") or ""
    return (priority, due, ident)


def format_due(due: str | None) -> str:
    """Compact due date display: '--' if missing, 'today' / 'tomorrow' / relative
    within a week, otherwise MM-DD."""
    if not due:
        return "    --    "
    try:
        from datetime import date, datetime
        today = date.today()
        d = datetime.fromisoformat(due.replace("Z", "+00:00")).date() if "T" in due else date.fromisoformat(due)
        delta = (d - today).days
        if delta < 0:
            return f"{-delta}d overdue".ljust(10)
        if delta == 0:
            return "today     "
        if delta == 1:
            return "tomorrow  "
        if delta <= 7:
            return f"in {delta}d    "[:10]
        return d.strftime("%b %d   ")[:10]
    except Exception:
        return (due[:10] + "  ")[:10]


def format_issue_row(node: dict) -> str:
    ident = node["identifier"]
    title = node["title"]
    state = node["state"]["name"]
    pri = PRIORITY_MAP.get(node.get("priority", 0), "-")
    due = format_due(node.get("dueDate"))
    labels = [l["name"] for l in node.get("labels", {}).get("nodes", [])]
    label_str = f"  [{', '.join(labels)}]" if labels else ""
    assignee = (node.get("assignee") or {}).get("name") or "unassigned"
    # Delegate (the agent handed the issue) rides alongside the human assignee:
    # "Muqsit → claude". Both are distinct Linear fields; showing only assignee
    # hid every delegation, so the column now carries whoever is actually on it.
    delegate = (node.get("delegate") or {}).get("name")
    who = f"{assignee} → {delegate}" if delegate else assignee
    return f"  {ident:<8} {pri:<7}  {state:<12}  {due}  {who:<22}  {title}{label_str}"


def print_by_milestone(nodes: list[dict]) -> None:
    """Group open issues by their milestone for the project / `--by-milestone`
    view. Each issue carries its milestone name and cycle via ISSUE_FIELDS, so
    this is a local group-by with no extra API calls.

    Issues matched to no milestone fall into an explicit 'No milestone' bucket,
    printed last — that bucket is the "what isn't matched to a deliverable"
    surface. Each row is annotated with its cycle, so you can read which
    iteration a deliverable's work is scheduled in (the milestone x cycle join).
    Progress % lives in `projects show` / `milestones list`, which count
    completed issues too; this view is open work only.
    """
    orphan = "— No milestone"
    groups: dict[str, list] = {}
    for n in nodes:
        name = (n.get("projectMilestone") or {}).get("name") or orphan
        groups.setdefault(name, []).append(n)
    ordered = sorted(k for k in groups if k != orphan)
    if orphan in groups:
        ordered.append(orphan)
    for name in ordered:
        issues = groups[name]
        print(f"{name}  ({len(issues)} open)")
        for n in issues:
            cyc = n.get("cycle")
            cyc_str = f"  ·{cycle_label(cyc)}" if cyc else ""
            print(f"{format_issue_row(n)}{cyc_str}")
        print()


def milestone_rollup(api_key: str, project_id: str) -> dict:
    """One paginated pass over a project's issues, all states, aggregated into
    {milestone_id: {"total": n, "done": n}} plus a "_none" key for issues in the
    project but no milestone. Slim field set (milestone id + state type) keeps a
    progress rollup one cheap query, not a fetch of every issue body.

    'done' counts issues whose workflow state type is 'completed'. Returns {} on
    API error (caller renders milestones without counts rather than aborting).
    """
    counts: dict = {}
    cursor = None
    pages = 0
    while True:
        data = gql(api_key, """
            query($pid: ID!, $cursor: String) {
                issues(first: 250, after: $cursor,
                       filter: { project: { id: { eq: $pid } } }) {
                    pageInfo { hasNextPage endCursor }
                    nodes { projectMilestone { id } state { type } }
                }
            }
        """, {"pid": project_id, "cursor": cursor})
        if check_errors(data):
            # Discard any partial counts. A mid-pagination failure would else
            # render an understated "% done" that looks complete; returning {}
            # makes the caller show "(0 issues)" — honestly-unknown, not wrong.
            return {}
        conn = (data.get("data") or {}).get("issues") or {}
        for n in conn.get("nodes", []):
            key = (n.get("projectMilestone") or {}).get("id") or "_none"
            c = counts.setdefault(key, {"total": 0, "done": 0})
            c["total"] += 1
            if (n.get("state") or {}).get("type") == "completed":
                c["done"] += 1
        info = conn.get("pageInfo") or {}
        cursor = info.get("endCursor")
        pages += 1
        if not info.get("hasNextPage"):
            break
        if pages >= _MAX_PAGES:
            print(
                f"Warning: milestone rollup stopped after {pages * 250} issues "
                f"(pagination safety rail); progress may be understated.",
                file=sys.stderr,
            )
            break
    return counts


def _fmt_rollup(entry: dict | None) -> str:
    """'(3/10 done, 30%)' from a milestone_rollup entry, or '(0 issues)'."""
    if not entry or not entry.get("total"):
        return "(0 issues)"
    total, done = entry["total"], entry["done"]
    return f"({done}/{total} done, {int(done * 100 / total)}%)"


# ---------------------------------------------------------------------------
# Paginated issue fetch
# ---------------------------------------------------------------------------

# Linear caps a connection at 50 nodes by default and 250 max per page. Without
# following pageInfo, any list silently truncates — which breaks "search before
# you create" and produces duplicate tickets. paginate_issues walks every page.
_PAGE_SIZE = 100
_MAX_PAGES = 100  # 10k-issue safety rail; warns if ever hit (no silent caps).


def paginate_issues(api_key: str, filter_str: str) -> list[dict] | None:
    """Fetch every issue matching a root-level filter, following pagination.

    `filter_str` is the inner body of the GraphQL `filter:` object (no braces).
    The opaque cursor is passed as a variable; only the static filter is
    interpolated. Returns None on API error (caller already printed it).
    """
    nodes: list[dict] = []
    cursor: str | None = None
    pages = 0
    query = f"""query($cursor: String) {{
        issues(first: {_PAGE_SIZE}, after: $cursor, filter: {{ {filter_str} }}) {{
            pageInfo {{ hasNextPage endCursor }}
            nodes {{ {ISSUE_FIELDS} }}
        }}
    }}"""
    while True:
        data = gql(api_key, query, {"cursor": cursor})
        if check_errors(data):
            return None
        conn = (data.get("data") or {}).get("issues") or {}
        nodes.extend(conn.get("nodes", []))
        pages += 1
        page_info = conn.get("pageInfo") or {}
        cursor = page_info.get("endCursor")
        if not page_info.get("hasNextPage"):
            break
        if pages >= _MAX_PAGES:
            print(
                f"Warning: stopped after {pages * _PAGE_SIZE} issues "
                f"(pagination safety rail). Narrow your filter to see the rest.",
                file=sys.stderr,
            )
            break
    return nodes


def paginate_connection(api_key: str, query: str, path: list[str],
                        variables: dict | None = None,
                        page_size: int = _PAGE_SIZE,
                        partial_out: dict | None = None) -> list[dict] | None:
    """Generic relay-cursor paginator for any connection (cycles, users,
    projects, labels, ...). Companion to paginate_issues for the issue path.

    `query` must declare `$cursor: String`, pass `after: $cursor` to the target
    connection, and select `pageInfo { hasNextPage endCursor }` next to `nodes`.
    `path` is the key sequence from the `data` root to the connection object
    (e.g. ["team", "projects"] or ["cycles"]). Returns every node across all
    pages, or None on API error (caller already printed it).

    `page_size` is the query's own `first:` value — it only sizes the safety
    rail message. When the rail stops the walk early, `partial_out` (if given)
    receives {"rows": <rows fetched>} so a caller emitting machine output can
    mark the result partial instead of passing off a truncated set as complete.
    """
    nodes: list[dict] = []
    cursor: str | None = None
    pages = 0
    base_vars = dict(variables or {})
    while True:
        v = dict(base_vars)
        v["cursor"] = cursor
        data = gql(api_key, query, v)
        if check_errors(data):
            return None
        conn: dict | None = data.get("data") or {}
        for key in path:
            conn = (conn or {}).get(key)
        conn = conn or {}
        nodes.extend(conn.get("nodes", []))
        pages += 1
        info = conn.get("pageInfo") or {}
        cursor = info.get("endCursor")
        if not info.get("hasNextPage"):
            break
        if pages >= _MAX_PAGES:
            print(
                f"Warning: stopped after {pages * page_size} rows "
                f"(pagination safety rail).",
                file=sys.stderr,
            )
            if partial_out is not None:
                partial_out["rows"] = pages * page_size
            break
    return nodes


def list_team_labels(api_key: str, team_id: str) -> list[dict]:
    """All labels visible to a team (team-scoped + workspace-wide), paginated.
    The single source of truth for label lookups across list/create/update and
    the label-resolution used by `update --label`."""
    query = """query($teamId: ID!, $cursor: String) {
        issueLabels(
            filter: {
                or: [
                    { team: { id: { eq: $teamId } } }
                    { team: { null: true } }
                ]
            }
            first: 100
            after: $cursor
        ) {
            pageInfo { hasNextPage endCursor }
            nodes { id name description color updatedAt }
        }
    }"""
    return paginate_connection(api_key, query, ["issueLabels"], {"teamId": team_id}) or []


def _select_cycle_node(cycles: list[dict], value: str, strict: bool = False) -> dict | None:
    matches = [c for c in cycles if (c.get("name") or "").lower() == value.lower()]
    if not matches and value.isdigit():
        matches = [c for c in cycles if str(c.get("number")) == value]
    if not matches:
        matches = [c for c in cycles if value.lower() in (c.get("name") or "").lower()]
    if not matches:
        msg = f"cycle '{value}' not found."
        if strict:
            raise LookupError(msg)
        print(f"Warning: {msg}", file=sys.stderr)
        return None
    matches = _sort_by_recent(matches)
    if len(matches) > 1:
        print(f"Note: '{value}' matched {len(matches)} cycles; "
              f"picked '{matches[0].get('name')}' (most recent).", file=sys.stderr)
    return matches[0]


def resolve_cycle_by_value(api_key: str, team_id: str, value: str,
                           strict: bool = False) -> dict | None:
    """Resolve a non-keyword --cycle value to a cycle node {id, name, ...}.
    Accepts a UUID (passthrough), a cycle number, or a fuzzy name match.
    Warns and returns None when nothing matches."""
    if not value:
        return None
    if len(value) == 36 and value.count("-") == 4:
        return {"id": value, "name": None, "startsAt": None, "endsAt": None}
    return _select_cycle_node(list_team_cycles(api_key, team_id), value, strict=strict)


def resolve_label_by_value(api_key: str, team_id: str, value: str,
                           strict: bool = False) -> dict | None:
    """Resolve a label id-or-name to a label node {id, name, ...}. UUID passes
    through; otherwise exact (case-insensitive) name wins, then substring.
    Ambiguity picks the most-recently-updated match and warns."""
    if not value:
        return None
    if len(value) == 36 and value.count("-") == 4:
        return {"id": value, "name": value}
    return _select_named_node(list_team_labels(api_key, team_id), value, "label",
                              strict=strict)


def build_cycle_scope_from_resolved(
    scope: str | None,
    active_meta: dict | None = None,
    next_meta: dict | None = None,
    cycle_node: dict | None = None,
) -> tuple[str | None, str | None, dict | None]:
    s = (scope or "active").lower()
    if s in ("active", "next"):
        meta = active_meta if s == "active" else next_meta
        if not meta:
            return None, None, None
        return f'cycle: {{ id: {{ eq: "{meta["id"]}" }} }}', cycle_label(meta), meta
    if s in ("none", "backlog"):
        return "cycle: { null: true }", "Backlog (no cycle)", None
    if s == "all":
        return "", "All issues", None
    if not cycle_node:
        return None, None, None
    label = cycle_label(cycle_node)
    if label == "no cycle":
        label = scope
    return f'cycle: {{ id: {{ eq: "{cycle_node["id"]}" }} }}', label, cycle_node


def build_cycle_scope(api_key: str, team_id: str,
                      scope: str | None) -> tuple[str | None, str | None, dict | None]:
    """Translate a --cycle value into an issues() filter fragment + display label.

    Returns (fragment, label, cycle_meta):
      - active/next  -> concrete cycle id filter, cycle name, meta dict
      - none/backlog -> `cycle: { null: true }`, "Backlog (no cycle)", None
      - all          -> "" (no cycle constraint), "All issues", None
      - <name>/<id>  -> concrete cycle id filter, matched name, node
    On an unresolvable active/next/named cycle, returns (None, None, None).
    """
    s = (scope or "active").lower()
    if s == "active":
        return build_cycle_scope_from_resolved(
            scope, active_meta=resolve_cycle_meta(api_key, team_id, s)
        )
    if s == "next":
        return build_cycle_scope_from_resolved(
            scope, next_meta=resolve_cycle_meta(api_key, team_id, s)
        )
    if s in ("none", "backlog", "all"):
        return build_cycle_scope_from_resolved(scope)
    return build_cycle_scope_from_resolved(
        scope, cycle_node=resolve_cycle_by_value(api_key, team_id, scope)
    )


def resolve_task_scope(cycle: str | None, project, milestone) -> str:
    """The effective --cycle scope for `linear tasks`. An explicit --cycle wins.
    With none set, scoping to a project or milestone widens to all cycles (you
    want the whole deliverable, not just this cycle's slice); a bare list still
    defaults to the active cycle."""
    if cycle is not None:
        return cycle
    return "all" if (project or milestone) else "active"


def cycle_date_to_iso(value: str, flag: str) -> str:
    """Cycle startsAt/endsAt are DateTimes. Accept a YYYY-MM-DD and anchor it to
    start-of-day UTC; pass an explicit ISO timestamp through untouched."""
    v = str(value).strip()
    if "T" in v:
        return v
    try:
        date.fromisoformat(v)
    except ValueError:
        raise SystemExit(f"Invalid {flag} '{value}'. Use YYYY-MM-DD (e.g. 2026-07-01).")
    return f"{v}T00:00:00.000Z"


def parse_timeless_date(value: str, flag: str) -> str | None:
    """Project/milestone start/target dates are Linear TimelessDate — a bare
    YYYY-MM-DD, no time or zone. 'none'/'' clears (returns None). Exits on a
    malformed date so a typo can't be sent as-is."""
    v = str(value or "").strip()
    if v.lower() in ("", "none"):
        return None
    try:
        date.fromisoformat(v[:10])
    except ValueError:
        raise SystemExit(f"Invalid {flag} '{value}'. Use YYYY-MM-DD (e.g. 2026-06-10).")
    return v[:10]


# ---------------------------------------------------------------------------
# setup
# ---------------------------------------------------------------------------

def apply_embeddings_choice(cfg: dict, backend: str, model: str | None) -> str:
    """Set (or clear, for "none") cfg["embeddings"]; returns the line to print.

    Only the backend name and model land in config.json. The Gemini key is read
    from GEMINI_API_KEY when a ranking runs and is never written here.
    """
    if backend == "none":
        cfg.pop("embeddings", None)
        return "Embeddings ranker: off"
    chosen = model or EMBEDDINGS_DEFAULT_MODEL[backend]
    cfg["embeddings"] = {"backend": backend, "model": chosen}
    return f"Embeddings ranker: {backend} ({chosen})"


def cmd_setup(args, cfg):
    if args.embeddings_model and not args.embeddings:
        print("--embeddings-model needs --embeddings ollama|gemini", file=sys.stderr)
        sys.exit(1)

    # Turning the ranker on is a one-key edit — don't re-run the team and agent
    # prompts (or spend an API call) on a box that is already configured.
    if args.embeddings and not args.api_key and cfg.get("teamId"):
        note = apply_embeddings_choice(cfg, args.embeddings, args.embeddings_model)
        if not save_config(cfg):
            print(f"Setup failed: could not persist {CONFIG_PATH}", file=sys.stderr)
            sys.exit(1)
        print(note)
        print(f"Config: {CONFIG_PATH}")
        return

    api_key = args.api_key or input("Linear API key: ").strip()
    if not api_key:
        print("API key required.")
        sys.exit(1)

    cfg["apiKey"] = api_key

    data = gql(api_key, "{ teams { nodes { id name key } } }")
    if check_errors(data):
        sys.exit(1)

    teams = data["data"]["teams"]["nodes"]
    if not teams:
        print("No teams found.")
        sys.exit(1)

    if len(teams) == 1:
        team = teams[0]
        print(f"Found team: {team['name']} ({team['key']})")
    else:
        print("Teams:")
        for i, t in enumerate(teams):
            print(f"  {i + 1}. {t['name']} ({t['key']})")
        choice = input(f"Select team [1-{len(teams)}]: ").strip()
        try:
            team = teams[int(choice) - 1]
        except (ValueError, IndexError):
            print("Invalid selection.")
            sys.exit(1)

    cfg["teamId"] = team["id"]
    cfg["teamKey"] = team["key"]

    agent = args.agent or input("Default agent identity (e.g. claude, codex) [skip]: ").strip()
    if agent:
        cfg["agent"] = agent

    embeddings_note = (apply_embeddings_choice(cfg, args.embeddings, args.embeddings_model)
                       if args.embeddings else None)

    cfg.pop("states", None)
    states = get_states(api_key, team["id"], cfg)

    # Here persistence IS the operation — a setup that didn't land is a failure,
    # not a warning, or the next command silently runs unconfigured.
    if not save_config(cfg):
        print(f"Setup failed: could not persist {CONFIG_PATH}", file=sys.stderr)
        sys.exit(1)
    print(f"\nConfigured: team={team['name']}, states={len(states)}")
    if agent:
        print(f"Agent identity: {agent}")
    if embeddings_note:
        print(embeddings_note)
    print(f"Config: {CONFIG_PATH}")


# ---------------------------------------------------------------------------
# similar tickets — `tasks --similar`
# ---------------------------------------------------------------------------

# Low-signal words dropped before comparing titles: grammar words, the generic
# action verbs that describe the work rather than the thing, and the project
# nouns that appear across most of the board. Mirrors the fleet's dedup guard so
# the two agree on what a "significant" token is.
SIMILAR_STOP = {
    "the", "a", "an", "to", "for", "of", "in", "on", "and", "or", "with",
    "without", "is", "are", "be", "by", "at", "from", "into", "via", "per",
    "so", "it", "this", "that", "then", "than", "not", "no", "as", "vs",
    "fix", "bug", "add", "update", "remove", "refactor", "investigate",
    "prix", "cloud", "rush", "agent", "agents",
}

# Ranker names in display order; the marks column is one letter each.
SIMILAR_RANKERS = ("semantic", "lexical", "embeddings")
_RANKER_MARK = {"semantic": "S", "lexical": "L", "embeddings": "E"}
SIMILAR_SEMANTIC_FIRST = 25
RRF_K = 60

# Extension point for optional rankers (embeddings). Each entry is
#   name -> fn(api_key, team_id, cfg, text, candidates) -> list[str] | None
# returning issue identifiers best-first, or None when the ranker is not
# available on this box (no backend configured, service unreachable). A None is
# skipped silently and the name is left out of the result's "rankers" list.
EXTRA_RANKERS: dict = {}


def sig_tokens(text: str | None) -> set[str]:
    words = re.findall(r"[a-z0-9]+", (text or "").lower())
    return {w for w in words if w not in SIMILAR_STOP and len(w) > 1}


def _jaccard(a: set[str], b: set[str]) -> float:
    return len(a & b) / len(a | b) if (a or b) else 0.0


def _overlap(a: set[str], b: set[str]) -> float:
    return len(a & b) / min(len(a), len(b)) if (a and b) else 0.0


def lexical_rank(text: str, issues: list[dict]) -> list[tuple[str, float]]:
    """Local, dependency-free ranker. Jaccard over significant title tokens
    catches a near-equal title; half an overlap coefficient over the description
    catches the same subject described under a different headline. Issues that
    share nothing are dropped."""
    query = sig_tokens(text)
    scored = []
    for node in issues:
        score = (_jaccard(query, sig_tokens(node.get("title")))
                 + 0.5 * _overlap(query, sig_tokens(node.get("description"))))
        if score > 0:
            scored.append((node["identifier"], score))
    scored.sort(key=lambda t: (-t[1], t[0]))
    return scored


def semantic_rank(api_key: str, team_id: str, text: str, *,
                  project_id: str | None = None,
                  first: int = SIMILAR_SEMANTIC_FIRST) -> list[dict] | None:
    """Linear's own vector search, one `searchIssues` call (rate limit: 30/min).
    Returns issue nodes best-first, or None on an API error (already printed)."""
    query = f"""query($term: String!, $first: Int, $teamId: String, $filter: IssueFilter) {{
        searchIssues(term: $term, first: $first, teamId: $teamId, filter: $filter,
                     includeArchived: false) {{
            nodes {{ {ISSUE_FIELDS} }}
        }}
    }}"""
    variables = {"term": text, "first": first, "teamId": team_id,
                 "filter": {"project": {"id": {"eq": project_id}}} if project_id else None}
    data = gql(api_key, query, variables)
    if check_errors(data):
        return None
    return ((data.get("data") or {}).get("searchIssues") or {}).get("nodes") or []


def rrf(rankings: dict[str, list[str]], k: int = RRF_K
        ) -> list[tuple[str, float, dict[str, int | None]]]:
    """Reciprocal rank fusion: each ranker contributes 1/(k + rank) per hit. Rows
    come back (identifier, fused score, {ranker: rank | None}) best-first, ties
    broken by identifier so output is stable."""
    scores: dict[str, float] = {}
    ranks: dict[str, dict[str, int | None]] = {}
    for name, ids in rankings.items():
        for pos, ident in enumerate(ids, start=1):
            scores[ident] = scores.get(ident, 0.0) + 1.0 / (k + pos)
            ranks.setdefault(ident, {})[name] = pos
    for ident in ranks:
        for name in rankings:
            ranks[ident].setdefault(name, None)
    return sorted(((i, s, ranks[i]) for i, s in scores.items()),
                  key=lambda t: (-t[1], t[0]))


# ---- board cache ------------------------------------------------------------
# `--similar` ranks over the whole team, done and canceled included. Paginating
# that board costs ~11 s for ~4k issues, so it lives in cache.sqlite and a run
# only pulls the issues whose `updatedAt` moved since the last sync. Archived
# issues and issues moved to another team never appear in an incremental pull
# (the connection excludes them), so a full refresh runs once a week to drop
# them. The watermark is inclusive (`gte`) and rows are upserted by id, so two
# issues stamped in the same instant, one of them not yet visible to the read,
# cost a re-fetch of that instant rather than a missed issue.
# `rm ~/.linear-cli/cache.sqlite` is the reset.
BOARD_FULL_REFRESH = timedelta(days=7)


def _board_rows(conn: sqlite3.Connection, team_id: str) -> list[dict]:
    return [json.loads(node) for (node,) in conn.execute(
        "SELECT node FROM issues WHERE team_id = ? ORDER BY issue_id", (team_id,))]


def cached_team_issues(api_key: str, team_id: str, *,
                       project_id: str | None = None) -> list[dict] | None:
    """The team's issues from the cache, refreshed incrementally first.

    Returns None only when nothing is cached and the API call failed (the
    caller already printed the error). A failed refresh over a warm cache
    answers from the cache and says so on stderr.
    """
    conn = _cache_db()
    try:
        row = conn.execute("SELECT last_updated_at, full_at FROM sync WHERE team_id = ?",
                           (team_id,)).fetchone()
        last_updated_at, full_at = row or (None, None)
        now = datetime.now(timezone.utc)
        full = True
        if full_at:
            try:
                full = now - datetime.fromisoformat(full_at) > BOARD_FULL_REFRESH
            except ValueError:
                full = True
        filters = [f'team: {{ id: {{ eq: "{team_id}" }} }}']
        if not full and last_updated_at:
            filters.append(f'updatedAt: {{ gte: "{last_updated_at}" }}')
        fetched = paginate_issues(api_key, ", ".join(filters))
        if fetched is None:
            cached = _board_rows(conn, team_id)
            if not cached:
                return None
            print(f"board cache: refresh failed, using the board as of {last_updated_at}",
                  file=sys.stderr)
        elif full and len(fetched) >= _PAGE_SIZE * _MAX_PAGES:
            # paginate_issues stopped at its safety rail and returned a partial
            # board; replacing the cache with it would drop everything past the
            # rail for good. Keep what we have and try again next run.
            cached = _board_rows(conn, team_id)
            print("board cache: full refresh hit the pagination rail; keeping the cached board",
                  file=sys.stderr)
        else:
            if full:
                conn.execute("DELETE FROM issues WHERE team_id = ?", (team_id,))
            conn.executemany(
                "INSERT OR REPLACE INTO issues(issue_id, team_id, updated_at, node) "
                "VALUES (?, ?, ?, ?)",
                [(n.get("id") or n["identifier"], team_id, n.get("updatedAt") or "", json.dumps(n))
                 for n in fetched if n.get("id") or n.get("identifier")])
            newest = max([n.get("updatedAt") or "" for n in fetched] + [last_updated_at or ""])
            conn.execute(
                "INSERT OR REPLACE INTO sync(team_id, last_updated_at, full_at) VALUES (?, ?, ?)",
                (team_id, newest or None, now.isoformat() if full else full_at))
            conn.commit()
            cached = _board_rows(conn, team_id)
    finally:
        conn.close()
    if project_id:
        cached = [n for n in cached if (n.get("project") or {}).get("id") == project_id]
    return cached


def rank_similar(api_key: str, team_id: str, cfg: dict, text: str, *,
                 project_id: str | None = None, limit: int = 10,
                 rankers: list[str] | None = None) -> dict:
    """Rank the team's tickets by relevance to `text`.

    Runs every available ranker (semantic, lexical, and whatever EXTRA_RANKERS
    registered), fuses them with RRF and returns
      {"rows": [{identifier, title, state, url, score, ranks: {name: rank|None}}],
       "rankers": [names that actually ran]}.
    A ranker that cannot run is skipped and left out of "rankers" — the caller
    never fails because one signal is missing. The candidate set is the whole
    team across every cycle including done and canceled issues: "this was
    already done" is a valid answer, and the state column says so.
    """
    wanted = list(rankers) if rankers is not None else list(
        dict.fromkeys(SIMILAR_RANKERS + tuple(EXTRA_RANKERS)))
    by_id: dict[str, dict] = {}
    rankings: dict[str, list[str]] = {}

    if "semantic" in wanted:
        nodes = semantic_rank(api_key, team_id, text, project_id=project_id)
        if nodes is not None:
            for n in nodes:
                by_id.setdefault(n["identifier"], n)
            rankings["semantic"] = [n["identifier"] for n in nodes]

    candidates: list[dict] | None = None
    needs_candidates = [r for r in wanted if r != "semantic"]
    if needs_candidates:
        candidates = cached_team_issues(api_key, team_id, project_id=project_id)
    if candidates is not None:
        for n in candidates:
            by_id.setdefault(n["identifier"], n)
        if "lexical" in wanted:
            rankings["lexical"] = [i for i, _ in lexical_rank(text, candidates)]
        for name in needs_candidates:
            fn = EXTRA_RANKERS.get(name)
            if fn is None:
                continue
            ids = fn(api_key, team_id, cfg, text, candidates)
            if ids is not None:
                rankings[name] = [i for i in ids if i in by_id]

    known = list(dict.fromkeys(SIMILAR_RANKERS + tuple(EXTRA_RANKERS)))
    rows = []
    for ident, score, ranks in rrf(rankings)[:max(1, limit)]:
        node = by_id.get(ident) or {}
        rows.append({
            "identifier": ident,
            "title": node.get("title") or "",
            "state": ((node.get("state") or {}).get("name")) or "",
            "url": node.get("url") or "",
            "score": round(score, 6),
            "ranks": {name: ranks.get(name) for name in known},
        })
    return {"rows": rows, "rankers": list(rankings)}


def format_similar_row(row: dict) -> str:
    marks = " ".join(_RANKER_MARK[name] if row["ranks"].get(name) else " "
                     for name in SIMILAR_RANKERS)
    return (f"  {row['score']:.4f}  {marks}  {row['identifier']:<9}  "
            f"{row['state']:<8} {row['title']}")


def similar_tasks(args, cfg, api_key, team_id):
    """`tasks --similar TEXT`: the check-before-create step. Honors --project
    (strict, like list_tasks); the other list filters are ignored on purpose:
    a duplicate in another cycle, status, or queue is exactly what this is for.
    Exit 0 even with no hits, because the answer "nothing similar" is the green
    light to create. Exit 1 when no ranker could run at all, so an outage never
    reads as that same green light."""
    pid = None
    if getattr(args, "project", None):
        try:
            pid = resolve_project_id(api_key, team_id, args.project, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    result = rank_similar(api_key, team_id, cfg, args.similar, project_id=pid,
                          limit=getattr(args, "limit", 10))
    if not result["rankers"]:
        print("Error: no ranker could run (Linear API unreachable or rejected "
              "the request; see above). This is not a no-match answer.",
              file=sys.stderr)
        sys.exit(1)
    if args.json:
        print(json.dumps(result, indent=2))
        return
    if not result["rows"]:
        print("No similar tickets.")
        return
    for row in result["rows"]:
        print(format_similar_row(row))


def advise_similar(api_key, team_id, cfg, title, description=None):
    """Print close existing tickets before a single `create`. Advisory only —
    never blocks, never prompts. Skipped by the caller for `--from-file`
    (searchIssues is rate-limited to 30/min)."""
    try:
        text = title + ("\n" + description if description else "")
        rows = rank_similar(api_key, team_id, cfg, text, limit=3)["rows"]
        hits = [r for r in rows if (r["ranks"].get("semantic") or 99) <= 3 or r["score"] >= 0.02]
        if hits:
            print("Similar existing tickets (consider enriching one instead of creating):", file=sys.stderr)
            for r in hits:
                print(f"  {r['identifier']:<10} {r['state']:<9} {r['title'][:70]}", file=sys.stderr)
        return hits
    except Exception as e:  # advisory: never block the create, whatever the row shape
        print(f"similar-check skipped: {e}", file=sys.stderr)
        return []


# ---------------------------------------------------------------------------
# embeddings ranker — optional third signal for `tasks --similar`
# ---------------------------------------------------------------------------

# Off unless `linear setup --embeddings ollama|gemini` wrote a backend. Every
# failure here is advisory: the ranker drops out with one stderr line and
# `--similar` still answers from semantic + lexical.

CACHE_DB_PATH = Path.home() / ".linear-cli" / "cache.sqlite"
# 0.24.0 shipped the vector cache under this name; it is renamed in place on first open.
_LEGACY_EMBEDDINGS_DB_PATH = Path.home() / ".linear-cli" / "embeddings.sqlite"
EMBEDDINGS_BACKENDS = ("ollama", "gemini")
# Defaults are the smallest models at the top of the quality curve (rationale and
# the alternatives worth passing to --embeddings-model are in PHNX-4105).
EMBEDDINGS_DEFAULT_MODEL = {
    "ollama": "qwen3-embedding:0.6b",
    "gemini": "gemini-embedding-001",
}
OLLAMA_DEFAULT_HOST = "http://localhost:11434"
GEMINI_API_ROOT = "https://generativelanguage.googleapis.com/v1beta/models"
# Matryoshka truncation: gemini-embedding-001 answers at 3072 by default and
# 768 is its cheapest well-behaved size. Never assume this is *the* dimension —
# what a vector actually measured is recorded per cached row.
GEMINI_OUTPUT_DIMS = 768
EMBED_BATCH = 64
EMBED_TIMEOUT = 20
# Descriptions run long; the first 2000 characters carry the subject.
EMBED_TEXT_CHARS = 2000
# SQLite's default parameter ceiling is 999 — chunk the id lookup under it.
_SQL_MAX_PARAMS = 500


class EmbeddingUnavailable(RuntimeError):
    """The configured backend could not be reached, or answered unusably.

    Carries a short human phrase (`ollama unreachable`), not the request URL —
    the Gemini key rides in the query string and must not reach a log.
    """


def embeddings_settings(cfg: dict) -> tuple[str, str] | None:
    """(backend, model) when a backend is configured, else None."""
    conf = cfg.get("embeddings") or {}
    backend = conf.get("backend")
    if backend not in EMBEDDINGS_BACKENDS:
        return None
    return backend, conf.get("model") or EMBEDDINGS_DEFAULT_MODEL[backend]


def _embed_post(url: str, payload: dict, backend: str) -> dict:
    """POST JSON, parse JSON. Every transport-level problem — refused connection,
    DNS, timeout, HTTP error, unparseable body — becomes EmbeddingUnavailable."""
    req = Request(url, data=json.dumps(payload).encode(), method="POST")
    req.add_header("Content-Type", "application/json")
    try:
        with urlopen(req, timeout=EMBED_TIMEOUT) as resp:
            return json.loads(resp.read())
    # URLError/HTTPError and socket timeouts are all OSError subclasses.
    except (OSError, ValueError) as e:
        # An answered-but-refusing backend (model not pulled, no embedding
        # support, bad key) is a different fix than a dead one — name the status.
        status = getattr(e, "code", None)
        raise EmbeddingUnavailable(
            f"{backend} unreachable" + (f" (HTTP {status})" if status else "")) from e


def _embed_ollama(model: str, texts: list[str]) -> list[list[float]]:
    host = (os.environ.get("OLLAMA_HOST") or OLLAMA_DEFAULT_HOST).strip().rstrip("/")
    # OLLAMA_HOST is commonly set bare ("127.0.0.1:11434") — urlopen needs a scheme.
    if "://" not in host:
        host = f"http://{host}"
    data = _embed_post(f"{host}/api/embed", {"model": model, "input": texts}, "ollama")
    return _as_vectors(data.get("embeddings"), len(texts), "ollama")


def _embed_gemini(model: str, texts: list[str]) -> list[list[float]]:
    key = (os.environ.get("GEMINI_API_KEY") or "").strip()
    if not key:
        raise EmbeddingUnavailable("gemini needs GEMINI_API_KEY")
    payload = {"requests": [{"model": f"models/{model}",
                             "content": {"parts": [{"text": t}]},
                             "outputDimensionality": GEMINI_OUTPUT_DIMS}
                            for t in texts]}
    data = _embed_post(f"{GEMINI_API_ROOT}/{model}:batchEmbedContents?key={key}",
                       payload, "gemini")
    raw = [(e or {}).get("values") for e in (data.get("embeddings") or [])]
    return _as_vectors(raw, len(texts), "gemini")


def _as_vectors(raw, expected: int, backend: str) -> list[list[float]]:
    """Validate a batch response: one non-empty float vector per input text, all
    the same width. A short or ragged batch would silently misalign vectors with
    issues, so it is treated as the backend being unusable."""
    try:
        vectors = [[float(x) for x in (v or [])] for v in (raw or [])]
    except (TypeError, ValueError) as e:  # a null inside a vector is still "unusable", not a crash
        raise EmbeddingUnavailable(f"{backend} returned an unusable batch") from e
    widths = {len(v) for v in vectors}
    if len(vectors) != expected or not widths or 0 in widths or len(widths) > 1:
        raise EmbeddingUnavailable(f"{backend} returned an unusable batch")
    return vectors


def embed_texts(cfg: dict, texts: list[str]) -> list[list[float]]:
    """Embed `texts` with the configured backend, in batches of EMBED_BATCH.

    Raises EmbeddingUnavailable when there is no backend or it cannot answer.
    """
    settings = embeddings_settings(cfg)
    if settings is None:
        raise EmbeddingUnavailable("no backend configured")
    backend, model = settings
    encode = _embed_ollama if backend == "ollama" else _embed_gemini
    vectors: list[list[float]] = []
    for start in range(0, len(texts), EMBED_BATCH):
        vectors.extend(encode(model, texts[start:start + EMBED_BATCH]))
    return vectors


def embedding_text(issue: dict) -> str:
    """What actually gets embedded: the title, then the head of the description."""
    return (f"{issue.get('title') or ''}\n"
            f"{(issue.get('description') or '')[:EMBED_TEXT_CHARS]}")


def _cache_db() -> sqlite3.Connection:
    """Open (creating it) the cache next to config.json: the team board that
    `--similar` ranks over, and the embedding vectors of the optional ranker.

    Ticket titles and descriptions live in here, so it gets the same private
    modes as the config that holds the API key.
    """
    CACHE_DB_PATH.parent.mkdir(mode=CONFIG_DIR_MODE, parents=True, exist_ok=True)
    if not CACHE_DB_PATH.exists() and _LEGACY_EMBEDDINGS_DB_PATH.exists():
        with contextlib.suppress(FileNotFoundError):  # a parallel first run already moved it
            os.replace(_LEGACY_EMBEDDINGS_DB_PATH, CACHE_DB_PATH)
    CACHE_DB_PATH.touch(mode=CONFIG_FILE_MODE, exist_ok=True)
    conn = sqlite3.connect(CACHE_DB_PATH)
    conn.execute("""CREATE TABLE IF NOT EXISTS vectors (
                        issue_id TEXT, model TEXT, updated_at TEXT,
                        dims INTEGER, vec BLOB,
                        PRIMARY KEY(issue_id, model))""")
    conn.execute("""CREATE TABLE IF NOT EXISTS issues (
                        issue_id TEXT PRIMARY KEY, team_id TEXT,
                        updated_at TEXT, node TEXT)""")
    conn.execute("CREATE INDEX IF NOT EXISTS issues_team ON issues(team_id)")
    conn.execute("""CREATE TABLE IF NOT EXISTS sync (
                        team_id TEXT PRIMARY KEY, last_updated_at TEXT, full_at TEXT)""")
    conn.commit()
    with contextlib.suppress(OSError):
        CACHE_DB_PATH.chmod(CONFIG_FILE_MODE)
    return conn


def _cached_vectors(conn: sqlite3.Connection, model: str,
                    issue_ids: list[str]) -> dict[str, tuple[str, list[float]]]:
    """{issue_id: (updated_at, vector)} for ids already embedded with `model`."""
    cached: dict[str, tuple[str, list[float]]] = {}
    for start in range(0, len(issue_ids), _SQL_MAX_PARAMS):
        chunk = issue_ids[start:start + _SQL_MAX_PARAMS]
        holes = ",".join("?" * len(chunk))
        rows = conn.execute(
            "SELECT issue_id, updated_at, vec FROM vectors "
            f"WHERE model = ? AND issue_id IN ({holes})", [model, *chunk])
        for issue_id, updated_at, vec in rows:
            cached[issue_id] = (updated_at or "", array.array("f", vec).tolist())
    return cached


def _embeddings_for(cfg: dict, issues: list[dict]) -> dict[str, list[float]]:
    """Vectors for `issues` keyed by issue id, embedding only what the cache is
    missing or holds against an older updatedAt. Raises EmbeddingUnavailable."""
    settings = embeddings_settings(cfg)
    if settings is None:
        raise EmbeddingUnavailable("no backend configured")
    _, model = settings
    conn = _cache_db()
    try:
        keyed = [i for i in issues if i.get("id")]
        cached = _cached_vectors(conn, model, [i["id"] for i in keyed])
        vectors: dict[str, list[float]] = {}
        stale: list[dict] = []
        for issue in keyed:
            hit = cached.get(issue["id"])
            if hit and hit[0] == (issue.get("updatedAt") or ""):
                vectors[issue["id"]] = hit[1]
            else:
                stale.append(issue)
        if stale:
            fresh = embed_texts(cfg, [embedding_text(i) for i in stale])
            rows = []
            for issue, vec in zip(stale, fresh):
                vectors[issue["id"]] = vec
                rows.append((issue["id"], model, issue.get("updatedAt") or "",
                             len(vec), array.array("f", vec).tobytes()))
            conn.executemany(
                "INSERT INTO vectors (issue_id, model, updated_at, dims, vec) "
                "VALUES (?, ?, ?, ?, ?) ON CONFLICT(issue_id, model) DO UPDATE SET "
                "updated_at = excluded.updated_at, dims = excluded.dims, "
                "vec = excluded.vec", rows)
            conn.commit()
        return vectors
    finally:
        conn.close()


def cosine(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    return dot / (na * nb) if na and nb else 0.0


def embedding_rank(text: str, issues: list[dict],
                   cfg: dict) -> list[tuple[str, float]] | None:
    """Rank `issues` by cosine similarity to `text`: (identifier, score) best-first.

    Returns None — not an empty list — when the backend is unreachable, so
    rank_similar leaves "embeddings" out of the rankers it reports instead of
    claiming a ranker ran and found nothing.
    """
    try:
        vectors = _embeddings_for(cfg, issues)
        query = embed_texts(cfg, [text])[0]
    except EmbeddingUnavailable as e:
        print(f"embeddings: {e}, skipped", file=sys.stderr)
        return None
    scored = []
    for issue in issues:
        vec = vectors.get(issue.get("id"))
        # A width mismatch means the cache holds a vector from another model
        # under the same name; zip would score it against a prefix. Drop it.
        if not vec or len(vec) != len(query):
            continue
        score = cosine(query, vec)
        if score > 0:
            scored.append((issue["identifier"], score))
    scored.sort(key=lambda t: (-t[1], t[0]))
    return scored


def embeddings_ranker(api_key: str, team_id: str, cfg: dict, text: str,
                      candidates: list[dict]) -> list[str] | None:
    """EXTRA_RANKERS adapter: identifiers best-first, or None when skipped."""
    ranked = embedding_rank(text, candidates, cfg)
    return None if ranked is None else [ident for ident, _ in ranked]


def register_embeddings_ranker(cfg: dict) -> None:
    """Register the ranker only when a backend is configured, so a box without
    one pays nothing and never sees "embeddings" in the rankers that ran."""
    if embeddings_settings(cfg):
        EXTRA_RANKERS["embeddings"] = embeddings_ranker
    else:
        EXTRA_RANKERS.pop("embeddings", None)


# ---------------------------------------------------------------------------
# tasks - list, view, board
# ---------------------------------------------------------------------------

def cmd_tasks(args, cfg, api_key, team_id):
    # If an identifier is given, show detail view
    if args.identifier:
        return show_issue_detail(args, api_key, team_id)

    # If --board, show team board
    if args.board:
        return show_board(args, api_key, team_id, cfg)

    # Otherwise, list tasks
    return list_tasks(args, cfg, api_key, team_id)


def list_tasks(args, cfg, api_key, team_id):
    """Default view: issues delegated to my agent + issues with no delegate
    (unowned). Ownership is Linear's native `delegate` field — nothing else.
    `--agent X` narrows to a specific agent. `--all` shows literally every task.
    `--cycle` scopes to the active cycle (default), next cycle, the whole team
    (`all`, cross-cycle + backlog), or the backlog only (`none`).
    `--assignee EMAIL|me|none` filters by who the issue is assigned to.
    `--project NAME|UUID` scopes to one project (strict: unknown name aborts).
    With no `--project` and no `--all`, the queue auto-scopes to the project
    bound to the current directory (via `agents projects`), so an agent launched
    inside a project folder sees that project's work, not the whole workspace.
    Tasks are sorted by priority, then due date (earliest first), then id.
    Results are fully paginated — no silent truncation at Linear's 50-issue cap.
    """
    # Project scope, resolved up front because it also widens the cycle scope (a
    # project view spans all cycles, not just the active one). Precedence:
    #   1. explicit --project wins (existing behavior).
    #   2. --all forces the whole-team view (auto-scope suppressed).
    #   3. otherwise, when `agents projects` binds this cwd to a project,
    #      auto-scope to it — cwd behaves like an implicit --project.
    # This is the fix for "an agent launched in a project subfolder lists every
    # project's tasks and picks the wrong one". Opt out with `autoScope: false`
    # in config.
    #
    # The two sources have DIFFERENT failure policies: a mistyped --project must
    # abort (RUSH-1496), but a stale/malformed cwd binding — which comes from
    # `agents projects`, outside this CLI's control — must fall back to the
    # unscoped view, never brick the single most-common command. So the explicit
    # path resolves strictly (crash on no-match) and the auto path validates
    # fail-open (drift -> None -> unscoped + a one-line stderr note).
    if getattr(args, "similar", None):
        return similar_tasks(args, cfg, api_key, team_id)

    explicit_project = getattr(args, "project", None)
    pid = None
    auto_project_name = None
    if explicit_project:
        try:
            pid = resolve_project_id(api_key, team_id, explicit_project, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    elif not args.all and cfg.get("autoScope", True) is not False:
        cwd_pid, cwd_pname = resolve_cwd_project()
        if cwd_pid:
            pid = resolve_auto_project_id(api_key, team_id, cwd_pid)
            if pid:
                auto_project_name = cwd_pname or cwd_pid
            else:
                print(f"linear: the current directory's project "
                      f"'{cwd_pname or cwd_pid}' (from agents projects) is not a "
                      f"live project on this team — showing the whole team. Use "
                      f"--project to scope explicitly.", file=sys.stderr)

    filters = [f'team: {{ id: {{ eq: "{team_id}" }} }}']

    # Cycle scope. active/next resolve to a concrete cycle id (and a header
    # name); `all` drops the cycle filter; `none`/`backlog` lists the backlog;
    # any other value is treated as a cycle name or id and resolved. A resolved
    # project scope (explicit or auto) widens to all cycles.
    scope = resolve_task_scope(args.cycle, pid,
                               getattr(args, "milestone", None))
    fragment, scope_label, cycle_meta = build_cycle_scope(api_key, team_id, scope)
    if fragment is None:
        print(f"Could not resolve --cycle '{scope}'.")
        return
    if fragment:
        filters.append(fragment)

    # Optional date floor: only issues created on/after --since (YYYY-MM-DD).
    since = getattr(args, "since", None)
    if since:
        try:
            date.fromisoformat(since)
        except ValueError:
            print(f"Invalid --since '{since}'. Use YYYY-MM-DD.", file=sys.stderr)
            return
        filters.append(f'createdAt: {{ gte: "{since}T00:00:00.000Z" }}')

    # Resolve which agent's queue to show (if any). `--all` overrides config default.
    if args.all:
        requested_agent = None
        include_unowned = False  # --all means show everything including other agents
    elif args.agent:
        requested_agent = args.agent
        include_unowned = False  # explicit agent request — just show their queue
    elif cfg.get("agent"):
        requested_agent = cfg.get("agent")
        include_unowned = True   # default view = my queue + unowned
    else:
        requested_agent = None
        include_unowned = False

    # Ownership is the native `delegate` field. Resolving the requested name
    # against the roster is a *check*, not a requirement: the filter compares
    # delegate names case-insensitively, so the raw name already matches.
    #
    # A typed `--agent` that doesn't resolve is a typo — exit, because silently
    # returning "no matching tasks" reads as an empty queue and unattended drains
    # act on that. The config-derived default is different: `get_agents` degrades
    # to [] on a transient API error (see its `return cached or []`), and bricking
    # the most-used command on a users-query blip is worse than the typo it would
    # catch. Warn and keep going there.
    explicit_agent = None
    if requested_agent:
        explicit_agent = resolve_agent_name(api_key, cfg, requested_agent)
        if not explicit_agent:
            names = ", ".join(a["name"] for a in get_agents(api_key, cfg)) or "none"
            if args.agent:
                print(f"Unknown agent '{requested_agent}'. Delegatable agents: {names}",
                      file=sys.stderr)
                sys.exit(1)
            print(f"Warning: configured agent '{requested_agent}' is not in the agent "
                  f"roster ({names}). Filtering on the name as given — "
                  f"re-run `linear setup --agent <name>` if it was renamed.",
                  file=sys.stderr)
            # `.strip()` to match what resolve_agent_name would have done — a
            # config value of " claude " must still filter.
            explicit_agent = requested_agent.strip()

    # `--label` composes with the agent queue. It used to be suppressed whenever
    # an agent filter was active, because ownership was itself a label and the two
    # label filters fought; ownership is the `delegate` field now, so they are
    # orthogonal and a caller asking for both gets both.
    if args.label:
        filters.append(f'labels: {{ name: {{ eq: "{args.label}" }} }}')

    # Project scope filter. `pid` was resolved up front: an explicit --project
    # crashed on a typo, an auto-detected cwd project fell back to None on drift.
    if pid:
        filters.append(f'project: {{ id: {{ eq: "{pid}" }} }}')

    # Milestone scope. Strict like --project: a mistyped milestone aborts rather
    # than silently listing the whole team. Pass --project alongside to
    # disambiguate a milestone name that exists in more than one project.
    milestone = getattr(args, "milestone", None)
    if milestone:
        try:
            mid = resolve_milestone_id(api_key, team_id, milestone, pid, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
        if not mid:
            print(f"No milestone matching '{milestone}'.", file=sys.stderr)
            sys.exit(1)
        filters.append(f'projectMilestone: {{ id: {{ eq: "{mid}" }} }}')

    # Assignee filter: 'me' = API-key owner, 'none' = unassigned, else email.
    assignee = getattr(args, "assignee", None)
    if assignee:
        if assignee.lower() == "none":
            filters.append("assignee: { null: true }")
        else:
            if assignee.lower() == "me":
                uid = get_viewer_id(api_key, cfg)
            else:
                uid = resolve_user_id_by_email(api_key, assignee)
            if not uid:
                print(f"Could not resolve assignee '{assignee}'.", file=sys.stderr)
                return
            filters.append(f'assignee: {{ id: {{ eq: "{uid}" }} }}')

    if getattr(args, "query", None):
        q = args.query.replace('"', '\\"')
        filters.append(
            f'or: [ {{ title: {{ containsIgnoreCase: "{q}" }} }}, '
            f'{{ description: {{ containsIgnoreCase: "{q}" }} }} ]'
        )

    if args.status:
        status = args.status.lower()
        if status == "open":
            filters.append(OPEN_STATE_FILTER)
        elif status == "done":
            filters.append('state: { type: { eq: "completed" } }')
        else:
            states = get_states(api_key, team_id, cfg)
            sid = resolve_state_id(states, status)
            if sid:
                filters.append(f'state: {{ id: {{ eq: "{sid}" }} }}')
            else:
                filters.append(f'state: {{ name: {{ eqi: "{status}" }} }}')
    else:
        filters.append(OPEN_STATE_FILTER)

    filter_str = ", ".join(filters)

    nodes = paginate_issues(api_key, filter_str)
    if nodes is None:
        return

    # Apply the delegate filter client-side so "my tasks + unowned" is expressible
    # in one pass (Linear can't express `delegate = X OR delegate IS NULL` in a
    # single server-side filter, and the page is already fetched).
    if explicit_agent:
        def match(n):
            if delegated_to(n, explicit_agent):
                return True
            return include_unowned and delegate_of(n) is None

        nodes = [n for n in nodes if match(n)]

    nodes.sort(key=issue_sort_key)

    if args.json:
        print(json.dumps({
            "scope": scope,
            "cycle": cycle_meta,
            # The effective project scope, so machine consumers can tell an
            # auto-scoped queue from the whole workspace. Null when unscoped.
            "project": ({"id": pid, "name": auto_project_name or explicit_project,
                         "auto": bool(auto_project_name)} if pid else None),
            "count": len(nodes),
            "issues": nodes,
        }, indent=2))
        return

    # When the cwd project auto-scoped the queue, say so — otherwise a short or
    # empty list reads as "no work" when it just means "none in THIS project".
    if auto_project_name:
        print(f"Scope: {auto_project_name}  (auto-detected from this directory; "
              f"--all for the whole team, --project X to override)")

    if not nodes:
        hint = (f" (scoped to {auto_project_name}; try --all, or --cycle all "
                f"for the backlog)") if auto_project_name else ""
        print(f"No matching tasks in {scope_label}.{hint}")
        return

    header_note = ""
    if explicit_agent and include_unowned:
        owned = sum(1 for n in nodes if delegated_to(n, explicit_agent))
        unowned = len(nodes) - owned
        header_note = f"  ({owned} yours, {unowned} unowned)"
    print(f"{scope_label} -- {len(nodes)} task(s){header_note}")
    print()
    # Group by milestone when asked (--by-milestone) or when scoped to a project
    # (the natural project view). The milestone name is already on each issue via
    # ISSUE_FIELDS, so this is a local group-by with no extra API calls.
    if getattr(args, "by_milestone", False) or bool(pid):
        print_by_milestone(nodes)
    else:
        for n in nodes:
            print(format_issue_row(n))


def show_board(args, api_key, team_id, cfg):
    # Full parity with list_tasks: an explicit --project resolves strictly and
    # aborts on a typo; --all shows every project; otherwise the board
    # auto-scopes to the cwd's project, validated fail-open so a stale/malformed
    # binding falls back to the whole team instead of rendering a
    # misleadingly-empty board. Resolved BEFORE the cycle scope, because — like
    # list_tasks — a project scope widens the cycle to the whole deliverable.
    explicit_project = getattr(args, "project", None)
    auto_project_name = None
    pid = None
    if explicit_project:
        try:
            pid = resolve_project_id(api_key, team_id, explicit_project, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    elif not getattr(args, "all", False) and cfg.get("autoScope", True) is not False:
        cwd_pid, cwd_pname = resolve_cwd_project()
        if cwd_pid:
            pid = resolve_auto_project_id(api_key, team_id, cwd_pid)
            if pid:
                auto_project_name = cwd_pname or cwd_pid
            else:
                print(f"linear: the current directory's project "
                      f"'{cwd_pname or cwd_pid}' (from agents projects) is not a "
                      f"live project on this team — showing the whole team. Use "
                      f"--project to scope explicitly.", file=sys.stderr)

    # Same cycle scoping as list_tasks (active/next/all/none/name/id), so the
    # board no longer hits the non-existent team.nextCycle field, and it
    # paginates instead of silently capping at 50 open issues. --cycle defaults
    # to None at the parser; a project scope widens to all cycles (the whole
    # deliverable), otherwise the board's default is the active cycle — and
    # normalizing None here keeps the --json "scope" field a string, not null.
    scope = args.cycle or ("all" if pid else "active")
    fragment, scope_label, cycle_meta = build_cycle_scope(api_key, team_id, scope)
    if fragment is None:
        print(f"Could not resolve --cycle '{scope}'.")
        return

    filters = [
        f'team: {{ id: {{ eq: "{team_id}" }} }}',
        OPEN_STATE_FILTER,
    ]
    if fragment:
        filters.append(fragment)
    if pid:
        filters.append(f'project: {{ id: {{ eq: "{pid}" }} }}')

    nodes = paginate_issues(api_key, ", ".join(filters))
    if nodes is None:
        return

    cycle_name = scope_label

    if args.json:
        print(json.dumps({
            "scope": scope,
            "cycle": cycle_meta,
            "project": ({"id": pid, "name": auto_project_name or explicit_project,
                         "auto": bool(auto_project_name)} if pid else None),
            "count": len(nodes),
            "issues": nodes,
        }, indent=2))
        return

    if auto_project_name:
        print(f"Scope: {auto_project_name}  (auto-detected from this directory; "
              f"--all for the whole team, --project X to override)")

    if not nodes:
        hint = f" (scoped to {auto_project_name}; --all for the whole team)" if auto_project_name else ""
        print(f"No open tasks in {cycle_name}.{hint}")
        return

    # Group by the native delegate. An issue has at most one delegate, so a board
    # column is exactly one agent's queue; no delegate means unassigned.
    agents: dict[str, list] = {}
    unassigned: list = []

    for n in nodes:
        dn = delegate_of(n)
        if dn:
            agents.setdefault(dn, []).append(n)
        else:
            unassigned.append(n)

    print(f"{cycle_name} -- {len(nodes)} task(s)")
    print()

    for agent_name in sorted(agents.keys()):
        issues = agents[agent_name]
        issues.sort(key=issue_sort_key)
        print(f"  @{agent_name} ({len(issues)})")
        for n in issues:
            pri = PRIORITY_MAP.get(n.get("priority", 0), "-")
            state = n["state"]["name"]
            due = format_due(n.get("dueDate"))
            print(f"    {n['identifier']:<8} {pri:<7}  {state:<12}  {due}  {n['title']}")
        print()

    if unassigned:
        unassigned.sort(key=issue_sort_key)
        print(f"  unassigned ({len(unassigned)})")
        for n in unassigned:
            pri = PRIORITY_MAP.get(n.get("priority", 0), "-")
            state = n["state"]["name"]
            due = format_due(n.get("dueDate"))
            labels = [l["name"] for l in n.get("labels", {}).get("nodes", [])]
            label_str = f"  [{', '.join(labels)}]" if labels else ""
            print(f"    {n['identifier']:<8} {pri:<7}  {state:<12}  {due}  {n['title']}{label_str}")
        print()


_ISSUE_COMMENTS_QUERY = """query($id: ID!, $cursor: String) {
    comments(
        filter: { issue: { id: { eq: $id } } }
        orderBy: createdAt
        first: 100
        after: $cursor
    ) {
        pageInfo { hasNextPage endCursor }
        nodes {
            id body createdAt url
            user { id name avatarUrl app }
        }
    }
}"""


def fetch_issue_comments(api_key: str, issue_id: str) -> list[dict] | None:
    """Every comment on one issue, fully paginated, in a deterministic order
    (createdAt ascending, then id) with stable comment ids. Returns None on an
    API error OR a truncated result so the caller exits non-zero — a comment
    thread that hit the pagination safety rail is NOT a complete detail view, so
    honor the paginator's partial_out and fail loudly rather than pass off a
    silently-cut list as whole."""
    capped: dict = {}
    nodes = paginate_connection(api_key, _ISSUE_COMMENTS_QUERY, ["comments"],
                                {"id": issue_id}, partial_out=capped)
    if nodes is None:
        return None
    if capped:
        print(f"Error: comment history truncated at {capped['rows']} "
              f"(pagination safety rail); refusing to show a partial thread.",
              file=sys.stderr)
        return None
    # Tie-break on id so equal timestamps still order deterministically.
    nodes.sort(key=lambda c: (c.get("createdAt") or "", c.get("id") or ""))
    return nodes


UPLOADS_DIR = Path.home() / ".linear-cli" / "uploads"
UPLOADS_MAX = 12
UPLOAD_MAX_BYTES = 20 * 1024 * 1024
UPLOAD_EXTENSIONS = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".heic", ".svg", ".bmp", ".tiff", ".img")
_UPLOAD_IMAGE_RE = re.compile(r'!\[([^\]]*)\]\(\s*<?(https://uploads\.linear\.app/[^\s)>]+)>?(?:\s+"[^"]*")?\s*\)')


def issue_images(api_key: str, texts: list[str]) -> list[dict]:
    """The Linear-hosted images a ticket's description and comments embed, in
    order, each downloaded once into UPLOADS_DIR: `{url, alt, path}`, `path`
    null when the download failed. uploads.linear.app needs the API key, so a
    reader that only has the URL cannot show the picture.
    """
    seen, images = set(), []
    for text in texts:
        for alt, url in _UPLOAD_IMAGE_RE.findall(text or ""):
            if url in seen or len(images) >= UPLOADS_MAX:
                continue
            seen.add(url)
            images.append({"url": url, "alt": alt, "path": _cached_upload(api_key, url)})
    return images


def _cached_upload(api_key: str, url: str) -> str | None:
    stem = hashlib.sha256(url.encode()).hexdigest()[:24]
    for ext in UPLOAD_EXTENSIONS:
        if (UPLOADS_DIR / f"{stem}{ext}").is_file():
            return str(UPLOADS_DIR / f"{stem}{ext}")
    req = Request(url)
    # Never forwarded on a redirect: the key is for uploads.linear.app alone.
    req.add_unredirected_header("Authorization", api_key)
    try:
        with urlopen(req, timeout=API_TIMEOUT) as resp:
            kind = (resp.headers.get_content_type() or "").lower()
            if not kind.startswith("image/"):
                return None
            body = resp.read(UPLOAD_MAX_BYTES + 1)
    except (URLError, TimeoutError, OSError):
        return None
    if len(body) > UPLOAD_MAX_BYTES:
        return None
    ext = mimetypes.guess_extension(kind) or ".img"
    if ext not in UPLOAD_EXTENSIONS:
        ext = ".img"
    target = UPLOADS_DIR / f"{stem}{ext}"
    partial = UPLOADS_DIR / f"{stem}.part"
    try:
        UPLOADS_DIR.mkdir(mode=CONFIG_DIR_MODE, parents=True, exist_ok=True)
        fd = os.open(partial, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, CONFIG_FILE_MODE)
        with os.fdopen(fd, "wb") as f:
            f.write(body)
        os.replace(partial, target)
    except OSError:
        with contextlib.suppress(OSError):
            partial.unlink()
        return None
    return str(target)


def show_issue_detail(args, api_key, team_id):
    identifier = args.identifier
    number = identifier.split("-")[-1]
    try:
        number = int(number)
    except ValueError:
        print(f"Invalid identifier: {identifier}", file=sys.stderr)
        sys.exit(1)

    data = gql(api_key, f"""{{
        issues(filter: {{
            team: {{ id: {{ eq: "{team_id}" }} }}
            number: {{ eq: {number} }}
        }}) {{
            nodes {{
                id identifier title description url
                state {{ name }}
                priority dueDate
                labels {{ nodes {{ name }} }}
                assignee {{ id name avatarUrl app }}
                delegate {{ id name avatarUrl app }}
                project {{ name }}
                projectMilestone {{ name }}
                cycle {{ number name }}
                parent {{ identifier title }}
                createdAt updatedAt
                relations {{ nodes {{ type relatedIssue {{ identifier title }} }} }}
                inverseRelations {{ nodes {{ type issue {{ identifier title }} }} }}
            }}
        }}
    }}""")

    # An API error is a hard failure — exit non-zero rather than return 0 with
    # no output, so a scripted caller can tell "errored" from "found nothing".
    if check_errors(data):
        sys.exit(1)

    nodes = data.get("data", {}).get("issues", {}).get("nodes", [])
    if not nodes:
        print(f"Issue {identifier} not found.")
        sys.exit(1)

    n = nodes[0]

    # Comments are fetched separately and FULLY paginated with stable ids and a
    # deterministic order — the inline `comments {{ nodes }}` above only ever
    # returned Linear's first page (a silent cutoff) and carried no comment id.
    comments = fetch_issue_comments(api_key, n["id"])
    if comments is None:
        sys.exit(1)  # paginate_connection already printed the API error
    n["comments"] = {"nodes": comments}

    if args.json:
        n["images"] = issue_images(api_key, [n.get("description") or ""] + [c.get("body") or "" for c in comments])
        print(json.dumps(n, indent=2))
        return

    pri = PRIORITY_MAP.get(n.get("priority", 0), "-")
    labels = ", ".join(l["name"] for l in n.get("labels", {}).get("nodes", []))
    assignee = n.get("assignee")
    assignee_name = assignee.get("name", "Unassigned") if assignee else "Unassigned"
    delegate = n.get("delegate")
    delegate_name = delegate.get("name") if delegate else None

    print(f"{n['identifier']}  {n['title']}")
    print(f"  Status:   {n['state']['name']}")
    print(f"  Priority: {pri}")
    print(f"  Labels:   {labels or '-'}")
    print(f"  Assignee: {assignee_name}")
    if delegate_name:
        print(f"  Delegate: {delegate_name}")
    if n.get("dueDate"):
        print(f"  Due:      {n['dueDate']}")
    if n.get("project"):
        print(f"  Project:  {n['project']['name']}")
    if n.get("projectMilestone"):
        print(f"  Milestone: {n['projectMilestone']['name']}")
    if n.get("cycle"):
        print(f"  Cycle:    {cycle_label(n['cycle'])}")
    if n.get("parent"):
        print(f"  Parent:   {n['parent']['identifier']}  {n['parent']['title']}")
    if n.get("url"):
        print(f"  URL:      {n['url']}")

    # Relations: forward edges (this blocks / relates to X) + inverse (X blocks this).
    rel_lines = []
    for r in (n.get("relations") or {}).get("nodes", []):
        other = r.get("relatedIssue") or {}
        verb = {"blocks": "Blocks", "related": "Related to", "duplicate": "Duplicate of"}.get(r.get("type"), r.get("type"))
        rel_lines.append(f"  {verb}: {other.get('identifier', '?')}  {other.get('title', '')}")
    for r in (n.get("inverseRelations") or {}).get("nodes", []):
        other = r.get("issue") or {}
        # "related" is symmetric; "blocks" inverts to "blocked by" and
        # "duplicate" to "duplicated by" (the other issue is the duplicate).
        verb = {"blocks": "Blocked by", "related": "Related to", "duplicate": "Duplicated by"}.get(r.get("type"), r.get("type"))
        rel_lines.append(f"  {verb}: {other.get('identifier', '?')}  {other.get('title', '')}")
    if rel_lines:
        print("  Relations:")
        for line in rel_lines:
            print(f"  {line}")
    print()

    desc = n.get("description", "")
    if desc:
        print("Description:")
        for line in desc.split("\n"):
            print(f"  {line}")
        print()

    comments = n.get("comments", {}).get("nodes", [])
    if comments:
        print(f"Comments ({len(comments)}):")
        for c in comments:
            user = c.get("user", {}).get("name", "Unknown")
            body = c.get("body", "").strip().replace("\n", "\n    ")
            print(f"  [{user}] {body}")
        print()


# ---------------------------------------------------------------------------
# update - all mutations on an existing issue
# ---------------------------------------------------------------------------

def collect_update_identifiers(identifiers, stdin_lines=None) -> list[str]:
    collected = list(identifiers or [])
    for line in stdin_lines or []:
        collected.extend(line.split())
    return list(dict.fromkeys(collected))


def run_bulk_update(args, cfg, api_key, team_id, identifiers: list[str],
                    project_id=_UNSET, milestone_id=_UNSET,
                    resolve_issue_fn=None, apply_update_fn=None):
    resolve_issue_fn = resolve_issue_fn or resolve_issue
    apply_update_fn = apply_update_fn or _apply_update
    total = len(identifiers)
    succeeded = 0
    failed: list[str] = []
    for i, ident in enumerate(identifiers, 1):
        issue = resolve_issue_fn(api_key, team_id, ident)
        if not issue:
            print(f"[{i}/{total}] {ident} -> not found")
            failed.append(ident)
            continue
        print(f"[{i}/{total}] {issue['identifier']}")
        try:
            if apply_update_fn(args, cfg, api_key, team_id, issue, single=False,
                               project_id=project_id, milestone_id=milestone_id):
                succeeded += 1
            else:
                failed.append(ident)
        except Exception as e:  # never let one ticket abort the batch
            print(f"  error: {e}", file=sys.stderr)
            failed.append(ident)

    summary = f"\n{succeeded}/{total} updated."
    if failed:
        summary += f" Failed: {', '.join(failed)}"
    sys.stdout.flush()  # keep per-issue progress ahead of the summary when piped
    print(summary, file=sys.stderr)
    if failed:
        sys.exit(1)


def cmd_update(args, cfg, api_key, team_id):
    """Dispatcher: apply the same set of changes to one or many issues.

    Identifiers come from positional args and/or --stdin (whitespace-separated,
    xargs-friendly). A single identifier keeps the original detailed output; two
    or more switch to compact per-issue progress with an error roll-up at the
    end so one bad ticket never aborts the batch.
    """
    # Require proof when marking done — checked once up front for the whole batch.
    if args.done and not args.proof:
        print("Cannot mark done without proof. Provide at least one --proof:", file=sys.stderr)
        print("  --proof /path/to/screenshot.png   (file upload)", file=sys.stderr)
        print("  --proof https://example.com/result (URL)", file=sys.stderr)
        print('  --proof "Sent 15/30 emails"        (text)', file=sys.stderr)
        sys.exit(1)

    stdin_lines = sys.stdin if getattr(args, "stdin", False) else None
    identifiers = collect_update_identifiers(args.identifier, stdin_lines)
    if not identifiers:
        print("No issue identifier given.", file=sys.stderr)
        sys.exit(1)

    # Next-invocation drain: a previous --done that hit a rate limit may have
    # left intents on disk. Clear them before adding a new close so the board
    # stays honest across invocations.
    if args.done:
        drain_queue(api_key, team_id, cfg, quiet=True)

    if getattr(args, "remind_at", None):
        args.remind_at = parse_remind_at(args.remind_at)

    # Resolve --project / --milestone ONCE, up front and strictly. An unknown
    # name aborts the whole run with a non-zero exit instead of silently no-op'ing
    # every issue in a batch while reporting success (RUSH-1496); resolving once
    # also avoids a per-issue project lookup on bulk updates.
    project_id = _UNSET
    milestone_id = _UNSET
    try:
        # Truthy (not `is not None`) so an empty `--project ""` is a no-op, not a
        # clear — only an explicit `--project none` clears the field.
        if args.project:
            project_id = (None if args.project.lower() == "none"
                          else resolve_project_id(api_key, team_id, args.project, strict=True))
        if args.milestone:
            scope = project_id if project_id not in (_UNSET, None) else None
            milestone_id = (None if args.milestone.lower() == "none"
                            else resolve_milestone_id(api_key, team_id, args.milestone, scope, strict=True))
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    if len(identifiers) == 1:
        issue = resolve_issue(api_key, team_id, identifiers[0])
        if not issue:
            print(f"Issue {identifiers[0]} not found.")
            sys.exit(1)
        ok = _apply_update(args, cfg, api_key, team_id, issue, single=True,
                           project_id=project_id, milestone_id=milestone_id)
        if not ok:
            sys.exit(1)
        return

    run_bulk_update(args, cfg, api_key, team_id, identifiers,
                    project_id=project_id, milestone_id=milestone_id)


def _apply_update(args, cfg, api_key, team_id, issue, single=True,
                  project_id=_UNSET, milestone_id=_UNSET) -> bool:
    """Apply every requested change to one resolved issue. Returns True if at
    least one change was made. Never mutates `args` (so it is safe to call in a
    loop) and never calls sys.exit (so a failure on one issue can be rolled up
    by the bulk caller). Hard per-issue failures return False.

    project_id / milestone_id are pre-resolved once by cmd_update: _UNSET means
    the flag was absent (leave the field alone), None means clear it, a string
    means set it."""
    identifier = issue.get("identifier") or args.identifier
    did_something = False

    # A local copy so the bulk loop doesn't see comment cleared after issue #1.
    comment = args.comment

    # Post proof as a comment before status change. When closing (--done),
    # proof is handled durably by the close queue so a rate limit cannot drop
    # it between the comment and the status change.
    if args.proof and not args.done:
        proof_body = build_proof_comment(api_key, args.proof)
        if comment:
            proof_body = f"{comment}\n\n{proof_body}"

        data = gql(api_key, """
            mutation($input: CommentCreateInput!) {
                commentCreate(input: $input) { success }
            }
        """, {"input": {"issueId": issue["id"], "body": proof_body}})

        if check_errors(data):
            return False
        if data["data"]["commentCreate"]["success"]:
            print(f"Proof posted to {identifier}.")
        else:
            print("Proof comment failed.", file=sys.stderr)
            return False
        did_something = True
        # Comment already included in proof, don't post separately
        comment = None

    # Status change
    target_state = None
    if args.done:
        target_state = "Done"
    elif args.pickup:
        target_state = "In Progress"
    elif args.todo:
        target_state = "Todo"
    elif args.status:
        target_state = args.status

    if target_state:
        if args.done:
            # Close is the one operation that must survive rate limits. Persist
            # the intent before trying, attempt immediately, and keep it on
            # transient failure for the next drain.
            ok = queue_close_and_try(
                api_key, team_id, cfg, issue, target_state,
                proof=args.proof,
                comment=comment,
                quiet=False,
            )
            if not ok:
                return False
            did_something = True
        else:
            states = get_states(api_key, team_id, cfg)
            state_id = resolve_state_id(states, target_state)
            if not state_id:
                print(f"State '{target_state}' not found. Available: {', '.join(states.keys())}", file=sys.stderr)
                return False

            data = gql(api_key, """
                mutation($id: String!, $stateId: String!) {
                    issueUpdate(id: $id, input: { stateId: $stateId }) {
                        success
                        issue { identifier title state { name } }
                    }
                }
            """, {"id": issue["id"], "stateId": state_id})

            if check_errors(data):
                return False

            result = data["data"]["issueUpdate"]
            if result["success"]:
                i = result["issue"]
                print(f"{i['identifier']} -> {i['state']['name']}")
            else:
                print("Status update failed.")
            did_something = True

    # Issue relations. Linear models all of these as one relation with a
    # direction: "A blocks B" is the same edge as "B blocked-by A", so
    # --blocked-by just flips which side is the issueId. type is blocks|related.
    rel_specs = []
    for ident in (args.blocks or []):
        rel_specs.append(("blocks", ident, "forward", "blocks"))
    for ident in (args.blocked_by or []):
        rel_specs.append(("blocked by", ident, "inverse", "blocks"))
    for ident in (args.relates or []):
        rel_specs.append(("related to", ident, "forward", "related"))
    for verb, ident, direction, rel_type in rel_specs:
        other = resolve_issue(api_key, team_id, ident)
        if not other:
            print(f"Related issue {ident} not found, skipping.", file=sys.stderr)
            continue
        if direction == "forward":
            issue_id, related_id = issue["id"], other["id"]
        else:
            issue_id, related_id = other["id"], issue["id"]
        rel_data = gql(api_key, """
            mutation($input: IssueRelationCreateInput!) {
                issueRelationCreate(input: $input) { success }
            }
        """, {"input": {"issueId": issue_id, "relatedIssueId": related_id, "type": rel_type}})
        if check_errors(rel_data):
            continue
        if (rel_data.get("data") or {}).get("issueRelationCreate", {}).get("success"):
            print(f"{identifier} {verb} {other['identifier']}")
            did_something = True
        else:
            print(f"Failed to link {identifier} {verb} {ident}.", file=sys.stderr)

    # Labels (add via --label, remove via --unlabel)
    if args.label or args.unlabel:
        all_labels = list_team_labels(api_key, team_id)
        if all_labels is not None:
            label_map = {l["name"]: l["id"] for l in all_labels}

            # Also get current labels on the issue
            issue_data = gql(api_key, f"""{{
                issue(id: "{issue['id']}") {{
                    labels {{ nodes {{ id name }} }}
                }}
            }}""")
            current_ids = []
            current_names_by_id = {}
            if not check_errors(issue_data):
                for l in issue_data["data"]["issue"]["labels"]["nodes"]:
                    current_ids.append(l["id"])
                    current_names_by_id[l["id"]] = l["name"]

            label_ids = list(current_ids)
            added = []
            for lbl in args.label or []:
                if lbl in label_map:
                    if label_map[lbl] not in label_ids:
                        label_ids.append(label_map[lbl])
                        added.append(lbl)
                else:
                    print("Warning: " + _unknown_label_message(lbl, list(label_map)), file=sys.stderr)

            removed = []
            for lbl in args.unlabel or []:
                if lbl in label_map and label_map[lbl] in label_ids:
                    label_ids.remove(label_map[lbl])
                    removed.append(lbl)
                else:
                    print(f"Warning: label '{lbl}' not on this issue, skipping.", file=sys.stderr)

            if added or removed:
                data = gql(api_key, """
                    mutation($id: String!, $labelIds: [String!]!) {
                        issueUpdate(id: $id, input: { labelIds: $labelIds }) {
                            success
                        }
                    }
                """, {"id": issue["id"], "labelIds": label_ids})

                if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
                    parts = []
                    if added:
                        parts.append(f"added {', '.join(added)}")
                    if removed:
                        parts.append(f"removed {', '.join(removed)}")
                    print(f"Labels on {identifier}: {'; '.join(parts)}")
                else:
                    print("Label update failed.")
                did_something = True

    # Cycle move: --cycle active | next | none
    if args.cycle:
        choice = args.cycle.lower()
        new_cycle_id: str | None = None
        if choice == "none":
            new_cycle_id = None
        elif choice in ("active", "next"):
            new_cycle_id = get_cycle_id(api_key, team_id, choice)
            if not new_cycle_id:
                print(f"Warning: no {choice} cycle found; skipping cycle move.", file=sys.stderr)
                new_cycle_id = ...  # sentinel — skip mutation below

        if new_cycle_id is not ...:
            data = gql(api_key, """
                mutation($id: String!, $cycleId: String) {
                    issueUpdate(id: $id, input: { cycleId: $cycleId }) {
                        success
                        issue { identifier cycle { number name } }
                    }
                }
            """, {"id": issue["id"], "cycleId": new_cycle_id})
            if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
                i = data["data"]["issueUpdate"]["issue"]
                cyc = cycle_label(i.get("cycle"))
                print(f"{i['identifier']} -> cycle: {cyc}")
                did_something = True
            else:
                print("Cycle update failed.")

    # Title / description: batched in one mutation
    title_desc_input: dict = {}
    if args.title is not None:
        title_desc_input["title"] = args.title
    description = read_description(args.description, args.description_file)
    if description is not None:
        title_desc_input["description"] = description
    if args.parent is not None:
        if args.parent.lower() == "none":
            title_desc_input["parentId"] = None
        else:
            parent_issue = resolve_issue(api_key, team_id, args.parent)
            if parent_issue:
                title_desc_input["parentId"] = parent_issue["id"]
                _warn_sub_issue(identifier, parent_issue.get("identifier") or args.parent)
            else:
                print(f"Warning: parent '{args.parent}' not found; skipping.", file=sys.stderr)
    # project_id / milestone_id are pre-resolved (strictly) by cmd_update.
    if project_id is not _UNSET:
        title_desc_input["projectId"] = project_id
    if milestone_id is not _UNSET:
        title_desc_input["projectMilestoneId"] = milestone_id
    if title_desc_input:
        data = gql(api_key, """
            mutation($id: String!, $input: IssueUpdateInput!) {
                issueUpdate(id: $id, input: $input) {
                    success
                    issue { identifier title }
                }
            }
        """, {"id": issue["id"], "input": title_desc_input})
        if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
            i = data["data"]["issueUpdate"]["issue"]
            changed = ", ".join(title_desc_input.keys())
            print(f"{i['identifier']} -> updated: {changed}")
            did_something = True
        else:
            print("Title/description update failed.")

    # Priority: --priority urgent|high|medium|low|none
    if args.priority is not None:
        pri = parse_priority(args.priority)
        data = gql(api_key, """
            mutation($id: String!, $priority: Int!) {
                issueUpdate(id: $id, input: { priority: $priority }) {
                    success
                    issue { identifier priority }
                }
            }
        """, {"id": issue["id"], "priority": pri})
        if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
            i = data["data"]["issueUpdate"]["issue"]
            name = PRIORITY_MAP.get(i.get("priority", 0), "-")
            print(f"{i['identifier']} -> priority: {name}")
            did_something = True
        else:
            print("Priority update failed.")

    # Assign: --assign email | none
    if args.assign is not None:
        if args.assign.lower() == "none":
            assignee_id = None
        else:
            assignee_id = resolve_assignee_id(api_key, args.assign)
            if not assignee_id:
                print(f"Warning: --assign '{args.assign}' matched no human "
                      f"(try an email, a name, or 'linear users'); skipping reassign.", file=sys.stderr)
                assignee_id = ...

        if assignee_id is not ...:
            data = gql(api_key, """
                mutation($id: String!, $assigneeId: String) {
                    issueUpdate(id: $id, input: { assigneeId: $assigneeId }) {
                        success
                        issue { identifier assignee { name } }
                    }
                }
            """, {"id": issue["id"], "assigneeId": assignee_id})
            if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
                i = data["data"]["issueUpdate"]["issue"]
                asg = (i.get("assignee") or {}).get("name") or "unassigned"
                print(f"{i['identifier']} -> assignee: {asg}")
                did_something = True
            else:
                print("Assign update failed.")

    # Delegate: --delegate <agent-name> | none. Agents can't be plain assignees
    # (Linear silently ignores an app user in assigneeId); delegation is the
    # supported path — the human stays assignee, the agent becomes delegate.
    if getattr(args, "delegate", None) is not None:
        if args.delegate.lower() == "none":
            delegate_id = None
        else:
            delegate_id = resolve_agent_id(api_key, cfg, args.delegate)
            if not delegate_id:
                names = ", ".join(a["name"] for a in get_agents(api_key, cfg)) or "none installed"
                print(f"Warning: agent '{args.delegate}' not found. Available agents: {names}", file=sys.stderr)
                delegate_id = ...

        if delegate_id is not ...:
            data = gql(api_key, """
                mutation($id: String!, $delegateId: String) {
                    issueUpdate(id: $id, input: { delegateId: $delegateId }) {
                        success
                        issue { identifier delegate { name } }
                    }
                }
            """, {"id": issue["id"], "delegateId": delegate_id})
            if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
                i = data["data"]["issueUpdate"]["issue"]
                dg = (i.get("delegate") or {}).get("name") or "no agent"
                print(f"{i['identifier']} -> delegate: {dg}")
                did_something = True
            else:
                print("Delegate update failed.")

    # Due date: --due-date YYYY-MM-DD | none
    if args.due_date is not None:
        due = parse_due_date(args.due_date)
        data = gql(api_key, """
            mutation($id: String!, $due: TimelessDate) {
                issueUpdate(id: $id, input: { dueDate: $due }) {
                    success
                    issue { identifier dueDate }
                }
            }
        """, {"id": issue["id"], "due": due})
        if not check_errors(data) and data["data"]["issueUpdate"]["success"]:
            i = data["data"]["issueUpdate"]["issue"]
            print(f"{i['identifier']} -> due: {i['dueDate'] or 'cleared'}")
            did_something = True
        else:
            print("Due date update failed.")

    # Reminder: --remind-at <time>, a notification to you at that moment
    if getattr(args, "remind_at", None):
        at = args.remind_at
        if set_issue_reminder(api_key, issue["id"], at):
            print(f"{issue.get('identifier', args.issue)} -> reminder: {at}")
            did_something = True
        else:
            print("Reminder update failed.")

    # Comment (a proof already folds its comment in above, clearing `comment`)
    if comment:
        data = gql(api_key, """
            mutation($input: CommentCreateInput!) {
                commentCreate(input: $input) { success }
            }
        """, {"input": {"issueId": issue["id"], "body": comment}})

        if check_errors(data):
            return False

        if data["data"]["commentCreate"]["success"]:
            print(f"Comment added to {identifier}.")
            did_something = True
        else:
            # A rejected commentCreate is a real failure: the one-field action
            # must exit non-zero so a caller (and the bulk roll-up) can tell it
            # apart from a posted comment. Previously this printed "Comment
            # failed" but still set did_something=True and returned 0.
            print("Comment failed.", file=sys.stderr)
            return False

    if not did_something and single:
        print("Nothing to do. Use --done, --pickup, --todo, --status, --cycle, --label, --priority, --assign, --delegate, --title, --description, --due-date, --remind-at, or --comment.")
    return did_something


# ---------------------------------------------------------------------------
# create
# ---------------------------------------------------------------------------

def tsv_field(value: str) -> str:
    """Make a value safe to place in one field of a tab-separated record.

    Bulk `--from-file` output is TSV, so a tab in a value would add a phantom
    column and a newline would split the row. Both come straight from the
    caller's JSONL (title, description) or from Linear (issue title), so
    neither is trustworthy — escape at the point of formatting.
    """
    return (str(value)
            .replace("\t", "\\t")
            .replace("\r\n", "\\n")
            .replace("\n", "\\n")
            .replace("\r", "\\n"))


def unassigned_error(title: str, reason: str, multiline: bool = True) -> str:
    """The message shown when a create would leave an issue with no assignee.

    Multi-line for interactive `linear create` (the operator reads it and fixes
    the command); single-line for `--from-file`, whose per-row output is a
    tab-separated record that a newline would corrupt.
    """
    if not multiline:
        return (f"unassigned issue refused: {reason}. Every ticket needs a clear owner — "
                f"set \"assign\" to an email or name on this row (see `linear users`), "
                f"or pass --force to create it unowned anyway.")
    return (
        f"Refusing to create an unassigned issue: {title!r}\n"
        f"  Why: {reason}.\n"
        f"  Every ticket needs a clear owner. An unassigned issue is nobody's job — it\n"
        f"  lands in the board's \"No assignee\" bucket and nobody picks it up.\n"
        f"  Fix:  --assign <email|name>   (run `linear users` to see who you can assign)\n"
        f"  Handing the work to an agent does not replace the owner: --delegate rides\n"
        f"  alongside --assign, it does not stand in for it.\n"
        f"  If it genuinely has no owner yet, say so explicitly: --force"
    )


def conflicting_milestone_flags_error(milestone: str, multiline: bool = True) -> str:
    """--milestone and --skip-milestone together is a hard error.

    The milestone value is user data (and a bulk-row field), so it is
    repr-escaped the same way unassigned_error echoes --assign.
    """
    if not multiline:
        return (f"--milestone {milestone!r} and --skip-milestone cannot be combined; "
                f"pass one, not both")
    return (
        f"Refusing to create: --milestone and --skip-milestone cannot be combined.\n"
        f"  You passed --milestone {milestone!r} and --skip-milestone.\n"
        f"  Pass one, not both."
    )


def _catalog_heading(catalog: list[dict], project_name: str | None) -> str | None:
    """A project name to put on 'Milestones in …' / `linear milestones list`.

    Prefer the resolved catalog's project name; fall back to the caller's
    --project string when the catalog is empty.
    """
    if not project_name:
        return None
    if catalog:
        n = (catalog[0].get("project") or {}).get("name")
        if n:
            return n
    return project_name


def _milestone_target_suffix(m: dict) -> str:
    tgt = (m.get("targetDate") or "")[:10]
    return f"     target {tgt}" if tgt else ""


def _grouped_catalog(catalog: list[dict]) -> list[tuple[str, list[dict]]]:
    grouped: dict[str, list[dict]] = {}
    for m in catalog:
        pname = (m.get("project") or {}).get("name") or "unknown"
        grouped.setdefault(pname, []).append(m)
    out = []
    for pname in sorted(grouped, key=str.lower):
        ms = sorted(grouped[pname],
                    key=lambda x: (x.get("sortOrder") or 0, x.get("name") or ""))
        out.append((pname, ms))
    return out


def missing_milestone_error(title: str, catalog: list[dict],
                            project_name: str | None = None,
                            multiline: bool = True,
                            backlog_milestone: str | None = None) -> str:
    """The message shown when a create would leave an issue with no milestone.

    Multi-line for interactive `linear create` (lists real milestones so an
    agent can pick one); single-line for `--from-file`, whose per-row output
    is a tab-separated record that a newline would corrupt. Catalog names
    are repr-escaped in the bulk form so a tab/newline cannot split the
    record — same as the owner guard's `!r` on --assign.
    """
    heading = _catalog_heading(catalog, project_name)
    if not multiline:
        msg = (f"issue without a milestone refused. Creating a task with no milestone "
               f"is bad practice — it lands in the project's \"No milestone\" bucket. "
               f"Set \"milestone\" on this row, or pass --skip-milestone. "
               f"{horizons_help(False, backlog_milestone)}")
        if catalog:
            names = [repr(m.get("name") or "") for m in catalog[:_MILESTONE_CATALOG_CAP]]
            msg += " Available: " + ", ".join(names) + "."
            if len(catalog) > _MILESTONE_CATALOG_CAP:
                if heading:
                    msg += f" Full list: linear milestones list {heading!r}."
                else:
                    msg += " Full list: linear projects."
        else:
            msg += " No milestones found — create one with `linear milestones create`."
        return msg

    lines = [
        f"Refusing to create an issue without a milestone: {title!r}",
        "  Creating a task with no milestone is bad practice — it lands in the project's",
        '  "No milestone" bucket and nobody can tell which deliverable it belongs to.',
        '  Attach it:  --milestone "<name>"   (resolved within the project)',
        "",
        horizons_help(True, backlog_milestone),
        "",
    ]
    if not catalog:
        lines.append("  No milestones found. Create one with `linear milestones create`.")
    else:
        if heading:
            groups = [(heading, sorted(
                catalog,
                key=lambda x: (x.get("sortOrder") or 0, x.get("name") or ""),
            ))]
        else:
            groups = _grouped_catalog(catalog)
        shown = 0
        truncated = False
        for pname, ms in groups:
            if shown >= _MILESTONE_CATALOG_CAP:
                truncated = True
                break
            lines.append(f"  Milestones in {pname}:")
            for m in ms:
                if shown >= _MILESTONE_CATALOG_CAP:
                    truncated = True
                    break
                lines.append(f"    {(m.get('name') or '')}{_milestone_target_suffix(m)}")
                shown += 1
            if truncated:
                break
        if truncated:
            lines.append("    …")
        lines.append("")
        if heading:
            # !r so a space in the project name stays one argv token
            # (`linear milestones list Rush App` would parse as project=Rush).
            lines.append(f"  Full list:  linear milestones list {heading!r}")
        else:
            lines.append("  Full list:  linear projects")
    lines.append("  If it genuinely has no milestone, say so explicitly: --skip-milestone")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Horizon policy: every created issue gets a due date, cycle and milestone by rule
# ---------------------------------------------------------------------------

DEFAULT_POLICY = {
    "sla": {
        "urgent": {"due": "today", "remind": "15m"},
        "high": {"due": "today", "remind": "1h"},
        "medium": {"due": "+1wd", "remind": "none"},
        "low": {"due": "cycle_end", "remind": "none"},
    },
    "backlogMilestone": "Q1 2027",
}

# "none" priority has no SLA row of its own; it is scheduled like medium.
_SLA_KEY = {1: "urgent", 2: "high", 3: "medium", 4: "low", 0: "medium"}


def load_policy(cfg: dict) -> dict:
    """`policy` from config over DEFAULT_POLICY, merged per priority and per key
    so a config that only changes `sla.low.due` keeps every other default."""
    user = cfg.get("policy") or {}
    if not isinstance(user, dict) or not isinstance(user.get("sla") or {}, dict):
        raise ValueError("policy and policy.sla must be JSON objects")
    sla = {}
    for key, rule in DEFAULT_POLICY["sla"].items():
        over = user.get("sla", {}).get(key) or {}
        if not isinstance(over, dict):
            raise ValueError(f'policy.sla.{key} must be an object like {{"due": "today", "remind": "15m"}}')
        sla[key] = {**rule, **over}
    return {
        "sla": sla,
        "backlogMilestone": user.get("backlogMilestone") or DEFAULT_POLICY["backlogMilestone"],
    }


def add_working_days(start: date, n: int) -> date:
    d = start
    while n > 0:
        d += timedelta(days=1)
        if d.weekday() < 5:
            n -= 1
    return d


def sla_due(rule: str, today: date, cycle_last: date | None) -> date | None:
    """A `policy.sla.<priority>.due` rule as a date: today | +Nwd | +Nd |
    cycle_end. None when the rule is cycle_end and there is no active cycle."""
    r = str(rule).strip().lower()
    if r == "today":
        return today
    if r == "cycle_end":
        return cycle_last
    m = re.fullmatch(r"\+(\d+)(wd|d)", r)
    if m:
        n = int(m.group(1))
        return add_working_days(today, n) if m.group(2) == "wd" else today + timedelta(days=n)
    raise ValueError(f"policy.sla due rule {rule!r} is not one of today, +Nwd, +Nd, cycle_end")


def sla_remind(rule: str, now: datetime) -> datetime | None:
    """A `policy.sla.<priority>.remind` rule (15m | 1h | none) as a time."""
    r = str(rule).strip().lower()
    if r in ("none", ""):
        return None
    m = re.fullmatch(r"(\d+)([mh])", r)
    if not m:
        raise ValueError(f"policy.sla remind rule {rule!r} is not Nm, Nh or none")
    n = int(m.group(1))
    return now + (timedelta(minutes=n) if m.group(2) == "m" else timedelta(hours=n))


def resolve_tz(name: str | None) -> tzinfo:
    """The team's timezone, so "today" and a cycle's last day are the team's,
    not the host's (fleet boxes run in UTC). Falls back to the host zone."""
    if name:
        try:
            from zoneinfo import ZoneInfo
            return ZoneInfo(name)
        except Exception:
            print(f"Note: timezone {name!r} is unavailable here (no tzdata?); "
                  f"dates use this machine's zone.", file=sys.stderr)
    return datetime.now().astimezone().tzinfo


def team_calendar(api_key: str, team_id: str) -> dict:
    """{"tz": tzinfo, "cycle": active cycle node or None} in one request."""
    data = gql(api_key, """query($id: String!) {
        team(id: $id) { timezone activeCycle { id number name startsAt endsAt } }
    }""", {"id": team_id})
    team = {} if check_errors(data) else ((data.get("data") or {}).get("team") or {})
    return {"tz": resolve_tz(team.get("timezone")), "cycle": team.get("activeCycle")}


def cycle_days(cycle: dict | None, tz: tzinfo) -> tuple[date, date] | None:
    """A cycle's first and last day in `tz`. Linear's endsAt is the exclusive
    instant the next cycle starts, so the last day is the one just before it."""
    if not cycle or not cycle.get("startsAt") or not cycle.get("endsAt"):
        return None
    start = datetime.fromisoformat(cycle["startsAt"].replace("Z", "+00:00"))
    end = datetime.fromisoformat(cycle["endsAt"].replace("Z", "+00:00"))
    return (start.astimezone(tz).date(),
            (end - timedelta(microseconds=1)).astimezone(tz).date())


def next_open_milestone(milestones: list[dict], exclude: str | None = None) -> dict | None:
    """The project's next deliverable: earliest targetDate still under 100%,
    never the parking milestone (policy.backlogMilestone)."""
    skip = (exclude or "").lower()
    open_ms = [m for m in milestones
               if m.get("targetDate") and (m.get("progress") or 0) < 100
               and (m.get("name") or "").lower() != skip]
    return min(open_ms, key=lambda m: m["targetDate"]) if open_ms else None


def plan_horizon(horizon: str, priority: int, policy: dict, today: date,
                 now: datetime, cycle: tuple[date, date] | None,
                 milestone: dict | None, due: str | None) -> dict:
    """Due date, reminder and cycle for one horizon. Pure: every input is
    already resolved, so the whole rule table is testable offline.

    Returns {"due": "YYYY-MM-DD"|None, "remind": datetime|None,
    "cycle": "active"|"none", "why": short reason for the due date}.
    An explicit `due` always wins and never carries an SLA reminder.
    """
    rule = policy["sla"][_SLA_KEY.get(priority, "medium")]
    cycle_last = cycle[1] if cycle else None
    out = {"due": due, "remind": None, "cycle": "none", "why": "explicit"}
    if horizon == "backlog":
        return out

    sla_date = sla_due(rule["due"], today, cycle_last)
    sla_at = sla_remind(rule["remind"], now)
    if horizon == "now":
        if due is None:
            out.update(due=sla_date.isoformat() if sla_date else None,
                       remind=sla_at, why=_SLA_KEY.get(priority, "medium"))
        out["cycle"] = "active"
        return out

    # milestone horizon
    if due is None:
        target = (milestone or {}).get("targetDate")
        target = date.fromisoformat(target[:10]) if target else None
        urgent = priority in (1, 2)
        if target and not (urgent and sla_date and sla_date < target):
            out.update(due=target.isoformat(), why="milestone target")
        else:
            out.update(due=sla_date.isoformat() if sla_date else None,
                       remind=sla_at if (urgent or not target) else None,
                       why=_SLA_KEY.get(priority, "medium"))
    if cycle and out["due"] and cycle[0] <= date.fromisoformat(out["due"]) <= cycle[1]:
        out["cycle"] = "active"
    return out


def horizons_help(multiline: bool = True, backlog_milestone: str | None = None) -> str:
    """The three horizons, as taught by every create refusal."""
    bm = backlog_milestone or DEFAULT_POLICY["backlogMilestone"]
    lines = [
        "now (default): due by priority (urgent/high today, medium next working day, "
        "low at cycle end), active cycle, the project's next open milestone",
        "--milestone M: due on M's target date (urgent/high keep an earlier SLA date), "
        "active cycle only when that date falls inside it",
        f"--backlog: Backlog state, no cycle, no due date, a future milestone "
        f"(default {bm!r})",
    ]
    if not multiline:
        return "Horizons: " + "; ".join(lines) + "."
    return "  Every issue lands in one horizon:\n" + "\n".join(f"    {l}" for l in lines)


def missing_project_error(title: str, projects: list[dict], multiline: bool = True,
                          backlog_milestone: str | None = None) -> str:
    names = [p.get("name") or "" for p in projects]
    if not multiline:
        shown = ", ".join(repr(n) for n in names[:_MILESTONE_CATALOG_CAP])
        return (f"issue without a project refused. Set \"project\" on this row, or run "
                f"from a directory `agents projects` binds to one. Projects: {shown}. "
                f"{horizons_help(False, backlog_milestone)}")
    lines = [
        f"Refusing to create an issue without a project: {title!r}",
        "  Pass --project <name>, or run from a repo `agents projects` binds to one.",
        "",
        horizons_help(True, backlog_milestone),
        "",
        "  Projects:",
    ]
    lines += [f"    {n}" for n in names[:_MILESTONE_CATALOG_CAP]]
    if len(names) > _MILESTONE_CATALOG_CAP:
        lines.append("    …  (full list: linear projects)")
    return "\n".join(lines)


# Project states that take no new issues (see project_state).
CLOSED_PROJECT_STATES = frozenset({"paused", "completed", "canceled"})


def closed_project_error(title: str, project: dict, multiline: bool = True) -> str:
    name, state = project.get("name") or project.get("id"), project.get("state")
    if not multiline:
        return (f"project {name!r} is {state} and takes no new issues. Comment on or "
                f"update an existing issue, or set \"force\" on this row to file anyway.")
    return "\n".join([
        f"Refusing to create {title!r}: project {name!r} is {state} in Linear.",
        "  A paused, completed or canceled project takes no new issues. Comment on or",
        "  update an existing issue instead; --force files it anyway.",
    ])


def _build_create_input(api_key: str, team_id: str, cfg: dict, fields: dict,
                        verbose: bool = True, ctx: dict | None = None
                        ) -> tuple[dict | None, str | None]:
    """Build an IssueCreateInput from a flat fields dict (keys match flag names:
    title, description, description_file, priority, parent, project,
    milestone, skip_milestone, status, cycle, assign, delegate, due_date, label,
    force, backlog, bug, blocker).

    Returns (input_obj, None) on success or (None, error) on hard failure.
    Warnings go to stderr when verbose=True. `ctx` memoizes the cwd project,
    team projects and team calendar across a bulk run, and receives two outputs per call:
    `applied` (the defaults line to print) and `remindAt` (the SLA reminder to
    set once the issue exists, or None).
    """
    def warn(msg):
        if verbose:
            print(f"Warning: {msg}", file=sys.stderr)

    description = read_description(fields.get("description"), fields.get("description_file"))

    title = fields.get("title")
    if not title:
        if not description:
            return None, "no title or description"
        title = derive_title(description)

    for image_path in fields.get("image") or []:
        asset_url = upload_file(api_key, image_path)
        if asset_url:
            description = (description or "") + f"\n\n![{Path(image_path).name}]({asset_url})"

    input_obj: dict = {"teamId": team_id, "title": title}
    if description:
        input_obj["description"] = description

    ctx = {} if ctx is None else ctx
    ctx["remindAt"] = None
    ctx["applied"] = ""

    due_explicit = None
    if fields.get("due_date"):
        try:
            due_explicit = parse_due_date(fields["due_date"])
        except SystemExit as e:
            return None, str(e)

    bug, blocker = bool(fields.get("bug")), bool(fields.get("blocker"))
    if blocker and not bug:
        return None, "--blocker qualifies a bug: pass --bug --blocker"
    if blocker and fields.get("backlog"):
        return None, "--bug --blocker is horizon now; it cannot be combined with --backlog"
    try:
        if fields.get("priority") is not None:
            priority = parse_priority(fields["priority"])
        else:
            priority = (2 if blocker else 4) if bug else 3
    except SystemExit as e:
        return None, str(e)
    input_obj["priority"] = priority

    if fields.get("backlog") or (bug and not blocker):
        horizon = "backlog"
    else:
        horizon = "milestone" if fields.get("milestone") else "now"
    applied = [f"horizon {horizon}"]
    try:
        policy = load_policy(cfg)
    except ValueError as e:
        return None, f"{e} (in {CONFIG_PATH})"
    if bug and fields.get("priority") is None:
        applied.append(f"bug → priority {PRIORITY_MAP[priority].lower()}")

    if fields.get("parent"):
        parent_issue = resolve_issue(api_key, team_id, fields["parent"])
        if parent_issue:
            input_obj["parentId"] = parent_issue["id"]
            if verbose:
                _warn_sub_issue(title, parent_issue.get("identifier") or fields["parent"])
        else:
            warn(f"parent '{fields['parent']}' not found; creating without parent.")

    # A named --project/--milestone that doesn't resolve is a hard error: creating
    # the issue without it would silently ignore what the user asked for (RUSH-1496).
    # With no --project the cwd's `agents projects` binding decides, as it does
    # for `linear tasks`; with neither, the issue would be nobody's deliverable.
    project_name = fields.get("project")
    if project_name:
        try:
            project_id = resolve_project_id(api_key, team_id, project_name, strict=True)
        except LookupError as e:
            return None, str(e)
    else:
        if "cwd_project" not in ctx:
            ctx["cwd_project"] = resolve_cwd_project()
        cwd_id, cwd_name = ctx["cwd_project"]
        project_id = (resolve_auto_project_id(api_key, team_id, cwd_id or "")
                      or resolve_auto_project_id(api_key, team_id, cwd_name or ""))
        if not project_id:
            return None, missing_project_error(title, list_team_projects(api_key, team_id),
                                               multiline=verbose,
                                               backlog_milestone=policy["backlogMilestone"])
        project_name = cwd_name or project_id
        applied.append(f"project: {project_name} (cwd)")
    input_obj["projectId"] = project_id

    # Pausing, completing or canceling a project in Linear is how the owner
    # closes it to new work; agents that land in its directory must not keep
    # filing into it. --force is the deliberate escape hatch.
    if "projects" not in ctx:
        ctx["projects"] = list_team_projects(api_key, team_id)
    project = next((p for p in ctx["projects"] if p.get("id") == project_id), None)
    if (project and project.get("state") in CLOSED_PROJECT_STATES
            and not fields.get("force")):
        return None, closed_project_error(title, project, multiline=verbose)

    skip_milestone = bool(fields.get("skip_milestone"))
    milestone = fields.get("milestone")
    if milestone and skip_milestone:
        return None, conflicting_milestone_flags_error(milestone, multiline=verbose)

    if "calendar" not in ctx:
        ctx["calendar"] = team_calendar(api_key, team_id)
    cal = ctx["calendar"]
    now = datetime.now(cal["tz"])
    today = now.date()

    catalog = list_milestone_catalog(api_key, team_id, project_id)
    project_name = _catalog_heading(catalog, project_name) or project_name
    chosen = None
    if milestone:
        if len(milestone) == 36 and milestone.count("-") == 4:
            chosen = next((m for m in catalog if m.get("id") == milestone), None)
            if not chosen:
                return None, f"milestone '{milestone}' is not in project {project_name!r}."
        else:
            try:
                chosen = _select_named_node(catalog, milestone, "milestone",
                                            strict=True, suggest=True)
            except LookupError as e:
                return None, str(e)
    elif skip_milestone:
        warn("--skip-milestone: this issue lands in the project's \"No milestone\" bucket.")
    elif horizon == "backlog":
        want = policy["backlogMilestone"]
        chosen = next((m for m in catalog
                       if (m.get("name") or "").lower() == want.lower()), None)
        if not chosen:
            return None, (f"--backlog needs a future milestone, and the backlog milestone "
                          f"{want!r} (policy.backlogMilestone) is not in project "
                          f"{project_name!r}. Pass --milestone <a future milestone>.")
        applied.append(f"milestone: {chosen['name']} (backlog)")
    else:
        chosen = next_open_milestone(catalog, policy["backlogMilestone"])
        if not chosen:
            return None, missing_milestone_error(title, catalog, project_name=project_name,
                                                 multiline=verbose,
                                                 backlog_milestone=policy["backlogMilestone"])
        applied.append(f"milestone: {chosen['name']} (next for {project_name})")
    if horizon == "backlog" and chosen:
        target = (chosen.get("targetDate") or "")[:10]
        if not target or target <= today.isoformat():
            return None, (f"--backlog needs a future milestone; {chosen.get('name')!r} "
                          f"targets {target or 'no date'}. Pass --milestone <a future milestone>.")
    if chosen:
        input_obj["projectMilestoneId"] = chosen["id"]

    try:
        plan = plan_horizon(horizon, priority, policy, today, now,
                            cycle_days(cal["cycle"], cal["tz"]), chosen, due_explicit)
    except ValueError as e:
        return None, f"{e} (set in {CONFIG_PATH})"
    if plan["due"]:
        input_obj["dueDate"] = plan["due"]
    if due_explicit is None:
        due_note = (f"{plan['why']} → due {plan['due']}" if plan["due"] else "no due date")
        if plan["remind"]:
            due_note += f", reminder {plan['remind'].astimezone(cal['tz']):%H:%M}"
            ctx["remindAt"] = (plan["remind"].astimezone(timezone.utc)
                               .isoformat(timespec="seconds").replace("+00:00", "Z"))
        applied.insert(1, due_note)

    desired_status = fields.get("status") or ("Backlog" if horizon == "backlog" else "Todo")
    states = get_states(api_key, team_id, cfg)
    sid = resolve_state_id(states, desired_status)
    if sid:
        input_obj["stateId"] = sid
    else:
        warn(f"state '{desired_status}' not found; letting Linear pick the default.")

    cycle_choice = (fields.get("cycle") or plan["cycle"]).lower()
    if cycle_choice == "active":
        cid = (cal["cycle"] or {}).get("id")
        if cid:
            input_obj["cycleId"] = cid
        else:
            warn("no active cycle found; creating without cycle.")
    elif cycle_choice == "next":
        cid = get_cycle_id(api_key, team_id, "next")
        if cid:
            input_obj["cycleId"] = cid
        else:
            warn("no next cycle found; creating without cycle.")
    if not fields.get("cycle"):
        applied.append("cycle: " + (cycle_label(cal["cycle"]) if "cycleId" in input_obj else "none"))

    assign_arg = fields.get("assign")
    unassigned_reason = ""
    if assign_arg is None:
        vid = get_viewer_id(api_key, cfg)
        if vid:
            input_obj["assigneeId"] = vid
        else:
            unassigned_reason = ("the API key owner could not be resolved, so there is "
                                 "no default assignee")
    elif assign_arg.lower() == "none":
        unassigned_reason = "you passed --assign none"
    else:
        uid = resolve_assignee_id(api_key, assign_arg)
        if uid:
            input_obj["assigneeId"] = uid
        else:
            # !r, not '{}': this value is echoed back into the bulk --from-file
            # error, which is one field of a tab-separated record. repr escapes
            # any tab/newline in the value so it can't split the record.
            unassigned_reason = (f"--assign {assign_arg!r} matched no human "
                                 f"(try an email, a name, or `linear users`)")

    # An issue with no assignee is nobody's job — it sits on the board unowned
    # and rots. Refuse by default; --force is the deliberate escape hatch.
    if "assigneeId" not in input_obj and not fields.get("force"):
        return None, unassigned_error(title, unassigned_reason, multiline=verbose)

    # The configured agent identity is who files the work, so it is the
    # delegate; the API-key owner (a human) stays the assignee.
    delegate_arg = fields.get("delegate")
    if delegate_arg is None and cfg.get("agent"):
        delegate_arg = cfg["agent"]
        applied.append(f"delegate: {delegate_arg}")
    if delegate_arg and delegate_arg.lower() != "none":
        did = resolve_agent_id(api_key, cfg, delegate_arg)
        if did:
            input_obj["delegateId"] = did
        else:
            names = ", ".join(a["name"] for a in get_agents(api_key, cfg)) or "none installed"
            warn(f"agent '{delegate_arg}' not found; creating without delegate. Available: {names}")

    label_names = fields.get("label") or []
    if isinstance(label_names, str):
        label_names = [label_names]
    label_names = list(label_names)
    if bug and not any(l.lower() == "bug" for l in label_names):
        label_names.append("Bug")
    if label_names:
        label_map = {l["name"]: l["id"] for l in list_team_labels(api_key, team_id)}
        label_ids = []
        for lbl in label_names:
            if lbl in label_map:
                label_ids.append(label_map[lbl])
            else:
                warn(_unknown_label_message(lbl, list(label_map)))
        if label_ids:
            input_obj["labelIds"] = label_ids

    ctx["applied"] = " · ".join(applied)
    return input_obj, None


def _send_issue_create(api_key: str, input_obj: dict) -> tuple[dict | None, str | None]:
    data = gql(api_key, """
        mutation($input: IssueCreateInput!) {
            issueCreate(input: $input) {
                success
                issue { id identifier title state { name } cycle { number name } assignee { name } delegate { name } }
            }
        }
    """, {"input": input_obj})
    if data.get("errors"):
        return None, data["errors"][0].get("message", "unknown error")
    result = data["data"]["issueCreate"]
    if not result["success"]:
        return None, "create failed"
    return result["issue"], None


def cmd_create(args, cfg, api_key, team_id):
    if args.from_file and getattr(args, "remind_at", None):
        print("--remind-at sets one issue's reminder; it cannot be combined with --from-file.", file=sys.stderr)
        sys.exit(2)
    if args.from_file:
        return _bulk_create(args.from_file, cfg, api_key, team_id,
                            force=args.force, skip_milestone=args.skip_milestone)

    fields = {
        "title": args.title,
        "description": args.description,
        "description_file": args.description_file,
        "priority": args.priority,
        "parent": args.parent,
        "project": args.project,
        "milestone": args.milestone,
        "skip_milestone": args.skip_milestone,
        "status": args.status,
        "cycle": args.cycle,
        "assign": args.assign,
        "delegate": args.delegate,
        "due_date": args.due_date,
        "label": args.label or [],
        "image": args.image or [],
        "force": args.force,
        "backlog": args.backlog,
        "bug": args.bug,
        "blocker": args.blocker,
    }
    remind_at = parse_remind_at(args.remind_at) if getattr(args, "remind_at", None) else None
    ctx: dict = {}
    input_obj, err = _build_create_input(api_key, team_id, cfg, fields, verbose=True, ctx=ctx)
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)
    explicit_remind = bool(remind_at)
    remind_at = remind_at or ctx["remindAt"]
    print(ctx["applied"], file=sys.stderr)
    advise_similar(api_key, team_id, cfg, input_obj.get("title") or "",
                   input_obj.get("description"))
    issue, err = _send_issue_create(api_key, input_obj)
    if err:
        print(f"Error: {err}", file=sys.stderr)
        return
    cyc = cycle_label(issue.get("cycle"))
    asg = (issue.get("assignee") or {}).get("name") or "unassigned"
    dg = (issue.get("delegate") or {}).get("name")
    meta = f"{cyc} | {asg}" + (f" | delegate: {dg}" if dg else "")
    print(f"Created {issue['identifier']}: {issue['title']}  [{meta}]")
    if remind_at:
        if set_issue_reminder(api_key, issue["id"], remind_at):
            print(f"{issue['identifier']} -> reminder: {remind_at}")
        else:
            print(f"Created {issue['identifier']}, but its reminder could not be set. "
                  f"Retry with: linear update {issue['identifier']} --remind-at {remind_at}", file=sys.stderr)
            # Only an asked-for reminder fails the command: the issue exists,
            # and a non-zero exit invites a retry that files a duplicate.
            if explicit_remind:
                sys.exit(1)


def _bulk_create(path: str, cfg, api_key, team_id, force: bool = False,
                 skip_milestone: bool = False):
    """Bulk create from JSONL. Each line is a JSON object with field names
    matching the single-create flags. Continue-on-error; tab-separated output:
    OK<TAB>IDENT<TAB>TITLE  or  ERROR<TAB>-<TAB>TITLE<TAB>REASON.

    `force` applies the --force owner-check bypass to every row; a row can also
    carry its own "force": true. `skip_milestone` is the same for the
    milestone check; a row can carry "skip_milestone": true."""
    if path == "-":
        stream = sys.stdin
    else:
        if not Path(path).exists():
            print(f"--from-file: {path} not found", file=sys.stderr)
            sys.exit(1)
        stream = open(path)

    ok_count = 0
    err_count = 0
    ctx: dict = {}
    for raw in stream:
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        try:
            fields = json.loads(line)
        except json.JSONDecodeError as e:
            print(f"ERROR\t-\t<unparseable>\t{e}")
            err_count += 1
            continue

        if force:
            fields["force"] = True
        if skip_milestone:
            fields["skip_milestone"] = True
        title_hint = tsv_field(
            fields.get("title")
            or (fields.get("description", "")[:60] if fields.get("description") else "<no title>")
        )
        input_obj, err = _build_create_input(api_key, team_id, cfg, fields, verbose=False, ctx=ctx)
        if err:
            print(f"ERROR\t-\t{title_hint}\t{err}")
            err_count += 1
            continue
        issue, err = _send_issue_create(api_key, input_obj)
        if err:
            print(f"ERROR\t-\t{title_hint}\t{err}")
            err_count += 1
            continue
        print(f"OK\t{issue['identifier']}\t{tsv_field(issue['title'])}")
        print(f"{issue['identifier']}: {ctx['applied']}", file=sys.stderr)
        if ctx["remindAt"] and not set_issue_reminder(api_key, issue["id"], ctx["remindAt"]):
            print(f"{issue['identifier']}: reminder could not be set", file=sys.stderr)
        ok_count += 1

    if path != "-":
        stream.close()
    print(f"\n{ok_count} created, {err_count} failed.", file=sys.stderr)


# ---------------------------------------------------------------------------
# cycles (list)
# ---------------------------------------------------------------------------

def cmd_cycles(args, cfg, api_key, team_id):
    """List cycles (default) or create / update one."""
    action = getattr(args, "cycles_action", None)
    if action == "create":
        return _cycle_create(args, api_key, team_id)
    if action == "update":
        return _cycle_update(args, api_key, team_id)

    cycles = list_team_cycles(api_key, team_id)
    if not cycles:
        print("No cycles found.")
        return

    if args.json:
        print(json.dumps(cycles, indent=2))
        return

    # Sort by startsAt descending (most recent first)
    cycles.sort(key=lambda c: c.get("startsAt") or "", reverse=True)

    # Find the currently-active cycle id so we can mark it in output.
    active_id = get_cycle_id(api_key, team_id, "active")
    next_id = get_cycle_id(api_key, team_id, "next")

    show_ids = getattr(args, "ids", False)

    # Format: status marker | name (number) | startsAt -> endsAt | issue count
    for c in cycles:
        cid = c.get("id")
        name = cycle_label(c)
        starts = (c.get("startsAt") or "")[:10]
        ends = (c.get("endsAt") or "")[:10]
        history = c.get("issueCountHistory") or []
        count = history[-1] if history else 0

        marker = "  "
        if cid == active_id:
            marker = "* "
        elif cid == next_id:
            marker = "> "
        elif c.get("completedAt"):
            marker = "- "

        id_str = f"  {cid}" if show_ids else ""
        print(f"{marker}{name:40s}  {starts} → {ends}  {count} issue(s){id_str}")

    print("\n(* = active, > = next, - = completed)")


def _cycle_create(args, api_key, team_id):
    """Create a cycle. Linear assigns the cycle number automatically."""
    input_obj = {
        "teamId": team_id,
        "name": args.name,
        "startsAt": cycle_date_to_iso(args.starts, "--starts"),
        "endsAt": cycle_date_to_iso(args.ends, "--ends"),
    }
    data = gql(api_key, """
        mutation($input: CycleCreateInput!) {
            cycleCreate(input: $input) {
                success
                cycle { id number name startsAt endsAt }
            }
        }
    """, {"input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("cycleCreate") or {}
    if not result.get("success"):
        print("Cycle create failed.", file=sys.stderr)
        sys.exit(1)
    c = result["cycle"]
    print(f"Created cycle {cycle_label(c)}  "
          f"{(c.get('startsAt') or '')[:10]} → {(c.get('endsAt') or '')[:10]}  ({c['id']})")


def _cycle_update(args, api_key, team_id):
    node = resolve_cycle_by_value(api_key, team_id, args.cycle)
    if not node:
        sys.exit(1)
    input_obj = {}
    if args.name is not None:
        input_obj["name"] = args.name
    if args.starts is not None:
        input_obj["startsAt"] = cycle_date_to_iso(args.starts, "--starts")
    if args.ends is not None:
        input_obj["endsAt"] = cycle_date_to_iso(args.ends, "--ends")
    if not input_obj:
        print("Nothing to update. Use --name, --starts, or --ends.", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!, $input: CycleUpdateInput!) {
            cycleUpdate(id: $id, input: $input) {
                success
                cycle { id number name startsAt endsAt }
            }
        }
    """, {"id": node["id"], "input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("cycleUpdate") or {}
    if not result.get("success"):
        print("Cycle update failed.", file=sys.stderr)
        sys.exit(1)
    c = result["cycle"]
    print(f"Updated cycle {cycle_label(c)}  "
          f"{(c.get('startsAt') or '')[:10]} → {(c.get('endsAt') or '')[:10]}  ({c['id']})")


# ---------------------------------------------------------------------------
# projects, labels, users (discovery)
# ---------------------------------------------------------------------------

def cmd_projects(args, cfg, api_key, team_id):
    """List projects (default), show one's detail + milestones, or create /
    update / archive / delete a project."""
    action = getattr(args, "projects_action", None)
    if action == "create":
        return _project_create(args, api_key, team_id)
    if action == "update":
        return _project_update(args, api_key, team_id)
    if action in ("archive", "delete"):
        return _project_delete(args, api_key, team_id)
    if action == "show":
        return _project_show(args, api_key, team_id)
    if action == "overview":
        return _project_overview(args, api_key, team_id)

    projects = list_team_projects(api_key, team_id)
    if not projects:
        print("No projects found.")
        return
    if args.json:
        print(json.dumps(projects, indent=2))
        return
    print(f"{len(projects)} project(s):")
    for p in projects:
        progress = int((p.get("progress") or 0) * 100)
        state = p.get("state") or "-"
        lead = (p.get("lead") or {}).get("name") or "-"
        count = int(p.get("scope") or 0)
        pri = PRIORITY_MAP.get(p.get("priority") or 0, "-")
        print(f"  {p['name']:40s}  {state:12s}  {pri:7s}  {progress:>3}%  "
              f"{count:>4} issue(s)  lead: {lead}")


def _project_show(args, api_key, team_id):
    """Detail view for one project. Accepts a UUID (queried directly) or a name
    with exact match preferred over substring, so 'Rush' stays addressable once
    'Rush App' / 'Rush CLI' also exist (RUSH-1497)."""
    value = args.name
    if len(value) == 36 and value.count("-") == 4:
        proj_id = value
    else:
        projects = list_team_projects(api_key, team_id)
        matches = [p for p in projects if p["name"].lower() == value.lower()]
        if not matches:
            matches = [p for p in projects if value.lower() in p["name"].lower()]
        if not matches:
            print(f"No project matching '{value}'.{_suggest(value, [p['name'] for p in projects])}")
            return
        if len(matches) > 1:
            print(f"Note: '{value}' matched {len(matches)} projects; "
                  f"showing '{matches[0]['name']}'. Use the exact name or id to disambiguate.",
                  file=sys.stderr)
        proj_id = matches[0]["id"]

    data = gql(api_key, """
        query($id: String!) {
            project(id: $id) {
                id name state status { name type } progress scope priority
                description
                startDate targetDate
                lead { name email }
                projectMilestones(first: 50) {
                    nodes { id name description targetDate sortOrder }
                }
            }
        }
    """, {"id": proj_id})
    if check_errors(data):
        return
    proj = (data.get("data") or {}).get("project") or {}
    if not proj:
        print(f"No project with id '{proj_id}'.")
        return
    proj["state"] = project_state(proj)
    if args.json:
        print(json.dumps(proj, indent=2))
        return
    lead = (proj.get("lead") or {}).get("name") or "-"
    count = int(proj.get("scope") or 0)
    print(f"{proj['name']}")
    print(f"  State:    {proj.get('state') or '-'}")
    print(f"  Priority: {PRIORITY_MAP.get(proj.get('priority') or 0, '-')}")
    print(f"  Progress: {int((proj.get('progress') or 0) * 100)}%")
    print(f"  Issues:   {count}")
    print(f"  Lead:     {lead}")
    print(f"  Start:    {proj.get('startDate') or '-'}")
    print(f"  Target:   {proj.get('targetDate') or '-'}")
    print(f"  ID:       {proj.get('id')}")
    desc = (proj.get("description") or "").strip()
    if desc:
        print(f"  Description:")
        for line in desc.splitlines() or [desc]:
            print(f"    {line}")
    milestones = (proj.get("projectMilestones") or {}).get("nodes", [])
    roll = milestone_rollup(api_key, proj_id)
    if milestones:
        milestones.sort(key=lambda m: m.get("sortOrder") or 0)
        print(f"\nMilestones ({len(milestones)}):")
        for m in milestones:
            tgt = m.get("targetDate") or "-"
            print(f"  - {m['name']:30s}  target: {tgt:12s}  {_fmt_rollup(roll.get(m['id']))}")
    none_entry = roll.get("_none")
    if none_entry and none_entry.get("total"):
        print(f"  - {'— No milestone':30s}  target: {'-':12s}  {_fmt_rollup(none_entry)}")


# ---------------------------------------------------------------------------
# projects overview — one call for a menu: projects → milestones → open issues
# ---------------------------------------------------------------------------

# Project priority as a word. Linear stores 0..4; a menu wants a label it can
# colour, not an int it has to map.
PROJECT_PRIORITY_WORD = {0: "none", 1: "urgent", 2: "high", 3: "medium", 4: "low"}

_PERSON_FIELDS = "id name displayName email avatarUrl app"

# The team's project list plus each project's lead and members, nested so the
# people cost no extra round trips. Linear caps a query at 10,000 complexity
# points; 100 projects x 50 members measured 14,401, so this walk pages 50
# projects at a time.
_OVERVIEW_PROJECT_PAGE = 50
_OVERVIEW_MEMBER_CAP = 50
_OVERVIEW_PROJECTS_QUERY = f"""query($teamId: String!, $cursor: String) {{
    team(id: $teamId) {{
        projects(first: {_OVERVIEW_PROJECT_PAGE}, after: $cursor) {{
            pageInfo {{ hasNextPage endCursor }}
            nodes {{
                id name state status {{ name type }} priority targetDate updatedAt
                lead {{ {_PERSON_FIELDS} }}
                members(first: {_OVERVIEW_MEMBER_CAP}) {{
                    pageInfo {{ hasNextPage }}
                    nodes {{ {_PERSON_FIELDS} }}
                }}
            }}
        }}
    }}
}}"""

_OVERVIEW_ISSUE_PAGE = 250
_OVERVIEW_ISSUE_QUERY = f"""query($ids: [ID!], $cursor: String) {{
    issues(first: {_OVERVIEW_ISSUE_PAGE}, after: $cursor,
           filter: {{ project: {{ id: {{ in: $ids }} }} }}) {{
        pageInfo {{ hasNextPage endCursor }}
        nodes {{
            identifier title url priority updatedAt createdAt dueDate startedAt
            state {{ name type }}
            assignee {{ {_PERSON_FIELDS} }}
            delegate {{ id name avatarUrl app }}
            cycle {{ number }}
            projectMilestone {{ id }}
            project {{ id }}
        }}
    }}
}}"""

_OVERVIEW_MILESTONE_QUERY = """query($ids: [ID!], $cursor: String) {
    projectMilestones(first: 100, after: $cursor,
                      filter: { project: { id: { in: $ids } } }) {
        pageInfo { hasNextPage endCursor }
        nodes { id name targetDate sortOrder project { id } }
    }
}"""

_CURRENT_CYCLE_QUERY = """query($id: String!) {
    team(id: $id) { activeCycle { id number name startsAt endsAt } }
}"""


def _identity(node: dict | None) -> dict | None:
    """A user identity object {id,name,avatarUrl,app} for a menu row, or None
    when the field is unset. `app` distinguishes an agent (delegate) from a human."""
    if not node:
        return None
    return {
        "id": node.get("id"),
        "name": node.get("name"),
        "avatarUrl": node.get("avatarUrl"),
        "app": bool(node.get("app")),
    }


def overview_members(project: dict, issues: list[dict]) -> list[dict]:
    """The people on a project, for a menu's avatar stack. Linear grants access
    by team, not by project, so "who can see it" is the whole team and says
    nothing. What a project does record is its lead and its explicit members;
    those come first (lead, then members by name). Most projects set neither,
    so when both are empty the list falls back to the humans assigned to the
    project's issues, any state, most issues first. `role` says which rule
    produced each entry. App users (agents) are left out: they show up as
    `delegate` on issue rows instead."""
    def person(node: dict, role: str) -> dict:
        return {"id": node.get("id"), "name": node.get("name"),
                "displayName": node.get("displayName"), "email": node.get("email"),
                "avatarUrl": node.get("avatarUrl"), "role": role}

    out: list[dict] = []
    seen: set = set()
    lead = project.get("lead")
    if lead and not lead.get("app"):
        out.append(person(lead, "lead"))
        seen.add(lead.get("id"))
    members = [m for m in ((project.get("members") or {}).get("nodes") or [])
               if not m.get("app") and m.get("id") not in seen]
    for m in sorted(members, key=lambda m: (m.get("name") or "").lower()):
        out.append(person(m, "member"))
        seen.add(m.get("id"))
    if out:
        return out

    counts: dict[str, int] = {}
    nodes: dict[str, dict] = {}
    for n in issues:
        a = n.get("assignee")
        if a and a.get("id") and not a.get("app"):
            counts[a["id"]] = counts.get(a["id"], 0) + 1
            nodes[a["id"]] = a
    ranked = sorted(nodes, key=lambda i: (-counts[i], (nodes[i].get("name") or "").lower()))
    return [person(nodes[i], "assignee") for i in ranked]


def overview_issue_row(n: dict) -> dict:
    """The fields a menu row needs, flattened: names instead of nested objects,
    and the cycle as its number so a caller can compare it against the top-level
    `cycle.number` to pick out "this cycle".

    `assignee` stays the human's NAME string (unchanged contract); `assigneeId`
    and `updatedAt` are added beside it, and `delegate` carries the full agent
    identity {id,name,avatarUrl,app} the menu draws an avatar from. `dueDate`
    (YYYY-MM-DD or null), `createdAt`, `startedAt` and `stateType` let a menu
    sort and group without a second read."""
    assignee = n.get("assignee") or {}
    return {
        "identifier": n.get("identifier"),
        "title": n.get("title"),
        "state": (n.get("state") or {}).get("name"),
        "cycle": (n.get("cycle") or {}).get("number"),
        "assignee": assignee.get("name"),
        "assigneeId": assignee.get("id"),
        "delegate": _identity(n.get("delegate")),
        "updatedAt": n.get("updatedAt"),
        "createdAt": n.get("createdAt"),
        "startedAt": n.get("startedAt"),
        "dueDate": n.get("dueDate"),
        "stateType": (n.get("state") or {}).get("type"),
        "priority": n.get("priority") or 0,
        "url": n.get("url"),
    }


def overview_issue_sort_key(row: dict) -> tuple:
    """Cycle number ascending (no cycle last), then priority (Urgent=1 first,
    no-priority=0 last), then identifier so ties are deterministic."""
    cyc = row.get("cycle")
    return (cyc is None, cyc or 0, row.get("priority") or 5, row.get("identifier") or "")


def overview_bucket(issues: list[dict]) -> dict:
    """Counts for one milestone (or a project's no-milestone bucket) plus its
    open rows. `open` is every state type that is not completed, canceled or
    duplicate (counted as canceled): backlog, unstarted, started and triage,
    so open + done + canceled == total always holds."""
    counts = {"total": 0, "open": 0, "done": 0, "canceled": 0}
    open_rows: list[dict] = []
    for n in issues:
        counts["total"] += 1
        state_type = (n.get("state") or {}).get("type")
        if state_type == "completed":
            counts["done"] += 1
        elif state_type in ("canceled", "duplicate"):
            counts["canceled"] += 1
        else:
            counts["open"] += 1
            open_rows.append(overview_issue_row(n))
    open_rows.sort(key=overview_issue_sort_key)
    return {"issues": counts, "open": open_rows}


def build_overview(cycle: dict | None, projects: list[dict], milestones: list[dict],
                   issues: list[dict], partial_reason: str | None = None,
                   generated_at: datetime | None = None) -> dict:
    """Assemble the overview document from already-fetched nodes. Pure — no API
    calls — so the shape and the ordering rules are testable without a
    workspace. Projects sort by priority (urgent first, none last) then name;
    milestones by sortOrder; open issues by cycle then priority."""
    ms_by_project: dict[str, list] = {}
    for m in milestones:
        ms_by_project.setdefault((m.get("project") or {}).get("id"), []).append(m)
    issues_by_ms: dict[str, list] = {}
    orphans_by_project: dict[str, list] = {}
    issues_by_project: dict[str, list] = {}
    for n in issues:
        issues_by_project.setdefault((n.get("project") or {}).get("id"), []).append(n)
        mid = (n.get("projectMilestone") or {}).get("id")
        if mid:
            issues_by_ms.setdefault(mid, []).append(n)
        else:
            orphans_by_project.setdefault((n.get("project") or {}).get("id"), []).append(n)

    out_projects = []
    for p in sorted(projects, key=lambda p: (p.get("priority") or 5,
                                             (p.get("name") or "").lower())):
        ms_nodes = sorted(ms_by_project.get(p["id"], []),
                          key=lambda m: m.get("sortOrder") or 0)
        out_projects.append({
            "id": p["id"],
            "name": p.get("name"),
            "priority": PROJECT_PRIORITY_WORD.get(p.get("priority") or 0, "none"),
            "state": project_state(p),
            "targetDate": p.get("targetDate"),
            "updatedAt": p.get("updatedAt"),
            "members": overview_members(p, issues_by_project.get(p["id"], [])),
            "milestones": [
                {"id": m["id"], "name": m.get("name"), "targetDate": m.get("targetDate"),
                 **overview_bucket(issues_by_ms.get(m["id"], []))}
                for m in ms_nodes
            ],
            "noMilestone": overview_bucket(orphans_by_project.get(p["id"], [])),
        })

    stamp = generated_at or datetime.now(timezone.utc)
    return {
        "generatedAt": stamp.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
        "cycle": ({"id": cycle.get("id"), "number": cycle.get("number"),
                   "name": cycle.get("name"), "startsAt": cycle.get("startsAt"),
                   "endsAt": cycle.get("endsAt")} if cycle else None),
        "projects": out_projects,
        "partial": partial_reason is not None,
        "partialReason": partial_reason,
    }


def _pick_overview_projects(projects: list[dict], wanted: list[str]) -> list[dict]:
    """Restrict the team's project list to the --project values, in the order
    given, deduplicated. A UUID must be on this team's list and a name must
    resolve strictly — a typo aborts rather than quietly returning less."""
    picked: list[dict] = []
    for value in wanted:
        if len(value) == 36 and value.count("-") == 4:
            match = next((p for p in projects if p["id"] == value), None)
            if not match:
                print(f"Error: no project with id '{value}' on this team.", file=sys.stderr)
                sys.exit(1)
        else:
            try:
                match = _select_named_node(projects, value, "project",
                                           strict=True, suggest=True)
            except LookupError as e:
                print(f"Error: {e}", file=sys.stderr)
                sys.exit(1)
        if match not in picked:
            picked.append(match)
    return picked


def _project_overview(args, api_key, team_id):
    """`projects overview`: the team's current cycle plus every project's
    milestones with issue counts and open rows, in one document. Four queries
    (projects with their people, active cycle, milestones, issues), each fully
    paginated; an API error on any of them exits non-zero rather than emitting
    a half document."""
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=2) as pool:
        project_read = pool.submit(
            paginate_connection, api_key, _OVERVIEW_PROJECTS_QUERY, ["team", "projects"],
            {"teamId": team_id}, page_size=_OVERVIEW_PROJECT_PAGE)
        cycle_read = pool.submit(gql, api_key, _CURRENT_CYCLE_QUERY, {"id": team_id})
        projects = project_read.result()
        data = cycle_read.result()
    if projects is None:
        sys.exit(1)
    wanted = getattr(args, "project", None) or []
    if wanted:
        projects = _pick_overview_projects(projects, wanted)
    partial_reasons: list[str] = [
        f"{p.get('name')} has more than {_OVERVIEW_MEMBER_CAP} members; members lists the first "
        f"{_OVERVIEW_MEMBER_CAP}"
        for p in projects
        if ((p.get("members") or {}).get("pageInfo") or {}).get("hasNextPage")
    ]

    if check_errors(data):
        sys.exit(1)
    cycle = ((data.get("data") or {}).get("team") or {}).get("activeCycle")

    ids = [p["id"] for p in projects]
    milestones: list[dict] = []
    issues: list[dict] = []
    if ids:
        milestone_cap: dict = {}
        issue_cap: dict = {}
        with ThreadPoolExecutor(max_workers=2) as pool:
            milestone_read = pool.submit(
                paginate_connection, api_key, _OVERVIEW_MILESTONE_QUERY,
                ["projectMilestones"], {"ids": ids}, partial_out=milestone_cap)
            issue_read = pool.submit(
                paginate_connection, api_key, _OVERVIEW_ISSUE_QUERY, ["issues"],
                {"ids": ids}, page_size=_OVERVIEW_ISSUE_PAGE, partial_out=issue_cap)
            milestones = milestone_read.result()
            issues = issue_read.result()
        if milestones is None:
            sys.exit(1)
        if milestone_cap:
            partial_reasons.append(
                f"milestones stopped at {milestone_cap['rows']} (pagination safety rail); "
                f"later milestones and their issues are missing")
        if issues is None:
            sys.exit(1)
        if issue_cap:
            partial_reasons.append(
                f"issues stopped at {issue_cap['rows']} (pagination safety rail); "
                f"counts and open lists are understated")

    doc = build_overview(cycle, projects, milestones, issues,
                         partial_reason="; ".join(partial_reasons) or None)
    if args.json:
        print(json.dumps(doc, indent=2))
        return
    print_overview(doc)


def print_overview(doc: dict) -> None:
    """Compact tree of the overview document: cycle header, then each project
    with its milestones (open/total) and open rows annotated with their cycle."""
    cyc = doc.get("cycle")
    if cyc:
        window = f"  {(cyc.get('startsAt') or '')[:10]} → {(cyc.get('endsAt') or '')[:10]}"
        print(f"Current cycle: {cycle_label(cyc)}{window}")
    else:
        print("Current cycle: none")
    if doc.get("partial"):
        print(f"Partial: {doc.get('partialReason')}")

    def rows(bucket: dict) -> None:
        for r in bucket["open"]:
            pri = PRIORITY_MAP.get(r["priority"] or 0, "-")
            who = r["assignee"] or "unassigned"
            cyc_str = f"  ·Cycle {r['cycle']}" if r["cycle"] is not None else ""
            print(f"      {r['identifier']:<9} {pri:<7} {r['state']:<12} {who:<16} "
                  f"{r['title']}{cyc_str}")

    for p in doc["projects"]:
        target = p["targetDate"] or "-"
        print(f"\n{p['name']}  [{p['priority']} · {p['state'] or '-'} · target {target}]")
        for m in p["milestones"]:
            c = m["issues"]
            print(f"    {m['name']}  ({c['open']} open / {c['total']} total)"
                  f"  target {m['targetDate'] or '-'}")
            rows(m)
        c = p["noMilestone"]["issues"]
        if c["total"]:
            print(f"    — No milestone  ({c['open']} open / {c['total']} total)")
            rows(p["noMilestone"])
    if not doc["projects"]:
        print("No projects.")


def _project_create(args, api_key, team_id):
    """Create a project on the current team. Lead is resolved by email; start /
    target are TimelessDates (YYYY-MM-DD)."""
    input_obj = {"teamIds": [team_id], "name": args.name}
    if args.description is not None:
        input_obj["description"] = args.description
    if args.lead is not None:
        lead_id = resolve_user_id_by_email(api_key, args.lead.strip())
        if not lead_id:
            print(f"Lead '{args.lead}' not found (pass an email).", file=sys.stderr)
            sys.exit(1)
        input_obj["leadId"] = lead_id
    start = parse_timeless_date(args.start, "--start")
    if start:
        input_obj["startDate"] = start
    target = parse_timeless_date(args.target, "--target")
    if target:
        input_obj["targetDate"] = target
    data = gql(api_key, """
        mutation($input: ProjectCreateInput!) {
            projectCreate(input: $input) {
                success
                project { id name url }
            }
        }
    """, {"input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("projectCreate") or {}
    if not result.get("success"):
        print("Project create failed.", file=sys.stderr)
        sys.exit(1)
    p = result["project"]
    print(f"Created project '{p['name']}'  ({p['id']})")
    if p.get("url"):
        print(f"  {p['url']}")


def _project_update(args, api_key, team_id):
    """Update a project via projectUpdate. Strict name resolution — a mistyped
    project aborts rather than no-opping. Supports description, name, lead,
    start/target TimelessDates, priority, and state (resolved to statusId)."""
    try:
        pid = resolve_project_id(api_key, team_id, args.project, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    input_obj: dict = {}
    if getattr(args, "name", None) is not None:
        input_obj["name"] = args.name

    description = read_description(
        getattr(args, "description", None),
        getattr(args, "description_file", None),
    )
    if description is not None:
        input_obj["description"] = description

    if getattr(args, "lead", None) is not None:
        lead_raw = (args.lead or "").strip()
        if lead_raw.lower() in ("", "none"):
            input_obj["leadId"] = None
        else:
            lead_id = resolve_user_id_by_email(api_key, lead_raw)
            if not lead_id:
                # Also try name/displayName via resolve_assignee_id for parity
                # with issue --assign, so --lead bisma works without an email.
                lead_id = resolve_assignee_id(api_key, lead_raw)
            if not lead_id:
                print(f"Lead '{args.lead}' not found (pass an email or name).",
                      file=sys.stderr)
                sys.exit(1)
            input_obj["leadId"] = lead_id

    if getattr(args, "start", None) is not None:
        # parse_timeless_date returns None for 'none'/'' — that clears the field.
        input_obj["startDate"] = parse_timeless_date(args.start, "--start")
    if getattr(args, "target", None) is not None:
        input_obj["targetDate"] = parse_timeless_date(args.target, "--target")

    if getattr(args, "priority", None) is not None:
        # Same names as issue --priority: urgent, high, medium, low, none.
        input_obj["priority"] = parse_priority(args.priority)

    if getattr(args, "state", None) is not None:
        try:
            input_obj["statusId"] = resolve_project_status_id(api_key, args.state)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)

    if not input_obj:
        print("Nothing to update. Use --description, --name, --lead, "
              "--start, --target, --priority, or --state.", file=sys.stderr)
        sys.exit(1)

    data = gql(api_key, """
        mutation($id: String!, $input: ProjectUpdateInput!) {
            projectUpdate(id: $id, input: $input) {
                success
                project {
                    id name state description priority
                    startDate targetDate
                    lead { name email }
                    status { id name type }
                }
            }
        }
    """, {"id": pid, "input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("projectUpdate") or {}
    if not result.get("success"):
        print("Project update failed.", file=sys.stderr)
        sys.exit(1)
    p = result["project"]
    lead = (p.get("lead") or {}).get("name") or "-"
    print(f"Updated project '{p['name']}'  ({p['id']})")
    print(f"  State:    {project_state(p) or '-'}")
    print(f"  Priority: {PRIORITY_MAP.get(p.get('priority') or 0, '-')}")
    print(f"  Lead:     {lead}")
    print(f"  Start:    {p.get('startDate') or '-'}")
    print(f"  Target:   {p.get('targetDate') or '-'}")
    desc = (p.get("description") or "").strip()
    if desc:
        # One-line summary so agents can confirm the write without re-querying.
        preview = desc.splitlines()[0]
        if len(preview) > 100:
            preview = preview[:97] + "..."
        print(f"  Description: {preview}")


def _project_delete(args, api_key, team_id):
    """Remove a project via projectDelete, which moves it to the workspace trash
    (recoverable in Linear's UI for ~30 days). Linear exposes no distinct
    'archive' mutation for projects, so `archive` and `delete` are the same
    operation — both trash the project."""
    try:
        pid = resolve_project_id(api_key, team_id, args.project, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!) { projectDelete(id: $id) { success } }
    """, {"id": pid})
    if check_errors(data) or not (data.get("data") or {}).get("projectDelete", {}).get("success"):
        print("Project remove failed.", file=sys.stderr)
        sys.exit(1)
    print(f"Removed project '{args.project}' (moved to trash; recoverable in Linear's UI).")


def cmd_milestones(args, cfg, api_key, team_id):
    """Manage project milestones: list, create, move, set-target-date, delete."""
    action = getattr(args, "milestones_action", None)
    if action == "create":
        return _milestone_create(args, api_key, team_id)
    if action == "move":
        return _milestone_move(args, api_key, team_id)
    if action == "set-target-date":
        return _milestone_set_target(args, api_key, team_id)
    if action == "delete":
        return _milestone_delete(args, api_key, team_id)
    if action == "list":
        return _milestone_list(args, api_key, team_id)
    print("Specify a subcommand: list <project> | create | move | "
          "set-target-date | delete.  See `linear milestones --help`.",
          file=sys.stderr)
    sys.exit(1)


def _milestone_list(args, api_key, team_id):
    """List a project's milestones (sorted by sortOrder)."""
    try:
        pid = resolve_project_id(api_key, team_id, args.project, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        query($id: String!) {
            project(id: $id) {
                name
                projectMilestones(first: 100) {
                    nodes { id name description targetDate sortOrder }
                }
            }
        }
    """, {"id": pid})
    if check_errors(data):
        return
    proj = (data.get("data") or {}).get("project") or {}
    nodes = (proj.get("projectMilestones") or {}).get("nodes", [])
    nodes.sort(key=lambda m: m.get("sortOrder") or 0)
    if args.json:
        print(json.dumps(nodes, indent=2))
        return
    if not nodes:
        print(f"No milestones in '{proj.get('name', args.project)}'.")
        return
    # Compute the rollup only for the human view (one extra issue query), so
    # --json neither pays for it nor leaks an undocumented key — same as the
    # sibling `projects show --json` path.
    roll = milestone_rollup(api_key, pid)
    print(f"{len(nodes)} milestone(s) in '{proj.get('name', args.project)}':")
    for m in nodes:
        tgt = m.get("targetDate") or "-"
        print(f"  {m['name']:30s}  target: {tgt:12s}  {_fmt_rollup(roll.get(m['id']))}")
    none_entry = roll.get("_none")
    if none_entry and none_entry.get("total"):
        print(f"  {'— No milestone':30s}  target: {'-':12s}  {_fmt_rollup(none_entry)}")


def _milestone_create(args, api_key, team_id):
    try:
        pid = resolve_project_id(api_key, team_id, args.project, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    input_obj = {"projectId": pid, "name": args.name}
    target = parse_timeless_date(args.target, "--target")
    if target:
        input_obj["targetDate"] = target
    if args.description is not None:
        input_obj["description"] = args.description
    data = gql(api_key, """
        mutation($input: ProjectMilestoneCreateInput!) {
            projectMilestoneCreate(input: $input) {
                success
                projectMilestone { id name targetDate }
            }
        }
    """, {"input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("projectMilestoneCreate") or {}
    if not result.get("success"):
        print("Milestone create failed.", file=sys.stderr)
        sys.exit(1)
    m = result["projectMilestone"]
    tgt = m.get("targetDate") or "no target"
    print(f"Created milestone '{m['name']}'  (target: {tgt})  ({m['id']})")


def _milestone_move(args, api_key, team_id):
    """Move a milestone to another project via projectMilestoneUpdate(projectId)."""
    src = None
    if getattr(args, "from_project", None):
        try:
            src = resolve_project_id(api_key, team_id, args.from_project, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    try:
        mid = resolve_milestone_id(api_key, team_id, args.milestone, src, strict=True)
        dest = resolve_project_id(api_key, team_id, args.to, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!, $input: ProjectMilestoneUpdateInput!) {
            projectMilestoneUpdate(id: $id, input: $input) {
                success
                projectMilestone { id name project { name } }
            }
        }
    """, {"id": mid, "input": {"projectId": dest}})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("projectMilestoneUpdate") or {}
    if not result.get("success"):
        print("Milestone move failed.", file=sys.stderr)
        sys.exit(1)
    m = result["projectMilestone"]
    proj = (m.get("project") or {}).get("name") or args.to
    print(f"Moved milestone '{m['name']}' -> project '{proj}'.")


def _milestone_set_target(args, api_key, team_id):
    """Set (or clear, with 'none') a milestone's target date — the reliable path
    that the web date-picker fought during the reorg (RUSH-1495)."""
    scope = None
    if getattr(args, "project", None):
        try:
            scope = resolve_project_id(api_key, team_id, args.project, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    try:
        mid = resolve_milestone_id(api_key, team_id, args.milestone, scope, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    target = parse_timeless_date(args.date, "target date")  # None when 'none'
    before = gql(api_key, "query($id: String!) { projectMilestone(id: $id) { targetDate } }",
                 {"id": mid})
    if check_errors(before):
        sys.exit(1)
    old_target = ((before.get("data") or {}).get("projectMilestone") or {}).get("targetDate")
    data = gql(api_key, """
        mutation($id: String!, $input: ProjectMilestoneUpdateInput!) {
            projectMilestoneUpdate(id: $id, input: $input) {
                success
                projectMilestone { id name targetDate }
            }
        }
    """, {"id": mid, "input": {"targetDate": target}})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("projectMilestoneUpdate") or {}
    if not result.get("success"):
        print("Set target date failed.", file=sys.stderr)
        sys.exit(1)
    m = result["projectMilestone"]
    print(f"Set '{m['name']}' target date -> {m.get('targetDate') or 'cleared'}.")
    if old_target and target and old_target != target:
        moved, failed = cascade_due_dates(api_key, mid, old_target, target)
        print(f"Moved {moved} open issue(s) due {old_target} -> {target}."
              + (f" {len(failed)} failed: {', '.join(failed)}" if failed else ""))
        if failed:
            sys.exit(1)


def cascade_due_dates(api_key: str, milestone_id: str, old: str, new: str
                      ) -> tuple[int, list[str]]:
    """Issues in the milestone that were due on its old target were dated by
    the milestone horizon, so a retarget carries them along. Open issues only:
    a done issue's due date is history. Returns (moved, failed identifiers)."""
    query = """query($mid: ID!, $due: TimelessDateOrDuration!, $cursor: String) {
        issues(first: 100, after: $cursor, filter: {
            projectMilestone: { id: { eq: $mid } }
            dueDate: { eq: $due }
            state: { type: { nin: ["completed", "canceled", "duplicate"] } }
        }) {
            pageInfo { hasNextPage endCursor }
            nodes { id identifier }
        }
    }"""
    nodes = paginate_connection(api_key, query, ["issues"], {"mid": milestone_id, "due": old}) or []
    moved, failed = 0, []
    for n in nodes:
        data = gql(api_key, """mutation($id: String!, $due: TimelessDate) {
            issueUpdate(id: $id, input: { dueDate: $due }) { success }
        }""", {"id": n["id"], "due": new})
        if check_errors(data) or not ((data.get("data") or {}).get("issueUpdate") or {}).get("success"):
            failed.append(n["identifier"])
        else:
            moved += 1
    return moved, failed


def _milestone_delete(args, api_key, team_id):
    scope = None
    if getattr(args, "project", None):
        try:
            scope = resolve_project_id(api_key, team_id, args.project, strict=True)
        except LookupError as e:
            print(f"Error: {e}", file=sys.stderr)
            sys.exit(1)
    try:
        mid = resolve_milestone_id(api_key, team_id, args.milestone, scope, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!) { projectMilestoneDelete(id: $id) { success } }
    """, {"id": mid})
    if check_errors(data):
        sys.exit(1)
    if (data.get("data") or {}).get("projectMilestoneDelete", {}).get("success"):
        print(f"Deleted milestone '{args.milestone}'.")
    else:
        print("Milestone delete failed.", file=sys.stderr)
        sys.exit(1)


def cmd_initiatives(args, cfg, api_key, team_id):
    """List initiatives (default), or show / create / update / link / unlink /
    archive one. Initiatives are workspace-scoped (not per-team)."""
    action = getattr(args, "initiatives_action", None)
    if action == "show":
        return _initiative_show(args, api_key, team_id)
    if action == "create":
        return _initiative_create(args, api_key, team_id)
    if action == "update":
        return _initiative_update(args, api_key, team_id)
    if action == "link":
        return _initiative_link(args, api_key, team_id)
    if action == "unlink":
        return _initiative_unlink(args, api_key, team_id)
    if action == "archive":
        return _initiative_archive(args, api_key, team_id)

    nodes = list_initiatives(api_key)
    if not nodes:
        print("No initiatives found.")
        return
    if args.json:
        print(json.dumps(nodes, indent=2))
        return
    print(f"{len(nodes)} initiative(s):")
    for n in nodes:
        status = n.get("status") or "-"
        target = n.get("targetDate") or "-"
        owner = (n.get("owner") or {}).get("name") or "-"
        print(f"  {n['name']:40s}  {status:12s}  target: {target:12s}  owner: {owner}")


def _initiative_show(args, api_key, team_id):
    """Detail for one initiative including linked projects."""
    try:
        iid = resolve_initiative_id(api_key, args.name, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        query($id: String!) {
            initiative(id: $id) {
                id name description status targetDate
                owner { name email }
                projects {
                    nodes { id name state status { name type } progress }
                }
            }
        }
    """, {"id": iid})
    if check_errors(data):
        sys.exit(1)
    init = (data.get("data") or {}).get("initiative") or {}
    if not init:
        print(f"No initiative with id '{iid}'.")
        sys.exit(1)
    for p in (init.get("projects") or {}).get("nodes") or []:
        p["state"] = project_state(p)
    if args.json:
        print(json.dumps(init, indent=2))
        return
    owner = (init.get("owner") or {}).get("name") or "-"
    print(f"{init['name']}")
    print(f"  Status:   {init.get('status') or '-'}")
    print(f"  Target:   {init.get('targetDate') or '-'}")
    print(f"  Owner:    {owner}")
    print(f"  ID:       {init.get('id')}")
    desc = (init.get("description") or "").strip()
    if desc:
        print(f"  Description:")
        for line in desc.splitlines() or [desc]:
            print(f"    {line}")
    projects = (init.get("projects") or {}).get("nodes") or []
    print(f"\nProjects ({len(projects)}):")
    if not projects:
        print("  (none linked)")
        return
    for p in projects:
        progress = int((p.get("progress") or 0) * 100)
        state = p.get("state") or "-"
        print(f"  - {p['name']:40s}  {state:12s}  {progress:>3}%")


def _initiative_create(args, api_key, team_id):
    """Create a workspace initiative. Status defaults to Linear's default
    (typically Planned) when --status is omitted."""
    input_obj: dict = {"name": args.name}
    description = read_description(
        getattr(args, "description", None),
        getattr(args, "description_file", None),
    )
    if description is not None:
        input_obj["description"] = description
    if getattr(args, "status", None) is not None:
        status = args.status.strip()
        # Linear's InitiativeStatus is PascalCase; accept any case.
        allowed = ("Proposed", "Planned", "Active", "Completed", "Canceled")
        matched = next((s for s in allowed if s.lower() == status.lower()), None)
        if not matched:
            print(f"Error: initiative status '{args.status}' not found. "
                  f"Use one of: {', '.join(allowed)}.", file=sys.stderr)
            sys.exit(1)
        input_obj["status"] = matched
    if getattr(args, "target", None) is not None:
        target = parse_timeless_date(args.target, "--target")
        if target:
            input_obj["targetDate"] = target
    if getattr(args, "owner", None) is not None:
        owner_raw = (args.owner or "").strip()
        owner_id = (resolve_user_id_by_email(api_key, owner_raw)
                    or resolve_assignee_id(api_key, owner_raw))
        if not owner_id:
            print(f"Owner '{args.owner}' not found (pass an email or name).",
                  file=sys.stderr)
            sys.exit(1)
        input_obj["ownerId"] = owner_id

    data = gql(api_key, """
        mutation($input: InitiativeCreateInput!) {
            initiativeCreate(input: $input) {
                success
                initiative { id name status targetDate }
            }
        }
    """, {"input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("initiativeCreate") or {}
    if not result.get("success"):
        print("Initiative create failed.", file=sys.stderr)
        sys.exit(1)
    n = result["initiative"]
    print(f"Created initiative '{n['name']}'  "
          f"status: {n.get('status') or '-'}  ({n['id']})")


def _initiative_update(args, api_key, team_id):
    try:
        iid = resolve_initiative_id(api_key, args.initiative, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    input_obj: dict = {}
    if getattr(args, "name", None) is not None:
        input_obj["name"] = args.name
    description = read_description(
        getattr(args, "description", None),
        getattr(args, "description_file", None),
    )
    if description is not None:
        input_obj["description"] = description
    if getattr(args, "status", None) is not None:
        status = args.status.strip()
        allowed = ("Proposed", "Planned", "Active", "Completed", "Canceled")
        matched = next((s for s in allowed if s.lower() == status.lower()), None)
        if not matched:
            print(f"Error: initiative status '{args.status}' not found. "
                  f"Use one of: {', '.join(allowed)}.", file=sys.stderr)
            sys.exit(1)
        input_obj["status"] = matched
    if getattr(args, "target", None) is not None:
        input_obj["targetDate"] = parse_timeless_date(args.target, "--target")
    if getattr(args, "owner", None) is not None:
        owner_raw = (args.owner or "").strip()
        if owner_raw.lower() in ("", "none"):
            input_obj["ownerId"] = None
        else:
            owner_id = (resolve_user_id_by_email(api_key, owner_raw)
                        or resolve_assignee_id(api_key, owner_raw))
            if not owner_id:
                print(f"Owner '{args.owner}' not found (pass an email or name).",
                      file=sys.stderr)
                sys.exit(1)
            input_obj["ownerId"] = owner_id

    if not input_obj:
        print("Nothing to update. Use --name, --description, --status, "
              "--target, or --owner.", file=sys.stderr)
        sys.exit(1)

    data = gql(api_key, """
        mutation($id: String!, $input: InitiativeUpdateInput!) {
            initiativeUpdate(id: $id, input: $input) {
                success
                initiative { id name status targetDate description
                             owner { name } }
            }
        }
    """, {"id": iid, "input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("initiativeUpdate") or {}
    if not result.get("success"):
        print("Initiative update failed.", file=sys.stderr)
        sys.exit(1)
    n = result["initiative"]
    print(f"Updated initiative '{n['name']}'  "
          f"status: {n.get('status') or '-'}  ({n['id']})")


def _initiative_link(args, api_key, team_id):
    """Link a project to an initiative (initiativeToProjectCreate). Idempotent
    enough: if the join already exists, report it instead of failing hard."""
    try:
        iid = resolve_initiative_id(api_key, args.initiative, strict=True)
        pid = resolve_project_id(api_key, team_id, args.project, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    existing = find_initiative_to_project_id(api_key, iid, pid)
    if existing:
        print(f"Already linked: initiative '{args.initiative}' <-> "
              f"project '{args.project}'  ({existing})")
        return

    data = gql(api_key, """
        mutation($input: InitiativeToProjectCreateInput!) {
            initiativeToProjectCreate(input: $input) {
                success
                initiativeToProject {
                    id
                    initiative { name }
                    project { name }
                }
            }
        }
    """, {"input": {"initiativeId": iid, "projectId": pid}})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("initiativeToProjectCreate") or {}
    if not result.get("success"):
        print("Initiative link failed.", file=sys.stderr)
        sys.exit(1)
    row = result["initiativeToProject"]
    iname = (row.get("initiative") or {}).get("name") or args.initiative
    pname = (row.get("project") or {}).get("name") or args.project
    print(f"Linked initiative '{iname}' <-> project '{pname}'  ({row['id']})")


def _initiative_unlink(args, api_key, team_id):
    """Remove the initiative↔project link via initiativeToProjectDelete."""
    try:
        iid = resolve_initiative_id(api_key, args.initiative, strict=True)
        pid = resolve_project_id(api_key, team_id, args.project, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    link_id = find_initiative_to_project_id(api_key, iid, pid)
    if not link_id:
        print(f"Error: initiative '{args.initiative}' is not linked to "
              f"project '{args.project}'.", file=sys.stderr)
        sys.exit(1)

    data = gql(api_key, """
        mutation($id: String!) {
            initiativeToProjectDelete(id: $id) { success }
        }
    """, {"id": link_id})
    if check_errors(data):
        sys.exit(1)
    if not (data.get("data") or {}).get("initiativeToProjectDelete", {}).get("success"):
        print("Initiative unlink failed.", file=sys.stderr)
        sys.exit(1)
    print(f"Unlinked initiative '{args.initiative}' from project '{args.project}'.")


def _initiative_archive(args, api_key, team_id):
    try:
        iid = resolve_initiative_id(api_key, args.initiative, strict=True)
    except LookupError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!) { initiativeArchive(id: $id) { success } }
    """, {"id": iid})
    if check_errors(data):
        sys.exit(1)
    if not (data.get("data") or {}).get("initiativeArchive", {}).get("success"):
        print("Initiative archive failed.", file=sys.stderr)
        sys.exit(1)
    print(f"Archived initiative '{args.initiative}'.")


# ---------------------------------------------------------------------------
# goals: company quarter (initiative) -> week (cycle name) -> my week / my day
# ---------------------------------------------------------------------------

WEEKLY_GOAL_LABEL = "Weekly goal"
DAILY_GOAL_LABEL = "Daily goal"
_UNNAMED_CYCLE = re.compile(r"^Cycle \d+$")
# Matches nothing: stands in for the active cycle's id when there is none, so
# the goals query keeps one shape.
_NO_CYCLE_ID = "00000000-0000-0000-0000-000000000000"

# Sized to stay under Linear's complexity cap (5 x 10 x 50).
_GOALS_COMPANY_QUERY = """{
    viewer { id name avatarUrl }
    organization { logoUrl }
    initiatives(first: 5, filter: { status: { eq: "Active" } }) {
        nodes {
            id name url startedAt createdAt targetDate
            projects(first: 10) {
                nodes { id name projectMilestones(first: 50) { nodes { id name targetDate progress } } }
            }
        }
    }
}"""

_GOAL_ISSUES_QUERY = """query($teamId: ID!, $cycleId: ID!, $today: TimelessDateOrDuration!,
                             $cursor: String) {
    issues(first: 25, after: $cursor, filter: {
        team: { id: { eq: $teamId } }
        state: { type: { nin: ["canceled", "duplicate"] } }
        or: [
            { labels: { some: { name: { eqIgnoreCase: "%s" } } }, cycle: { id: { eq: $cycleId } } }
            { labels: { some: { name: { eqIgnoreCase: "%s" } } }, dueDate: { eq: $today } }
        ]
    }) {
        pageInfo { hasNextPage endCursor }
        nodes {
            id identifier title url dueDate createdAt
            assignee { id } cycle { id } project { id }
            labels(first: 10) { nodes { name } }
            projectMilestone { id name targetDate }
            children(first: 50) { nodes { id identifier title url state { type } } }
        }
    }
}""" % (WEEKLY_GOAL_LABEL, DAILY_GOAL_LABEL)


def fetch_goal_issues(api_key: str, team_id: str, cycle_id: str | None,
                      today: date) -> list[dict]:
    """Every Weekly goal in the active cycle and Daily goal due today, all people."""
    return paginate_connection(api_key, _GOAL_ISSUES_QUERY, ["issues"], {
        "teamId": team_id, "cycleId": cycle_id or _NO_CYCLE_ID, "today": today.isoformat(),
    }) or []


def goals_state(api_key: str, team_id: str) -> dict:
    """Everything `linear goals` reads, fetched once: four requests."""
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=3) as pool:
        company_read = pool.submit(gql, api_key, _GOALS_COMPANY_QUERY)
        calendar_read = pool.submit(team_calendar, api_key, team_id)
        users_read = pool.submit(list_workspace_users, api_key)
        data = company_read.result()
        if check_errors(data):
            sys.exit(1)
        cal = calendar_read.result()
        today = datetime.now(cal["tz"]).date()
        nodes = fetch_goal_issues(api_key, team_id, (cal["cycle"] or {}).get("id"), today)
        users = users_read.result()
    d = data.get("data") or {}
    active = (d.get("initiatives") or {}).get("nodes") or []
    # Several Active initiatives: the company goal is the one due first.
    active.sort(key=lambda i: i.get("targetDate") or "9999-12-31")
    return {
        "generatedAt": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
        "initiative": active[0] if active else None,
        "logoUrl": (d.get("organization") or {}).get("logoUrl"),
        "viewer": d.get("viewer") or {},
        "users": users,
        "cycle": cal["cycle"],
        "tz": cal["tz"],
        "today": today,
        "nodes": nodes,
    }


def _has_label(node: dict, name: str) -> bool:
    return any((l.get("name") or "").lower() == name.lower()
               for l in (node.get("labels") or {}).get("nodes") or [])


def pick_goals(nodes: list[dict], user_id: str | None, cycle_id: str | None,
               today: date) -> tuple[dict | None, dict | None]:
    """(Weekly goal, Daily goal) for one person; the newest wins a duplicate."""
    mine = sorted((n for n in nodes if (n.get("assignee") or {}).get("id") == user_id),
                  key=lambda n: n.get("createdAt") or "", reverse=True)
    week = next((n for n in mine if _has_label(n, WEEKLY_GOAL_LABEL) and cycle_id
                 and (n.get("cycle") or {}).get("id") == cycle_id), None)
    day = next((n for n in mine if _has_label(n, DAILY_GOAL_LABEL)
                and n.get("dueDate") == today.isoformat()), None)
    return week, day


def goal_object(node: dict | None) -> dict | None:
    """The contract's Goal: the issue plus its sub-issues as to-dos."""
    if not node:
        return None
    kids = [c for c in (node.get("children") or {}).get("nodes") or []
            if (c.get("state") or {}).get("type") not in ("canceled", "duplicate")]
    todos = [{"id": c["id"], "identifier": c["identifier"], "title": c["title"],
              "url": c.get("url"), "completed": (c.get("state") or {}).get("type") == "completed"}
             for c in kids]
    ms = node.get("projectMilestone")
    return {
        "id": node["id"], "identifier": node["identifier"], "title": node["title"],
        "url": node.get("url"), "dueDate": node.get("dueDate"),
        "milestone": {"name": ms.get("name"), "targetDate": ms.get("targetDate")} if ms else None,
        "done": sum(1 for t in todos if t["completed"]), "total": len(todos), "todos": todos,
    }


def initiative_milestones(initiative: dict | None, exclude: str | None = None) -> list[dict]:
    """The quarter's key results: open milestones of the initiative's projects
    due by its target date, never the parking milestone. Each carries its
    project as {id, name} so a goal can be filed into it."""
    if not initiative:
        return []
    until = initiative.get("targetDate") or "9999-12-31"
    skip = (exclude or "").lower()
    out = []
    for p in (initiative.get("projects") or {}).get("nodes") or []:
        for m in (p.get("projectMilestones") or {}).get("nodes") or []:
            tgt = m.get("targetDate")
            if (tgt and tgt <= until and (m.get("progress") or 0) < 100
                    and (m.get("name") or "").lower() != skip):
                out.append({**m, "project": {"id": p["id"], "name": p["name"]}})
    return sorted(out, key=lambda m: (m["targetDate"], m.get("name") or ""))


def _person(u: dict) -> dict:
    return {"id": u.get("id"), "name": u.get("name"), "avatarUrl": u.get("avatarUrl")}


def build_goals(state: dict, backlog_milestone: str | None = None) -> dict:
    """The `linear goals --json` document from goals_state(). Pure."""
    ini = state.get("initiative")
    cycle = state.get("cycle")
    cid = (cycle or {}).get("id")
    today = state["today"]
    viewer = state.get("viewer") or {}
    name = ((cycle or {}).get("name") or "").strip()

    def goals_for(uid):
        week, day = pick_goals(state.get("nodes") or [], uid, cid, today)
        return goal_object(week), goal_object(day)

    my_week, my_day = goals_for(viewer.get("id"))
    team = []
    for u in sorted(state.get("users") or [], key=lambda u: (u.get("name") or "").lower()):
        if u.get("app") or u.get("guest") or u.get("id") == viewer.get("id"):
            continue
        w, dg = goals_for(u.get("id"))
        team.append({"user": _person(u), "weekGoal": w, "dayGoal": dg})
    return {
        "generatedAt": state.get("generatedAt"),
        "company": {
            "initiative": {
                "id": ini["id"], "name": ini["name"], "url": ini.get("url"),
                "startDate": (ini.get("startedAt") or ini.get("createdAt") or "")[:10] or None,
                "targetDate": ini.get("targetDate"),
            } if ini else None,
            "logoUrl": state.get("logoUrl"),
            "milestones": [
                {"id": m["id"], "name": m["name"], "project": m["project"]["name"],
                 "targetDate": m.get("targetDate"), "progress": round(m.get("progress") or 0)}
                for m in initiative_milestones(ini, backlog_milestone)
            ],
        },
        "week": {
            "cycle": {"id": cycle["id"], "number": cycle.get("number"), "name": cycle.get("name"),
                      "startsAt": cycle.get("startsAt"), "endsAt": cycle.get("endsAt")} if cycle else None,
            "goal": name if name and not _UNNAMED_CYCLE.match(name) else None,
        },
        "me": {"user": _person(viewer), "weekGoal": my_week, "dayGoal": my_day},
        "team": team,
    }


def _goal_line(g: dict | None, level: str) -> str:
    if not g:
        return f'not set: linear goals set {level} "<text>"'
    ms = f" · {g['milestone']['name']}" if g.get("milestone") else ""
    return f"{g['identifier']} {g['title']} ({g['done']}/{g['total']}) · due {g['dueDate']}{ms}"


def print_goals(doc: dict, today: date, tz: tzinfo) -> None:
    ini = doc["company"]["initiative"]
    if ini:
        left = ""
        if ini.get("targetDate"):
            days = (date.fromisoformat(ini["targetDate"]) - today).days
            left = f" · {days} days left (target {ini['targetDate']})"
        print(f"Company   {ini['name']}{left}")
    else:
        print("Company   no Active initiative")
    for m in doc["company"]["milestones"]:
        print(f"            {m['name']}  {m['targetDate']}  {m['progress']}%")
    cyc = doc["week"]["cycle"]
    if cyc:
        span = cycle_days(cyc, tz)
        ends = f" · ends {span[1]} ({(span[1] - today).days} days)" if span else ""
        goal = doc["week"]["goal"] or 'no week goal set: linear goals set company-week "<text>"'
        print(f"Week      {goal} · Cycle {cyc.get('number')}{ends}")
    else:
        print("Week      no active cycle")
    print(f"My week   {_goal_line(doc['me']['weekGoal'], 'week')}")
    day = doc["me"]["dayGoal"]
    print(f"My day    {_goal_line(day, 'day')}")
    for t in (day or {}).get("todos") or []:
        print(f"            [{'x' if t['completed'] else ' '}] {t['identifier']} {t['title']}")
    if doc["team"]:
        print("Team")
        for row in doc["team"]:
            dg = row["dayGoal"]
            day_txt = f"{dg['identifier']} {dg['title']} ({dg['done']}/{dg['total']})" if dg else "no day goal"
            wg = row["weekGoal"]
            week_txt = f" · week: {wg['title']}" if wg else ""
            print(f"  {row['user']['name']}: {day_txt}{week_txt}")


def _die(msg: str) -> None:
    print(msg, file=sys.stderr)
    sys.exit(1)


def _goal_milestone(api_key: str, team_id: str, cfg: dict, state: dict,
                    value: str | None) -> dict:
    """--milestone resolved across the team, else the policy default: the
    company's next open key-result milestone. Returns a catalog node with
    `project: {id, name}`."""
    if value:
        try:
            return _select_named_node(list_milestone_catalog(api_key, team_id), value,
                                      "milestone", strict=True, suggest=True)
        except LookupError as e:
            _die(str(e))
    m = next_open_milestone(initiative_milestones(state["initiative"]),
                            load_policy(cfg)["backlogMilestone"])
    if not m:
        _die("No open milestone in the Active initiative's projects to file the goal in. "
             "Pass --milestone <name>.")
    return m


def _goal_label_id(api_key: str, team_id: str, name: str) -> str:
    for l in list_team_labels(api_key, team_id):
        if (l.get("name") or "").lower() == name.lower():
            return l["id"]
    _die(f"label {name!r} not found. Goals are issues with that label; ask the owner to add it.")


def _issue_write(api_key: str, issue_id: str | None, input_obj: dict) -> dict:
    if issue_id:
        data = gql(api_key, """mutation($id: String!, $input: IssueUpdateInput!) {
            issueUpdate(id: $id, input: $input) { success issue { id identifier } }
        }""", {"id": issue_id, "input": input_obj})
        key = "issueUpdate"
    else:
        data = gql(api_key, """mutation($input: IssueCreateInput!) {
            issueCreate(input: $input) { success issue { id identifier } }
        }""", {"input": input_obj})
        key = "issueCreate"
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get(key) or {}
    if not result.get("success"):
        _die(f"{key} failed.")
    return result["issue"]


def _emit_goal(args, api_key: str, team_id: str, state: dict, level: str, verb: str) -> None:
    """Re-read the goals after a write and print the resulting Goal."""
    cid = (state["cycle"] or {}).get("id")
    nodes = fetch_goal_issues(api_key, team_id, cid, state["today"])
    week, day = pick_goals(nodes, state["viewer"].get("id"), cid, state["today"])
    goal = goal_object(week if level == "week" else day)
    if getattr(args, "json", False):
        print(json.dumps(goal, indent=2))
        return
    print(f"{verb}: {_goal_line(goal, level)}")
    for t in (goal or {}).get("todos") or [] if level == "day" else []:
        print(f"  [{'x' if t['completed'] else ' '}] {t['identifier']} {t['title']}")


def _goals_set(args, cfg, api_key, team_id):
    if args.level == "company-week":
        cal = team_calendar(api_key, team_id)
        if not cal["cycle"]:
            _die("No active cycle to name. The company week goal is the active cycle's name.")
        data = gql(api_key, """mutation($id: String!, $name: String!) {
            cycleUpdate(id: $id, input: { name: $name }) {
                success cycle { id number name startsAt endsAt }
            }
        }""", {"id": cal["cycle"]["id"], "name": args.text})
        if check_errors(data):
            sys.exit(1)
        c = ((data.get("data") or {}).get("cycleUpdate") or {}).get("cycle")
        if not c:
            _die("cycleUpdate failed.")
        week = build_goals({"cycle": c, "today": datetime.now(cal["tz"]).date()})["week"]
        if getattr(args, "json", False):
            print(json.dumps(week, indent=2))
        else:
            print(f"Week goal: {week['goal'] or 'cleared'} (cycle {c.get('number')})")
        return

    state = goals_state(api_key, team_id)
    viewer_id = state["viewer"].get("id")
    cycle = state["cycle"]
    week, day = pick_goals(state["nodes"], viewer_id, (cycle or {}).get("id"), state["today"])
    existing = week if args.level == "week" else day
    input_obj: dict = {"title": args.text}
    if args.milestone or not existing:
        m = _goal_milestone(api_key, team_id, cfg, state, args.milestone)
        input_obj.update(projectId=m["project"]["id"], projectMilestoneId=m["id"])
    if args.level == "day" and week:
        input_obj["parentId"] = week["id"]
    if not existing:
        if args.level == "week":
            span = cycle_days(cycle, state["tz"])
            if not span:
                _die("No active cycle. A Weekly goal belongs to the active cycle and is due at its end.")
            due = span[1]
        else:
            due = state["today"]
        label = WEEKLY_GOAL_LABEL if args.level == "week" else DAILY_GOAL_LABEL
        input_obj.update(teamId=team_id, assigneeId=viewer_id, dueDate=due.isoformat(),
                         labelIds=[_goal_label_id(api_key, team_id, label)])
        if cycle:
            input_obj["cycleId"] = cycle["id"]
        sid = resolve_state_id(get_states(api_key, team_id, cfg), "Todo")
        if sid:
            input_obj["stateId"] = sid
    _issue_write(api_key, (existing or {}).get("id"), input_obj)
    _emit_goal(args, api_key, team_id, state, args.level,
               "Updated" if existing else "Created")


def _goals_todo(args, cfg, api_key, team_id):
    if bool(args.text) == bool(args.done):
        _die('Pass a to-do text, or --done <ID>: linear goals todo "<text>" | linear goals todo --done PHNX-12')
    state = goals_state(api_key, team_id)
    cycle = state["cycle"]
    _, day = pick_goals(state["nodes"], state["viewer"].get("id"),
                        (cycle or {}).get("id"), state["today"])
    if args.done:
        todos = {t["identifier"].upper(): t for t in (goal_object(day) or {}).get("todos") or []}
        issue = todos.get(args.done.strip().upper())
        if not issue:
            _die(f"{args.done} is not a to-do of today's Daily goal"
                 + (f" ({day['identifier']})." if day else " (none is set).")
                 + " To close another issue: linear update <ID> --done --proof <…>")
        sid = resolve_state_id(get_states(api_key, team_id, cfg), "Done")
        if not sid:
            _die("No 'Done' workflow state on this team.")
        _issue_write(api_key, issue["id"], {"stateId": sid})
        _emit_goal(args, api_key, team_id, state, "day", f"Done {issue['identifier']}")
        return
    if not day:
        _die("No Daily goal for today, so there is nothing to add a to-do to. "
             'Set one first: linear goals set day "<text>"')
    input_obj = {"teamId": team_id, "title": args.text, "parentId": day["id"],
                 "assigneeId": state["viewer"].get("id"), "dueDate": state["today"].isoformat()}
    if (day.get("project") or {}).get("id"):
        input_obj["projectId"] = day["project"]["id"]
    if (day.get("projectMilestone") or {}).get("id"):
        input_obj["projectMilestoneId"] = day["projectMilestone"]["id"]
    if cycle:
        input_obj["cycleId"] = cycle["id"]
    sid = resolve_state_id(get_states(api_key, team_id, cfg), "Todo")
    if sid:
        input_obj["stateId"] = sid
    created = _issue_write(api_key, None, input_obj)
    _emit_goal(args, api_key, team_id, state, "day", f"Added {created['identifier']}")


def cmd_goals(args, cfg, api_key, team_id):
    """Company quarter, week and day goals; `set` and `todo` write yours."""
    try:
        load_policy(cfg)
    except ValueError as e:
        _die(f"{e} (in {CONFIG_PATH})")
    action = getattr(args, "goals_action", None)
    if action == "set":
        return _goals_set(args, cfg, api_key, team_id)
    if action == "todo":
        return _goals_todo(args, cfg, api_key, team_id)
    state = goals_state(api_key, team_id)
    doc = build_goals(state, load_policy(cfg)["backlogMilestone"])
    if getattr(args, "json", False):
        print(json.dumps(doc, indent=2))
        return
    print_goals(doc, state["today"], state["tz"])


def cmd_labels(args, cfg, api_key, team_id):
    """List labels (default) or create / update / delete one."""
    action = getattr(args, "labels_action", None)
    if action == "create":
        return _label_create(args, api_key, team_id)
    if action == "update":
        return _label_update(args, api_key, team_id)
    if action == "delete":
        return _label_delete(args, api_key, team_id)

    labels = list_team_labels(api_key, team_id)
    labels.sort(key=lambda l: l["name"].lower())
    if args.json:
        print(json.dumps(labels, indent=2))
        return
    print(f"{len(labels)} label(s):")
    for l in labels:
        desc = l.get("description") or ""
        print(f"  {l['name']:30s}  {desc}")


def _label_create(args, api_key, team_id):
    input_obj = {"teamId": team_id, "name": args.name}
    if args.color is not None:
        input_obj["color"] = args.color
    if args.description is not None:
        input_obj["description"] = args.description
    data = gql(api_key, """
        mutation($input: IssueLabelCreateInput!) {
            issueLabelCreate(input: $input) {
                success
                issueLabel { id name color }
            }
        }
    """, {"input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("issueLabelCreate") or {}
    if not result.get("success"):
        print("Label create failed.", file=sys.stderr)
        sys.exit(1)
    l = result["issueLabel"]
    print(f"Created label '{l['name']}'  ({l['id']})")


def _label_update(args, api_key, team_id):
    node = resolve_label_by_value(api_key, team_id, args.label)
    if not node:
        sys.exit(1)
    input_obj = {}
    if args.name is not None:
        input_obj["name"] = args.name
    if args.color is not None:
        input_obj["color"] = args.color
    if args.description is not None:
        input_obj["description"] = args.description
    if not input_obj:
        print("Nothing to update. Use --name, --color, or --description.", file=sys.stderr)
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!, $input: IssueLabelUpdateInput!) {
            issueLabelUpdate(id: $id, input: $input) {
                success
                issueLabel { id name }
            }
        }
    """, {"id": node["id"], "input": input_obj})
    if check_errors(data):
        sys.exit(1)
    result = (data.get("data") or {}).get("issueLabelUpdate") or {}
    if not result.get("success"):
        print("Label update failed.", file=sys.stderr)
        sys.exit(1)
    print(f"Updated label '{node.get('name')}' -> '{result['issueLabel']['name']}'")


def _label_delete(args, api_key, team_id):
    node = resolve_label_by_value(api_key, team_id, args.label)
    if not node:
        sys.exit(1)
    data = gql(api_key, """
        mutation($id: String!) {
            issueLabelDelete(id: $id) { success }
        }
    """, {"id": node["id"]})
    if check_errors(data):
        sys.exit(1)
    if (data.get("data") or {}).get("issueLabelDelete", {}).get("success"):
        print(f"Deleted label '{node.get('name') or node['id']}'.")
    else:
        print("Label delete failed.", file=sys.stderr)
        sys.exit(1)


# ---------------------------------------------------------------------------
# migrate-agent-labels — one-time retirement of the legacy `agent:<name>` labels
# ---------------------------------------------------------------------------

AGENT_LABEL_PREFIX = "agent:"


def classify_agent_labels(node: dict, label_names: list[str], roster: list[str]) -> tuple[str, str]:
    """Decide what to do with ONE issue and ALL the legacy `agent:*` labels on it.

    Deciding per issue rather than per label is the whole point: an issue can
    carry two of these labels, and `issueUpdate(labelIds:)` replaces the entire
    label set, so two writes computed from the same pre-mutation snapshot would
    resurrect each other's stripped label and the second delegate would silently
    overwrite the first.

    Returns (verdict, detail) where verdict is one of:
      migrate    — the labels agree on one real agent and the issue has no delegate
      already    — that agent is already the delegate; just strip the labels
      conflict   — the labels disagree, or they name someone other than the
                   current delegate. Never overwrite; a human decides.
      unresolved — at least one label suffix is not a delegatable agent

    Pure decision logic, no I/O, so every branch is testable without touching a
    workspace.
    """
    if not label_names:
        return "already", ""      # nothing to do; caller filters these out
    suffixes = [n[len(AGENT_LABEL_PREFIX):].strip() for n in label_names]
    resolved, unknown = [], []
    for s in suffixes:
        match = next((r for r in roster if r.lower() == s.lower()), None)
        (resolved if match else unknown).append(match or s)

    if unknown:
        # Mixed state needs a human: leave the issue exactly as it is, including
        # any sibling label that would have resolved.
        return "unresolved", ", ".join(f"'{s}' is not a delegatable agent" for s in unknown)

    targets = sorted(set(resolved), key=str.lower)
    if len(targets) > 1:
        return "conflict", f"labels name {' and '.join(targets)} — cannot pick one"

    target = targets[0]
    current = delegate_of(node)
    if current is None:
        return "migrate", target
    if current.lower() == target.lower():
        return "already", target
    return "conflict", f"delegated to {current}, label says {target}"


CARRIER_PAGE = 200


def label_carrier_ids(api_key: str, label_id: str) -> tuple[set[str], bool] | None:
    """(ids of the issues carrying this label, truncated?) — workspace-wide.
    None on API error.

    Deliberately not team-scoped: `list_team_labels` also returns workspace-wide
    labels (`team: { null: true }`), so a label this team has finished with may
    still be carrying ownership on another team's issues. Deleting on the
    strength of a team-scoped scan would strip those silently.

    `includeArchived: true` matters — the connection excludes archived issues by
    default, so a label carried only by archived issues would look unused and be
    deleted.

    The truncation flag is load-bearing, not bookkeeping. The caller subtracts
    the issues this run clears from this set, so a *partial* page whose every
    entry happens to be cleared subtracts to empty and reads as "nothing carries
    it" — deleting a label that hundreds of untouched issues still use. Asking
    for one more than the page size is how we can tell.
    """
    data = gql(api_key, """
        query($id: ID!, $first: Int!) {
            issues(filter: { labels: { id: { eq: $id } } },
                   includeArchived: true, first: $first) {
                nodes { id }
            }
        }
    """, {"id": label_id, "first": CARRIER_PAGE + 1})
    if check_errors(data):
        return None
    nodes = ((data.get("data") or {}).get("issues") or {}).get("nodes", [])
    return {n["id"] for n in nodes[:CARRIER_PAGE]}, len(nodes) > CARRIER_PAGE


def cmd_migrate_agent_labels(args, cfg, api_key, team_id):
    """Migrate legacy `agent:<name>` ownership labels to native delegates.

    Dry run by default; `--apply` writes. An issue already delegated to a
    different agent is never overwritten — it is reported as a conflict and left
    exactly as it was. A label whose suffix is not a delegatable agent is
    reported as unresolved and left in place.

    `--apply` exits non-zero when anything is left behind, so an unattended run
    cannot mistake a partial migration for a clean one. A dry run exits 0: it is
    an inspection, and reporting work-to-do as a failure is what made every
    future run red.
    """
    roster = [a["name"] for a in get_agents(api_key, cfg)]
    if not roster:
        print("No agent members found in this workspace; nothing can be resolved.",
              file=sys.stderr)
        sys.exit(1)

    # Every issue in the team carrying any `agent:*` label, in any state — a Done
    # issue keeps its label until we strip it, and the label can't be deleted
    # while it is still in use.
    issue_filter = {
        "team": {"id": {"eq": team_id}},
        "labels": {"name": {"startsWith": AGENT_LABEL_PREFIX}},
    }
    query = """query($filter: IssueFilter!, $cursor: String) {
        issues(filter: $filter, first: 100, after: $cursor) {
            pageInfo { hasNextPage endCursor }
            nodes {
                id identifier title
                state { name }
                delegate { name }
                labels { nodes { id name } }
            }
        }
    }"""
    nodes = paginate_connection(api_key, query, ["issues"], {"filter": issue_filter})
    if nodes is None:
        sys.exit(1)

    # One row per issue: (node, [agent labels on it], verdict, detail).
    rows = []
    for n in nodes:
        labels = [l for l in n.get("labels", {}).get("nodes", [])
                  if l["name"].startswith(AGENT_LABEL_PREFIX)]
        if not labels:
            continue
        verdict, detail = classify_agent_labels(n, [l["name"] for l in labels], roster)
        rows.append((n, labels, verdict, detail))

    def of(kind):
        return [r for r in rows if r[2] == kind]

    mode = "APPLY" if args.apply else "DRY RUN"
    print(f"[{mode}] {len(rows)} issue(s) carry a legacy '{AGENT_LABEL_PREFIX}*' label")
    print()

    def names(labels):
        return ",".join(l["name"] for l in labels)

    for n, labels, _, agent in of("migrate"):
        print(f"  migrate    {n['identifier']:<10} {names(labels):<24} -> delegate: {agent}")
    for n, labels, _, agent in of("already"):
        print(f"  strip      {n['identifier']:<10} {names(labels):<24} (already delegated to {agent})")
    for n, labels, _, why in of("conflict"):
        print(f"  CONFLICT   {n['identifier']:<10} {names(labels):<24} {why}", file=sys.stderr)
    for n, labels, _, why in of("unresolved"):
        print(f"  UNRESOLVED {n['identifier']:<10} {names(labels):<24} {why}", file=sys.stderr)
    print()

    # Counters must reflect what actually happened, not what was planned — the
    # summary line is what a person or a script reads to call the job done.
    migrated = stripped = 0
    write_failures = 0
    blocked_labels = {l["name"] for _, labels, v, _ in rows
                      if v in ("conflict", "unresolved") for l in labels}

    # Issues whose agent labels this run removes. In a dry run that is every
    # planned row: the delete gate has to reason about the state the migration
    # WOULD leave behind, or the preview reports every label as still in use —
    # by the very issues it just said it would strip.
    cleared_ids = {n["id"] for n, _, v, _ in rows if v in ("migrate", "already")}

    if args.apply:
        for n, labels, verdict, detail in rows:
            if verdict not in ("migrate", "already"):
                continue
            drop = {l["id"] for l in labels}
            keep = [x["id"] for x in n["labels"]["nodes"] if x["id"] not in drop]
            input_obj = {"labelIds": keep}
            if verdict == "migrate":
                aid = resolve_agent_id(api_key, cfg, detail)
                if not aid:
                    print(f"  FAILED     {n['identifier']} — no agent id for {detail}",
                          file=sys.stderr)
                    write_failures += 1
                    blocked_labels.update(l["name"] for l in labels)
                    continue
                input_obj["delegateId"] = aid
            data = gql(api_key, """
                mutation($id: String!, $input: IssueUpdateInput!) {
                    issueUpdate(id: $id, input: $input) {
                        success
                        issue { identifier delegate { name } }
                    }
                }
            """, {"id": n["id"], "input": input_obj})
            if check_errors(data) or not ((data.get("data") or {}).get("issueUpdate") or {}).get("success"):
                print(f"  FAILED     {n['identifier']} — issueUpdate did not succeed",
                      file=sys.stderr)
                write_failures += 1
                blocked_labels.update(l["name"] for l in labels)
                continue
            issue = data["data"]["issueUpdate"]["issue"]
            dg = (issue.get("delegate") or {}).get("name") or "none"
            print(f"  done       {issue['identifier']:<10} delegate={dg}")
            if verdict == "migrate":
                migrated += 1
            else:
                stripped += 1

    # Retire every `agent:*` label that nothing carries any more. A label still
    # on a conflicting, unresolved, or failed issue is deliberately kept, and so
    # is one still carried outside the scanned team.
    delete_failures = 0
    labels = [l for l in list_team_labels(api_key, team_id)
              if l["name"].startswith(AGENT_LABEL_PREFIX)]
    for l in sorted(labels, key=lambda x: x["name"]):
        if l["name"] in blocked_labels:
            print(f"  keep       label {l['name']} — still in use by an unmigrated issue")
            continue
        found = label_carrier_ids(api_key, l["id"])
        if found is None:
            print(f"  FAILED     label {l['name']} — could not look up what carries it",
                  file=sys.stderr)
            delete_failures += 1
            continue
        carriers, truncated = found
        if truncated:
            print(f"  keep       label {l['name']} — more than {CARRIER_PAGE} issues "
                  f"carry it; too many to confirm it is free")
            continue
        # Whatever this run clears no longer counts against the label. In a dry
        # run that projection is the whole point of the preview. The remainder
        # may be on another team, or archived, or simply not scanned — the one
        # thing worth saying is that this run did not clear it.
        if carriers - cleared_ids:
            print(f"  keep       label {l['name']} — still carried by an issue "
                  f"this run did not migrate")
            continue
        if not args.apply:
            print(f"  delete     label {l['name']} (nothing carries it after migration)")
            continue
        data = gql(api_key, """
            mutation($id: String!) { issueLabelDelete(id: $id) { success } }
        """, {"id": l["id"]})
        if check_errors(data) or not (data.get("data") or {}).get("issueLabelDelete", {}).get("success"):
            print(f"  FAILED     label {l['name']} — delete did not succeed", file=sys.stderr)
            delete_failures += 1
            continue
        print(f"  deleted    label {l['name']}")

    print()
    print(f"migrated={migrated} stripped={stripped} "
          f"conflicts={len(of('conflict'))} unresolved={len(of('unresolved'))} "
          f"failures={write_failures + delete_failures}")

    left_behind = of("conflict") or of("unresolved") or write_failures or delete_failures
    if not args.apply:
        # These are independent: a lookup failure must not swallow the guidance
        # for conflicts a human still has to resolve.
        if of("conflict") or of("unresolved"):
            print("Dry run only. The entries above need a human before --apply can "
                  "finish: resolve each conflict, and rename or remove any label "
                  "whose suffix is not an agent.")
        if delete_failures:
            print("Could not reach the API to check what still carries some labels "
                  "— re-run once it responds.")
        if not left_behind:
            print("Dry run only. Re-run with --apply to write.")
        return
    if left_behind:
        print("Migration incomplete — resolve the entries above by hand, then re-run.",
              file=sys.stderr)
        sys.exit(1)


def cmd_users(args, cfg, api_key, team_id):
    """List workspace members, grouped into humans (assign with --assign) and
    agents (delegate with --delegate) — the two are distinct in Linear."""
    users = list_workspace_users(api_key)
    if args.json:
        users.sort(key=lambda u: u["name"].lower())
        print(json.dumps(users, indent=2))
        return
    humans = sorted((u for u in users if not u.get("app")), key=lambda u: u["name"].lower())
    agents = sorted((u for u in users if u.get("app")), key=lambda u: u["name"].lower())
    print(f"Humans ({len(humans)})  — assign with --assign <email or name>:")
    for u in humans:
        print(f"  {u['name']:25s}  {u.get('email') or '-'}")
    print(f"\nAgents ({len(agents)})  — delegate with --delegate <name>:")
    for u in agents:
        print(f"  {u['name']:25s}  (app user)")


def cmd_states(args, cfg, api_key, team_id):
    """List the team's workflow states — the valid values for --status / --done.

    Without this, agents guess state names and only learn they were wrong when a
    mutation fails. Bypasses the cfg cache so the list is always current.
    """
    fresh = {**cfg, "states": None}
    states = get_states(api_key, team_id, fresh)
    if not states:
        return
    # Order by workflow type, then name, so the progression reads top-to-bottom.
    type_order = {"backlog": 0, "unstarted": 1, "started": 2, "completed": 3, "canceled": 4}
    items = sorted(states.items(), key=lambda kv: (type_order.get(kv[1]["type"], 9), kv[0]))
    if args.json:
        print(json.dumps(
            [{"name": name, "type": meta["type"], "id": meta["id"]} for name, meta in items],
            indent=2,
        ))
        return
    print(f"{len(items)} workflow state(s):")
    for name, meta in items:
        print(f"  {name:20s}  {meta['type']}")


_INBOX_TYPE_LABEL = {
    "issueNewComment": "comment",
    "issueCommentMention": "mention",
    "issueMention": "mention",
    "issueCommentReaction": "reaction",
    "issueAssignedToYou": "assigned",
    "issueUnassignedFromYou": "unassigned",
    "issueStatusChanged": "status",
    "issueDue": "due",
    "issueBlocking": "blocking",
    "issueEmojiReaction": "reaction",
    "issueSubscribed": "subscribed",
}


def _mark_notifications_read(api_key, ids):
    """Mark specific notifications read via notificationUpdate. Returns (ok, failed)."""
    import datetime
    now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    ok, failed = [], []
    for nid in ids:
        data = gql(api_key, """
            mutation($id: String!, $at: DateTime!) {
              notificationUpdate(id: $id, input: { readAt: $at }) { success }
            }
        """, {"id": nid, "at": now})
        # check_errors prints "Error: <message>" to stderr on a GraphQL error,
        # matching the other per-item mutation loops in this file.
        had_error = check_errors(data)
        success = (((data.get("data") or {}).get("notificationUpdate") or {}).get("success"))
        (ok if success and not had_error else failed).append(nid)
    return ok, failed


def cmd_inbox(args, cfg, api_key, team_id):
    """Show your Linear inbox — notifications (comments, mentions, assignments,
    status changes). Acts too: --read <id> marks notification(s) read, --read-all
    marks everything read. Unread only by default; --all lists read as well.

    Agents: `--json` returns full metadata per notification — the issue
    (identifier + url + state), the comment (id + url + body), the thread parent,
    and the actor — enough to follow up. Reply on a ticket with
    `linear update <identifier> --comment "..."`; dismiss with
    `linear inbox --read <notification-id>`."""
    # --- actions take precedence over listing ---
    if getattr(args, "read_all", False):
        data = gql(api_key, "mutation { notificationMarkReadAll { success } }")
        ok = not check_errors(data) and (((data.get("data") or {}).get("notificationMarkReadAll") or {}).get("success"))
        if ok:
            print("Marked all notifications read.")
        else:
            print("Failed to mark all read.", file=sys.stderr)
            sys.exit(1)
        return
    read_ids = getattr(args, "read", None) or []
    if read_ids:
        ok, failed = _mark_notifications_read(api_key, read_ids)
        if ok:
            print(f"Marked read ({len(ok)}): {', '.join(ok)}")
        if failed:
            print(f"Failed to mark read: {', '.join(failed)}", file=sys.stderr)
            sys.exit(1)
        return

    # --- list ---
    limit = getattr(args, "limit", 30)  # argparse always supplies a default; honor an explicit value
    if limit < 1:
        print("inbox: --limit must be >= 1", file=sys.stderr)
        return
    data = gql(api_key, """
        query($n: Int!) {
          notifications(first: $n) {
            nodes {
              id __typename createdAt readAt
              ... on IssueNotification {
                type url inboxUrl
                issue { identifier title url state { name type } }
                comment { id url body }
                parentComment { id url }
                actor { name displayName }
              }
            }
          }
        }
    """, {"n": limit})
    if check_errors(data):
        return
    nodes = ((data.get("data") or {}).get("notifications") or {}).get("nodes") or []
    # Only issue notifications carry an issue; ignore project/other kinds for now.
    items = [n for n in nodes if n.get("issue")]
    if not getattr(args, "all", False):
        items = [n for n in items if not n.get("readAt")]

    if getattr(args, "json", False):
        print(json.dumps(items, indent=2))
        return

    if not items:
        hint = "" if getattr(args, "all", False) else " (no unread — use --all to include read)"
        print(f"Inbox empty{hint}.")
        return

    unread = sum(1 for n in items if not n.get("readAt"))
    scope = "" if getattr(args, "all", False) else " unread"
    print(f"Inbox — {len(items)}{scope} ({unread} unread):")
    for n in items:
        iss = n.get("issue") or {}
        actor = n.get("actor") or {}
        who = actor.get("name") or actor.get("displayName") or "-"
        label = _INBOX_TYPE_LABEL.get(n.get("type"), n.get("type") or "note")
        dot = " " if n.get("readAt") else "*"
        title = (iss.get("title") or "")[:48]
        print(f"  {dot} {iss.get('identifier','-'):11.11s}{label:11.11s}{who:14.14s}{title}")
        body = ((n.get("comment") or {}).get("body") or "").replace("\n", " ").strip()
        if body and label in ("comment", "mention"):
            print(f'      "{body[:96]}"')
    print('\nReply:  linear update <ID> --comment "..."    ·    '
          'Dismiss:  linear inbox --read <notification-id>  (id in --json)')


def cmd_agents(args, cfg, api_key, team_id):
    """List the workspace's agent members — the app users you can `--delegate` to.

    Auto-detected from Linear (the `app` user flag), cached, and refreshed every
    6h. `--refresh` forces an immediate re-fetch — run it after installing or
    removing an agent app so the roster (and `--delegate <name>`) stays current.
    """
    agents = get_agents(api_key, cfg, force=args.refresh)
    if args.json:
        print(json.dumps(agents, indent=2))
        return
    if not agents:
        print("No agent members found. Install agent apps in Linear "
              "(Settings -> API -> Applications), then: linear agents --refresh")
        return
    print(f"{len(agents)} agent(s)  (auto-refreshes every 6h; --refresh to force):")
    for a in agents:
        print(f"  {a['name']:20s}  delegate: linear update <ID> --delegate {a['name'].lower()}")


def print_queue_list(detailed: bool = False) -> None:
    """Print queued close intents (shared by bare `queue` and `queue list`)."""
    intents = list_queue_intents()
    if not intents:
        print("Queue empty.")
        return
    print(f"{len(intents)} close intent(s) queued:")
    for intent in intents:
        ident = intent.get("identifier", "?")
        state = intent.get("state", "Done")
        attempts = intent.get("attempts", 0) or 0
        if detailed:
            nxt = intent.get("next_attempt")
            status = f"attempts={attempts}"
            if nxt:
                status += f", next {nxt}"
            print(f"  {ident} -> {state} ({status})")
        else:
            print(f"  {ident} -> {state} (attempts={attempts})")
    print("Run `linear queue drain` to apply them.")


def cmd_queue(args, cfg, api_key, team_id):
    """Manage the durable local queue for closes that could not be written
    immediately (typically because Linear returned a rate limit)."""
    action = getattr(args, "queue_action", None)
    if action == "drain":
        if getattr(args, "dry_run", False):
            intents = list_queue_intents()
            if not intents:
                print("Queue empty.")
                return
            for intent in intents:
                ident = intent.get("identifier", "?")
                state = intent.get("state", "Done")
                attempts = intent.get("attempts", 0) or 0
                nxt = intent.get("next_attempt")
                status = f"retry {attempts}"
                if nxt:
                    status += f", next {nxt}"
                print(f"{ident} -> {state} ({status})")
            return
        once = bool(getattr(args, "once", False))
        applied, remaining = drain_queue(
            api_key, team_id, cfg, quiet=False, once=once,
        )
        if remaining == 0:
            print(f"Drained {applied} intent(s). Queue empty.")
        else:
            print(f"Applied {applied}, {remaining} pending retry.")
        return

    # Bare `linear queue` and `linear queue list` both list intents.
    if action in (None, "list"):
        print_queue_list(detailed=(action == "list"))
        return

    print(f"Unknown queue action: {action}", file=sys.stderr)
    sys.exit(2)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

# Every verb each group's subparser knows. A bare name that is not one of these
# gets the implicit `show` inserted by rewrite_group_argv; a verb added to the
# parser but not here would be rewritten to `show <verb>` and be unreachable.
_PROJECTS_VERBS = frozenset({"show", "overview", "create", "update", "archive", "delete"})
_INITIATIVES_VERBS = frozenset({"show", "create", "update", "link", "unlink", "archive"})


def rewrite_group_argv(argv: list[str]) -> list[str]:
    """Back-compat + ergonomics: `projects <name>` / `initiatives <name>` (a bare
    name, not a verb) is shorthand for `<group> show <name>`. Insert the implicit
    `show` so the detail view keeps working alongside the other subcommands."""
    argv = list(argv)
    if len(argv) < 2:
        return argv
    verbs = {"projects": _PROJECTS_VERBS, "initiatives": _INITIATIVES_VERBS}.get(argv[0])
    if not verbs:
        return argv
    for i in range(1, len(argv)):
        if not argv[i].startswith("-"):
            if argv[i] not in verbs:
                argv = argv[:i] + ["show"] + argv[i:]
            break
    return argv


def main():
    parser = argparse.ArgumentParser(
        prog="linear",
        description="Task management CLI for AI agent teams",
    )
    parser.add_argument("--version", action="version", version=f"linear-cli {__version__}")
    parser.add_argument("--team", default=None,
                        help="Override team for this invocation (key like 'ANT' or UUID)")
    sub = parser.add_subparsers(dest="command")

    # setup
    p_setup = sub.add_parser("setup", help="Configure Linear connection and agent identity")
    p_setup.add_argument("--api-key", help="Linear API key")
    p_setup.add_argument("--agent", help="Default agent identity")
    p_setup.add_argument("--embeddings", choices=["ollama", "gemini", "none"],
                         help="Optional embeddings ranker for `tasks --similar`: "
                              "ollama (local, OLLAMA_HOST), gemini (GEMINI_API_KEY "
                              "from the environment), or none to turn it off")
    p_setup.add_argument("--embeddings-model", metavar="NAME",
                         help="Embedding model for --embeddings (default: "
                              "qwen3-embedding:0.6b for ollama, "
                              "gemini-embedding-001 for gemini)")

    # tasks (list + view + board)
    p_tasks = sub.add_parser("tasks", help="List tasks, view details, or show team board")
    p_tasks.add_argument("identifier", nargs="?", help="Issue identifier for detail view (e.g. ANT-42)")
    p_tasks.add_argument("--agent", help="Filter to the agent this issue is delegated to (e.g. --agent claude). See: linear agents")
    p_tasks.add_argument("--all", action="store_true", help="Ignore default agent filter AND cwd project auto-scope; show the whole team")
    p_tasks.add_argument("--board", action="store_true", help="Team board grouped by agent")
    p_tasks.add_argument("--label", help="Filter by any label")
    p_tasks.add_argument("--project", help="Filter by project (name or UUID). Defaults to the cwd's project when `agents projects` binds one; --all disables that.")
    p_tasks.add_argument("--milestone",
                         help="Filter by milestone (name or UUID). Pair with "
                              "--project to disambiguate a name across projects.")
    p_tasks.add_argument("--by-milestone", action="store_true",
                         help="Group the list by milestone, with a 'No milestone' "
                              "bucket for issues matched to a project but no milestone. "
                              "Auto-on when --project scopes the list.")
    p_tasks.add_argument("--status", help="Filter: backlog, todo, progress, done, open")
    p_tasks.add_argument("--cycle", default=None, metavar="SCOPE",
                         help="Scope: active (default), next, all (whole team, "
                              "cross-cycle + backlog), none/backlog (no cycle), "
                              "or a specific cycle by name or id. When --project or "
                              "--milestone is set, defaults to all cycles (the whole "
                              "deliverable, not just this cycle's slice).")
    p_tasks.add_argument("--since", metavar="YYYY-MM-DD",
                         help="Only issues created on or after this date")
    p_tasks.add_argument("--assignee",
                         help="Filter by assignee: email, 'me', or 'none' (unassigned)")
    p_tasks.add_argument("--query", help="Search title + description (case-insensitive)")
    p_tasks.add_argument("--similar", metavar="TEXT",
                         help="Rank the team's tickets by relevance to TEXT (a title or "
                              "paragraph) — Linear semantic search + local lexical match, "
                              "fused. Whole team, every cycle, done included. Honors "
                              "--project only; other list filters are ignored. "
                              "Use before creating a ticket.")
    p_tasks.add_argument("--limit", type=int, default=10,
                         help="Rows to show with --similar (default 10, minimum 1)")
    p_tasks.add_argument("--json", action="store_true", help="JSON output")

    # update (all mutations on one or many issues)
    p_update = sub.add_parser("update", help="Update one or many issues (status, comment, ...)")
    p_update.add_argument("identifier", nargs="*",
                          help="Issue identifier(s), e.g. ANT-42 ANT-43. "
                               "Omit and use --stdin to pipe ids.")
    p_update.add_argument("--stdin", action="store_true",
                          help="Read additional identifiers (whitespace-separated) from stdin")
    status_group = p_update.add_mutually_exclusive_group()
    status_group.add_argument("--done", action="store_true", help="Mark Done")
    status_group.add_argument("--pickup", action="store_true", help="Mark In Progress")
    status_group.add_argument("--todo", action="store_true", help="Mark Todo")
    status_group.add_argument("--status", help="Move to any state by name")
    p_update.add_argument("--comment", help="Add a comment")
    p_update.add_argument("--proof", action="append",
                          help="Proof of completion (repeatable). File path, URL, or text.")
    p_update.add_argument("--label", action="append", help="Add label(s) (repeatable)")
    p_update.add_argument("--unlabel", action="append",
                          help="Remove label(s) (repeatable)")
    p_update.add_argument("--priority", type=str, default=None,
                          help="urgent, high, medium, low, or none")
    p_update.add_argument("--assign", default=None,
                          help="Reassign to a human by email or name, or 'none' to unassign")
    p_update.add_argument("--delegate", default=None,
                          help="Delegate to an agent by name (e.g. --delegate claude); "
                               "'none' to clear. See: linear agents")
    p_update.add_argument("--parent", default=None,
                          help="Set parent issue (e.g. ANT-42), or 'none' to detach")
    p_update.add_argument("--project", default=None,
                          help="Move to project (name or ID), or 'none' to detach")
    p_update.add_argument("--milestone", default=None,
                          help="Set milestone (name), or 'none' to detach")
    p_update.add_argument("--title", default=None, help="Rename the issue")
    p_update.add_argument("--description", default=None,
                          help="Replace the issue description")
    p_update.add_argument("--description-file", dest="description_file", default=None,
                          help="Read description from a file path, or '-' for stdin")
    p_update.add_argument("--due-date", dest="due_date",
                          help="Set due date (YYYY-MM-DD), or 'none' to clear")
    p_update.add_argument("--remind-at", dest="remind_at",
                          help="Remind you at an exact time: YYYY-MM-DDTHH:MM (local) or with a zone")
    p_update.add_argument("--cycle", choices=["active", "next", "none"],
                          help="Move issue: active cycle, next cycle, or remove from cycle")
    p_update.add_argument("--blocks", action="append",
                          help="Mark this issue as blocking another (e.g. ANT-42). Repeatable.")
    p_update.add_argument("--blocked-by", dest="blocked_by", action="append",
                          help="Mark this issue as blocked by another (e.g. ANT-42). Repeatable.")
    p_update.add_argument("--relates", action="append",
                          help="Mark a non-blocking 'related' link to another issue. Repeatable.")

    # create
    p_create = sub.add_parser("create", help="Create a new issue")
    p_create.add_argument("title", nargs="?", default=None,
                          help="Issue title (optional if --description provided)")
    p_create.add_argument("--label", action="append", help="Label(s) to add (repeatable)")
    p_create.add_argument("--priority", type=str, default=None,
                          help="urgent, high, medium (default), low, or none")
    p_create.add_argument("--description", help="Issue description")
    p_create.add_argument("--description-file", dest="description_file", default=None,
                          help="Read description from a file path, or '-' for stdin")
    p_create.add_argument("--image", action="append",
                          help="Image file path to upload and embed (repeatable)")
    p_create.add_argument("--status", help="Initial status (default: Todo; Backlog with --backlog)")
    p_create.add_argument("--cycle", default=None,
                          choices=["active", "next", "none"],
                          help="Cycle assignment (default: set by the horizon)")
    p_create.add_argument("--due-date", dest="due_date",
                          help="Due date (YYYY-MM-DD). Always wins over the horizon's date.")
    p_create.add_argument("--backlog", action="store_true",
                          help="Horizon backlog: Backlog state, no cycle, no due date, a "
                               "future milestone (default policy.backlogMilestone)")
    p_create.add_argument("--bug", action="store_true",
                          help="Label Bug. Alone it implies --backlog and priority low")
    p_create.add_argument("--blocker", action="store_true",
                          help="With --bug: horizon now, priority high, the next milestone")
    p_create.add_argument("--remind-at", dest="remind_at",
                          help="Remind you at an exact time: YYYY-MM-DDTHH:MM (local) or with a zone")
    p_create.add_argument("--assign", default=None,
                          help="Assign to a human by email or name (default: API key owner). "
                               "'none' leaves it unowned and requires --force.")
    p_create.add_argument("--delegate", default=None,
                          help="Delegate the new issue to an agent by name (e.g. --delegate claude)")
    p_create.add_argument("--from-file", dest="from_file", default=None,
                          help="Bulk create from JSONL (one issue per line). '-' for stdin.")
    p_create.add_argument("--parent", default=None,
                          help="Parent issue identifier for sub-issues (e.g. ANT-42)")
    p_create.add_argument("--project", default=None,
                          help="Project name or ID (default: the cwd's `agents projects` binding)")
    p_create.add_argument("--milestone", default=None,
                          help="Project milestone name. Horizon milestone: due on its "
                               "target date. Default: the project's next open milestone.")
    p_create.add_argument("--skip-milestone", action="store_true",
                          help="Escape hatch: create with no milestone (warns). The issue "
                               "lands in the project's \"No milestone\" bucket.")
    p_create.add_argument("--force", action="store_true",
                          help="Create even with no assignee, or in a paused, completed or "
                               "canceled project. Without it, both are refused.")

    # cycles (list / create / update). No delete: archiving a cycle discards
    # sprint history, so it's intentionally not exposed here.
    p_cycles = sub.add_parser("cycles", help="List, create, or update cycles")
    p_cycles.add_argument("--json", action="store_true", help="JSON output")
    p_cycles.add_argument("--ids", action="store_true", help="Show cycle IDs in the plain list")
    cyc_sub = p_cycles.add_subparsers(dest="cycles_action")
    c_create = cyc_sub.add_parser("create", help="Create a cycle")
    c_create.add_argument("--name", required=True, help="Cycle name")
    c_create.add_argument("--starts", required=True, metavar="YYYY-MM-DD", help="Start date")
    c_create.add_argument("--ends", required=True, metavar="YYYY-MM-DD", help="End date")
    c_update = cyc_sub.add_parser("update", help="Update a cycle (by id or name)")
    c_update.add_argument("cycle", help="Cycle id, number, or name")
    c_update.add_argument("--name", default=None, help="New name")
    c_update.add_argument("--starts", default=None, metavar="YYYY-MM-DD", help="New start date")
    c_update.add_argument("--ends", default=None, metavar="YYYY-MM-DD", help="New end date")

    # projects (list / show / create / update / archive / delete)
    p_projects = sub.add_parser("projects",
                                help="List, show, create, update, archive, or delete projects")
    p_projects.add_argument("--json", action="store_true", help="JSON output (list)")
    proj_sub = p_projects.add_subparsers(dest="projects_action")
    ps_show = proj_sub.add_parser("show", help="Show a project's detail + milestones")
    ps_show.add_argument("name", help="Project name (exact match preferred) or id")
    # SUPPRESS: when absent, don't overwrite a parent `projects --json` (the shim
    # can place --json before the injected `show`); when present, still sets json.
    ps_show.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                         help="JSON output")
    ps_overview = proj_sub.add_parser(
        "overview",
        help="One document for menus: current cycle + projects → milestones → open issues")
    ps_overview.add_argument("--project", action="append", default=None, metavar="NAME|ID",
                             help="Restrict to a project (repeatable; exact name or id)")
    ps_overview.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                             help="JSON output")
    ps_create = proj_sub.add_parser("create", help="Create a project")
    ps_create.add_argument("--name", required=True, help="Project name")
    ps_create.add_argument("--description", default=None, help="Project description")
    ps_create.add_argument("--lead", default=None, metavar="EMAIL", help="Lead by email")
    ps_create.add_argument("--start", default=None, metavar="YYYY-MM-DD", help="Start date")
    ps_create.add_argument("--target", default=None, metavar="YYYY-MM-DD", help="Target date")
    ps_update = proj_sub.add_parser("update", help="Update a project (description, lead, dates, priority, state)")
    ps_update.add_argument("project", help="Project name or id (strict)")
    ps_update.add_argument("--name", default=None, help="Rename the project")
    ps_update.add_argument("--description", default=None, help="Set project description")
    ps_update.add_argument("--description-file", default=None, metavar="PATH",
                           help="Read description from file ('-' = stdin)")
    ps_update.add_argument("--lead", default=None, metavar="EMAIL|NAME",
                           help="Set lead by email or name; 'none' clears")
    ps_update.add_argument("--start", default=None, metavar="YYYY-MM-DD",
                           help="Start date; 'none' clears")
    ps_update.add_argument("--target", default=None, metavar="YYYY-MM-DD",
                           help="Target date; 'none' clears")
    ps_update.add_argument("--priority", default=None,
                           help="Project priority: urgent, high, medium, low, or none")
    ps_update.add_argument("--state", default=None,
                           help="Project state/status: backlog|planned|started|paused|"
                                "completed|canceled, or a status name")
    ps_archive = proj_sub.add_parser("archive", aliases=["delete"],
                                     help="Remove a project (moves to trash; recoverable in Linear's UI)")
    ps_archive.add_argument("project", help="Project name or id")

    # milestones (list / create / move / set-target-date / delete)
    p_ms = sub.add_parser("milestones",
                          help="List, create, move, retarget, or delete project milestones")
    p_ms.add_argument("--json", action="store_true", help="JSON output")
    ms_sub = p_ms.add_subparsers(dest="milestones_action")
    ms_list = ms_sub.add_parser("list", help="List a project's milestones")
    ms_list.add_argument("project", help="Project name or id")
    ms_list.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                         help="JSON output")
    ms_create = ms_sub.add_parser("create", help="Create a milestone in a project")
    ms_create.add_argument("--project", required=True, help="Project name or id")
    ms_create.add_argument("--name", required=True, help="Milestone name")
    ms_create.add_argument("--target", default=None, metavar="YYYY-MM-DD", help="Target date")
    ms_create.add_argument("--description", default=None, help="Milestone description")
    ms_move = ms_sub.add_parser("move", help="Move a milestone to another project")
    ms_move.add_argument("milestone", help="Milestone name or id")
    ms_move.add_argument("--to", required=True, dest="to", help="Destination project name or id")
    ms_move.add_argument("--from", default=None, dest="from_project",
                         help="Source project to disambiguate a same-named milestone")
    ms_target = ms_sub.add_parser("set-target-date", help="Set or clear a milestone's target date")
    ms_target.add_argument("milestone", help="Milestone name or id")
    ms_target.add_argument("date", help="YYYY-MM-DD, or 'none' to clear")
    ms_target.add_argument("--project", default=None, help="Project to disambiguate the milestone")
    ms_delete = ms_sub.add_parser("delete", help="Delete a milestone")
    ms_delete.add_argument("milestone", help="Milestone name or id")
    ms_delete.add_argument("--project", default=None, help="Project to disambiguate the milestone")

    # initiatives (list / show / create / update / link / unlink / archive)
    p_init = sub.add_parser("initiatives",
                            help="List, show, create, update, link, unlink, or archive initiatives")
    p_init.add_argument("--json", action="store_true", help="JSON output (list)")
    init_sub = p_init.add_subparsers(dest="initiatives_action")
    ini_show = init_sub.add_parser("show", help="Show an initiative + linked projects")
    ini_show.add_argument("name", help="Initiative name or id")
    ini_show.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                          help="JSON output")
    ini_create = init_sub.add_parser("create", help="Create an initiative")
    ini_create.add_argument("--name", required=True, help="Initiative name")
    ini_create.add_argument("--description", default=None, help="Description")
    ini_create.add_argument("--description-file", default=None, metavar="PATH",
                            help="Read description from file ('-' = stdin)")
    ini_create.add_argument("--status", default=None,
                            help="Proposed|Planned|Active|Completed|Canceled")
    ini_create.add_argument("--target", default=None, metavar="YYYY-MM-DD",
                            help="Target date")
    ini_create.add_argument("--owner", default=None, metavar="EMAIL|NAME",
                            help="Owner by email or name")
    ini_update = init_sub.add_parser("update", help="Update an initiative")
    ini_update.add_argument("initiative", help="Initiative name or id (strict)")
    ini_update.add_argument("--name", default=None, help="Rename")
    ini_update.add_argument("--description", default=None, help="Set description")
    ini_update.add_argument("--description-file", default=None, metavar="PATH",
                            help="Read description from file ('-' = stdin)")
    ini_update.add_argument("--status", default=None,
                            help="Proposed|Planned|Active|Completed|Canceled")
    ini_update.add_argument("--target", default=None, metavar="YYYY-MM-DD",
                            help="Target date; 'none' clears")
    ini_update.add_argument("--owner", default=None, metavar="EMAIL|NAME",
                            help="Owner by email or name; 'none' clears")
    ini_link = init_sub.add_parser("link", help="Link a project to an initiative")
    ini_link.add_argument("initiative", help="Initiative name or id")
    ini_link.add_argument("--project", required=True, help="Project name or id")
    ini_unlink = init_sub.add_parser("unlink", help="Unlink a project from an initiative")
    ini_unlink.add_argument("initiative", help="Initiative name or id")
    ini_unlink.add_argument("--project", required=True, help="Project name or id")
    ini_archive = init_sub.add_parser("archive", help="Archive an initiative")
    ini_archive.add_argument("initiative", help="Initiative name or id")

    # goals (read: company / week / my week / my day / team; write: set, todo)
    p_goals = sub.add_parser("goals",
                             help="Company, week and day goals; set yours and add to-dos")
    p_goals.add_argument("--json", action="store_true", help="JSON output")
    goals_sub = p_goals.add_subparsers(dest="goals_action")
    g_set = goals_sub.add_parser("set", help="Set your week or day goal, or the company week goal")
    g_set.add_argument("level", choices=["week", "day", "company-week"])
    g_set.add_argument("text", help="The goal")
    g_set.add_argument("--milestone", default=None,
                       help="Milestone it moves (default: the company's next open milestone)")
    g_set.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                       help="Print the resulting goal as JSON")
    g_todo = goals_sub.add_parser("todo", help="Add a to-do under today's Daily goal, or finish one")
    g_todo.add_argument("text", nargs="?", default=None, help="The to-do")
    g_todo.add_argument("--done", metavar="ID", default=None, help="Mark this to-do done")
    g_todo.add_argument("--json", action="store_true", default=argparse.SUPPRESS,
                        help="Print today's goal as JSON")

    # labels (list / create / update / delete)
    p_labels = sub.add_parser("labels", help="List, create, update, or delete labels")
    p_labels.add_argument("--json", action="store_true", help="JSON output")
    lbl_sub = p_labels.add_subparsers(dest="labels_action")
    l_create = lbl_sub.add_parser("create", help="Create a label")
    l_create.add_argument("name", help="Label name")
    l_create.add_argument("--color", default=None, metavar="#RRGGBB", help="Hex color")
    l_create.add_argument("--description", default=None, help="Label description")
    l_update = lbl_sub.add_parser("update", help="Update a label (by id or name)")
    l_update.add_argument("label", help="Label id or name")
    l_update.add_argument("--name", default=None, help="New name")
    l_update.add_argument("--color", default=None, metavar="#RRGGBB", help="New hex color")
    l_update.add_argument("--description", default=None, help="New description")
    l_delete = lbl_sub.add_parser("delete", help="Delete a label (by id or name)")
    l_delete.add_argument("label", help="Label id or name")

    # users
    p_users = sub.add_parser("users", help="List assignable users")
    p_users.add_argument("--json", action="store_true", help="JSON output")

    # agents (app users you can delegate to; auto-detected + cached)
    p_agents = sub.add_parser("agents", help="List agent members you can --delegate to")
    p_agents.add_argument("--refresh", action="store_true",
                          help="Force an immediate re-fetch (after installing/removing an agent app)")
    p_agents.add_argument("--json", action="store_true", help="JSON output")

    # migrate-agent-labels (one-time: legacy `agent:<name>` labels -> delegates)
    p_migrate = sub.add_parser(
        "migrate-agent-labels",
        help="One-time: move legacy agent:<name> ownership labels onto native delegates")
    p_migrate.add_argument("--apply", action="store_true",
                           help="Write the changes (default: dry run)")

    # states (workflow states for the team)
    p_states = sub.add_parser("states", help="List the team's workflow states")
    p_states.add_argument("--json", action="store_true", help="JSON output")

    # inbox (Linear notifications: comments, mentions, assignments, status changes)
    p_inbox = sub.add_parser("inbox", help="Show your Linear inbox (notifications), or mark them read")
    p_inbox.add_argument("--all", action="store_true",
                         help="Include already-read notifications (default: unread only)")
    p_inbox.add_argument("--limit", type=int, default=30,
                         help="Max notifications to fetch (default: 30)")
    p_inbox.add_argument("--json", action="store_true", help="JSON output")
    p_inbox.add_argument("--read", action="append", metavar="NOTIFICATION_ID",
                         help="Mark the given notification id(s) read (repeatable). id is in --json output.")
    p_inbox.add_argument("--read-all", action="store_true",
                         help="Mark ALL notifications read")

    # queue (durable closes when Linear is rate-limited)
    p_queue = sub.add_parser("queue",
                             help="Manage the durable local close queue")
    queue_sub = p_queue.add_subparsers(dest="queue_action")
    queue_sub.add_parser("list",
                         help="List queued close intents (same as bare queue)")
    q_drain = queue_sub.add_parser("drain",
                                   help="Apply queued closes with backoff")
    q_drain.add_argument("--dry-run", action="store_true",
                         help="List queued intents without applying them")
    q_drain.add_argument("--once", action="store_true",
                         help="Apply at most one due intent, then stop")

    args = parser.parse_args(rewrite_group_argv(sys.argv[1:]))

    if not args.command:
        parser.print_help()
        sys.exit(0)

    cfg = load_config()
    register_embeddings_ranker(cfg)

    if args.command == "setup":
        cmd_setup(args, cfg)
        return

    # All other commands need credentials
    api_key = get_api_key(cfg)
    team_id = get_team_id(cfg)

    if not api_key:
        print("No API key configured. Run: linear setup", file=sys.stderr)
        sys.exit(1)

    if args.team:
        override = resolve_team_id_by_key(api_key, args.team)
        if not override:
            sys.exit(1)
        team_id = override
        # Use a volatile cfg copy — don't pollute on-disk config with the
        # other team's states cache.
        cfg = {**cfg, "states": {}, "__volatile__": True}

    if not team_id:
        print("No team configured. Run: linear setup", file=sys.stderr)
        sys.exit(1)

    if args.command == "tasks":
        cmd_tasks(args, cfg, api_key, team_id)
    elif args.command == "update":
        cmd_update(args, cfg, api_key, team_id)
    elif args.command == "create":
        cmd_create(args, cfg, api_key, team_id)
    elif args.command == "cycles":
        cmd_cycles(args, cfg, api_key, team_id)
    elif args.command == "projects":
        cmd_projects(args, cfg, api_key, team_id)
    elif args.command == "milestones":
        cmd_milestones(args, cfg, api_key, team_id)
    elif args.command == "initiatives":
        cmd_initiatives(args, cfg, api_key, team_id)
    elif args.command == "goals":
        cmd_goals(args, cfg, api_key, team_id)
    elif args.command == "labels":
        cmd_labels(args, cfg, api_key, team_id)
    elif args.command == "users":
        cmd_users(args, cfg, api_key, team_id)
    elif args.command == "agents":
        cmd_agents(args, cfg, api_key, team_id)
    elif args.command == "migrate-agent-labels":
        cmd_migrate_agent_labels(args, cfg, api_key, team_id)
    elif args.command == "states":
        cmd_states(args, cfg, api_key, team_id)
    elif args.command == "inbox":
        cmd_inbox(args, cfg, api_key, team_id)
    elif args.command == "queue":
        cmd_queue(args, cfg, api_key, team_id)


if __name__ == "__main__":
    main()