"""
Delimit Ledger Manager — Strategy + Operational ledger as first-class MCP tools.

Two ledgers per project:
- Strategy: consensus decisions, positioning, pricing, product direction
- Operational: tasks, bugs, features — the "keep building" items

Ledger lives at {project}/.delimit/ledger/ (project-local).
Ventures auto-registered at ~/.delimit/ventures.json on first use.
"""

import base64
import json
import hashlib
import os
import re
import subprocess
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

GLOBAL_DIR = Path.home() / ".delimit"
VENTURES_FILE = GLOBAL_DIR / "ventures.json"

# LED-1145 Phase 2 #3: P0 quota soft warning. The soft block fires when an
# add_item call would push the unresolved-P0 count over the quota. Item is
# still added — this is policy nudge, not enforcement, per the strategic
# deliberation's hierarchy ("policy after primitives"). Override via env var:
#   DELIMIT_P0_SOFT_QUOTA=80     # raise the gate
#   DELIMIT_P0_SOFT_QUOTA=0      # disable warning entirely
P0_SOFT_QUOTA_DEFAULT = 50


def _p0_soft_quota() -> int:
    """Resolve the active P0 quota threshold from env. 0 disables warnings."""
    raw = os.environ.get("DELIMIT_P0_SOFT_QUOTA", "")
    if raw == "":
        return P0_SOFT_QUOTA_DEFAULT
    try:
        n = int(raw)
        return max(0, n)
    except (TypeError, ValueError):
        return P0_SOFT_QUOTA_DEFAULT


def _count_unresolved_p0(project_path: str = ".") -> int:
    """Count P0 items currently in an unresolved state (open / in_progress /
    blocked) across both ops and strategy ledgers. Excludes done / cancelled /
    archived. Used by the add_item soft-quota nudge."""
    ledger_dir = _project_ledger_dir(project_path)
    unresolved = {"open", "in_progress", "blocked"}
    count = 0
    for filename in ("operations.jsonl", "strategy.jsonl"):
        path = ledger_dir / filename
        if not path.exists():
            continue
        items = _read_ledger(path)
        # Replay events to current state
        state: Dict[str, Dict[str, Any]] = {}
        for item in items:
            iid = item.get("id", "")
            if not iid:
                continue
            if item.get("type") == "update":
                if iid in state:
                    if "status" in item:
                        state[iid]["status"] = item["status"]
                    if "priority" in item:
                        state[iid]["priority"] = item["priority"]
            else:
                state[iid] = {**item}
        for it in state.values():
            if it.get("priority") == "P0" and it.get("status") in unresolved:
                count += 1
    return count


def _detect_venture(project_path: str = ".") -> Dict[str, str]:
    """Auto-detect venture/project info from the directory."""
    p = Path(project_path).resolve()
    info = {"name": p.name, "path": str(p)}

    # Try package.json
    pkg = p / "package.json"
    if pkg.exists():
        try:
            d = json.loads(pkg.read_text())
            info["name"] = d.get("name", p.name)
            info["type"] = "node"
        except Exception:
            pass

    # Try pyproject.toml
    pyproj = p / "pyproject.toml"
    if pyproj.exists():
        try:
            text = pyproj.read_text()
            for line in text.splitlines():
                if line.strip().startswith("name"):
                    name = line.split("=", 1)[1].strip().strip('"').strip("'")
                    if name:
                        info["name"] = name
                        info["type"] = "python"
                        break
        except Exception:
            pass

    # Try git remote
    try:
        remote = subprocess.run(
            ["git", "remote", "get-url", "origin"],
            capture_output=True, text=True, timeout=3, cwd=str(p)
        )
        if remote.returncode == 0:
            url = _strip_url_userinfo(remote.stdout.strip())
            # Extract repo name from URL
            repo = url.rstrip("/").split("/")[-1].replace(".git", "")
            info["repo"] = url
            if not info.get("type"):
                info["name"] = repo
    except Exception:
        pass

    return info


# LED-3733: the canonical guard logic now lives in the reusable
# ai.registry_guards module so the swarm venture registry (and, later, the
# souls / agents-tasks stores) apply the identical, tested guard instead of
# a drifting copy. These thin private wrappers are kept for backward
# compatibility with existing imports (tests, callers).
from ai.registry_guards import strip_url_userinfo as _strip_url_userinfo  # noqa: E402
from ai.registry_guards import is_ephemeral_path as _is_ephemeral_path  # noqa: E402


def _register_venture(info: Dict[str, str]):
    """Silently register a venture in the global registry.

    Guard (STR-2169 / LED-3733, 2026-07-01): refuse to persist ephemeral
    paths (anything under /tmp or matching pytest/tempfile patterns) and
    strip any embedded credentials from the remote URL before writing.
    Previously only bare "/tmp" was rejected, so ~250 pytest tmp_path
    ventures accumulated and PAT-embedded remotes were stored verbatim.
    The guard fails-silently — tests that pass tmp_path to functions
    which auto-register simply don't pollute the registry going forward.
    """
    GLOBAL_DIR.mkdir(parents=True, exist_ok=True)
    ventures = {}
    if VENTURES_FILE.exists():
        try:
            ventures = json.loads(VENTURES_FILE.read_text())
        except Exception:
            pass

    name = info["name"]
    path = info.get("path", "")
    # Never auto-register transient test/scratch paths into the shared registry.
    if _is_ephemeral_path(path):
        return
    if name not in ventures:
        ventures[name] = {
            "path": path,
            "repo": _strip_url_userinfo(info.get("repo", "")),
            "type": info.get("type", ""),
            "registered_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        }
        VENTURES_FILE.write_text(json.dumps(ventures, indent=2))


# LED-1188 / Plan-C: env-aware home so DELIMIT_HOME / DELIMIT_NAMESPACE_ROOT
# overrides apply to the ledger paths same as everywhere else. Falls back
# to ~/.delimit when neither env var is set (back-compat with v4.5.1 and
# all prior versions).
def _delimit_home() -> Path:
    for env_key in ("DELIMIT_HOME", "DELIMIT_NAMESPACE_ROOT"):
        val = os.environ.get(env_key, "").strip()
        if val:
            return Path(val)
    return Path.home() / ".delimit"


CENTRAL_LEDGER_DIR = _delimit_home() / "ledger"
LEDGER_V2_DIR = _delimit_home() / "ledger-v2"

# LED-1188 D3 (deliberation att_f86e1f51110e8ed6 follow-up, 2026-04-28):
# Plan-C migration partitions the central ledger into per-venture sub-ledgers
# under ledger-v2/<slug>/. The resolver below auto-detects which layout is
# present and reads from it. Slugs match the migration script's canonical
# names so a v4.5.2 install picks up an existing Plan-C-staged tree without
# requiring the swap to happen first.
_VENTURE_CANONICAL = {
    "delimit-mcp": "delimit",
    "delimit-action": "delimit",
    "delimit-ui": "delimit",
    "delimit-cli": "delimit",        # npm package name
    "delimit-gateway": "delimit",    # gateway repo
    ".delimit": "delimit",
    "wirereport": "wire-report",
    "stakeone": "stake-one",
}
_KNOWN_VENTURE_SLUGS = {
    "delimit", "wire-report", "domainvested",
    "livetube", "stake-one", "root", "unsorted",
}


def _canonical_venture_slug(name: str) -> Optional[str]:
    """Map a detected venture name to a canonical sub-ledger slug.

    Returns None when the name doesn't match any known venture; callers
    treat that as "no per-venture sub-ledger, use the central layout."
    """
    if not name:
        return None
    n = name.lower().strip()
    n = _VENTURE_CANONICAL.get(n, n)
    return n if n in _KNOWN_VENTURE_SLUGS else None


def _detect_model() -> str:
    """Auto-detect which AI model is running this session.

    Checks environment variables set by various AI coding assistants:
    - CLAUDE_MODEL / CLAUDE_CODE_MODEL: Claude Code
    - CODEX_MODEL: OpenAI Codex CLI
    - GEMINI_MODEL: Gemini CLI
    - MCP_CLIENT_NAME: Generic MCP client identifier
    Falls back to "unknown" if none are set.
    """
    # Claude Code
    for var in ("CLAUDE_MODEL", "CLAUDE_CODE_MODEL"):
        val = os.environ.get(var)
        if val:
            return val

    # OpenAI Codex
    val = os.environ.get("CODEX_MODEL")
    if val:
        return val

    # Gemini
    val = os.environ.get("GEMINI_MODEL")
    if val:
        return val

    # Generic MCP client
    val = os.environ.get("MCP_CLIENT_NAME")
    if val:
        return val

    return "unknown"


def _project_ledger_dir(project_path: str = ".") -> Path:
    """Resolve the ledger directory for a project, with Plan-C auto-detect.

    Resolution order (LED-1188 D3, deliberation att_f86e1f51110e8ed6):
      1. Detect venture from project_path -> canonical slug (delimit,
         wire-report, domainvested, livetube, stake-one).
      2. If LEDGER_V2_DIR / <slug> / operations.jsonl exists, return that
         per-venture sub-ledger. (Plan-C staged but not yet swapped.)
      3. If CENTRAL_LEDGER_DIR / <slug> / operations.jsonl exists, return
         that per-venture sub-ledger. (Plan-C swapped.)
      4. Fall back to CENTRAL_LEDGER_DIR (legacy single-file layout).

    Cross-model handoff fix (still enforced): Codex and Gemini were writing to
    $PWD/.delimit/ledger/ which caused ledger fragmentation. The central
    ~/.delimit/ledger/ tree (or its Plan-C-partitioned form) remains the
    single source of truth — per-project .delimit/ dirs are for policies and
    config only.
    """
    # Quick exit: legacy callers that pass venture="" / project="." and
    # have no Plan-C tree on disk get the original single-file layout.
    if not LEDGER_V2_DIR.exists() and not (CENTRAL_LEDGER_DIR / "delimit").exists():
        return CENTRAL_LEDGER_DIR

    info = _detect_venture(project_path)
    slug = _canonical_venture_slug(info.get("name", ""))
    if slug is None:
        return CENTRAL_LEDGER_DIR

    # Plan-C staged: ledger-v2/<slug>/
    staged = LEDGER_V2_DIR / slug
    if (staged / "operations.jsonl").exists():
        return staged

    # Plan-C swapped: ledger/<slug>/
    swapped = CENTRAL_LEDGER_DIR / slug
    if (swapped / "operations.jsonl").exists():
        return swapped

    # No partitioned tree for this venture — fall back to the central
    # legacy layout (operations.jsonl + strategy.jsonl directly in ledger/).
    return CENTRAL_LEDGER_DIR


def _resolve_namespace(project_path: str = ".") -> Dict[str, str]:
    """Report which physical store a project_path resolves to (LED-3720).

    Read-only. Returns the detected venture name, the canonical namespace
    slug (or "" when the name maps to no known venture and the central
    legacy layout is used), and the absolute store directory that
    list/query/health will actually read. Exposed so callers and tests can
    prove that two aliases for the same project resolve to the SAME store —
    the fragmentation that let an item be reachable via one alias and not
    another. Does NOT move or mutate any data.
    """
    info = _detect_venture(project_path)
    slug = _canonical_venture_slug(info.get("name", "")) or ""
    return {
        "venture": info.get("name", ""),
        "namespace": slug,
        "store_path": str(_project_ledger_dir(project_path)),
    }


def _replay_status_counts(
    ledger_dir: Path,
    ledgers=("operations.jsonl", "strategy.jsonl"),
) -> Dict[str, int]:
    """Deterministic, event-sourced status tally for a resolved ledger dir.

    Single source of truth for the open / in_progress / blocked / done / …
    counts (LED-3720). Replays create + update rows in file order, honoring
    the LATEST status per item id, so an item whose status was updated to
    ``done`` is counted as done — NOT open. This mirrors ``list_items``'
    event-sourcing exactly (an update row is only applied to an item whose
    create row was already seen in the SAME store), so the number it returns
    equals an independent recount of the same store and matches the
    ``summary`` ``list_items`` computes.

    Read-only. Never counts a status value of ``None`` (an item with no
    status field lands in no bucket), matching list_items' breakdown.
    """
    counts: Dict[str, int] = {}
    for filename in ledgers:
        path = ledger_dir / filename
        if not path.exists():
            continue
        state: Dict[str, Optional[str]] = {}
        for item in _read_ledger(path):
            iid = item.get("id", "")
            if not iid:
                continue
            if item.get("type") == "update":
                if iid in state and "status" in item:
                    state[iid] = item.get("status")
            else:
                state[iid] = item.get("status")
        for status in state.values():
            if status is None:
                continue
            counts[status] = counts.get(status, 0) + 1
    return counts


def _ensure(project_path: str = "."):
    ledger_dir = _project_ledger_dir(project_path)
    ledger_dir.mkdir(parents=True, exist_ok=True)
    for name in ["strategy.jsonl", "operations.jsonl"]:
        f = ledger_dir / name
        if not f.exists():
            f.write_text("")

    # Auto-register venture on first use
    info = _detect_venture(project_path)
    _register_venture(info)


def _read_ledger(path: Path) -> List[Dict]:
    items = []
    if not path.exists():
        return items
    for line in path.read_text().splitlines():
        line = line.strip()
        if line:
            try:
                items.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return items


def _append(path: Path, entry: Dict) -> Dict:
    path.parent.mkdir(parents=True, exist_ok=True)
    if not path.exists():
        path.write_text("")
    items = _read_ledger(path)
    prev_hash = items[-1].get("hash", "genesis") if items else "genesis"
    entry["hash"] = hashlib.sha256(f"{prev_hash}{json.dumps(entry, sort_keys=True)}".encode()).hexdigest()[:16]
    entry["created_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ")

    with open(path, "a") as f:
        f.write(json.dumps(entry) + "\n")
    return entry


# ── LED-877 signal guard ─────────────────────────────────────────────


# Sources that originate from sensed observations (social/strategy scans).
# Centralized so the guard logic is in one place even if more prefixes are
# added later (e.g. github_sense, reddit_sense).
_SENSED_SOURCE_PREFIXES = ("social_scan", "social_strategy")


def _check_source_is_ledger_item(
    source: str,
    *,
    purpose: str = "promote_to_ledger",
    title: str = "",
    ledger: str = "",
) -> None:
    """LED-877 signal guard.

    Sensed observations (``source='social_scan:...'``) MUST NOT land in
    the ledger by default — they belong in the intel signal store.

    LED-216 Phase 1 split: callers can declare *why* they are checking.

    ``purpose='promote_to_ledger'`` (default, original strict behavior)
        Used by ``add_item`` and any code path that actually writes a
        ledger row. Raises ``ValueError`` on a sensed source unless the
        ``_DELIMIT_SIGNAL_PROMOTED_BY`` bypass env var is set (which is
        the explicit promote-to-ledger path).

    ``purpose='draft_only'``
        Used by code paths that produce a reply draft from a sensed
        observation but do NOT promote the signal to the ledger. Drafts
        are an acceptable consumer of sensed sources, so the guard is a
        no-op for this purpose.

    Both purposes still respect ``DELIMIT_SIGNAL_GUARD=shadow`` for the
    shadow-log fallback used during the LED-877 rollout.
    """
    _src_norm = (source or "").strip().lower()
    if not any(_src_norm.startswith(p) for p in _SENSED_SOURCE_PREFIXES):
        return  # Not a sensed source; nothing to guard against.

    if purpose == "draft_only":
        # Drafts may legitimately reference a sensed observation. The
        # guard exists to prevent ledger writes, not draft generation.
        return

    if purpose != "promote_to_ledger":
        # Defensive: unknown purpose ⇒ default to strict behavior so a
        # typo can't accidentally weaken the guard.
        pass

    _promoted_by = os.environ.get("_DELIMIT_SIGNAL_PROMOTED_BY", "")
    if _promoted_by:
        return  # Explicit promote_to_ledger path; bypass authorized.

    msg = (
        f"LED-877 guard: source={source!r} is a sensed observation, not "
        f"a ledger item. Use ai.sensing.signal_store.ingest() instead. "
        f"Promote explicitly via promote_to_ledger(signal_id=...)."
    )

    _guard_mode = os.environ.get("DELIMIT_SIGNAL_GUARD", "enforce").lower()
    if _guard_mode == "shadow":
        try:
            _shadow_log = Path.home() / ".delimit" / "logs" / "signal_guard_shadow.jsonl"
            _shadow_log.parent.mkdir(parents=True, exist_ok=True)
            with _shadow_log.open("a") as _f:
                _f.write(json.dumps({
                    "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
                    "title": title,
                    "source": source,
                    "ledger": ledger,
                    "purpose": purpose,
                    "msg": msg,
                }) + "\n")
        except Exception:
            pass
        # fall through (shadow mode does not raise)
        return

    raise ValueError(msg)


def add_item(
    title: str,
    ledger: str = "ops",
    type: str = "task",
    priority: str = "P1",
    description: str = "",
    source: str = "session",
    project_path: str = ".",
    tags: Optional[List[str]] = None,
    acceptance_criteria: Optional[List[str]] = None,
    context: str = "",
    tools_needed: Optional[List[str]] = None,
    estimated_complexity: str = "",
    worked_by: str = "",
    decision_card: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Add a new item to the project's strategy or operational ledger.

    LED-189: Items can have acceptance_criteria (testable "done when" conditions).
    LED-190: Items can have context, tools_needed, and estimated_complexity
    for agent-executable task format.
    LED-877: Signal guard — rejects source='social_scan' writes so sensed
    observations cannot land in the ledger. Observations belong in the intel
    signal store (ai/sensing/signal_store.py). Bypass via env var for the
    promote_to_ledger path: _DELIMIT_SIGNAL_PROMOTED_BY=<who>.

    LED-216 Phase 1: the guard is now reusable via
    ``_check_source_is_ledger_item(..., purpose='draft_only')`` for code
    paths that produce reply drafts from sensed observations without
    promoting the underlying signal to the ledger.
    """
    _check_source_is_ledger_item(
        source,
        purpose="promote_to_ledger",
        title=title,
        ledger=ledger,
    )

    _ensure(project_path)
    venture = _detect_venture(project_path)
    ledger_dir = _project_ledger_dir(project_path)
    path = ledger_dir / ("strategy.jsonl" if ledger == "strategy" else "operations.jsonl")

    # LED-824: ID-collision fix. The Plan-C resolver routes delimit-context
    # queries to ledger-v2/<slug>/. Per-venture ID counters used to scan
    # only the active sub-ledger, so newly-created items could collide with
    # IDs already used in the legacy CENTRAL_LEDGER_DIR root files. Now we
    # union all known IDs across (a) the resolved sub-ledger AND (b) every
    # peer sub-ledger AND (c) the legacy root, then pick the next free.
    items = _read_ledger(path)
    prefix = "STR" if ledger == "strategy" else "LED"
    existing_ids = {i.get("id", "") for i in items if i.get("type") != "update"}

    # Union with all peer files in ledger-v2/* and the legacy root files,
    # for both strategy and operations ledgers (an LED-N could collide
    # whether it lives in operations or strategy in any sub-ledger).
    filename = "strategy.jsonl" if ledger == "strategy" else "operations.jsonl"
    candidate_paths: list[Path] = []
    if LEDGER_V2_DIR.exists():
        for sub in LEDGER_V2_DIR.iterdir():
            if sub.is_dir():
                candidate_paths.append(sub / filename)
    candidate_paths.append(CENTRAL_LEDGER_DIR / filename)

    for cand in candidate_paths:
        if cand == path:
            continue  # already scanned
        if not cand.exists():
            continue
        try:
            for entry in _read_ledger(cand):
                if entry.get("type") == "update":
                    continue
                eid = entry.get("id", "")
                if eid:
                    existing_ids.add(eid)
        except Exception:
            # Best-effort: a malformed peer file shouldn't block id assignment
            continue

    # Walk forward from len()+1 until we find a non-colliding slot.
    num = len(items) + 1
    while f"{prefix}-{num:03d}" in existing_ids:
        num += 1
    item_id = f"{prefix}-{num:03d}"

    entry = {
        "id": item_id,
        "title": title,
        "type": type,
        "priority": priority,
        "description": description,
        "source": source,
        "venture": venture["name"],
        "status": "open",
        "tags": tags or [],
        "worked_by": worked_by or _detect_model(),
    }
    # LED-189: Optional acceptance criteria
    if acceptance_criteria:
        entry["acceptance_criteria"] = acceptance_criteria
    # LED-190: Optional agent-executable fields
    if context:
        entry["context"] = context
    if tools_needed:
        entry["tools_needed"] = tools_needed
    if estimated_complexity:
        entry["estimated_complexity"] = estimated_complexity
    # Phase-0 autonomous-org (mem-b2a9762b78d8): a decision card is a ledger
    # item of type="decision" carrying the founder-facing option set. The
    # payload is stored on the initial (non-update) event so list_items' replay
    # ({**item}) preserves it. Additive — items without a decision_card are
    # unchanged. Only attached when type == "decision" to avoid polluting
    # ordinary tasks.
    if decision_card and type == "decision":
        entry["decision_card"] = decision_card

    result = _append(path, entry)

    # Sync to Supabase for dashboard visibility
    try:
        from ai.supabase_sync import sync_ledger_item
        sync_ledger_item(result)
    except Exception:
        pass  # Never let cloud sync break ledger operations

    response: Dict[str, Any] = {
        "added": result,
        "ledger": ledger,
        "venture": venture["name"],
        "total_items": len(_read_ledger(path)),
    }

    # LED-1145 Phase 2 #3: P0 soft quota nudge. Soft (item still added),
    # not hard. Surfaces a warning when the unresolved-P0 count crosses
    # the quota — gives the founder a signal to groom before piling on.
    if priority == "P0":
        quota = _p0_soft_quota()
        if quota > 0:
            current_p0 = _count_unresolved_p0(project_path)
            if current_p0 > quota:
                response["warning"] = (
                    f"P0 quota soft-block: {current_p0} unresolved P0 items "
                    f"(threshold {quota}). Item was still added. Consider running "
                    f"delimit_ledger_groom to triage existing P0s before adding more, "
                    f"or set DELIMIT_P0_SOFT_QUOTA={current_p0 + 50} to raise the gate."
                )
                response["p0_count"] = current_p0
                response["p0_quota"] = quota

    return response


def _find_item_in_ledger_dir(item_id: str, ledger_dir: Path) -> Optional[Dict[str, Any]]:
    """Search a ledger directory for an item by ID. Returns (ledger_name, path) or None."""
    for ledger_name, filename in [("ops", "operations.jsonl"), ("strategy", "strategy.jsonl")]:
        path = ledger_dir / filename
        items = _read_ledger(path)
        for item in items:
            if item.get("id") == item_id and item.get("type") != "update":
                return {"ledger_name": ledger_name, "path": path}
    return None


def update_item(
    item_id: str,
    status: Optional[str] = None,
    note: Optional[str] = None,
    priority: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    assignee: Optional[str] = None,
    due_date: Optional[str] = None,
    labels: Optional[List[str]] = None,
    blocked_by: Optional[str] = None,
    blocks: Optional[str] = None,
    project_path: str = ".",
    worked_by: str = "",
    commit_sha: Optional[str] = None,
    pr_url: Optional[str] = None,
) -> Dict[str, Any]:
    """Update an existing ledger item's fields.

    LED-1408 Phase 1: when `status="done"` is requested, callers MAY provide
    `commit_sha` and/or `pr_url` as proof that the work shipped to main.
    The proof is recorded on the update event under `ship_proof` with a
    `verified: bool` flag. Phase 1 does NOT enforce — items still
    transition to `done` even without proof — but the flag lets future
    audits and the Phase 2 reconciler find unverified-done items.
    """
    _ensure(project_path)
    ledger_dir = _project_ledger_dir(project_path)

    # First, search the specified project's ledger
    found = _find_item_in_ledger_dir(item_id, ledger_dir)

    # If not found, search all registered ventures as a fallback
    if not found:
        try:
            ventures = {}
            if VENTURES_FILE.exists():
                ventures = json.loads(VENTURES_FILE.read_text())
        except Exception:
            ventures = {}

        searched = {str(ledger_dir)}
        for _name, info in ventures.items():
            vpath = info.get("path", "")
            if not vpath:
                continue
            candidate_dir = Path(vpath) / ".delimit" / "ledger"
            if str(candidate_dir) in searched:
                continue
            searched.add(str(candidate_dir))
            found = _find_item_in_ledger_dir(item_id, candidate_dir)
            if found:
                break

    if found:
        ledger_name = found["ledger_name"]
        path = found["path"]
        update = {
            "id": item_id,
            "type": "update",
            "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "worked_by": worked_by or _detect_model(),
        }
        if status:
            update["status"] = status
        if note:
            update["note"] = note
        if priority:
            update["priority"] = priority
        if title:
            update["title"] = title
        if description:
            update["description"] = description
        if assignee:
            update["assignee"] = assignee
        if due_date:
            update["due_date"] = due_date
        if labels is not None:
            # LED-2221: write to both `labels` and `tags`. The list_items
            # reconstruction (around line ~870) merges update events into
            # current state by checking the `tags` key only. Writing only
            # `labels` silently drops the update at read time, which in
            # particular meant the build daemon's `autonomous-build` tag
            # check could never see tags written through the MCP. Keeping
            # `labels` for any external consumer that reads the raw event
            # stream; adding `tags` so the live state aggregator picks it up.
            update["labels"] = labels
            update["tags"] = labels
        if blocked_by:
            update["blocked_by"] = blocked_by
        if blocks:
            update["blocks"] = blocks

        # LED-1408 Phase 1: attach ship_proof block when status transitions to
        # `done` or `shipped_pending`. Verified=True iff commit_sha or pr_url
        # was supplied (directly or scraped from the note). Phase 2's
        # reconciler will use this to distinguish "trustworthy done" from
        # "marked done but never verified on main."
        if status in ("done", "shipped_pending"):
            try:
                from ai.ledger_proof import build_ship_proof
                update["ship_proof"] = build_ship_proof(
                    commit_sha=commit_sha,
                    pr_url=pr_url,
                    note=note,
                )
                # LED-1420 Phase 2 strict-mode flip: when DELIMIT_LEDGER_STRICT_DONE=1,
                # an unverified `done` transition is downgraded to `shipped_pending`
                # so the nightly reconciler (scripts/delimit_ledger_reconciler.py)
                # can promote it to `done` once a commit-trailer match shows up on
                # origin/main. Off by default so existing workflows keep closing
                # items without hitting an unexpected gate; flip when the
                # reconciler has been observed running for ~1 week without
                # surprises.
                strict = os.environ.get("DELIMIT_LEDGER_STRICT_DONE") == "1"
                if (
                    strict
                    and status == "done"
                    and not update["ship_proof"].get("verified")
                ):
                    update["status"] = "shipped_pending"
                    existing_note = update.get("note") or ""
                    suffix = (
                        "[LED-1420 strict-mode: downgraded done → shipped_pending — "
                        "no commit_sha/pr_url proof; reconciler will upgrade to "
                        "done when it finds a Ledger-Item: " + item_id + " trailer "
                        "on origin/main]"
                    )
                    update["note"] = (existing_note + " " + suffix).strip() if existing_note else suffix
            except Exception:
                # Soft-fail: a ship_proof bug must not break ledger close.
                # The unverified state will be re-detectable from the missing
                # key on the next audit pass.
                pass

        _append(path, update)

        # Sync to Supabase for dashboard visibility
        try:
            from ai.supabase_sync import sync_ledger_update
            sync_ledger_update(item_id, status=status or "", note=note or "")
        except Exception:
            pass  # Never let cloud sync break ledger operations

        return {"updated": item_id, "changes": update, "ledger": ledger_name}

    return {"error": f"Item {item_id} not found in project ledger"}


# LED-1145 Phase 1 PR-A: known-good slim projection for AI agent triage.
# Strips description / acceptance_criteria / context / tags / hash etc. so the
# response fits in an MCP tool result without truncation. Default behaviour
# stays full (callers depending on description still get it); pass
# fields="slim" or an explicit allowlist to opt in.
SLIM_FIELDS = ("id", "title", "status", "priority", "type", "venture", "updated_at")
_VALID_FIELDS = SLIM_FIELDS + (
    "description", "acceptance_criteria", "context", "tags", "created_at",
    "worked_by", "last_worked_by", "last_note", "hash", "source", "tools_needed",
    "estimated_complexity", "ledger",
)
_VALID_SORT = ("updated_at", "created_at", "priority")
_VALID_ORDER = ("asc", "desc")
_PRIORITY_ORDER = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}


def _normalize_filter_list(value):
    """Accept None / str / list and produce a list of strings (or None)."""
    if value is None:
        return None
    if isinstance(value, str):
        if not value.strip():
            return None
        return [v.strip() for v in value.split(",") if v.strip()]
    return list(value) if value else None


def _resolve_fields(fields):
    """Map the `fields` parameter to a concrete projection set.

    Returns:
        (projection: set[str] | None, error: str | None)
        - projection=None means "return all fields" (backward-compat default)
        - error is set when an unknown field name was requested
    """
    if fields is None or fields == "":
        return None, None
    if isinstance(fields, str):
        if fields == "slim":
            return set(SLIM_FIELDS), None
        if fields == "*":
            return None, None
        # comma-separated string from the MCP boundary
        names = [f.strip() for f in fields.split(",") if f.strip()]
    else:
        names = list(fields)
    if not names:
        return None, None
    if names == ["*"]:
        return None, None
    if names == ["slim"]:
        return set(SLIM_FIELDS), None
    unknown = [n for n in names if n not in _VALID_FIELDS and n != "*"]
    if unknown:
        return None, f"unknown field(s) requested: {sorted(unknown)}; valid: {sorted(_VALID_FIELDS)}"
    return set(names), None


def _ts_to_iso(value):
    """Coerce a timestamp value into a comparable ISO string. Empty → ''."""
    return value or ""


def _compare_iso(a: str, b: str, op: str) -> bool:
    """Lexical ISO comparison; both sides must be naively-comparable strings.
    Empty string sorts before any real timestamp so '<X' is True for missing."""
    if op == "before":
        return a < b
    if op == "after":
        return a > b
    return False


def _make_cursor(offset: int, filter_sig: str) -> str:
    payload = f"{offset}:{filter_sig}"
    return base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii")


def _decode_cursor(cursor: str) -> tuple[int, str]:
    try:
        decoded = base64.urlsafe_b64decode(cursor.encode("ascii")).decode("utf-8")
        offset_str, filter_sig = decoded.split(":", 1)
        return int(offset_str), filter_sig
    except (ValueError, UnicodeDecodeError):
        return 0, ""


def _filter_signature(spec: Dict[str, Any]) -> str:
    """Short, deterministic hash of the filter spec; used to invalidate cursors
    when the caller changes filters between page requests."""
    canonical = json.dumps(spec, sort_keys=True, default=str)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12]


def list_items(
    ledger: str = "both",
    # Backward-compat single-value filters:
    status: Optional[str] = None,
    priority: Optional[str] = None,
    # New multi-value filters:
    status__in=None,
    priority__in=None,
    tags__contains_all=None,
    text: Optional[str] = None,
    linked_external_id: Optional[str] = None,
    created_before: Optional[str] = None,
    created_after: Optional[str] = None,
    updated_before: Optional[str] = None,
    updated_after: Optional[str] = None,
    # Sort + projection + pagination:
    sort: str = "updated_at",
    order: str = "desc",
    fields=None,
    limit: int = 50,
    cursor: Optional[str] = None,
    project_path: str = ".",
) -> Dict[str, Any]:
    """List ledger items with optional filters, sort, projection, and cursor pagination.

    LED-1145 Phase 1 PR-A: extended from the original 3-filter signature.
    Backward compatible — old callers passing only `status` / `priority`
    continue to work without change.

    Args:
        ledger: "ops" | "strategy" | "both".
        status: single-value status filter (back-compat).
        priority: single-value priority filter (back-compat).
        status__in: list (or comma-separated string) of statuses to match.
        priority__in: list (or comma-separated string) of priorities to match.
        tags__contains_all: list (or comma-separated string); item must contain ALL these tags.
        text: case-insensitive substring match against title + description.
        linked_external_id: exact substring match in description or tags
            (e.g. "github.com/owner/repo/issues/123").
        created_before / created_after / updated_before / updated_after:
            ISO timestamps (e.g. "2026-04-01T00:00:00Z"). Lexical compare.
        sort: "updated_at" | "created_at" | "priority". Default updated_at.
        order: "asc" | "desc". Default desc.
        fields: response projection. None or "*" = full (default, back-compat).
            "slim" = SLIM_FIELDS only. List/CSV of field names = those only.
            Unknown field names → ERROR (no silent no-op).
        limit: page size (default 50).
        cursor: opaque pagination token from a prior call's `next_cursor`.
            If filters change between calls, the cursor is invalidated and the
            response begins at offset 0 with `cursor_invalidated=True`.
        project_path: ledger root (auto-detect when ".").

    Returns:
        {
            "venture": str,
            "items": {"ops": [...], "strategy": [...]},
            "summary": {"total": int, "open": int, "done": int, "in_progress": int},
            "next_cursor": str | None,
            "cursor_invalidated": bool (only when True),
        }
    """
    _ensure(project_path)
    ledger_dir = _project_ledger_dir(project_path)
    venture = _detect_venture(project_path)

    if sort not in _VALID_SORT:
        return {"error": f"sort must be one of {list(_VALID_SORT)}"}
    if order not in _VALID_ORDER:
        return {"error": f"order must be one of {list(_VALID_ORDER)}"}

    projection, projection_err = _resolve_fields(fields)
    if projection_err:
        return {"error": projection_err}

    # Normalise filter list params (accept str / list / None)
    status_list = _normalize_filter_list(status__in)
    priority_list = _normalize_filter_list(priority__in)
    tags_list = _normalize_filter_list(tags__contains_all)

    # Backward compat: status="open" → status_list=["open"]; same for priority.
    if status and not status_list:
        status_list = [status]
    if priority and not priority_list:
        priority_list = [priority]

    # Filter signature for cursor invalidation
    filter_spec = {
        "ledger": ledger,
        "status_list": sorted(status_list) if status_list else None,
        "priority_list": sorted(priority_list) if priority_list else None,
        "tags_list": sorted(tags_list) if tags_list else None,
        "text": text,
        "linked_external_id": linked_external_id,
        "created_before": created_before,
        "created_after": created_after,
        "updated_before": updated_before,
        "updated_after": updated_after,
        "sort": sort,
        "order": order,
    }
    current_sig = _filter_signature(filter_spec)

    # Decode cursor if provided
    start_offset = 0
    cursor_invalidated = False
    if cursor:
        decoded_offset, decoded_sig = _decode_cursor(cursor)
        if decoded_sig == current_sig:
            start_offset = decoded_offset
        else:
            cursor_invalidated = True

    text_lower = (text or "").lower() if text else None

    # LED-3739: the status breakdown in `summary` must be independent of the
    # `status`/`status__in` filter, otherwise callers that filter by status
    # (get_context uses status="open"; health_summary uses
    # status__in=[open,in_progress,blocked]) get a self-referential breakdown
    # where the filtered-out statuses read 0. That produced the reported
    # divergence (health said 43 in_progress, context said 0 in_progress on the
    # SAME store). We accumulate items that pass every filter EXCEPT the status
    # filter and compute the breakdown over that scoped set, so the status
    # distribution is authoritative regardless of the status filter.
    breakdown_pool: list = []

    results: Dict[str, list] = {}
    for ledger_name, filename in [("ops", "operations.jsonl"), ("strategy", "strategy.jsonl")]:
        if ledger not in ("both", ledger_name):
            continue

        path = ledger_dir / filename
        items = _read_ledger(path)

        # Build current state by replaying events (event-sourced)
        state: Dict[str, Dict[str, Any]] = {}
        for item in items:
            item_id = item.get("id", "")
            if item.get("type") == "update":
                if item_id in state:
                    if "status" in item:
                        state[item_id]["status"] = item["status"]
                    if "note" in item:
                        state[item_id]["last_note"] = item["note"]
                    if "priority" in item:
                        state[item_id]["priority"] = item["priority"]
                    if "worked_by" in item:
                        state[item_id]["last_worked_by"] = item["worked_by"]
                    if "tags" in item and item["tags"] is not None:
                        # Tag updates replace the existing tag set when present
                        state[item_id]["tags"] = item["tags"]
                    state[item_id]["updated_at"] = item.get("updated_at")
            else:
                state[item_id] = {**item}

        filtered = list(state.values())

        # Apply every filter EXCEPT status first, so the status breakdown can be
        # computed over this scoped-but-status-agnostic set.
        if priority_list:
            priorities = set(priority_list)
            filtered = [i for i in filtered if i.get("priority") in priorities]
        if tags_list:
            required_tags = set(tags_list)
            filtered = [
                i for i in filtered
                if required_tags.issubset(set(i.get("tags") or []))
            ]
        if text_lower:
            filtered = [
                i for i in filtered
                if text_lower in (i.get("title") or "").lower()
                or text_lower in (i.get("description") or "").lower()
            ]
        if linked_external_id:
            needle = linked_external_id
            filtered = [
                i for i in filtered
                if needle in (i.get("description") or "")
                or needle in " ".join(i.get("tags") or [])
                or needle in (i.get("context") or "")
            ]
        if created_before:
            filtered = [i for i in filtered if _compare_iso(_ts_to_iso(i.get("created_at")), created_before, "before")]
        if created_after:
            filtered = [i for i in filtered if _compare_iso(_ts_to_iso(i.get("created_at")), created_after, "after")]
        if updated_before:
            filtered = [i for i in filtered if _compare_iso(_ts_to_iso(i.get("updated_at") or i.get("created_at")), updated_before, "before")]
        if updated_after:
            filtered = [i for i in filtered if _compare_iso(_ts_to_iso(i.get("updated_at") or i.get("created_at")), updated_after, "after")]

        # Snapshot the scoped set (all filters except status) for the breakdown.
        breakdown_pool.extend(filtered)

        # Now apply the status filter for the returned items + pagination total.
        if status_list:
            statuses = set(status_list)
            filtered = [i for i in filtered if i.get("status") in statuses]

        # Sort
        reverse = order == "desc"
        if sort == "priority":
            filtered.sort(
                key=lambda x: (
                    _PRIORITY_ORDER.get(x.get("priority", "P2"), 9),
                    x.get("created_at", ""),
                ),
                reverse=reverse,
            )
        else:
            sort_key = "updated_at" if sort == "updated_at" else "created_at"
            filtered.sort(
                key=lambda x: x.get(sort_key) or x.get("created_at") or "",
                reverse=reverse,
            )

        results[ledger_name] = filtered

    # Apply projection + pagination across the combined result.
    # Combine per-ledger lists in stable order (ops first, then strategy).
    combined: list = []
    for ledger_name in ("ops", "strategy"):
        if ledger_name in results:
            for it in results[ledger_name]:
                # Tag each item with its ledger source so the projected
                # response retains the routing info even when "ledger" itself
                # isn't part of the original record.
                if "ledger" not in it:
                    it = {**it, "ledger": ledger_name}
                combined.append(it)

    total_pre_page = len(combined)
    page = combined[start_offset:start_offset + limit]
    next_offset = start_offset + len(page)
    has_more = next_offset < total_pre_page
    next_cursor = _make_cursor(next_offset, current_sig) if has_more else None

    # Apply projection to page items.
    if projection is not None:
        page = [{k: v for k, v in i.items() if k in projection} for i in page]

    # Re-bucket projected page items back into ops / strategy for response shape.
    paged_results: Dict[str, list] = {"ops": [], "strategy": []}
    if ledger != "strategy":
        paged_results.setdefault("ops", [])
    if ledger != "ops":
        paged_results.setdefault("strategy", [])
    # Walk the page using the still-tagged combined data to know which bucket;
    # we kept "ledger" in the projection step only when it was already there.
    for src, dst in zip(combined[start_offset:start_offset + limit], page):
        bucket = src.get("ledger") or "ops"
        paged_results.setdefault(bucket, []).append(dst)

    summary_total = total_pre_page
    # Status breakdown is computed over breakdown_pool (all filters EXCEPT the
    # status filter) so it is authoritative and identical for callers that
    # differ only in their status filter (LED-3739). `total` remains the
    # post-status-filter count for pagination + the health "unresolved" count.
    response = {
        "venture": venture["name"],
        "items": {k: v for k, v in paged_results.items() if k in results},
        "summary": {
            "total": summary_total,
            "open": sum(1 for i in breakdown_pool if i.get("status") == "open"),
            "done": sum(1 for i in breakdown_pool if i.get("status") == "done"),
            "in_progress": sum(1 for i in breakdown_pool if i.get("status") == "in_progress"),
            "blocked": sum(1 for i in breakdown_pool if i.get("status") == "blocked"),
            "archived": sum(1 for i in breakdown_pool if i.get("status") == "archived"),
        },
        "next_cursor": next_cursor,
    }
    if cursor_invalidated:
        response["cursor_invalidated"] = True
    return response


def get_context(project_path: str = ".") -> Dict[str, Any]:
    """Get a concise ledger summary for AI context — what's open, what's next."""
    venture = _detect_venture(project_path)
    result = list_items(status="open", project_path=project_path)
    open_items = []
    for ledger_items in result["items"].values():
        open_items.extend(ledger_items)

    priority_order = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}
    open_items.sort(key=lambda x: priority_order.get(x.get("priority", "P2"), 9))

    return {
        "venture": venture["name"],
        "open_items": len(open_items),
        "next_up": [{"id": i["id"], "title": i["title"], "priority": i["priority"]}
                     for i in open_items[:5]],
        "summary": result["summary"],
    }


def list_ventures() -> Dict[str, Any]:
    """List all registered ventures/projects."""
    if not VENTURES_FILE.exists():
        return {"ventures": {}, "count": 0}
    try:
        ventures = json.loads(VENTURES_FILE.read_text())
        return {"ventures": ventures, "count": len(ventures)}
    except Exception:
        return {"ventures": {}, "count": 0}


# ═══════════════════════════════════════════════════════════════════════
#  LEDGER QUERY (Natural language → structured queries)
# ═══════════════════════════════════════════════════════════════════════

def query_ledger(query: str, project_path: str = ".") -> Dict[str, Any]:
    """Answer natural language questions about the ledger.

    Supports: "what shipped this week?", "what's blocked?", "show P0s",
    "how many items completed?", "what's next?", etc.
    """
    q = query.lower().strip()

    # Route to appropriate data based on query intent
    if any(w in q for w in ["shipped", "completed", "done", "finished", "closed"]):
        result = list_items(status="done", project_path=project_path, limit=50)
        items = []
        for v in result.get("items", {}).values():
            items.extend(v)

        # Filter by time if mentioned
        if "today" in q:
            today = time.strftime("%Y-%m-%d")
            items = [i for i in items if i.get("updated_at", "").startswith(today) or i.get("created_at", "").startswith(today)]
        elif "week" in q or "7 day" in q:
            cutoff = time.time() - 7 * 86400
            items = [i for i in items if _parse_ts(i.get("updated_at", "")) > cutoff]
        elif "month" in q or "30 day" in q:
            cutoff = time.time() - 30 * 86400
            items = [i for i in items if _parse_ts(i.get("updated_at", "")) > cutoff]

        return {"query": query, "intent": "completed", "items": [{"id": i["id"], "title": i["title"]} for i in items], "count": len(items)}

    elif any(w in q for w in ["blocked", "blocking", "stuck"]):
        result = list_items(status="open", project_path=project_path, limit=50)
        items = []
        for v in result.get("items", {}).values():
            items.extend(v)
        # Check for items with blocked_by links
        blocked = []
        for i in items:
            links = get_links(i["id"], project_path)
            has_blocker = any(l.get("type") == "blocked_by" for l in links.get("links", []))
            if has_blocker or i.get("status") == "blocked":
                blocked.append(i)
        return {"query": query, "intent": "blocked", "items": [{"id": i["id"], "title": i["title"]} for i in blocked], "count": len(blocked)}

    elif any(w in q for w in ["next", "should i", "what to work", "priority", "urgent"]):
        return get_context(project_path)

    elif "p0" in q:
        result = list_items(priority="P0", project_path=project_path, limit=20)
        items = []
        for v in result.get("items", {}).values():
            items.extend(v)
        return {"query": query, "intent": "priority_filter", "priority": "P0", "items": [{"id": i["id"], "title": i["title"], "status": i.get("status", "open")} for i in items], "count": len(items)}

    elif "p1" in q:
        result = list_items(priority="P1", project_path=project_path, limit=20)
        items = []
        for v in result.get("items", {}).values():
            items.extend(v)
        return {"query": query, "intent": "priority_filter", "priority": "P1", "items": [{"id": i["id"], "title": i["title"], "status": i.get("status", "open")} for i in items], "count": len(items)}

    elif any(w in q for w in ["how many", "count", "total", "stats", "summary"]):
        result = list_items(project_path=project_path, limit=500)
        all_items = []
        for v in result.get("items", {}).values():
            all_items.extend(v)
        by_status = {}
        by_priority = {}
        by_venture = {}
        for i in all_items:
            s = i.get("status", "open")
            by_status[s] = by_status.get(s, 0) + 1
            p = i.get("priority", "P1")
            by_priority[p] = by_priority.get(p, 0) + 1
            v = i.get("venture", "unknown")
            by_venture[v] = by_venture.get(v, 0) + 1
        return {"query": query, "intent": "stats", "total": len(all_items), "by_status": by_status, "by_priority": by_priority, "by_venture": by_venture}

    elif any(w in q for w in ["open", "todo", "remaining", "left"]):
        return get_context(project_path)

    else:
        # Default: search by keyword in titles
        result = list_items(project_path=project_path, limit=100)
        all_items = []
        for v in result.get("items", {}).values():
            all_items.extend(v)
        words = q.split()
        matches = [i for i in all_items if any(w in i.get("title", "").lower() for w in words)]
        return {"query": query, "intent": "search", "items": [{"id": i["id"], "title": i["title"], "status": i.get("status")} for i in matches[:20]], "count": len(matches)}


def _parse_ts(ts_str: str) -> float:
    """Parse ISO timestamp to epoch seconds."""
    try:
        import datetime
        dt = datetime.datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
        return dt.timestamp()
    except Exception:
        return 0


# ═══════════════════════════════════════════════════════════════════════
#  LEDGER LINKS (Dependencies, Blockers, Parent-Child)
# ═══════════════════════════════════════════════════════════════════════

LINKS_FILE_NAME = "links.jsonl"
VALID_LINK_TYPES = {"blocks", "blocked_by", "parent", "child", "relates_to", "duplicates", "supersedes", "superseded_by"}


def link_items(
    from_id: str,
    to_id: str,
    link_type: str = "blocks",
    note: str = "",
    project_path: str = ".",
) -> Dict[str, Any]:
    """Create a relationship between two ledger items."""
    if link_type not in VALID_LINK_TYPES:
        return {"error": f"Invalid link_type '{link_type}'. Use: {', '.join(sorted(VALID_LINK_TYPES))}"}

    _ensure(project_path)
    ledger_dir = _project_ledger_dir(project_path)
    links_file = ledger_dir / LINKS_FILE_NAME

    link = {
        "from": from_id,
        "to": to_id,
        "type": link_type,
        "note": note,
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }

    with open(links_file, "a") as f:
        f.write(json.dumps(link) + "\n")

    # Auto-create reverse link for bidirectional types
    reverse_map = {
        "blocks": "blocked_by",
        "blocked_by": "blocks",
        "parent": "child",
        "child": "parent",
        "supersedes": "superseded_by",
        "superseded_by": "supersedes",
    }
    if link_type in reverse_map:
        reverse = {
            "from": to_id,
            "to": from_id,
            "type": reverse_map[link_type],
            "note": note,
            "created_at": link["created_at"],
            "auto_reverse": True,
        }
        with open(links_file, "a") as f:
            f.write(json.dumps(reverse) + "\n")

    return {"linked": True, "from": from_id, "to": to_id, "type": link_type}


def get_links(
    item_id: str,
    project_path: str = ".",
) -> Dict[str, Any]:
    """Get all links/relationships for a ledger item."""
    _ensure(project_path)
    ledger_dir = _project_ledger_dir(project_path)
    links_file = ledger_dir / LINKS_FILE_NAME

    if not links_file.exists():
        return {"item_id": item_id, "links": [], "count": 0}

    links = []
    try:
        for line in links_file.read_text().strip().split("\n"):
            if not line.strip():
                continue
            link = json.loads(line)
            if link.get("from") == item_id or link.get("to") == item_id:
                links.append(link)
    except Exception:
        pass

    return {"item_id": item_id, "links": links, "count": len(links)}


def unlink_items(
    from_id: str,
    to_id: str,
    project_path: str = ".",
) -> Dict[str, Any]:
    """Remove all links between two items."""
    _ensure(project_path)
    ledger_dir = _project_ledger_dir(project_path)
    links_file = ledger_dir / LINKS_FILE_NAME

    if not links_file.exists():
        return {"unlinked": False, "reason": "No links file"}

    kept = []
    removed = 0
    for line in links_file.read_text().strip().split("\n"):
        if not line.strip():
            continue
        link = json.loads(line)
        if (link.get("from") == from_id and link.get("to") == to_id) or \
           (link.get("from") == to_id and link.get("to") == from_id):
            removed += 1
        else:
            kept.append(line)

    links_file.write_text("\n".join(kept) + "\n" if kept else "")
    return {"unlinked": True, "removed": removed}


# ═══════════════════════════════════════════════════════════════════════
#  SESSION HANDOFF
# ═══════════════════════════════════════════════════════════════════════

SESSIONS_DIR = GLOBAL_DIR / "sessions"


def session_handoff(
    summary: str,
    items_completed: Optional[List[str]] = None,
    items_added: Optional[List[str]] = None,
    key_decisions: Optional[List[str]] = None,
    blockers: Optional[List[str]] = None,
    files_changed: Optional[List[str]] = None,
    venture: str = "",
    project_path: str = "",
    source_model: str = "session_handoff",
    refresh_soul: bool = True,
) -> Dict[str, Any]:
    """Store a session summary for cross-session continuity.

    Called at end of a productive session so the next session can load context.

    LED-3731: the handoff store (``~/.delimit/sessions/``) and the souls store
    (``~/.delimit/souls/<project-hash>/``, read by ``delimit_revive``) used to
    be DISJOINT — a fresh handoff never surfaced in revive, so revive could
    return a month-old soul instead of the state written minutes earlier (this
    bit a real wire-report session). To close the gap, a handoff now ALSO
    refreshes a lightweight pointer-soul in the souls store for the SAME
    project via ``session_phoenix.capture_soul`` (no soul-schema duplication),
    keyed by the same project identity ``revive`` uses. That makes the newest
    handoff win the next revive.

    ``project_path`` (additive, default = cwd) selects which project's soul is
    refreshed so handoff + revive agree on identity. ``refresh_soul=False``
    skips the pointer-soul (used by the deterministic-floor orphan-salvage
    path, which manages its own capture stamping). The refresh is
    best-effort and NEVER raises into the handoff write — matching
    session_phoenix's "never blocks revive" contract. The ephemeral-path guard
    (LED-3759) is inherited automatically: ``capture_soul`` routes through
    ``session_phoenix._project_dir``, which redirects ephemeral (/tmp/pytest)
    paths to a throwaway store, so a test handoff never pollutes the real
    souls store. No storage-format change to either store.
    """
    SESSIONS_DIR.mkdir(parents=True, exist_ok=True)

    session_id = f"session_{time.strftime('%Y%m%d_%H%M%S')}"
    handoff = {
        "id": session_id,
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "venture": venture or "all",
        "summary": summary,
        "items_completed": items_completed or [],
        "items_added": items_added or [],
        "key_decisions": key_decisions or [],
        "blockers": blockers or [],
        "files_changed": files_changed or [],
    }

    path = SESSIONS_DIR / f"{session_id}.json"
    path.write_text(json.dumps(handoff, indent=2))

    # LED-1705: stamp the deterministic-floor coordinator so the Stop hook
    # treats this model-invoked handoff as the fresh, richer artifact and
    # skips writing a deterministic floor over it.
    try:
        try:
            from ai.last_capture import stamp_capture
        except ImportError:  # pragma: no cover - flat import layout
            from last_capture import stamp_capture
        stamp_capture(source="model", session_id=session_id)
    except Exception:
        pass

    # LED-3731: refresh a pointer-soul so the NEXT revive for this project
    # returns THIS handoff's state, not an older soul. Best-effort; a failure
    # here must never break the handoff write.
    soul_id = ""
    if refresh_soul:
        try:
            try:
                from ai.session_phoenix import capture_soul
            except ImportError:  # pragma: no cover - flat import layout
                from session_phoenix import capture_soul  # type: ignore
            soul = capture_soul(
                active_task=summary,
                decisions=list(key_decisions or []),
                key_context=list(items_completed or []),
                blockers=list(blockers or []),
                next_steps=list(items_added or []),
                source_model=source_model,
                project_path=project_path or "",
                task_status="in_progress",
            )
            soul_id = getattr(soul, "soul_id", "") or ""
        except Exception:
            soul_id = ""

    result = {"saved": session_id, "path": str(path), "handoff": handoff}
    if soul_id:
        result["soul_id"] = soul_id
    return result


def session_history(limit: int = 5) -> Dict[str, Any]:
    """Load recent session handoffs for context recovery."""
    if not SESSIONS_DIR.exists():
        return {"sessions": [], "count": 0}

    files = sorted(SESSIONS_DIR.glob("session_*.json"), reverse=True)[:limit]
    sessions = []
    for f in files:
        try:
            sessions.append(json.loads(f.read_text()))
        except Exception:
            continue

    return {"sessions": sessions, "count": len(sessions)}


# ── LED-1145 Phase 1 PR-B: bulk_action ───────────────────────────────────

# Single allowlisted enum keeps the API surface tiny and predictable.
# `archive` is a soft transition (status="archived", appended to JSONL); items
# stay in replay forever. NO hard delete. Per-item failures don't block others.
BULK_ACTIONS = ("archive", "set_status", "set_priority", "add_tag", "mark_done", "cancel")
# LED-1408: `shipped_pending` is the intermediate state between "committed" and
# "verified on main." Items transition to shipped_pending when a worker reports
# completion (commit exists somewhere) but the orchestrator hasn't yet verified
# the commit is reachable from origin/main. The reconciler (Phase 2) promotes
# shipped_pending → done once reachability is confirmed.
_VALID_BULK_STATUSES = ("open", "in_progress", "blocked", "shipped_pending", "done", "cancelled", "archived", "completed")
_VALID_BULK_PRIORITIES = ("P0", "P1", "P2", "P3")


def _normalize_id_list(value):
    """Accept None / str (CSV) / list and return list[str]. Used by bulk_action
    and the MCP tool wrapper."""
    if value is None:
        return []
    if isinstance(value, str):
        return [v.strip() for v in value.split(",") if v.strip()]
    return [str(v).strip() for v in value if str(v).strip()]


def _replay_current_state(item_id: str, ledger_dir: Path) -> Optional[Dict[str, Any]]:
    """Walk the ledger and return the CURRENT replayed state of `item_id`,
    or None if the item doesn't exist. Used by bulk_action for dry-run
    diff preview AND by auto_close_linked_external to verify state changes.

    Replays the same fields list_items does so callers see a consistent view."""
    for ledger_name, filename in [("ops", "operations.jsonl"), ("strategy", "strategy.jsonl")]:
        path = ledger_dir / filename
        items = _read_ledger(path)
        state: Optional[Dict[str, Any]] = None
        for item in items:
            if item.get("id") != item_id:
                continue
            if item.get("type") == "update":
                if state is not None:
                    if "status" in item:
                        state["status"] = item["status"]
                    if "priority" in item:
                        state["priority"] = item["priority"]
                    if "tags" in item and item["tags"] is not None:
                        state["tags"] = item["tags"]
                    if "note" in item:
                        state["last_note"] = item["note"]
                    if "worked_by" in item:
                        state["last_worked_by"] = item["worked_by"]
                    if "updated_at" in item:
                        state["updated_at"] = item["updated_at"]
            else:
                state = {**item}
        if state is not None:
            return state
    return None


def _apply_field_update(
    item_id: str,
    field: str,
    new_value: Any,
    note: Optional[str],
    ledger_dir: Path,
) -> None:
    """Append a single-field update event directly to the item's file within
    a specific ledger_dir. Used by bulk_action when a ``ledger_dir_override``
    is given so priority/status changes land in the correct per-venture
    sub-ledger even when the item isn't reachable from the caller's
    project_path. Byte-compatible with the events update_item writes, so the
    change is fully reversible (append a counter-event to restore).
    """
    found = _find_item_in_ledger_dir(item_id, ledger_dir)
    if not found:
        raise RuntimeError(f"item {item_id} not found in {ledger_dir}")
    update_event = {
        "id": item_id,
        "type": "update",
        field: new_value,
        "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "worked_by": _detect_model(),
    }
    if note:
        update_event["note"] = note
    _append(found["path"], update_event)


def bulk_action(
    item_ids,
    action: str,
    dry_run: bool = True,
    note: Optional[str] = None,
    new_status: Optional[str] = None,
    new_priority: Optional[str] = None,
    tag: Optional[str] = None,
    project_path: str = ".",
    ledger_dir_override: Optional[Path] = None,
) -> Dict[str, Any]:
    """Apply one action to many items. Default `dry_run=True` returns what
    would change without writing. Per-item failures are reported but don't
    block other items in the batch.

    LED-1145 Phase 1 PR-B. The deliberation locked these defaults: a single
    enum-dispatched tool (not bulk_close/bulk_done/...), dry_run=True default,
    no hard delete, archive is a soft status transition.

    Args:
        item_ids: list of LED-XXX ids (str or list-of-str; CSV string accepted).
        action: one of BULK_ACTIONS.
        dry_run: when True (default), don't write. When False, apply.
        note: optional note attached to every successful update event.
        new_status: required when action="set_status".
        new_priority: required when action="set_priority".
        tag: required when action="add_tag" (single tag string).
        project_path: ledger root.

    Returns:
        {
            "dry_run": bool,
            "action": str,
            "would_change": [{id, field, old, new}, ...]  # if dry_run
            "changed":      [{id, field, old, new}, ...]  # if not dry_run
            "errors":       [{id, reason}]
            "summary": {"requested": int, "would_change": int (or "changed"), "errors": int}
        }
    """
    if action not in BULK_ACTIONS:
        return {
            "error": f"unknown action {action!r}; allowed: {list(BULK_ACTIONS)}",
        }

    ids = _normalize_id_list(item_ids)
    if not ids:
        return {"error": "item_ids must contain at least one id"}

    # Per-action argument validation
    if action == "set_status":
        if not new_status:
            return {"error": "set_status requires new_status"}
        if new_status not in _VALID_BULK_STATUSES:
            return {"error": f"new_status must be one of {list(_VALID_BULK_STATUSES)}"}
    if action == "set_priority":
        if not new_priority:
            return {"error": "set_priority requires new_priority"}
        if new_priority not in _VALID_BULK_PRIORITIES:
            return {"error": f"new_priority must be one of {list(_VALID_BULK_PRIORITIES)}"}
    if action == "add_tag":
        if not tag or not str(tag).strip():
            return {"error": "add_tag requires a non-empty tag"}

    if ledger_dir_override is not None:
        ledger_dir = Path(ledger_dir_override)
    else:
        _ensure(project_path)
        ledger_dir = _project_ledger_dir(project_path)

    # Build the per-item change description.
    # archive    → status: <current> → archived
    # set_status → status: <current> → <new_status>
    # set_priority → priority: <current> → <new_priority>
    # add_tag    → tags: [..current..] → [..current.., <tag>] (if not already present)
    # mark_done  → status: <current> → done
    # cancel     → status: <current> → cancelled
    changes: List[Dict[str, Any]] = []
    errors: List[Dict[str, str]] = []

    for item_id in ids:
        state = _replay_current_state(item_id, ledger_dir)
        if state is None:
            errors.append({"id": item_id, "reason": "not_found"})
            continue

        if action in ("archive", "mark_done", "cancel", "set_status"):
            new_val = {
                "archive": "archived",
                "mark_done": "done",
                "cancel": "cancelled",
                "set_status": new_status,
            }[action]
            old_val = state.get("status")
            if old_val == new_val:
                # No-op; skip silently (idempotent action). Don't record.
                continue
            changes.append({"id": item_id, "field": "status", "old": old_val, "new": new_val})
        elif action == "set_priority":
            old_val = state.get("priority")
            if old_val == new_priority:
                continue
            changes.append({"id": item_id, "field": "priority", "old": old_val, "new": new_priority})
        elif action == "add_tag":
            existing_tags = state.get("tags") or []
            if tag in existing_tags:
                continue
            changes.append({
                "id": item_id, "field": "tags",
                "old": list(existing_tags),
                "new": list(existing_tags) + [tag],
            })

    if dry_run:
        return {
            "dry_run": True,
            "action": action,
            "would_change": changes,
            "errors": errors,
            "summary": {
                "requested": len(ids),
                "would_change": len(changes),
                "errors": len(errors),
            },
        }

    # Apply mode: write one update event per change.
    applied: List[Dict[str, Any]] = []
    for change in changes:
        item_id = change["id"]
        field = change["field"]
        try:
            if ledger_dir_override is not None:
                # Write directly into the resolved sub-ledger so items that
                # aren't reachable from the caller's project_path still update.
                _apply_field_update(item_id, field, change["new"], note, ledger_dir)
            elif field == "status":
                update_item(item_id=item_id, status=change["new"], note=note, project_path=project_path)
            elif field == "priority":
                update_item(item_id=item_id, priority=change["new"], note=note, project_path=project_path)
            elif field == "tags":
                _apply_tag_update(item_id, change["new"], note=note, project_path=project_path)
            applied.append(change)
        except Exception as exc:  # noqa: BLE001 — per-item isolation
            errors.append({"id": item_id, "reason": f"write_failed: {exc}"})

    return {
        "dry_run": False,
        "action": action,
        "changed": applied,
        "errors": errors,
        "summary": {
            "requested": len(ids),
            "changed": len(applied),
            "errors": len(errors),
        },
    }


def _apply_tag_update(item_id: str, new_tags: List[str], note: Optional[str], project_path: str) -> None:
    """Append a tags-update event for an item. Used by bulk_action(add_tag).

    update_item() doesn't support a tags param today, so we write the update
    event directly through the same path it uses. The replay logic in
    list_items already handles `tags` updates.
    """
    ledger_dir = _project_ledger_dir(project_path)
    found = _find_item_in_ledger_dir(item_id, ledger_dir)
    if not found:
        raise RuntimeError(f"item {item_id} disappeared during apply")
    path = found["path"]
    update_event = {
        "id": item_id,
        "type": "update",
        "tags": new_tags,
        "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    }
    if note:
        update_event["note"] = note
    _append(path, update_event)


# ── LED-1145 Phase 2 #1: linked-external auto-close ──────────────────────

import re as _re

# Long form: https://github.com/<owner>/<repo>/(issues|pull)/<num>
_GH_URL_RE = _re.compile(r"github\.com/([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)/(?:issues|pull)/(\d+)")
# Short form: <owner>/<repo>#<num> (avoid matching plain markdown headings)
_GH_SHORT_RE = _re.compile(r"\b([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)#(\d+)\b")
# Explicit tag form: gh:<owner>/<repo>/<num> — for callers that want unambiguous linkage
_GH_TAG_RE = _re.compile(r"^gh:([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)/(\d+)$")


def _extract_external_link(item: Dict[str, Any]) -> Optional[tuple]:
    """Find the FIRST github issue/PR reference inside an item.

    Scans description, context, last_note, and tags (in that order). Returns
    `(owner, repo, number)` or None. Long URLs > short forms > explicit tags.
    """
    haystacks = []
    for field in ("description", "context", "last_note"):
        v = item.get(field)
        if isinstance(v, str) and v:
            haystacks.append(v)
    # Tag matches: each tag string standalone; explicit gh: form first
    for t in (item.get("tags") or []):
        if not isinstance(t, str):
            continue
        m = _GH_TAG_RE.match(t)
        if m:
            return m.group(1), m.group(2), int(m.group(3))
        haystacks.append(t)

    text = "\n".join(haystacks)
    if not text:
        return None

    # Long URL takes precedence
    m = _GH_URL_RE.search(text)
    if m:
        return m.group(1), m.group(2), int(m.group(3))
    m = _GH_SHORT_RE.search(text)
    if m:
        return m.group(1), m.group(2), int(m.group(3))
    return None


def _gh_fetch_issue_state(owner: str, repo: str, number: int, _runner=None) -> Dict[str, Any]:
    """Query the GitHub API for issue/PR state. Returns:
        {ok: bool, state: str, merged: bool|None, closed_at: str|None,
         merge_commit_sha: str|None, state_reason: str|None}
    or {ok: False, error: str} on failure.

    `_runner` is for test injection; defaults to subprocess.run.
    """
    if _runner is None:
        _runner = subprocess.run

    try:
        result = _runner(
            ["gh", "api", f"/repos/{owner}/{repo}/issues/{number}",
             "--jq", '{state, closed_at, state_reason, pull_request: (.pull_request != null)}'],
            capture_output=True, text=True, timeout=15, check=False,
        )
        if result.returncode != 0:
            return {"ok": False, "error": (result.stderr or result.stdout)[:200]}
        meta = json.loads(result.stdout)
    except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError) as exc:
        return {"ok": False, "error": str(exc)[:200]}

    out = {
        "ok": True,
        "state": meta.get("state"),
        "closed_at": meta.get("closed_at"),
        "state_reason": meta.get("state_reason"),
        "merged": None,
        "merge_commit_sha": None,
    }

    # If it's a PR, fetch the merge bit separately. We skip this on
    # non-PR issues to save an API call.
    if meta.get("pull_request"):
        try:
            pr_result = _runner(
                ["gh", "api", f"/repos/{owner}/{repo}/pulls/{number}",
                 "--jq", '{merged, merge_commit_sha}'],
                capture_output=True, text=True, timeout=15, check=False,
            )
            if pr_result.returncode == 0:
                pr_meta = json.loads(pr_result.stdout)
                out["merged"] = bool(pr_meta.get("merged"))
                out["merge_commit_sha"] = pr_meta.get("merge_commit_sha")
        except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError):
            pass
    return out


def _resolve_action_for_external(state_meta: Dict[str, Any]) -> Optional[Dict[str, str]]:
    """Map GitHub issue/PR state -> (action, note) for bulk_action.

    Returns None when the LED should be left alone (still open / fetch error).

    Per LED-1146 deliberation:
      - PR merged → mark_done with merge SHA
      - issue/PR closed with state_reason='completed' → mark_done with closed_at
      - issue/PR closed with state_reason='not_planned' (or no reason) → archive
      - state='open' → None (leave alone)
      - fetch error → None (leave alone)
    """
    if not state_meta.get("ok"):
        return None
    if state_meta.get("state") != "closed":
        return None

    # PR with merged=True is unambiguous "we shipped this"
    if state_meta.get("merged"):
        sha = (state_meta.get("merge_commit_sha") or "")[:8]
        return {"action": "mark_done", "note": f"AUTO-CLOSE: merged {sha}".rstrip()}

    state_reason = state_meta.get("state_reason")
    closed_at = (state_meta.get("closed_at") or "")[:19]

    if state_reason == "completed":
        return {"action": "mark_done", "note": f"AUTO-CLOSE: closed-completed {closed_at}".rstrip()}

    # not_planned, duplicate, bot triage, no-reason → archive
    return {"action": "archive", "note": f"AUTO-CLOSE: closed-not-planned {closed_at}".rstrip()}


def auto_close_linked_external(
    project_path: str = ".",
    dry_run: bool = True,
    max_items: int = 200,
    _gh_runner=None,
) -> Dict[str, Any]:
    """Walk open ledger items, detect linked GitHub issues/PRs, and propose
    closing any whose external counterpart already resolved.

    LED-1145 Phase 2 #1. Built on top of bulk_action() from PR-B (uses its
    `mark_done` and `archive` actions).

    Args:
        project_path: ledger root.
        dry_run: True (default) returns a plan without writing.
        max_items: hard cap on the number of items processed in one call.
            When the candidate set exceeds this, we process the first N and
            mark `truncated=True` in the response.
        _gh_runner: test-only hook for stubbing the gh CLI.

    Returns:
        {
            "dry_run": bool,
            "scanned": int,             # items walked
            "linked": int,              # items with a recognised github reference
            "would_close" or "closed": [
                {"id", "external": "owner/repo#num", "action", "note", "state"},
                ...
            ],
            "left_open": [{"id", "external", "reason"}],   # external still open / fetch error
            "errors": [{"id", "reason"}],
            "truncated": bool,
            "summary": {...},
        }
    """
    _ensure(project_path)
    listing = list_items(status__in=["open", "in_progress", "blocked"], limit=10_000, project_path=project_path)

    candidates: List[Dict[str, Any]] = []
    for ledger_name in ("ops", "strategy"):
        candidates.extend(listing.get("items", {}).get(ledger_name, []))

    truncated = False
    if len(candidates) > max_items:
        candidates = candidates[:max_items]
        truncated = True

    # Per-call cache so the same external URL referenced by multiple LEDs
    # only triggers one gh API call.
    fetch_cache: Dict[tuple, Dict[str, Any]] = {}

    would_close: List[Dict[str, Any]] = []
    left_open: List[Dict[str, Any]] = []
    errors: List[Dict[str, Any]] = []
    scanned = 0
    linked = 0

    for item in candidates:
        scanned += 1
        ext = _extract_external_link(item)
        if ext is None:
            continue
        linked += 1
        owner, repo, number = ext
        external_label = f"{owner}/{repo}#{number}"

        cache_key = (owner.lower(), repo.lower(), number)
        if cache_key not in fetch_cache:
            fetch_cache[cache_key] = _gh_fetch_issue_state(owner, repo, number, _runner=_gh_runner)
        state_meta = fetch_cache[cache_key]

        if not state_meta.get("ok"):
            errors.append({
                "id": item.get("id"),
                "external": external_label,
                "reason": f"gh_api_failed: {state_meta.get('error', 'unknown')}",
            })
            continue

        if state_meta.get("state") != "closed":
            left_open.append({
                "id": item.get("id"),
                "external": external_label,
                "reason": "external_still_open",
            })
            continue

        action_plan = _resolve_action_for_external(state_meta)
        if action_plan is None:
            left_open.append({
                "id": item.get("id"),
                "external": external_label,
                "reason": "no_action_resolved",
            })
            continue

        would_close.append({
            "id": item.get("id"),
            "external": external_label,
            "action": action_plan["action"],
            "note": action_plan["note"],
            "state": state_meta.get("state"),
            "merged": state_meta.get("merged"),
        })

    if dry_run:
        return {
            "dry_run": True,
            "scanned": scanned,
            "linked": linked,
            "would_close": would_close,
            "left_open": left_open,
            "errors": errors,
            "truncated": truncated,
            "max_items": max_items,
            "summary": {
                "scanned": scanned,
                "linked": linked,
                "would_close": len(would_close),
                "left_open": len(left_open),
                "errors": len(errors),
                "truncated": truncated,
            },
        }

    # Apply mode: dispatch each plan through bulk_action one item at a time
    # (so per-item action and note can vary).
    closed_results: List[Dict[str, Any]] = []
    for plan in would_close:
        ba = bulk_action(
            item_ids=[plan["id"]],
            action=plan["action"],
            dry_run=False,
            note=plan["note"],
            project_path=project_path,
        )
        if ba.get("summary", {}).get("changed"):
            closed_results.append(plan)
        else:
            errors.append({
                "id": plan["id"],
                "reason": f"bulk_action_failed: {ba.get('errors') or ba.get('error') or 'unknown'}",
            })

    return {
        "dry_run": False,
        "scanned": scanned,
        "linked": linked,
        "closed": closed_results,
        "left_open": left_open,
        "errors": errors,
        "truncated": truncated,
        "max_items": max_items,
        "summary": {
            "scanned": scanned,
            "linked": linked,
            "closed": len(closed_results),
            "left_open": len(left_open),
            "errors": len(errors),
            "truncated": truncated,
        },
    }


# ── LED-1145 Phase 2 #2: proposal-first grooming ─────────────────────────

import datetime as _dt
from collections import defaultdict as _defaultdict


def _is_garbage_venture(name: str) -> bool:
    """Detect test/scratch venture buckets that should be archived in bulk.

    Pattern matches: tmp*, test_*, venture_<single-letter>, custom-venture.
    Deliberately does NOT match "unknown" — those are orphaned items that
    warrant a separate review pass, not auto-archive.
    """
    if not name:
        return False
    if name.startswith(("tmp", "test_")):
        return True
    if name == "custom-venture":
        return True
    # venture_a / venture_b / venture_z (test fixture pattern)
    if _re.match(r"^venture_[a-z]$", name):
        return True
    return False


def _title_prefix(title: str, length: int = 50) -> str:
    """Normalise a title for fuzzy duplicate detection.

    Strips bracketed prefixes like '[DELIMIT]' and lowercases so
    "[DELIMIT] GitHub outreach: foo" and "GitHub outreach: foo" group.
    """
    if not title:
        return ""
    cleaned = _re.sub(r"^\s*\[[^\]]+\]\s*", "", title)
    return cleaned.strip().lower()[:length]


def _parse_iso(ts: str) -> Optional[_dt.datetime]:
    """Parse an ISO-8601 timestamp. Tolerates the trailing 'Z' suffix and
    naive timestamps. Returns None for unparseable input."""
    if not ts:
        return None
    try:
        # 2026-04-13T08:00:00Z and 2026-04-13T08:00:00 both supported
        v = ts.rstrip("Z")
        # Strip fractional seconds if present
        if "." in v:
            v = v.split(".", 1)[0]
        return _dt.datetime.fromisoformat(v).replace(tzinfo=_dt.timezone.utc)
    except (ValueError, TypeError):
        return None


def groom_proposal(
    project_path: str = ".",
    stale_days: int = 30,
    dup_min_count: int = 3,
    max_per_category: int = 50,
) -> Dict[str, Any]:
    """Read-only grooming proposal: surfaces stale / duplicate / garbage-venture
    items as a structured plan for the founder to review and apply.

    LED-1145 Phase 2 #2. The deliberation explicitly said "risky operations
    like deduplication and mass-cancellation must not be a single atomic
    action. The AI proposes a plan; the founder approves; execution goes
    through the safe bulk_action tool." This function is the proposal half.

    Categories detected:
      - stale_open: status open|in_progress|blocked AND updated_at older
        than `stale_days`. Suggested action: archive.
      - duplicate_titles: groups of >= `dup_min_count` items sharing the
        same normalised title prefix (50 chars, brackets stripped). Suggested
        action: archive (de-duped). The first item in the group is kept.
      - garbage_venture: items whose venture matches the test/scratch
        pattern (tmp*, test*, venture_<letter>, custom-venture, unknown).
        Suggested action: archive.

    Out of scope (separate detectors / future PRs):
      - linked-external auto-close — already shipped as a separate tool
        (delimit_ledger_auto_close_external)
      - P0 inflation review — surfaced separately via list_items
      - cross-venture orphan cleanup — needs portfolio policy

    Args:
        project_path: ledger root.
        stale_days: threshold for "stale_open" detector. Default 30.
        dup_min_count: minimum group size for "duplicate_titles". Default 3.
        max_per_category: cap per category in the response. Default 50.

    Returns:
        {
            "proposals": [
                {
                    "category": str,
                    "rationale": str,
                    "items": [{"id", "title", "venture", "status", "updated_at"}, ...],
                    "suggested_action": str,
                    "ready_to_apply": str,  # copy-pasteable bulk_action invocation hint
                    "truncated": bool,
                    "total_in_category": int,
                },
                ...
            ],
            "summary": {
                "total_categories": int,
                "total_items": int,
                "stale_open": int,
                "duplicate_titles": int,
                "garbage_venture": int,
            }
        }
    """
    listing = list_items(
        status__in=["open", "in_progress", "blocked"],
        limit=10_000,
        project_path=project_path,
    )
    candidates: List[Dict[str, Any]] = []
    for ledger_name in ("ops", "strategy"):
        candidates.extend(listing.get("items", {}).get(ledger_name, []))

    now = _dt.datetime.now(_dt.timezone.utc)
    stale_threshold = now - _dt.timedelta(days=stale_days)

    stale_items: List[Dict[str, Any]] = []
    garbage_items: List[Dict[str, Any]] = []
    title_groups: Dict[str, List[Dict[str, Any]]] = _defaultdict(list)

    for item in candidates:
        # garbage_venture (highest precedence — short-circuits other categories)
        if _is_garbage_venture(item.get("venture", "") or ""):
            garbage_items.append(item)
            continue

        # stale_open
        ts = item.get("updated_at") or item.get("created_at")
        parsed = _parse_iso(ts)
        if parsed and parsed < stale_threshold:
            stale_items.append(item)

        # duplicate_titles (always grouped — overlapping with stale is OK,
        # the apply-side de-dupe relies on the founder picking one category)
        prefix = _title_prefix(item.get("title", ""))
        if prefix:
            title_groups[prefix].append(item)

    duplicate_groups = {
        prefix: items for prefix, items in title_groups.items() if len(items) >= dup_min_count
    }

    def _trim(items):
        total = len(items)
        truncated = total > max_per_category
        return items[:max_per_category], total, truncated

    proposals = []

    if stale_items:
        sliced, total, truncated = _trim(stale_items)
        proposals.append({
            "category": "stale_open",
            "rationale": f"{total} items have been open for >{stale_days} days with no update; "
                         f"either the work needs revival or the LED was set-and-forget debt.",
            "items": [
                {
                    "id": i.get("id"),
                    "title": (i.get("title") or "")[:80],
                    "venture": i.get("venture"),
                    "status": i.get("status"),
                    "updated_at": i.get("updated_at") or i.get("created_at"),
                }
                for i in sliced
            ],
            "suggested_action": "archive",
            "ready_to_apply": (
                f"delimit_ledger_bulk(item_ids='{','.join(i.get('id', '') for i in sliced)}', "
                f"action='archive', dry_run=True)"
            ),
            "truncated": truncated,
            "total_in_category": total,
        })

    if duplicate_groups:
        # One proposal entry per group so the founder can decide group-by-group
        for prefix, group in sorted(duplicate_groups.items()):
            sliced, total, truncated = _trim(group)
            # Keep the most-recent (or first) item, archive the rest
            sliced_sorted = sorted(
                sliced,
                key=lambda x: x.get("updated_at") or x.get("created_at") or "",
                reverse=True,
            )
            keep = sliced_sorted[0]
            archive = sliced_sorted[1:]
            proposals.append({
                "category": "duplicate_titles",
                "rationale": (
                    f"{total} items share the title prefix {prefix!r}. "
                    f"Suggesting we keep {keep.get('id')} (most recent) "
                    f"and archive the other {len(archive)}."
                ),
                "items": [
                    {
                        "id": i.get("id"),
                        "title": (i.get("title") or "")[:80],
                        "venture": i.get("venture"),
                        "status": i.get("status"),
                        "updated_at": i.get("updated_at") or i.get("created_at"),
                        "_role": "keep" if i.get("id") == keep.get("id") else "archive",
                    }
                    for i in sliced_sorted
                ],
                "suggested_action": "archive",
                "ready_to_apply": (
                    f"delimit_ledger_bulk(item_ids='{','.join(i.get('id', '') for i in archive)}', "
                    f"action='archive', dry_run=True)"
                ),
                "truncated": truncated,
                "total_in_category": total,
            })

    if garbage_items:
        sliced, total, truncated = _trim(garbage_items)
        proposals.append({
            "category": "garbage_venture",
            "rationale": (
                f"{total} items belong to test/scratch venture buckets "
                f"(tmp*, test*, venture_<letter>, custom-venture, unknown) that "
                f"shouldn't be in production data. Safe to archive in bulk."
            ),
            "items": [
                {
                    "id": i.get("id"),
                    "title": (i.get("title") or "")[:80],
                    "venture": i.get("venture"),
                    "status": i.get("status"),
                    "updated_at": i.get("updated_at") or i.get("created_at"),
                }
                for i in sliced
            ],
            "suggested_action": "archive",
            "ready_to_apply": (
                f"delimit_ledger_bulk(item_ids='{','.join(i.get('id', '') for i in sliced)}', "
                f"action='archive', dry_run=True)"
            ),
            "truncated": truncated,
            "total_in_category": total,
        })

    summary = {
        "total_categories": len(proposals),
        "total_items": sum(p["total_in_category"] for p in proposals),
        "stale_open": len(stale_items),
        "duplicate_titles": sum(len(g) for g in duplicate_groups.values()),
        "garbage_venture": len(garbage_items),
    }

    return {
        "proposals": proposals,
        "summary": summary,
    }


# ── LED-1145 Phase 2 #4: stale-TTL auto-cancel ────────────────────────────

# After this many days of dormancy, an open item is considered stale enough
# that auto-archive is the safe default. groom_proposal still flags items
# at 30d for triage; this stricter cap is for items that crossed the
# triage threshold but never got reviewed. Override via env or arg.
STALE_TTL_DEFAULT_DAYS = 60


def _stale_ttl_default() -> int:
    """Resolve the active stale-TTL threshold from env. 0 disables auto-cancel."""
    raw = os.environ.get("DELIMIT_STALE_TTL_DAYS", "")
    if raw == "":
        return STALE_TTL_DEFAULT_DAYS
    try:
        n = int(raw)
        return max(0, n)
    except (TypeError, ValueError):
        return STALE_TTL_DEFAULT_DAYS


def auto_cancel_stale(
    project_path: str = ".",
    threshold_days: Optional[int] = None,
    dry_run: bool = True,
    max_items: int = 200,
) -> Dict[str, Any]:
    """Auto-archive open items that have been dormant past the stale TTL.

    LED-1145 Phase 2 #4. Composes Phase 2 #2's stale-detection logic with
    Phase 1 PR-B's bulk_action(action="archive"). Same dry_run-default
    pattern as auto_close_linked_external — caller passes dry_run=False
    explicitly to apply.

    Distinct from groom_proposal's stale_open category because:
      - The threshold is stricter (default 60d vs groom's 30d)
      - It auto-applies on dry_run=False (groom is purely propose)
      - It's intended for nightly automation; groom is for interactive use

    Args:
        project_path: ledger root.
        threshold_days: dormancy threshold. Default reads
            DELIMIT_STALE_TTL_DAYS env (60 if unset). 0 disables.
        dry_run: True (default) returns a plan; False applies via bulk_action.
        max_items: cap per call. Items beyond the cap surface in
            `truncated=True` with `total_candidates` set.

    Returns:
        {
            "dry_run": bool,
            "threshold_days": int,
            "would_cancel" or "cancelled": [{id, title, venture, status, last_seen}, ...],
            "errors": [...],
            "summary": {"scanned": int, "stale": int, "cancelled": int, "errors": int, "truncated": bool},
        }
    """
    if threshold_days is None:
        threshold_days = _stale_ttl_default()
    if threshold_days == 0:
        return {
            "dry_run": dry_run,
            "threshold_days": 0,
            "would_cancel" if dry_run else "cancelled": [],
            "errors": [],
            "summary": {
                "scanned": 0, "stale": 0, "cancelled": 0,
                "errors": 0, "truncated": False,
                "note": "DELIMIT_STALE_TTL_DAYS=0 disables auto-cancel",
            },
        }

    listing = list_items(
        status__in=["open", "in_progress", "blocked"],
        limit=10_000,
        project_path=project_path,
    )
    candidates: List[Dict[str, Any]] = []
    for ledger_name in ("ops", "strategy"):
        candidates.extend(listing.get("items", {}).get(ledger_name, []))

    now = _dt.datetime.now(_dt.timezone.utc)
    stale_threshold = now - _dt.timedelta(days=threshold_days)

    stale_items: List[Dict[str, Any]] = []
    for item in candidates:
        ts = item.get("updated_at") or item.get("created_at")
        parsed = _parse_iso(ts)
        if parsed and parsed < stale_threshold:
            stale_items.append(item)

    total_candidates = len(stale_items)
    truncated = total_candidates > max_items
    stale_items = stale_items[:max_items]

    summary_records = [
        {
            "id": i.get("id"),
            "title": (i.get("title") or "")[:80],
            "venture": i.get("venture"),
            "status": i.get("status"),
            "last_seen": i.get("updated_at") or i.get("created_at"),
        }
        for i in stale_items
    ]

    if dry_run:
        return {
            "dry_run": True,
            "threshold_days": threshold_days,
            "would_cancel": summary_records,
            "errors": [],
            "summary": {
                "scanned": len(candidates),
                "stale": total_candidates,
                "would_cancel": len(stale_items),
                "errors": 0,
                "truncated": truncated,
            },
        }

    # Apply via bulk_action(action="archive") so the same audit / no-hard-
    # delete invariants used everywhere else apply here too.
    if not stale_items:
        return {
            "dry_run": False,
            "threshold_days": threshold_days,
            "cancelled": [],
            "errors": [],
            "summary": {
                "scanned": len(candidates), "stale": 0,
                "cancelled": 0, "errors": 0, "truncated": False,
            },
        }

    ids = [i.get("id") for i in stale_items if i.get("id")]
    bulk_result = bulk_action(
        item_ids=ids,
        action="archive",
        dry_run=False,
        note=f"auto-cancel: dormant > {threshold_days} days",
        project_path=project_path,
    )

    return {
        "dry_run": False,
        "threshold_days": threshold_days,
        "cancelled": summary_records,
        "errors": bulk_result.get("errors", []),
        "summary": {
            "scanned": len(candidates),
            "stale": total_candidates,
            "cancelled": bulk_result.get("summary", {}).get("changed", 0),
            "errors": len(bulk_result.get("errors", [])),
            "truncated": truncated,
        },
    }


# ── LED-1145 Phase 2 capstone: ledger health summary ─────────────────────

# Traffic-light thresholds for each category. Tuned from observed
# real-world ledger sizes — ~50 stale items is when the "groom me"
# signal becomes loud; ~20 is yellow heads-up.
_HEALTH_STALE_RED = 50
_HEALTH_STALE_YELLOW = 20
_HEALTH_GARBAGE_RED = 10
_HEALTH_GARBAGE_YELLOW = 1
_HEALTH_DUPE_RED = 5
_HEALTH_DUPE_YELLOW = 2


def _grade(value: int, yellow: int, red: int) -> str:
    if value >= red:
        return "red"
    if value >= yellow:
        return "yellow"
    return "green"


def _worst_grade(*grades: str) -> str:
    """Worst-of N traffic lights: red > yellow > green."""
    if "red" in grades:
        return "red"
    if "yellow" in grades:
        return "yellow"
    return "green"


def health_summary(
    project_path: str = ".",
    stale_days: int = 30,
    dup_min_count: int = 3,
) -> Dict[str, Any]:
    """One-shot health check for the ledger. Composes list_items +
    groom_proposal + the P0 quota helper into a single traffic-light
    verdict and a list of concrete next actions.

    LED-1145 capstone — closes the loop on the entire ledger-tooling
    refactor. Designed for nightly/weekly review or for the founder's
    session-start status snapshot.

    Args:
        project_path: ledger root.
        stale_days: threshold passed through to groom_proposal.
        dup_min_count: threshold passed through to groom_proposal.

    LED-3720: the reported counts are now derived from a deterministic
    event-sourced recount of the SINGLE store this project_path resolves to
    (``_replay_status_counts``), and the store's provenance is surfaced via
    ``namespace`` + ``store_path``. Previously the open-count's origin was
    invisible, so a call that auto-detected the ``root`` catch-all store
    (thousands of never-closed sensed signals) looked identical to one that
    read the clean ``delimit`` store — the "garbage open-count" symptom.
    The count itself was always an accurate replay of whichever store was
    read; the ambiguity was WHICH store. ``namespace``/``store_path`` make
    that explicit and ``open_recount`` proves the number against the store.

    Returns:
        {
            "venture": str,
            "namespace": str,          # canonical slug read ("" = central legacy)
            "store_path": str,         # absolute dir the counts came from
            "totals": {"unresolved": int, "open": int, "in_progress": int, "blocked": int, ...},
            "open_recount": int,       # independent deterministic recount == totals.open
            "p0":         {"count": int, "quota": int, "health": str},
            "stale":      {"count": int, "health": str},
            "duplicates": {"groups": int, "items": int, "health": str},
            "garbage":    {"count": int, "health": str},
            "overall_health": "green" | "yellow" | "red",
            "next_actions": [
                {"reason": str, "tool": str, "args": dict},
                ...
            ],
        }
    """
    # Deterministic, event-sourced recount of the ONE store this path
    # resolves to. Authoritative source for the totals below (LED-3720).
    ns = _resolve_namespace(project_path)
    ledger_dir = _project_ledger_dir(project_path)
    status_counts = _replay_status_counts(ledger_dir)

    listing = list_items(
        status__in=["open", "in_progress", "blocked"],
        limit=10_000,
        project_path=project_path,
    )
    venture = listing.get("venture", "unknown")

    p0_count = _count_unresolved_p0(project_path=project_path)
    p0_quota = _p0_soft_quota()
    p0_health = "green"
    if p0_quota > 0:
        if p0_count > p0_quota * 1.5:
            p0_health = "red"
        elif p0_count > p0_quota:
            p0_health = "yellow"

    proposal = groom_proposal(
        project_path=project_path,
        stale_days=stale_days,
        dup_min_count=dup_min_count,
        max_per_category=10_000,
    )
    proposal_summary = proposal.get("summary", {})
    stale_count = proposal_summary.get("stale_open", 0)
    dup_items = proposal_summary.get("duplicate_titles", 0)
    garbage_count = proposal_summary.get("garbage_venture", 0)
    dup_groups = sum(
        1 for p in proposal.get("proposals", [])
        if p.get("category") == "duplicate_titles"
    )

    stale_health = _grade(stale_count, _HEALTH_STALE_YELLOW, _HEALTH_STALE_RED)
    dup_health = _grade(dup_groups, _HEALTH_DUPE_YELLOW, _HEALTH_DUPE_RED)
    garbage_health = _grade(garbage_count, _HEALTH_GARBAGE_YELLOW, _HEALTH_GARBAGE_RED)

    overall = _worst_grade(p0_health, stale_health, dup_health, garbage_health)

    # Build next-actions list ordered by impact
    next_actions: List[Dict[str, Any]] = []
    if garbage_count > 0:
        next_actions.append({
            "reason": f"{garbage_count} item(s) in test/scratch venture buckets",
            "tool": "delimit_ledger_groom",
            "args": {"venture": venture},
            "follow_up": "Apply the garbage_venture proposal via delimit_ledger_bulk(action='archive')",
        })
    if dup_groups > 0:
        next_actions.append({
            "reason": f"{dup_groups} duplicate-title group(s) covering {dup_items} items",
            "tool": "delimit_ledger_groom",
            "args": {"venture": venture, "dup_min_count": dup_min_count},
            "follow_up": "Review duplicate_titles proposals; archive all but most-recent in each group",
        })
    if stale_count > 0:
        next_actions.append({
            "reason": f"{stale_count} item(s) dormant >{stale_days}d",
            "tool": "delimit_ledger_auto_cancel_stale",
            "args": {"venture": venture, "threshold_days": 60, "dry_run": True},
            "follow_up": "Run with dry_run=False after reviewing the plan",
        })
    if p0_quota > 0 and p0_count > p0_quota:
        next_actions.append({
            "reason": f"P0 inflation: {p0_count} unresolved P0s (threshold {p0_quota})",
            "tool": "delimit_ledger_list",
            "args": {"venture": venture, "priority_in": "P0", "status_in": "open,in_progress,blocked", "fields": "slim"},
            "follow_up": "Triage each: ship, demote to P1, or archive",
        })
    if not next_actions:
        next_actions.append({
            "reason": "All categories green",
            "tool": None,
            "args": {},
            "follow_up": "No grooming required",
        })

    open_ct = status_counts.get("open", 0)
    in_progress_ct = status_counts.get("in_progress", 0)
    blocked_ct = status_counts.get("blocked", 0)
    return {
        "venture": venture,
        "namespace": ns["namespace"],
        "store_path": ns["store_path"],
        "totals": {
            "unresolved": open_ct + in_progress_ct + blocked_ct,
            "open": open_ct,
            "in_progress": in_progress_ct,
            "blocked": blocked_ct,
        },
        "open_recount": open_ct,
        "p0": {"count": p0_count, "quota": p0_quota, "health": p0_health},
        "stale": {"count": stale_count, "health": stale_health, "threshold_days": stale_days},
        "duplicates": {"groups": dup_groups, "items": dup_items, "health": dup_health},
        "garbage": {"count": garbage_count, "health": garbage_health},
        "overall_health": overall,
        "next_actions": next_actions,
    }