#!/usr/bin/env python3
"""claude-obsidian-sync -- summarize Claude conversations into an Obsidian vault.

Reads Claude Code transcripts (~/.claude/projects/<project>/<session>.jsonl) and claude.ai
data exports, condenses each conversation, redacts secrets, summarizes it with an
OpenAI-compatible LLM endpoint (e.g. LiteLLM), and writes one note per conversation into
Obsidian through the Local REST API plugin. Standard library only.

  claude-obsidian-sync stats                 # sizes and counts, no network
  claude-obsidian-sync sync [--dry-run]      # Claude Code sessions (incremental)
  claude-obsidian-sync import-claude-ai EXPORT.zip|conversations.json

Configuration comes from the environment, falling back to ~/.config/claude-obsidian/config.env
(KEY=VALUE lines). Nothing environment-specific lives in this file. See README.md.
"""
from __future__ import annotations

import argparse
import concurrent.futures as cf
import datetime as dt
import fcntl
import hashlib
import json
import os
import pathlib
import re
import shutil
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile

HOME = pathlib.Path.home()
CONFIG_DIR = HOME / ".config" / "claude-obsidian"


# --------------------------------------------------------------------------- config
def load_config() -> dict:
    cfg: dict[str, str] = {}
    f = CONFIG_DIR / "config.env"
    if f.exists():
        for line in f.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                k, v = line.split("=", 1)
                v = re.split(r"\s+#", v, maxsplit=1)[0]            # inline comment; a '#' inside a value survives
                cfg[k.strip()] = v.strip().strip('"').strip("'")
    cfg.update({k: v for k, v in os.environ.items() if k in DEFAULTS or k.startswith("LLM_") or k.startswith("OBSIDIAN_")})
    for k, v in DEFAULTS.items():
        cfg.setdefault(k, v)
    return cfg


DEFAULTS = {
    "OBSIDIAN_API_URL": "http://127.0.0.1:27123",
    "OBSIDIAN_API_KEY_FILE": str(CONFIG_DIR / "api-key"),
    "LLM_BACKEND": "claude-cli",           # claude-cli | openai
    "CLAUDE_BIN": "claude",
    "LLM_BASE_URL": "",
    "LLM_API_KEY_FILE": "",
    "LLM_API_KEY_ENV": "LITELLM_KEY",
    "LLM_MODEL": "haiku",
    "VAULT_FOLDER": "Claude",
    "SETTLE_MINUTES": "30",
    "CLAUDE_PROJECTS_DIR": str(HOME / ".claude" / "projects"),
    "STATE_FILE": str(HOME / ".local" / "state" / "claude-obsidian" / "state.json"),
    "JOBS": "1",
    # Cap per run. Summaries draw on the same plan limits as interactive Claude use, so the
    # timer drips a backlog instead of bursting it (a 107-session burst exhausted a session limit).
    "MAX_PER_RUN": "10",
    # Spend guards, in API list-price USD as reported by `claude -p` (total_cost_usd). Not the
    # plan's own meter -- that is only visible in Claude Code's /usage -- but a faithful proxy
    # for how much was consumed. Hitting one stops the run like a usage limit (exit 75).
    "BUDGET_USD_PER_RUN": "0.50",
    "BUDGET_USD_PER_DAY": "2.00",
    # Pre-call estimate prices (USD per million tokens, list). A call is refused BEFORE it runs
    # if the estimate would cross a budget. Deliberately NOT `claude -p --max-budget-usd`: that
    # aborts after the tokens are spent and throws the result away (measured: $0.26 for nothing).
    "LLM_PRICE_IN_PER_M": "1.0",
    "LLM_PRICE_OUT_PER_M": "5.0",
}


def secret(cfg: dict, file_key: str, env_key: str | None = None) -> str:
    path = cfg.get(file_key)
    if path:
        p = pathlib.Path(os.path.expanduser(path))
        if p.exists():
            return p.read_text().strip()
    if env_key and os.environ.get(cfg.get(env_key, "") or ""):
        return os.environ[cfg[env_key]].strip()
    raise SystemExit(f"missing secret: set {file_key} (or {env_key})")


# --------------------------------------------------------------------------- redaction
# Applied to everything sent to the LLM AND to the note written. Over-redaction is the
# accepted failure mode: a summary missing a token is fine, a vault holding one is not.
_REDACTIONS = [
    (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.S), "[REDACTED PRIVATE KEY]"),
    (re.compile(r"(?i)\b(bearer|token)\s+[A-Za-z0-9._~+/=-]{16,}"), r"\1 [REDACTED]"),
    (re.compile(r"(?i)((?:password|passwd|pwd|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|client[_-]?secret|NB_API_KEY|PAT)[\"']?\s*[:=]\s*[\"']?)([^\s\"'`,;]{6,})"), r"\1[REDACTED]"),
    (re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}"), "[REDACTED JWT]"),
    (re.compile(r"\b(?:sk|pk|rk)-[A-Za-z0-9_-]{20,}|\bgh[pousr]_[A-Za-z0-9]{30,}|\bglpat-[A-Za-z0-9_-]{20,}|\bxox[abpr]-[A-Za-z0-9-]{10,}"), "[REDACTED TOKEN]"),
    # >=48 hex: API keys / sha256 tokens. Deliberately leaves 40-hex git SHAs and 32-hex md5s.
    (re.compile(r"\b[a-fA-F0-9]{48,}\b"), "[REDACTED HEX]"),
    # Random mixed-case+digit strings (generated passwords). No '/' so paths survive.
    (re.compile(r"(?<![A-Za-z0-9_./-])(?=[A-Za-z0-9+]*[0-9])(?=[A-Za-z0-9+]*[a-z])(?=[A-Za-z0-9+]*[A-Z])[A-Za-z0-9+]{20,}={0,2}(?![A-Za-z0-9_./-])"), "[REDACTED]"),
]


def redact(text: str) -> str:
    for pat, repl in _REDACTIONS:
        text = pat.sub(repl, text)
    return text


# --------------------------------------------------------------------------- parsing
_NOISE_BLOCKS = re.compile(
    r"<(system-reminder|local-command-stdout|local-command-stderr|local-command-caveat|task-notification)>.*?</\1>",
    re.S,
)
_CMD = re.compile(r"<command-name>(.*?)</command-name>.*?(?:<command-args>(.*?)</command-args>)?", re.S)
MAX_MSG = 6000


def _clip(text: str) -> str:
    return text if len(text) <= MAX_MSG else text[:4000] + "\n[...clipped...]\n" + text[-1500:]


def _clean_user(text: str) -> str:
    text = _NOISE_BLOCKS.sub("", text)
    m = _CMD.search(text)
    if m:
        args = (m.group(2) or "").strip()
        text = f"{m.group(1).strip()} {args}".strip()
    text = re.sub(r"</?command-(message|args|name)>", "", text)
    return text.strip()


def _tool_line(block: dict) -> str:
    name = block.get("name", "tool")
    inp = block.get("input") or {}
    desc = inp.get("description") or inp.get("command") or inp.get("file_path") or inp.get("query") \
        or inp.get("url") or inp.get("prompt") or inp.get("pattern") or ""
    desc = " ".join(str(desc).split())[:160]
    return f"[{name}: {desc}]" if desc else f"[{name}]"


def parse_claude_code(path: pathlib.Path) -> dict:
    meta = {"session_id": path.stem, "cwd": None, "branch": None, "first": None, "last": None,
            "title": None, "version": None, "prompts": 0}
    turns: list[list[str]] = []
    for line in path.open(encoding="utf-8", errors="replace"):
        try:
            r = json.loads(line)
        except json.JSONDecodeError:
            continue
        t = r.get("type")
        if t == "ai-title" and r.get("aiTitle"):
            meta["title"] = r["aiTitle"]
        if t == "custom-title" and r.get("customTitle"):
            meta["title"] = r["customTitle"]
        if t not in ("user", "assistant") or r.get("isSidechain") or r.get("isMeta"):
            continue
        ts = r.get("timestamp")
        if ts:
            meta["first"] = meta["first"] or ts
            meta["last"] = ts
        meta["cwd"] = meta["cwd"] or r.get("cwd")
        meta["branch"] = r.get("gitBranch") or meta["branch"]
        meta["version"] = r.get("version") or meta["version"]
        content = (r.get("message") or {}).get("content")
        if t == "user":
            if isinstance(content, str):
                text = content
            elif isinstance(content, list):
                text = "\n".join(b.get("text", "") for b in content if b.get("type") == "text")
            else:
                continue
            text = _clean_user(text)
            if text:
                role = "SUMMARY-OF-EARLIER-CONTEXT" if r.get("isCompactSummary") else "USER"
                turns.append([role, _clip(text)])
                meta["prompts"] += role == "USER"
        else:
            parts = []
            for b in content if isinstance(content, list) else []:
                if b.get("type") == "text" and b.get("text", "").strip():
                    parts.append(_clip(b["text"].strip()))
                elif b.get("type") == "tool_use":
                    parts.append(_tool_line(b))
            if not parts:
                continue
            if turns and turns[-1][0] == "ASSISTANT":       # a tool loop is many records
                turns[-1][1] += "\n" + "\n".join(parts)
            else:
                turns.append(["ASSISTANT", "\n".join(parts)])
    meta["transcript"] = "\n\n".join(f"{role}: {text}" for role, text in turns)
    return meta


def parse_claude_ai_export(src: pathlib.Path) -> list[dict]:
    """claude.ai 'Export data' -> conversations.json. Tolerant of field-name drift."""
    if src.suffix == ".zip":
        with zipfile.ZipFile(src) as z:
            name = next((n for n in z.namelist() if n.endswith("conversations.json")), None)
            if not name:
                raise SystemExit(f"no conversations.json inside {src}")
            data = json.loads(z.read(name))
    else:
        data = json.loads(src.read_text())
    out = []
    for c in data if isinstance(data, list) else data.get("conversations", []):
        turns, count = [], 0
        for m in c.get("chat_messages") or c.get("messages") or []:
            text = m.get("text") or ""
            if not text and isinstance(m.get("content"), list):
                text = "\n".join(b.get("text", "") for b in m["content"] if b.get("type") == "text")
            for a in m.get("attachments") or []:
                if a.get("extracted_content"):
                    text += f"\n[attachment {a.get('file_name', '')}]: {a['extracted_content'][:1500]}"
            if not text.strip():
                continue
            role = "USER" if (m.get("sender") or m.get("role")) in ("human", "user") else "ASSISTANT"
            turns.append(f"{role}: {_clip(text.strip())}")
            count += role == "USER"
        if not turns:
            continue
        out.append({
            "session_id": c.get("uuid") or hashlib.sha1(json.dumps(c, sort_keys=True).encode()).hexdigest(),
            "title": c.get("name") or None, "first": c.get("created_at"), "last": c.get("updated_at"),
            "cwd": None, "branch": None, "version": None, "prompts": count,
            "transcript": "\n\n".join(turns),
        })
    return out


# --------------------------------------------------------------------------- LLM
SYSTEM = """You write concise knowledge-base notes summarizing a conversation between a user and Claude.
The reader is the same user, months later, searching their notes. Capture what happened and
what was learned, not a play-by-play. Be specific: name the systems, files, commands, errors
and decisions. Never include passwords, API keys, tokens or other secrets, even if present;
refer to them generically ("the API key"). Text marked [REDACTED] must stay redacted.

Reply with ONLY a JSON object, no prose, with these keys:
  "title": short specific title (<= 70 chars)
  "summary": 3-6 sentences
  "outcomes": list of what was actually built, fixed or answered
  "decisions": list of choices made and why
  "facts": list of reusable facts, gotchas, root causes worth remembering
  "open_threads": list of unfinished work or follow-ups
  "tags": 3-8 lowercase kebab-case topic tags
Use empty lists where nothing applies."""

CHUNK_SYSTEM = """You are condensing one part of a long conversation between a user and Claude so that a
later step can summarize the whole. Output dense bullet notes: goals, actions taken, results,
errors and their causes, decisions, open questions. Be specific. Never reproduce secrets."""

CHUNK_CHARS = 300_000


class Ledger:
    """Append-only record of every model call: tokens and list-price cost. Also enforces budgets."""

    def __init__(self, cfg: dict):
        self.path = pathlib.Path(os.path.expanduser(cfg["STATE_FILE"])).parent / "usage.jsonl"
        self.run_budget = float(cfg["BUDGET_USD_PER_RUN"])
        self.day_budget = float(cfg["BUDGET_USD_PER_DAY"])
        self.lock = threading.Lock()
        self.run_cost = 0.0
        self.run_calls = 0
        self.run_in = self.run_out = 0

    def entries(self, since: dt.datetime | None = None) -> list[dict]:
        if not self.path.exists():
            return []
        out = []
        for line in self.path.read_text().splitlines():
            try:
                e = json.loads(line)
            except json.JSONDecodeError:
                continue
            if since is None or dt.datetime.fromisoformat(e["ts"]) >= since:
                out.append(e)
        return out

    def spent_since(self, hours: float) -> float:
        since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=hours)
        return sum(e.get("cost_usd") or 0 for e in self.entries(since))

    def check(self, estimate: float = 0.0) -> None:
        """Called BEFORE each model call with that call's estimated cost -- so a budget is never overshot."""
        with self.lock:
            if self.run_cost + estimate > self.run_budget:
                raise BudgetExhausted(f"run budget ${self.run_budget:.2f}: ${self.run_cost:.3f} spent, "
                                      f"next call est. ${estimate:.3f}")
        day = self.spent_since(24)
        if day + estimate > self.day_budget:
            raise BudgetExhausted(f"24h budget ${self.day_budget:.2f}: ${day:.3f} spent, next call est. ${estimate:.3f}")

    def record(self, label: str, model: str, usage: dict, cost: float | None) -> None:
        e = {"ts": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"), "item": label, "model": model,
             "input": (usage.get("input_tokens") or 0) + (usage.get("cache_creation_input_tokens") or 0)
                      + (usage.get("cache_read_input_tokens") or 0),
             "output": usage.get("output_tokens") or 0, "cost_usd": cost}
        with self.lock:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            with self.path.open("a") as f:
                f.write(json.dumps(e) + "\n")
            self.run_calls += 1
            self.run_cost += cost or 0
            self.run_in += e["input"]
            self.run_out += e["output"]

    def report(self) -> str:
        return (f"usage this run: {self.run_calls} model calls, {self.run_in:,} in / {self.run_out:,} out tokens, "
                f"${self.run_cost:.3f} list-price equiv | last 24h ${self.spent_since(24):.3f} of ${self.day_budget:.2f}")


class BudgetExhausted(RuntimeError):
    """A configured spend budget is reached -- stop the run the same way as a usage limit."""


class QuotaExhausted(RuntimeError):
    """The plan/API usage limit is hit. Every further call would fail the same way -- stop the run."""


_QUOTA = re.compile(r"(?i)hit your (?:session |usage |weekly )?limit|usage limit|quota|credit balance|resets? (?:at )?\d")


class LLM:
    """Two backends.

    claude-cli (default): headless Claude Code on the user's own plan -- no API key to manage.
      --no-session-persistence is load-bearing: without it every summary writes a new transcript
      into ~/.claude/projects and the next sync would summarize its own summarizing.
    openai: any OpenAI-compatible /chat/completions endpoint (LiteLLM, local models).
    """

    def __init__(self, cfg: dict, ledger: "Ledger | None" = None):
        self.ledger = ledger or Ledger(cfg)
        self.price_in = float(cfg["LLM_PRICE_IN_PER_M"]) / 1e6
        self.price_out = float(cfg["LLM_PRICE_OUT_PER_M"]) / 1e6
        self.label = threading.local()
        self.backend = cfg["LLM_BACKEND"]
        self.model = cfg["LLM_MODEL"]
        self.resolved = cfg["LLM_MODEL"]
        if self.backend == "claude-cli":
            self.bin = shutil.which(os.path.expanduser(cfg["CLAUDE_BIN"]))
            if not self.bin:
                raise SystemExit(f"claude CLI not found ({cfg['CLAUDE_BIN']})")
            self.workdir = pathlib.Path(os.path.expanduser(cfg["STATE_FILE"])).parent  # no project CLAUDE.md here
            self.workdir.mkdir(parents=True, exist_ok=True)
        elif self.backend == "openai":
            if not cfg["LLM_BASE_URL"]:
                raise SystemExit("LLM_BASE_URL is not set (see config.env.example)")
            self.url = cfg["LLM_BASE_URL"].rstrip("/") + "/chat/completions"
            self.key = secret(cfg, "LLM_API_KEY_FILE", "LLM_API_KEY_ENV")
        else:
            raise SystemExit(f"unknown LLM_BACKEND {self.backend!r}")

    def estimate(self, system: str, user: str, max_tokens: int) -> float:
        # ~3 chars/token over-estimates input (measured 3.1 on transcripts); assume the full output;
        # +20% margin for anything unmeasured. Errs high on purpose: refusing early beats overshooting.
        return 1.2 * ((len(system) + len(user)) / 3.0 * self.price_in + max_tokens * self.price_out)

    def chat(self, system: str, user: str, max_tokens: int = 2500) -> str:
        self.ledger.check(self.estimate(system, user, max_tokens))
        return self._chat_cli(system, user) if self.backend == "claude-cli" else self._chat_openai(system, user, max_tokens)

    def _chat_cli(self, system: str, user: str) -> str:
        # Cost levers, measured on one session: $0.057 -> $0.015 with identical summary content.
        #   alwaysThinkingEnabled=false  -- extended thinking was ~85% of output tokens
        #   DISABLE_PROMPT_CACHING=1     -- Claude Code writes prompts to the 1h cache at 2x input
        #                                   price; a one-shot call never reads it back
        #   CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 -- otherwise a hidden side request re-reads the
        #                                   whole prompt (~20 tokens out, title-generation shaped),
        #                                   doubling billed input: 9,638 vs 4,663 visible tokens.
        cmd = [self.bin, "-p", "--model", self.model, "--no-session-persistence", "--tools", "",
               "--strict-mcp-config", "--output-format", "json",
               "--settings", json.dumps({"alwaysThinkingEnabled": False}), "--system-prompt", system]
        env = dict(os.environ, DISABLE_PROMPT_CACHING="1", CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC="1")
        last = ""
        for attempt in range(5):
            try:
                proc = subprocess.run(cmd, input=user, capture_output=True, text=True, timeout=900, cwd=self.workdir, env=env)
            except subprocess.TimeoutExpired:
                last = "timeout"
            else:
                try:
                    d = json.loads(proc.stdout)
                except json.JSONDecodeError:
                    d = {"is_error": True, "result": (proc.stdout or proc.stderr)[-300:]}
                if d.get("usage") or d.get("total_cost_usd"):             # failed calls can still cost
                    # modelUsage is what was BILLED across every request the CLI made; `usage` is only
                    # the visible turn and undercounted by half before the nonessential-traffic fix.
                    mu = list((d.get("modelUsage") or {}).values())
                    billed = {"input_tokens": sum((v.get("inputTokens") or 0) + (v.get("cacheReadInputTokens") or 0)
                                                  + (v.get("cacheCreationInputTokens") or 0) for v in mu),
                              "output_tokens": sum(v.get("outputTokens") or 0 for v in mu)} if mu else (d.get("usage") or {})
                    self.ledger.record(getattr(self.label, "v", "?"), self.model, billed, d.get("total_cost_usd"))
                if not d.get("is_error") and d.get("result"):
                    used = list((d.get("modelUsage") or {}).keys())
                    if used:
                        self.resolved = used[0]
                    return d["result"]
                last = str(d.get("result") or proc.stderr)[-300:]
                if _QUOTA.search(last):
                    raise QuotaExhausted(last.strip())
                if not re.search(r"(?i)rate.?limit|overloaded|429|529|timeout|temporar", last):
                    break                                              # not transient -- do not hammer
            time.sleep(2 ** attempt * 10)
        raise RuntimeError(f"claude -p failed: {last}")

    def _chat_openai(self, system: str, user: str, max_tokens: int) -> str:
        body = json.dumps({"model": self.model, "temperature": 0.2, "max_tokens": max_tokens,
                           "messages": [{"role": "system", "content": system},
                                        {"role": "user", "content": user}]}).encode()
        for attempt in range(5):
            req = urllib.request.Request(self.url, data=body, method="POST", headers={
                "Authorization": f"Bearer {self.key}", "Content-Type": "application/json"})
            try:
                with urllib.request.urlopen(req, timeout=600) as r:
                    d = json.load(r)
                u = d.get("usage") or {}
                self.ledger.record(getattr(self.label, "v", "?"), self.model,
                                   {"input_tokens": u.get("prompt_tokens"), "output_tokens": u.get("completion_tokens")}, None)
                return d["choices"][0]["message"]["content"]
            except urllib.error.HTTPError as e:
                detail = e.read()[:300].decode(errors="replace")
                if _QUOTA.search(detail):
                    raise QuotaExhausted(detail) from None
                if e.code in (429, 500, 502, 503, 504, 529) and attempt < 4:
                    time.sleep(2 ** attempt * 5)
                    continue
                raise RuntimeError(f"LLM HTTP {e.code}: {detail}") from None
            except (urllib.error.URLError, TimeoutError) as e:
                if attempt < 4:
                    time.sleep(2 ** attempt * 5)
                    continue
                raise RuntimeError(f"LLM unreachable: {e}") from None
        raise RuntimeError("LLM retries exhausted")

    def summarize(self, meta: dict) -> dict:
        text = redact(meta["transcript"])
        if len(text) > CHUNK_CHARS:                       # map-reduce for long sessions
            chunks, cur = [], []
            size = 0
            for part in text.split("\n\n"):
                if size + len(part) > CHUNK_CHARS and cur:
                    chunks.append("\n\n".join(cur)); cur, size = [], 0
                cur.append(part); size += len(part) + 2
            chunks.append("\n\n".join(cur))
            notes = [self.chat(CHUNK_SYSTEM, f"Part {i + 1} of {len(chunks)}:\n\n{c}", 3000)
                     for i, c in enumerate(chunks)]
            text = "Condensed notes from each part of a long conversation, in order:\n\n" + \
                   "\n\n---\n\n".join(notes)
        header = f"Working directory: {meta.get('cwd')}\nGit branch: {meta.get('branch')}\n" \
                 f"Existing title: {meta.get('title')}\n\n"
        raw = self.chat(SYSTEM, header + text)
        m = re.search(r"\{.*\}", raw, re.S)
        if not m:
            raise RuntimeError(f"LLM did not return JSON: {raw[:200]!r}")
        out = json.loads(m.group(0))
        for k in ("outcomes", "decisions", "facts", "open_threads", "tags"):
            out[k] = [redact(str(x)) for x in (out.get(k) or []) if str(x).strip()]
        out["summary"] = redact(str(out.get("summary", "")))
        out["title"] = redact(str(out.get("title") or meta.get("title") or meta["session_id"]))
        return out


# --------------------------------------------------------------------------- vault
class Vault:
    def __init__(self, cfg: dict):
        # OBSIDIAN_API_URL may list several URLs, comma-separated, tried in order at ping() -- e.g. the
        # permanent address first and a port-forward second, so one config works either way.
        self.candidates = [u.strip().rstrip("/") for u in cfg["OBSIDIAN_API_URL"].split(",") if u.strip()]
        self.base = self.candidates[0]
        self.key = secret(cfg, "OBSIDIAN_API_KEY_FILE")

    def put(self, path: str, content: str) -> None:
        req = urllib.request.Request(f"{self.base}/vault/{urllib.parse.quote(path)}",
                                     data=content.encode(), method="PUT", headers={
                                         "Authorization": f"Bearer {self.key}",
                                         "Content-Type": "text/markdown"})
        with urllib.request.urlopen(req, timeout=60) as r:
            if r.status not in (200, 201, 204):
                raise RuntimeError(f"vault PUT {path}: HTTP {r.status}")

    def ping(self) -> None:
        """Checked BEFORE any summarizing, so an unreachable vault never spends usage limits."""
        errors = []
        for base in self.candidates:
            req = urllib.request.Request(f"{base}/", headers={"Authorization": f"Bearer {self.key}"})
            try:
                with urllib.request.urlopen(req, timeout=10) as r:
                    info = json.load(r)
                self.base = base
                break
            except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
                errors.append(f"{base}: {getattr(e, 'reason', e)}")
        else:
            raise SystemExit("Obsidian REST API unreachable -- " + "; ".join(errors))
        print(f"vault: {self.base}", flush=True)
        if not info.get("authenticated"):
            raise SystemExit("Obsidian REST API reachable but the API key was rejected")


_BAD = re.compile(r'[\\/:*?"<>|#^\[\]]+')


def note_path(cfg: dict, kind: str, meta: dict, title: str) -> str:
    when = (meta.get("first") or dt.datetime.now(dt.timezone.utc).isoformat())[:10]
    slug = " ".join(_BAD.sub(" ", title).split())[:60].strip() or "untitled"
    return f"{cfg['VAULT_FOLDER']}/{kind}/{when[:4]}/{when} {slug} ({meta['session_id'][:8]}).md"


def render(meta: dict, s: dict, source: str, model: str, src_file: str | None) -> str:
    project = pathlib.Path(meta["cwd"]).name if meta.get("cwd") else None
    tags = ["claude", source] + [t for t in s["tags"] if t not in ("claude", source)]
    fm = {
        "type": "claude-conversation", "source": source, "session_id": meta["session_id"],
        "title": s["title"], "project": project, "cwd": meta.get("cwd"), "branch": meta.get("branch"),
        "started": meta.get("first"), "ended": meta.get("last"), "prompts": meta.get("prompts"),
        "summarized_by": model, "synced": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "tags": tags,
    }
    lines = ["---"] + [f"{k}: {json.dumps(v)}" for k, v in fm.items() if v not in (None, "")] + ["---", "",
             f"# {s['title']}", "", "> [!summary]"] + [f"> {l}" for l in s["summary"].splitlines() or [""]]
    for heading, key in (("Outcomes", "outcomes"), ("Decisions", "decisions"),
                         ("Facts & gotchas", "facts"), ("Open threads", "open_threads")):
        if s[key]:
            lines += ["", f"## {heading}", ""] + [f"- {x}" for x in s[key]]
    lines += ["", "---"]
    if source == "claude-code":
        lines.append(f"Resume: `claude --resume {meta['session_id']}`" + (f" · transcript `{src_file}`" if src_file else ""))
    else:
        lines.append(f"claude.ai conversation `{meta['session_id']}`")
    return "\n".join(lines) + "\n"


# --------------------------------------------------------------------------- state
class State:
    def __init__(self, path: str):
        self.path = pathlib.Path(os.path.expanduser(path))
        self.lock = threading.Lock()
        self.data = json.loads(self.path.read_text()) if self.path.exists() else {}

    def save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(self.data, indent=1, sort_keys=True))
        tmp.replace(self.path)

    def record(self, key: str, value: dict) -> None:
        with self.lock:
            self.data[key] = value
            self.save()


def write_index(cfg: dict, vault: Vault, state: State) -> None:
    rows = sorted((v for v in state.data.values() if v.get("note")), key=lambda v: v.get("started") or "", reverse=True)
    out = ["---", 'type: "claude-index"', "---", "", "# Claude conversations", "",
           f"{len(rows)} conversations. Regenerated by claude-obsidian-sync.", "",
           "| Date | Title | Project | Source |", "|---|---|---|---|"]
    for v in rows:
        link = v["note"][:-3]
        title = v.get("title", "").replace("|", "-")
        out.append(f"| {(v.get('started') or '')[:10]} | [[{link}\\|{title}]] | {v.get('project') or ''} | {v.get('source')} |")
    vault.put(f"{cfg['VAULT_FOLDER']}/Index.md", "\n".join(out) + "\n")


# --------------------------------------------------------------------------- commands
def top_level_sessions(cfg: dict) -> list[pathlib.Path]:
    root = pathlib.Path(os.path.expanduser(cfg["CLAUDE_PROJECTS_DIR"]))
    return sorted(root.glob("*/*.jsonl"), key=lambda p: p.stat().st_mtime)   # nested = subagents


def process(cfg, llm, vault, state, key, meta, source, kind, fingerprint, src_file, dry_run) -> str:
    # Cache the summary until the note is written, so a vault failure never pays for it twice.
    cache = pathlib.Path(os.path.expanduser(cfg["STATE_FILE"])).parent / "cache" / \
        f"{hashlib.sha1(key.encode()).hexdigest()[:16]}-{hashlib.sha1(fingerprint.encode()).hexdigest()[:8]}.json"
    llm.label.v = key
    if cache.exists():
        s = json.loads(cache.read_text())
    else:
        s = llm.summarize(meta)
        cache.parent.mkdir(parents=True, exist_ok=True)
        cache.write_text(json.dumps(s))
    prev = state.data.get(key, {})
    path = prev.get("note") or note_path(cfg, kind, meta, s["title"])        # stable across re-syncs
    body = render(meta, s, source, llm.resolved, src_file)
    if dry_run == "prefetch":
        return f"summary cached ({path})"
    if dry_run:
        print(f"\n===== {path} =====\n{body}")
        return path
    vault.put(path, body)
    cache.unlink(missing_ok=True)
    state.record(key, {"note": path, "title": s["title"], "fingerprint": fingerprint, "source": source,
                       "started": meta.get("first"),
                       "project": pathlib.Path(meta["cwd"]).name if meta.get("cwd") else None})
    return path


SKIPPED = "(empty, skipped)"
EX_TEMPFAIL = 75                    # "try again later" -- systemd logs it, the next timer run retries


def run_jobs(cfg, work, fn) -> tuple[int, int, str | None]:
    ok = failed = 0
    quota = None
    stop = threading.Event()        # Future.cancel() cannot stop a job a worker already picked up

    def guarded(*w):
        if stop.is_set():
            raise cf.CancelledError()
        try:
            return fn(*w)
        except (QuotaExhausted, BudgetExhausted):
            stop.set()
            raise

    with cf.ThreadPoolExecutor(max_workers=int(cfg["JOBS"])) as ex:
        futs = {ex.submit(guarded, *w): w for w in work}
        for fut in cf.as_completed(futs):
            label = futs[fut][0]
            try:
                result = fut.result()
                if result == SKIPPED:
                    print(f"skip {label} (no conversation content)", flush=True)
                else:
                    print(f"ok   {label} -> {result}", flush=True); ok += 1
            except (QuotaExhausted, BudgetExhausted) as e:
                if quota is None:
                    quota = str(e)
                    for f in futs:                                   # don't launch the rest
                        f.cancel()
            except cf.CancelledError:
                pass
            except Exception as e:                                   # keep going; next run retries
                print(f"FAIL {label}: {e}", file=sys.stderr, flush=True); failed += 1
    return ok, failed, quota


def finish(ok: int, failed: int, quota: str | None, dry_run, remaining: int, ledger: "Ledger | None" = None) -> None:
    verb = {"prefetch": "summarized into cache (vault untouched)"}.get(dry_run) or \
        ("rendered (dry run, nothing written)" if dry_run else "written")
    print(f"done: {ok} {verb}, {failed} failed, {remaining} still queued")
    if ledger:
        print(ledger.report())
    if quota:
        print(f"stopped: {quota}", file=sys.stderr)
        sys.exit(EX_TEMPFAIL)
    sys.exit(1 if failed else 0)


def cmd_usage(cfg, args) -> None:
    led = Ledger(cfg)
    rows = led.entries()
    if not rows:
        print("no model calls recorded yet"); return
    for label, hours in (("last 5h", 5), ("last 24h", 24), ("last 7d", 168)):
        since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=hours)
        es = [e for e in rows if dt.datetime.fromisoformat(e["ts"]) >= since]
        print(f"{label:>9}: {len(es):4d} calls  {sum(e['input'] for e in es):>10,} in  "
              f"{sum(e['output'] for e in es):>8,} out  ${sum(e.get('cost_usd') or 0 for e in es):.3f}")
    print(f"budgets: ${led.run_budget:.2f}/run, ${led.day_budget:.2f}/24h (list-price equiv; plan meter: /usage in Claude Code)")


def cmd_stats(cfg, args) -> None:
    total = 0
    rows = []
    for p in top_level_sessions(cfg):
        m = parse_claude_code(p)
        n = len(redact(m["transcript"]))
        total += n
        rows.append((n, p.stat().st_size, m["prompts"], m["title"] or p.stem))
    rows.sort(reverse=True)
    for n, raw, msgs, title in rows[: args.top]:
        print(f"{raw / 1e6:7.2f} MB raw -> {n / 1e3:7.1f} KB condensed  {msgs:4d} prompts  {title[:60]}")
    print(f"\n{len(rows)} sessions, {total / 1e6:.1f} MB condensed (~{total / 4 / 1e6:.2f}M tokens in), "
          f"{sum(1 for r in rows if r[0] > CHUNK_CHARS)} need map-reduce")


def run_lock(cfg: dict):
    """One run at a time: a timer run overlapping a manual one would race on state and ledger files."""
    path = pathlib.Path(os.path.expanduser(cfg["STATE_FILE"])).parent / "run.lock"
    path.parent.mkdir(parents=True, exist_ok=True)
    fh = path.open("w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        print("another claude-obsidian-sync run is in progress -- exiting")
        sys.exit(0)
    return fh                                                         # held until the process exits


def cmd_sync(cfg, args) -> None:
    _lock = run_lock(cfg)
    state = State(cfg["STATE_FILE"])
    settle = dt.timedelta(minutes=int(cfg["SETTLE_MINUTES"]))
    now = time.time()
    work = []
    for p in top_level_sessions(cfg):
        if args.session and p.stem != args.session:
            continue
        st = p.stat()
        fp = f"{st.st_size}:{int(st.st_mtime)}"
        key = f"claude-code:{p.stem}"
        if not args.force and state.data.get(key, {}).get("fingerprint") == fp:
            continue
        if not (args.force or args.session) and now - st.st_mtime < settle.total_seconds():
            continue                                                    # still being written
        work.append((p, fp, key))
    total = len(work)
    cap = args.limit or int(cfg["MAX_PER_RUN"])
    work = work[-cap:] if not args.oldest else work[:cap]            # newest first by default
    print(f"{total} session(s) pending, summarizing {len(work)} this run", flush=True)
    if not work:
        return
    prefetch = getattr(args, "prefetch", False)
    llm, vault = LLM(cfg), (None if args.dry_run or prefetch else Vault(cfg))
    if vault:
        vault.ping()
    mode = "prefetch" if prefetch else args.dry_run

    def one(p, fp, key):
        meta = parse_claude_code(p)
        if not meta["transcript"].strip():
            if not args.dry_run and not prefetch:
                state.record(key, {"note": None, "fingerprint": fp, "source": "claude-code"})
            return SKIPPED
        src = str(p).replace(str(HOME), "~")
        return process(cfg, llm, vault, state, key, meta, "claude-code", "Code", fp, src, mode)

    ok, failed, quota = run_jobs(cfg, [(w[0].name, *w) for w in work], lambda _l, p, fp, key: one(p, fp, key))
    if vault and ok:
        write_index(cfg, vault, state)
    finish(ok, failed, quota, mode, total - ok, llm.ledger)


def cmd_import(cfg, args) -> None:
    _lock = run_lock(cfg)
    state = State(cfg["STATE_FILE"])
    convs = parse_claude_ai_export(pathlib.Path(args.export))
    work = []
    for m in convs:
        key = f"claude-ai:{m['session_id']}"
        fp = m.get("last") or str(len(m["transcript"]))
        if args.force or state.data.get(key, {}).get("fingerprint") != fp:
            work.append((m, fp, key))
    total = len(work)
    work = work[: args.limit or int(cfg["MAX_PER_RUN"])]
    print(f"{len(convs)} conversations in export, {total} new or changed, summarizing {len(work)} this run", flush=True)
    if not work:
        return
    llm, vault = LLM(cfg), (None if args.dry_run else Vault(cfg))
    if vault:
        vault.ping()
    ok, failed, quota = run_jobs(cfg, [((m.get("title") or m["session_id"])[:50], m, fp, key) for m, fp, key in work],
                          lambda _l, m, fp, key: process(cfg, llm, vault, state, key, m, "claude-ai", "Web",
                                                         fp, None, args.dry_run))
    if vault and ok:
        write_index(cfg, vault, state)
    finish(ok, failed, quota, args.dry_run, total - ok, llm.ledger)


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("usage", help="model calls, tokens and spend from the local ledger")
    s = sub.add_parser("stats", help="condensed sizes, no network")
    s.add_argument("--top", type=int, default=15)
    s = sub.add_parser("sync", help="summarize Claude Code sessions")
    s.add_argument("--dry-run", action="store_true", help="print notes instead of writing them")
    s.add_argument("--prefetch", action="store_true",
                   help="summarize into the local cache without touching the vault; a later sync writes them at no extra cost")
    s.add_argument("--force", action="store_true", help="ignore state and the settle window")
    s.add_argument("--session", help="only this session id")
    s.add_argument("--limit", type=int, help="max sessions this run (default MAX_PER_RUN)")
    s.add_argument("--oldest", action="store_true", help="take the oldest pending sessions instead of the newest")
    s = sub.add_parser("import-claude-ai", help="import a claude.ai data export")
    s.add_argument("export", help="the export .zip or its conversations.json")
    s.add_argument("--dry-run", action="store_true")
    s.add_argument("--force", action="store_true")
    s.add_argument("--limit", type=int, help="max conversations this run (default MAX_PER_RUN)")
    args = ap.parse_args()
    cfg = load_config()
    {"stats": cmd_stats, "usage": cmd_usage, "sync": cmd_sync, "import-claude-ai": cmd_import}[args.cmd](cfg, args)


if __name__ == "__main__":
    main()
