noconnorandClaude Opus 5
2bfa7e9a2d
The user asked for usage to be monitored closely, and measuring it turned up that most of the spend was avoidable. Metering: every model call is appended to usage.jsonl with BILLED tokens (modelUsage -- the visible `usage` turn undercounted by half) and the CLI's list-price cost. `usage` reports 5h/24h/7d. Each call's cost is estimated first and refused if it would cross the per-run or rolling 24h budget, so budgets cannot be overshot. Removed --max-budget-usd: it aborts after spending and discards the result ($0.26 lost in one call). Cost, measured on one session ($0.057 -> $0.010, same summary content): - alwaysThinkingEnabled=false: extended thinking was ~85% of output. - DISABLE_PROMPT_CACHING=1: one-shot prompts were being written to the 1h cache at 2x input price. - CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1: a hidden side request re-read the whole prompt, doubling billed input (9,638 vs 4,663). Summaries are cached until their note is written, so a vault failure never re-pays; --prefetch summarizes while the vault is unreachable. Offline tests (fake claude binary): budget refusal pre-flight, cache reuse with zero calls, map-reduce, env/argv levers, billed-token ledger. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KF322NzozeaFjpgWvh4F4g
772 lines
37 KiB
Python
Executable File
772 lines
37 KiB
Python
Executable File
#!/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 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)
|
|
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):
|
|
self.base = cfg["OBSIDIAN_API_URL"].rstrip("/")
|
|
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."""
|
|
req = urllib.request.Request(f"{self.base}/", headers={"Authorization": f"Bearer {self.key}"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as r:
|
|
info = json.load(r)
|
|
except (urllib.error.URLError, TimeoutError, OSError) as e:
|
|
raise SystemExit(f"Obsidian REST API unreachable at {self.base}: {getattr(e, 'reason', e)}") from None
|
|
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 cmd_sync(cfg, args) -> None:
|
|
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:
|
|
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()
|