claude-mode/linux/cm-json.py

#!/usr/bin/env python3
"""
JSON engine for claude-mode (POSIX port).

Bash cannot safely read or rewrite JSON, and jq is not installed often enough to
depend on. Everything that touches structured data goes through here, so the
shell script only ever handles flat lines of text.

Subcommands:
  apply <settings> <state> <mode> [preset] [helper]  rewrite the managed keys
  summary <preset>                                   human lines for the UI
  models <preset>                                    "tier<TAB>model" lines
  set-tier <preset> <tier> <model>                   edit one tier in place
  scaffold <provider>                                print a blank preset
  get <file> <dotted.path>                           print one value
  presets <dir>                                      "name<TAB>provider<TAB>desc"
  scan-sessions <projects> [max-age-days] [ignored]  broken transcripts, as JSON
  ignore-session <ignored> <projects> add|remove|clear|list [id]
  cache-models <cache> <provider> <1|0> [baseUrl]    store a TSV catalogue (stdin)
  preset-rename <presets-dir> <old> <new> <state>    rename, repointing state.json
  auth-of <preset>                                   "mode<TAB>keyRef"; fails if missing
"""

import glob
import json
import os
import re
import shlex
import sys

# Matches both the qualified gateway id (anthropic/claude-opus-5) and the bare
# internal id Claude Code persists (claude-opus-5, claude-haiku-4-5-20251001).
ANTHROPIC_MODEL_RE = re.compile(r"(^|/)claude[-.]|^anthropic/")


def is_anthropic_model(model_id):
    return bool(model_id) and bool(ANTHROPIC_MODEL_RE.search(str(model_id)))

# Must stay in lockstep with the Windows build's $script:BaseManagedEnvKeys.
BASE_MANAGED = [
    "ANTHROPIC_BASE_URL",
    "ANTHROPIC_AUTH_TOKEN",
    "ANTHROPIC_API_KEY",
    "ANTHROPIC_DEFAULT_OPUS_MODEL",
    "ANTHROPIC_DEFAULT_SONNET_MODEL",
    "ANTHROPIC_DEFAULT_HAIKU_MODEL",
    "ANTHROPIC_DEFAULT_FABLE_MODEL",
    # Both pin a concrete model outside the tier mapping. A stale value in either
    # survives a switch, and the small/fast slot is used by background
    # summarisation - a known cause of "works normally, dies on compaction".
    "ANTHROPIC_MODEL",
    "ANTHROPIC_SMALL_FAST_MODEL",
    "CLAUDE_CODE_SUBAGENT_MODEL",
    "CLAUDE_CODE_DISABLE_1M_CONTEXT",
    "CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY",
    "CLAUDE_CODE_ATTRIBUTION_HEADER",
    "CLAUDE_CODE_AUTO_COMPACT_WINDOW",
    "CLAUDE_CODE_MAX_CONTEXT_TOKENS",
    "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC",
    "API_TIMEOUT_MS",
]

TIERS = ["opus", "sonnet", "haiku", "fable"]


def load(path, default=None):
    if not os.path.exists(path):
        return default if default is not None else {}
    with open(path, "r", encoding="utf-8") as fh:
        text = fh.read().strip()
    if not text:
        return default if default is not None else {}
    return json.loads(text)


def save(path, data):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(data, fh, indent=2)
        fh.write("\n")
    os.replace(tmp, path)


def cmd_apply(argv):
    settings_path, state_path, mode = argv[0], argv[1], argv[2]
    preset_path = argv[3] if len(argv) > 3 and argv[3] else None
    helper = argv[4] if len(argv) > 4 and argv[4] else None

    settings = load(settings_path)
    state = load(state_path)

    # Clear the baseline keys plus whatever the previous switch actually wrote,
    # so a preset's custom extraEnv key cannot outlive the preset that added it.
    doomed = set(BASE_MANAGED) | set(state.get("writtenEnvKeys") or [])
    env = settings.get("env")
    if isinstance(env, dict):
        for k in doomed:
            env.pop(k, None)
        if not env:
            settings.pop("env", None)
    settings.pop("apiKeyHelper", None)

    written = []

    if mode != "anthropic":
        if not preset_path:
            raise SystemExit("apply: a gateway mode needs a preset")
        preset = load(preset_path)
        if preset.get("provider") != mode:
            raise SystemExit(
                "preset declares provider '%s', not '%s'" % (preset.get("provider"), mode)
            )

        models = preset.get("models") or {}

        # Cost guard. Gateways resell Anthropic models at full list price with no
        # subscription discount, so routing a tier there is almost never intended.
        # Opt in per preset with "allowAnthropicModels": true.
        if not preset.get("allowAnthropicModels"):
            offenders = []
            for tier in TIERS:
                if is_anthropic_model(models.get(tier)):
                    offenders.append("%s -> %s" % (tier, models[tier]))
            if is_anthropic_model(preset.get("subagentModel")):
                offenders.append("subagent -> %s" % preset["subagentModel"])
            if offenders:
                raise SystemExit(
                    "refusing to switch: preset routes a tier at an Anthropic model "
                    "through '%s' (%s). Gateways bill these at full price. Add "
                    '"allowAnthropicModels": true to the preset if deliberate.'
                    % (mode, "; ".join(offenders))
                )

        block = {}
        block["ANTHROPIC_BASE_URL"] = str(preset["baseUrl"])
        # Explicitly empty, not absent: a cached Anthropic login can otherwise
        # override the gateway config and surface as a model-not-found error.
        # Removed entirely when switching back to anthropic.
        block["ANTHROPIC_API_KEY"] = ""
        for tier in TIERS:
            val = models.get(tier)
            if val:
                block["ANTHROPIC_DEFAULT_%s_MODEL" % tier.upper()] = str(val)

        if preset.get("subagentModel"):
            block["CLAUDE_CODE_SUBAGENT_MODEL"] = str(preset["subagentModel"])
        if preset.get("gatewayModelDiscovery"):
            block["CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY"] = "1"

        # Behind a custom base URL Claude Code cannot resolve a third-party model
        # id to a context length, so it guesses low and auto-compacts early.
        ctx = preset.get("contextTokens")
        if ctx:
            block["CLAUDE_CODE_MAX_CONTEXT_TOKENS"] = str(int(ctx))
            block["CLAUDE_CODE_AUTO_COMPACT_WINDOW"] = str(int(ctx))

        for k, v in (preset.get("extraEnv") or {}).items():
            block[k] = str(v)

        auth = preset.get("auth") or {"mode": "vault", "keyRef": "openrouter"}
        if auth.get("mode") == "vault":
            if not helper:
                raise SystemExit("apply: vault auth needs the key-helper path")
            # Claude Code runs this value as a shell command line, so a $HOME
            # containing a space (common on macOS) has to arrive quoted or the
            # shell splits it and tries to execute the first word. shlex.quote
            # leaves an ordinary path untouched, so nothing churns.
            settings["apiKeyHelper"] = shlex.quote(helper)
        else:
            block["ANTHROPIC_AUTH_TOKEN"] = str(auth.get("token") or "lmstudio")

        env = settings.get("env")
        if isinstance(env, dict):
            env.update(block)
        else:
            settings["env"] = block
        written = list(block.keys())

    save(settings_path, settings)

    state["mode"] = mode
    state["preset"] = os.path.splitext(os.path.basename(preset_path))[0] if preset_path else ""
    state["writtenEnvKeys"] = written
    save(state_path, state)

    for k in written:
        print(k)


def cmd_summary(argv):
    p = load(argv[0])
    if p.get("description"):
        print(p["description"])
    models = p.get("models") or {}
    bits = ["%s=%s" % (t, models[t]) for t in TIERS if models.get(t)]
    if bits:
        print("   ".join(bits))
    ctx = p.get("contextTokens")
    if ctx:
        print("context {:,} tokens   base {}".format(int(ctx), p.get("baseUrl", "")))
    else:
        print("base %s" % p.get("baseUrl", ""))


def cmd_models(argv):
    p = load(argv[0])
    models = p.get("models") or {}
    for t in TIERS:
        print("%s\t%s" % (t, models.get(t, "")))
    print("subagent\t%s" % (p.get("subagentModel") or ""))


def cmd_set_tier(argv):
    path, tier, model = argv[0], argv[1], argv[2]
    p = load(path)
    if tier == "subagent":
        p["subagentModel"] = model
    elif tier in TIERS:
        p.setdefault("models", {})[tier] = model
    else:
        raise SystemExit("unknown tier '%s'" % tier)
    save(path, p)


def cmd_scaffold(argv):
    provider = argv[0]
    base = {"provider": provider, "description": "new preset"}
    if provider == "openrouter":
        base["baseUrl"] = "https://openrouter.ai/api"
        base["auth"] = {"mode": "vault", "keyRef": "openrouter"}
    elif provider == "zai":
        base["baseUrl"] = "https://api.z.ai/api/anthropic"
        base["auth"] = {"mode": "vault", "keyRef": "zai"}
    elif provider == "lmstudio":
        base["baseUrl"] = "http://127.0.0.1:1234"
        base["auth"] = {"mode": "literal", "token": "lmstudio"}
    else:
        raise SystemExit("unknown provider '%s'" % provider)

    base["models"] = {t: "" for t in TIERS}
    base["subagentModel"] = "inherit"
    base["gatewayModelDiscovery"] = provider == "openrouter"
    base["contextTokens"] = 262144 if provider == "lmstudio" else 1000000
    if provider == "lmstudio":
        base["extraEnv"] = {"CLAUDE_CODE_ATTRIBUTION_HEADER": "0"}
    if provider == "zai":
        base["extraEnv"] = {
            "API_TIMEOUT_MS": "3000000",
            "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
        }
    print(json.dumps(base, indent=2))


def cmd_auth_of(argv):
    """auth-of <preset> - 'mode<TAB>keyRef' for the key helper, defaults filled in.

    Exits non-zero when the file is missing, which load() would otherwise
    hide as an empty preset - and an empty preset reads as vault/openrouter,
    so a preset renamed mid-read would quietly hand out the wrong key.
    """
    try:
        with open(argv[0], encoding="utf-8") as fh:
            text = fh.read().strip()
    except OSError:
        sys.exit(1)
    p = json.loads(text) if text else {}
    auth = p.get("auth") or {}
    print("%s\t%s" % (auth.get("mode") or "vault", auth.get("keyRef") or "openrouter"))


def cmd_preset_rename(argv):
    """preset-rename <presets-dir> <old> <new> <state.json>

    Every running session's key helper resolves the active preset through
    state.json on each key fetch, so at no instant may state.json name a file
    that is not there. The new name is linked in first, state.json repointed,
    and only then does the old name go. Prints {"renamed", "active"}.
    """
    d, old, new, state_path = argv[:4]
    src = os.path.join(d, old + ".json")
    dst = os.path.join(d, new + ".json")
    if not os.path.isfile(src):
        raise SystemExit("preset '%s' not found" % old)
    if os.path.exists(dst):
        raise SystemExit("preset '%s' already exists" % new)
    try:
        os.link(src, dst)
    except OSError:
        __import__("shutil").copy2(src, dst)     # filesystems without hard links

    state = load(state_path, {})
    active = isinstance(state, dict) and state.get("preset") == old
    if active:
        state["preset"] = new
        save(state_path, state)

    os.unlink(src)
    print(json.dumps({"renamed": True, "active": active}))


def cmd_get(argv):
    data = load(argv[0])
    cur = data
    for part in argv[1].split("."):
        if isinstance(cur, dict) and part in cur:
            cur = cur[part]
        else:
            return
    if isinstance(cur, (dict, list)):
        print(json.dumps(cur))
    elif isinstance(cur, bool):
        print("true" if cur else "false")
    elif cur is not None:
        print(cur)


def cmd_presets(argv):
    d = argv[0]
    if not os.path.isdir(d):
        return
    for name in sorted(os.listdir(d)):
        if not name.endswith(".json"):
            continue
        try:
            p = load(os.path.join(d, name))
        except Exception:
            continue
        print("%s\t%s\t%s" % (name[:-5], p.get("provider", "openrouter"), p.get("description", "")))


def cmd_managed(argv):
    state = load(argv[0]) if argv else {}
    for k in sorted(set(BASE_MANAGED) | set(state.get("writtenEnvKeys") or [])):
        print(k)


def cmd_settings_env(argv):
    """Print 'KEY=VALUE' for every managed key currently in settings.json."""
    settings = load(argv[0])
    state = load(argv[1]) if len(argv) > 1 else {}
    keys = set(BASE_MANAGED) | set(state.get("writtenEnvKeys") or [])
    if settings.get("apiKeyHelper"):
        print("apiKeyHelper=%s" % settings["apiKeyHelper"])
    env = settings.get("env") or {}
    for k in sorted(keys):
        if k in env:
            print("%s=%s" % (k, env[k]))


def cmd_or_models(argv):
    """OpenRouter /v1/models on stdin -> 'id<TAB>ctx<TAB>$in<TAB>$out' lines."""
    data = json.loads(sys.stdin.read())
    for m in sorted(data.get("data") or [], key=lambda x: x.get("id", "")):
        pricing = m.get("pricing") or {}

        def per_m(key):
            try:
                return round(float(pricing.get(key)) * 1e6, 3)
            except (TypeError, ValueError):
                return ""

        print("%s\t%s\t%s\t%s" % (m.get("id", ""), m.get("context_length", ""), per_m("prompt"), per_m("completion")))


def cmd_cache_models(argv):
    """cache-models <cache.json> <provider> <1|0> [baseUrl]   TSV catalogue on stdin

    Keeps the last catalogue each provider returned, for a reader that cannot
    afford a network round trip - the bar panel's model picker. One node per
    provider with its own fetchedAt and ok, so a failed OpenRouter fetch never
    invalidates a fresh LM Studio list. A failed fetch keeps the previous list
    (a stale list beats none when the network blinks) and marks it ok=false.

    Model ids, context lengths and prices only: never a key reference or token.
    Prints the number of models stored by this call.
    """
    path, provider, ok = argv[0], argv[1], argv[2] == "1"
    base = argv[3].rstrip("/") if len(argv) > 3 else ""

    def num(s, cast):
        try:
            return cast(s)
        except (TypeError, ValueError):
            return None

    rows = []
    for line in sys.stdin.read().splitlines():
        parts = line.split("\t")
        mid = parts[0].strip() if parts else ""
        if not mid:
            continue
        m = {"id": mid}
        if provider == "openrouter":
            ctx, pin, pout = (parts[1:] + ["", "", ""])[:3]
            m["contextTokens"] = num(ctx, int)
            m["priceIn"] = num(pin, float)
            m["priceOut"] = num(pout, float)
        elif provider == "lmstudio":
            state, ctx = (parts[1:] + ["", ""])[:2]
            m["state"] = state
            m["contextTokens"] = num(ctx, int)
        else:
            m["note"] = parts[1] if len(parts) > 1 else ""
        rows.append(m)

    try:
        data = load(path, {})
    except (OSError, ValueError):
        data = {}
    providers = data.get("providers") if isinstance(data, dict) else None
    if not isinstance(providers, dict):
        providers = {}
    node = providers.get(provider)
    if not isinstance(node, dict):
        node = {}

    now = __import__("datetime").datetime.now(
        __import__("datetime").timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    if ok:
        node = {"fetchedAt": now, "ok": True, "models": rows}
        if provider == "lmstudio":
            node["baseUrl"] = base
    else:
        # A list from a different LM Studio server is not stale, it is wrong.
        if provider == "lmstudio" and node.get("baseUrl", "") != base:
            node = {"baseUrl": base, "models": []}
        node["ok"] = False
        node["failedAt"] = now

    providers[provider] = node
    save(path, {"schema": 1, "providers": providers})
    print(len(rows) if ok else 0)


def cmd_lms_models(argv):
    """LM Studio /api/v0/models on stdin -> 'id<TAB>state<TAB>ctx' lines."""
    data = json.loads(sys.stdin.read())
    for m in sorted(data.get("data") or [], key=lambda x: x.get("id", "")):
        print("%s\t%s\t%s" % (m.get("id", ""), m.get("state", "unknown"), m.get("max_context_length", "")))


# A model id carrying a bracket suffix - e.g. `claude-fable-5[1m]` - is Claude
# Code's extended-context marker. It belongs to Anthropic's 1M models and no
# gateway recognises it. A session that had it can carry the tag onto a new id
# after a switch, producing something like
# `~deepseek/deepseek-v4-flash-latest[1m]` that only fails at compaction, because
# compaction re-resolves the model from session state.
TAGGED_MODEL_RE = re.compile(r"\[[0-9]+[a-zA-Z]\]$")


def walk_model_ids(node, out):
    """Collect model-ish strings anywhere in the config.

    Claude Code caches resolved models under more than one key - both
    clientDataCacheSlots and additionalModelOptionsCache have been observed - so
    scan rather than reach for a fixed path.
    """
    if isinstance(node, dict):
        for k, v in node.items():
            if isinstance(v, str):
                if k in ("model", "value") and v and (v[0].isalnum() or v[0] == "~"):
                    out.add(v)
            else:
                walk_model_ids(v, out)
    elif isinstance(node, list):
        for item in node:
            walk_model_ids(item, out)


def cmd_stale_models(argv):
    """Cached model ids worth flagging, as 'kind<TAB>id' lines.

    kind is 'anthropic' (would bill at gateway list price) or 'tagged' (carries a
    [1m]-style marker that breaks compaction on a gateway).
    """
    ids = set()
    walk_model_ids(load(argv[0], {}), ids)
    for m in sorted(i for i in ids if is_anthropic_model(i)):
        print("anthropic\t" + m)
    for m in sorted(i for i in ids if TAGGED_MODEL_RE.search(i)):
        print("tagged\t" + m)


def cmd_strip_tags(argv):
    """Remove [1m]-style suffixes from cached model ids.

    The tag is NOT junk everywhere: on an Anthropic id it is how Claude Code
    selects the 1M variant, so stripping it silently downgrades that choice to
    200k. On a gateway id the same tag is meaningless and breaks compaction.
    Default is gateway ids only; pass 'all' as argv[1] to include Anthropic ids.

    Prints 'strip<TAB>id' and 'keep<TAB>id' lines.
    """
    path = argv[0]
    strip_all = len(argv) > 1 and argv[1] == "all"

    raw = open(path, encoding="utf-8").read()
    tagged = sorted(set(re.findall(r'"([^"]*\[[0-9]+[a-zA-Z]\])"', raw)))
    if not tagged:
        return

    target = [t for t in tagged if strip_all or not is_anthropic_model(t)]
    kept = [t for t in tagged if t not in target]

    for k in kept:
        print("keep\t" + k)
    if not target:
        return

    fixed = raw
    for t in target:
        fixed = fixed.replace('"' + t + '"', '"' + re.sub(r"\[[0-9]+[a-zA-Z]\]$", "", t) + '"')
    json.loads(fixed)  # refuse to write anything that is not valid JSON
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(fixed)
    for t in target:
        print("strip\t" + t)


def cmd_health(argv):
    """Write health.json - the machine-readable state a fleet reader consumes.

    usage: health <root> <claude.json> <mode> <preset> <guardrail> <keyBackend> <version>

    Contract, deliberately narrow:
      * NO key material. keysConfigured is names only.
      * Model lists are [{id, anthropic}] so a reader never re-derives the
        Anthropic matcher. A tagged *Anthropic* id is normal (that is how the 1M
        variant is selected) and must not render as a fault.
      * guardrailStatus is tri-state or null; never collapse it to a boolean.
    """
    root, cfg_path, mode, preset, guardrail, key_backend, version = (argv + [""] * 7)[:7]

    ids = set()
    walk_model_ids(load(cfg_path, {}), ids)
    try:
        raw = open(cfg_path, encoding="utf-8").read()
        ids.update(re.findall(r'"([^"]*\[[0-9]+[a-zA-Z]\])"', raw))
    except OSError:
        pass

    def entries(seq):
        return [{"id": i, "anthropic": bool(is_anthropic_model(i))} for i in sorted(seq)]

    h = {
        "schema": 1,
        "tool": "claude-mode",
        "version": version or "0.0.0",
        "updatedAt": __import__("datetime").datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
        "os": "posix",
        "mode": mode,
        "preset": "" if mode == "anthropic" else preset,
    }

    tagged = entries(i for i in ids if TAGGED_MODEL_RE.search(i))

    if mode == "anthropic":
        h["staleModelIds"] = []
        h["taggedModelIds"] = tagged
    else:
        p = load(os.path.join(root, "presets", preset + ".json"), {})
        h["provider"] = p.get("provider", "")
        h["baseUrl"] = p.get("baseUrl", "")
        h["models"] = {t: p.get("models", {}).get(t, "") for t in TIERS if p.get("models", {}).get(t)}
        h["subagentModel"] = p.get("subagentModel", "")
        h["contextTokens"] = int(p["contextTokens"]) if p.get("contextTokens") else None
        h["gatewayDiscovery"] = bool(p.get("gatewayModelDiscovery"))
        auth = p.get("auth") or {}
        if auth.get("mode", "vault") == "vault":
            h["keyBackend"] = key_backend or "unknown"
            h["keysConfigured"] = [auth.get("keyRef", "openrouter")]
        else:
            h["keyBackend"] = "inline"
            h["keysConfigured"] = []
        h["costGuardPassed"] = True
        h["guardrailStatus"] = guardrail or None
        h["staleModelIds"] = entries(i for i in ids if is_anthropic_model(i))
        h["taggedModelIds"] = tagged

    # The switchable catalogue, so a reader (the Omarchy bar widget) can offer
    # every preset without re-walking the presets directory itself. Names,
    # providers, descriptions and model maps only - the same shape already
    # published for the active preset, and no key material rides along.
    catalogue = []
    pdir = os.path.join(root, "presets")
    if os.path.isdir(pdir):
        for fname in sorted(os.listdir(pdir)):
            if not fname.endswith(".json"):
                continue
            pp = load(os.path.join(pdir, fname), {})
            if not isinstance(pp, dict):
                continue
            pauth = pp.get("auth") or {}
            pmode = pauth.get("mode", "vault")
            catalogue.append({
                "name": fname[:-5],
                "provider": pp.get("provider", "openrouter"),
                "description": pp.get("description", ""),
                "contextTokens": int(pp["contextTokens"]) if pp.get("contextTokens") else None,
                "models": {t: pp.get("models", {}).get(t, "")
                           for t in TIERS if pp.get("models", {}).get(t)},
                # Enough for a reader to render an editing form without opening
                # the preset file: where the server is, and whether it is set to
                # send a real credential. The credential itself never appears -
                # only which named slot it would come from.
                "baseUrl": pp.get("baseUrl", ""),
                "authMode": pmode,
                "keyRef": pauth.get("keyRef", "") if pmode == "vault" else "",
            })
    h["presets"] = catalogue

    save(os.path.join(root, "health.json"), h)



def cmd_preflight_json(argv):
    """Emit a preflight verdict as JSON.

    usage: preflight-json <ok|blocked> <mode> <preset> <code> <title> <detail>
                          <remedy> <remedyKind> <keyRef> <baseUrl>

    The shell side does the checking; this exists so the strings reach a reader
    correctly quoted rather than through hand-rolled escaping in bash.
    """
    (status, mode, preset, code, title, detail,
     remedy, kind, key_ref, base_url) = (argv + [""] * 10)[:10]
    out = {
        "ok": status == "ok",
        "mode": mode,
        "preset": preset,
        "code": code or ("ok" if status == "ok" else "blocked"),
        "title": title,
        "detail": detail,
        "remedy": remedy,
        "remedyKind": kind,
        "keyRef": key_ref,
        "baseUrl": base_url,
    }
    print(json.dumps(out))


def cmd_sessions_json(argv):
    """Convert the session TSV on stdin to JSON.

    Columns, in order: pid, ppid, tty, busy, cwd, self, parentCmd. `busy` is a
    sampled-CPU heuristic, not a promise, and is reported as such.
    """
    rows = []
    for line in sys.stdin.read().splitlines():
        if not line.strip():
            continue
        parts = line.split("\t")
        parts += [""] * (7 - len(parts))
        pid, ppid, tty, busy, cwd, is_self, pcmd = parts[:7]
        try:
            pid_i = int(pid)
        except ValueError:
            continue
        rows.append({
            "pid": pid_i,
            "ppid": int(ppid) if ppid.isdigit() else 0,
            "tty": tty,
            "busy": busy == "yes",
            "cwd": cwd,
            "self": is_self == "yes",
            "parentCmd": pcmd,
        })
    print(json.dumps({"count": len(rows), "busy": sum(1 for r in rows if r["busy"]),
                      "sessions": rows}))


def cmd_set_url(argv):
    """set-url <preset> <baseUrl>

    LM Studio is not necessarily on this machine. It can be another box on the
    LAN, or something reached through a tunnel or a reverse proxy, so the base
    URL is editable rather than fixed at the loopback address it ships with.
    """
    path, url = argv[0], argv[1].strip().rstrip("/")
    if not re.match(r"^https?://[^\s/]+", url):
        raise SystemExit("base URL must start with http:// or https://")
    p = load(path)
    p["baseUrl"] = url
    save(path, p)
    print(url)


def cmd_set_auth(argv):
    """set-auth <preset> none|key [keyRef]

    `none` is LM Studio's out-of-the-box state: it accepts any token, so the
    literal placeholder is written inline and is explicitly not a secret. `key`
    is for a server with authentication switched on, where the token is a real
    credential and belongs in the vault like every other one - settings.json
    never sees it either way.
    """
    path, mode = argv[0], argv[1]
    key_ref = argv[2] if len(argv) > 2 and argv[2] else "lmstudio"
    p = load(path)
    if mode == "none":
        p["auth"] = {"mode": "literal", "token": "lmstudio"}
    elif mode == "key":
        p["auth"] = {"mode": "vault", "keyRef": key_ref}
    else:
        raise SystemExit("auth mode must be 'none' or 'key'")
    save(path, p)
    print(json.dumps(p["auth"]))


def cmd_set_flag(argv):
    """set-flag <preset> <key> true|false

    Only used for `configured` so far. Kept generic because a preset-level
    boolean written by hand is exactly the kind of thing that ends up as a
    string "false", which is truthy everywhere that matters.
    """
    path, key, val = argv[0], argv[1], argv[2]
    p = load(path)
    p[key] = (val == "true")
    save(path, p)


def cmd_set_all(argv):
    """set-all <preset> <model-id> - point every tier and the subagent at one id."""
    path, model = argv[0], argv[1]
    p = load(path)
    p["models"] = {t: model for t in TIERS}
    if p.get("subagentModel") and p["subagentModel"] != "inherit":
        p["subagentModel"] = model
    save(path, p)
    print(model)


# ---------------------------------------------------------------------------
# Session transcript repair
#
# Native Anthropic requires previous_message_id to be an id it issued itself -
# one starting `msg_`. A session that took even one completion from a gateway
# while the mode was switched under it has that provider's id format in its
# transcript instead (OpenRouter issues `gen-<epoch>-<rand>`), and every attempt
# to resume it afterwards fails with a 400 naming previous_message_id. Client-
# side error placeholders, written as model `<synthetic>` with a UUID for an id,
# do the same thing when one is last.
#
# The transcript is newline-delimited JSON, one independent object per line, so
# rolling back to the last message Anthropic actually issued is a truncation.
# Everything after it is lost - which is the cost, and why nothing here runs
# without being asked twice.
# ---------------------------------------------------------------------------

def _msg_id(obj):
    m = obj.get("message")
    if isinstance(m, dict) and m.get("id"):
        return str(m["id"])
    return None



def _blocks(msg):
    c = msg.get("content")
    if isinstance(c, str):
        return [{"type": "text", "text": c}]
    return c if isinstance(c, list) else []


def _clip(t, n):
    t = " ".join(str(t).split())
    return t if len(t) <= n else t[:n - 1] + "\u2026"


def summarise_dropped(lines, per_block=1400):
    """Render the turns a repair is about to discard as readable Markdown.

    Tool *results* are deliberately left out. They are the bulk of a transcript
    by volume and the least useful part of a summary - what matters on the way
    back in is what was asked, what was said, and what was run.
    """
    out, tools = [], 0
    for line in lines:
        if not line.strip():
            continue
        try:
            d = json.loads(line)
        except Exception:
            continue
        t, msg = d.get("type"), (d.get("message") or {})
        if t == "user" and not d.get("isMeta"):
            for b in _blocks(msg):
                if b.get("type") == "text" and b.get("text", "").strip():
                    out.append("### User\n\n" + _clip(b["text"], per_block))
        elif t == "assistant":
            if d.get("isApiErrorMessage"):
                for b in _blocks(msg):
                    if b.get("type") == "text":
                        out.append("> **API error:** " + _clip(b.get("text", ""), 300))
                continue
            said, ran = [], []
            for b in _blocks(msg):
                if b.get("type") == "text" and b.get("text", "").strip():
                    said.append(_clip(b["text"], per_block))
                elif b.get("type") == "tool_use":
                    inp = b.get("input") or {}
                    hint = inp.get("command") or inp.get("file_path") or inp.get("pattern") or inp.get("path") or ""
                    ran.append("`%s`%s" % (b.get("name", "tool"),
                                           (" \u2014 " + _clip(hint, 120)) if hint else ""))
                    tools += 1
            if said:
                out.append("### Claude\n\n" + "\n\n".join(said))
            if ran:
                out.append("Ran: " + ", ".join(ran[:12]) + (" \u2026" if len(ran) > 12 else ""))
    return "\n\n".join(out), tools


def _meta_entry(template, parent_uuid, text):
    """A user-role entry marked isMeta, which Claude Code treats as context
    rather than as something to answer - the same marker it uses for its own
    local-command caveats. It carries no message.id, so it cannot affect the
    previous_message_id that made the session unresumable in the first place.
    """
    now = __import__("datetime").datetime.now(
        __import__("datetime").timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    e = {
        "parentUuid": parent_uuid,
        "isSidechain": False,
        "userType": template.get("userType", "external"),
        "cwd": template.get("cwd", ""),
        "sessionId": template.get("sessionId", ""),
        "version": template.get("version", ""),
        "gitBranch": template.get("gitBranch", ""),
        "type": "user",
        "isMeta": True,
        "uuid": str(__import__("uuid").uuid4()),
        "timestamp": now,
        "message": {"role": "user", "content": text},
    }
    return {k: v for k, v in e.items() if v != "" or k in ("gitBranch",)}

def cmd_repair_session(argv):
    """repair-session <transcript.jsonl> [--apply] [--no-reinject]

    Prints a JSON verdict. With --apply, writes the dropped turns out as
    Markdown, backs the file up, truncates it to the last Anthropic-issued
    message, and (unless --no-reinject) appends those turns back as a single
    isMeta note so the resumed session still knows what it did.
    """
    path = argv[0]
    apply_it = "--apply" in argv[1:]
    reinject = "--no-reinject" not in argv[1:]

    with open(path, encoding="utf-8") as fh:
        lines = fh.read().splitlines()

    last_good = -1          # index of the last line carrying a msg_ id
    foreign, synthetic = [], []
    for i, line in enumerate(lines):
        if not line.strip():
            continue
        try:
            obj = json.loads(line)
        except Exception:
            continue
        mid = _msg_id(obj)
        if mid is None:
            continue
        if mid.startswith("msg_"):
            last_good = i
        elif mid.startswith("gen-"):
            foreign.append({"line": i + 1, "id": mid,
                            "model": str((obj.get("message") or {}).get("model", ""))})
        else:
            synthetic.append({"line": i + 1, "id": mid,
                              "apiError": bool(obj.get("isApiErrorMessage"))})

    # Only the *last* id matters for resuming: an error placeholder in the
    # middle of a long-finished turn is history, not a blocker.
    tail_ids = [i for i in range(len(lines) - 1, last_good, -1)
                if lines[i].strip() and _msg_id(_safe(lines[i])) is not None]
    healthy = (last_good >= 0 and not tail_ids)

    # Not every transcript without an Anthropic id is damaged, and saying so
    # turns the scan into noise. A session that ran start to finish on a gateway
    # has `gen-` ids throughout by design: it resumes perfectly well under the
    # provider it was born on, has nothing to truncate back to, and is only a
    # problem if you try to resume it as Anthropic. Likewise a session that never
    # got a reply at all has no ids and nothing wrong with it.
    if healthy:
        kind = "healthy"
    elif last_good >= 0:
        kind = "repairable"
    elif foreign:
        kind = "gateway-native"
    elif synthetic:
        kind = "synthetic-only"
    else:
        kind = "no-messages"

    out = {
        "path": path,
        "lines": len(lines),
        "lastGoodLine": last_good + 1 if last_good >= 0 else 0,
        "dropLines": 0 if healthy or last_good < 0 else len(lines) - (last_good + 1),
        "foreignIds": foreign,
        "syntheticIds": synthetic,
        "healthy": healthy,
        "kind": kind,
        "repairable": (not healthy) and last_good >= 0,
        "applied": False,
        "backup": "",
        "recovered": "",
        "reinjected": False,
        "droppedTools": 0,
    }

    if apply_it and out["repairable"]:
        stamp = __import__("datetime").datetime.now().strftime("%Y%m%d-%H%M%S")
        backup = "%s.pre-repair-backup-%s" % (path, stamp)
        dropped = lines[last_good + 1:]
        keep = lines[:last_good + 1]

        # The turns being cut are the work itself. Losing the conversation that
        # produced a morning's changes is most of the damage, so they are written
        # out as readable Markdown before anything is deleted...
        digest, ntools = summarise_dropped(dropped)
        recovered = "%s.recovered-%s.md" % (path.rsplit(".jsonl", 1)[0], stamp)
        header = ("# Recovered turns\n\n"
                  "Cut from `%s` on %s, because the transcript could no longer be\n"
                  "resumed. Everything below happened; none of it is in the session any more.\n\n"
                  "---\n\n" % (os.path.basename(path), stamp))
        with open(recovered, "w", encoding="utf-8") as fh:
            fh.write(header + (digest or "_No readable messages in the dropped turns._\n"))

        with open(backup, "w", encoding="utf-8") as fh:
            fh.write("\n".join(lines) + ("\n" if lines else ""))

        # ...and then handed back to the session, so the agent that resumes it
        # knows what it just did rather than waking with a gap in its memory.
        if reinject and digest:
            template = _safe(keep[last_good])
            body = digest if len(digest) <= 24000 else (
                digest[:24000] + "\n\n\u2026 truncated; the full text is in " + recovered)
            note = ("<recovered-transcript>\n"
                    "Context, not an instruction - do not act on it or reply to it.\n\n"
                    "This session was rolled back to its last resumable point after a "
                    "provider switch left it unable to resume. The turns below were part "
                    "of this conversation and are no longer in it. The full text is at "
                    + recovered + "\n\n" + body + "\n</recovered-transcript>")
            keep.append(json.dumps(_meta_entry(template, template.get("uuid"), note)))
            out["reinjected"] = True

        with open(path, "w", encoding="utf-8") as fh:
            fh.write("\n".join(keep) + "\n")

        out["applied"] = True
        out["backup"] = backup
        out["recovered"] = recovered
        out["droppedTools"] = ntools

    print(json.dumps(out))


def _safe(line):
    try:
        return json.loads(line)
    except Exception:
        return {}


def _last_msg_id_from_tail(path, window=262144):
    """The id of the last message-bearing line, read from the end of the file.

    Classifying a transcript needs two facts: what the last message id is, and
    whether any Anthropic id exists at all. The first decides the common case on
    its own - if the last one is `msg_`, the transcript is healthy and the rest
    of the file never has to be touched. Since that is 40-odd of the 59 here,
    and some are 10MB, reading the tail first is the difference between a scan
    that can run on a timer and one that cannot.

    Returns (id_or_None, saw_whole_file).
    """
    size = os.path.getsize(path)
    with open(path, "rb") as fh:
        start = max(0, size - window)
        fh.seek(start)
        chunk = fh.read()
    whole = start == 0
    if not whole:
        nl = chunk.find(b"\n")          # drop the partial first line
        chunk = chunk[nl + 1:] if nl >= 0 else b""
    for raw in reversed(chunk.splitlines()):
        if not raw.strip():
            continue
        try:
            d = json.loads(raw.decode("utf-8", "replace"))
        except Exception:
            continue
        mid = _msg_id(d)
        if mid:
            return mid, whole
    return None, whole


def _broken_entry(path):
    """The scan's record for one transcript if it needs repair, else None.

    Raises OSError when the file cannot be read, which the scan counts as
    skipped rather than as healthy.
    """
    last, whole = _last_msg_id_from_tail(path)
    # Healthy is decided by the tail alone, and is the common case.
    if last is not None and last.startswith("msg_"):
        return None
    if last is None and whole:
        return None                           # no replies at all; nothing to fix

    with open(path, encoding="utf-8", errors="replace") as fh:
        lines = fh.read().splitlines()

    last_good, foreign = -1, []
    for i, line in enumerate(lines):
        if not line.strip():
            continue
        try:
            d = json.loads(line)
        except Exception:
            continue
        mid = _msg_id(d)
        if not mid:
            continue
        if mid.startswith("msg_"):
            last_good = i
        elif mid.startswith("gen-"):
            foreign.append(str((d.get("message") or {}).get("model", "")))

    if last_good < 0:
        return None                           # gateway-native or reply-less
    tail = [i for i in range(len(lines) - 1, last_good, -1)
            if lines[i].strip() and _msg_id(_safe(lines[i])) is not None]
    if not tail:
        return None

    return {
        "sessionId": os.path.basename(path)[:-6],
        "project": os.path.basename(os.path.dirname(path)),
        "path": path,
        "lines": len(lines),
        "dropLines": len(lines) - (last_good + 1),
        "providers": sorted(set(f for f in foreign if f)),
        "mtime": int(os.path.getmtime(path)),
    }


# ---------------------------------------------------------------------------
# Dismissed sessions
#
# Some damage is never going to be repaired - a throwaway session, or one whose
# work was finished another way - and without a way to say so it holds the
# bar's warning dot on forever, which teaches you to ignore the dot. So a broken
# session can be dismissed, and one untouched for longer than a threshold is
# hidden on its own. Neither is deleted or forgotten: both move to `ignored[]`,
# with the reason, and stay one command away.
#
# Its own file because nothing else can hold authored state: state.json is
# rewritten wholesale on every switch, health.json on every `health`.
# ---------------------------------------------------------------------------

IGNORE_AGE_DAYS = 7
SESSION_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$")


def _load_ignored(path):
    try:
        data = load(path, {})
    except (OSError, ValueError):
        data = {}                             # a corrupt file must not break the scan
    sessions = data.get("sessions") if isinstance(data, dict) else None
    return sessions if isinstance(sessions, dict) else {}


def _save_ignored(path, sessions):
    save(path, {"schema": 1, "sessions": sessions})


def _find_transcript(projects, sid):
    hits = sorted(glob.glob(os.path.join(projects, "*", sid + ".jsonl")))
    return hits[0] if hits else None


def _prune_ignored(projects, sessions):
    """Drop entries whose transcript has gone, so a session id that is later
    reused - or restored from a backup - does not come back already hidden."""
    gone = [sid for sid in sessions if not _find_transcript(projects, sid)]
    for sid in gone:
        del sessions[sid]
    return len(gone)


def cmd_ignore_session(argv):
    """ignore-session <ignored.json> <projects-dir> add|remove|clear|list [id]

    Prints JSON. `list` reconciles every entry against the disk: whether the
    transcript still exists, and whether it is still broken at all.
    """
    path, projects, action = argv[0], argv[1], argv[2]
    sid = argv[3] if len(argv) > 3 else ""
    if sid.endswith(".jsonl"):
        sid = sid[:-6]
    if action in ("add", "remove") and not SESSION_ID_RE.match(sid):
        raise SystemExit("not a session id: '%s'" % sid)

    sessions = _load_ignored(path)

    if action == "list":
        rows = []
        for key in sorted(sessions, key=lambda k: str(sessions[k].get("ignoredAt", ""))):
            found = _find_transcript(projects, key)
            try:
                still = bool(found and _broken_entry(found))
            except OSError:
                still = False
            rows.append({"sessionId": key,
                         "project": sessions[key].get("project", ""),
                         "ignoredAt": sessions[key].get("ignoredAt", ""),
                         "exists": bool(found), "broken": still})
        print(json.dumps({"sessions": rows, "count": len(rows)}))
        return

    pruned = _prune_ignored(projects, sessions)
    changed = False

    if action == "add":
        found = _find_transcript(projects, sid)
        if not found:
            raise SystemExit("no transcript for session '%s'" % sid)
        if sid not in sessions:
            sessions[sid] = {
                "project": os.path.basename(os.path.dirname(found)),
                "ignoredAt": __import__("datetime").datetime.now(
                    __import__("datetime").timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            }
            changed = True
    elif action == "remove":
        changed = sessions.pop(sid, None) is not None
    elif action == "clear":
        changed = bool(sessions)
        sessions = {}
    else:
        raise SystemExit("ignore-session: unknown action '%s'" % action)

    if changed or pruned:
        _save_ignored(path, sessions)
    print(json.dumps({"action": action, "sessionId": sid, "changed": changed,
                      "pruned": pruned, "count": len(sessions)}))


def cmd_scan_sessions(argv):
    """scan-sessions <projects-dir> [max-age-days] [ignored.json]

    JSON list of transcripts worth repairing. Only `repairable` is reported: a
    transcript carrying a genuine Anthropic message with a different provider's
    output after it. A session that ran entirely on a gateway, or never got a
    reply, is not damage.

    Broken transcripts that were dismissed, or not written to for longer than
    max-age-days (default 7; 0 turns the age rule off), are reported under
    `ignored[]` with a reason instead of under `broken[]`.
    """
    root = argv[0]
    age_arg = argv[1] if len(argv) > 1 else ""
    try:
        max_age = int(age_arg) if age_arg != "" else IGNORE_AGE_DAYS
    except ValueError:
        raise SystemExit("max age must be a whole number of days, not '%s'" % age_arg)
    dismissed = _load_ignored(argv[2]) if len(argv) > 2 and argv[2] else {}
    cutoff = __import__("time").time() - max_age * 86400 if max_age > 0 else None

    broken, ignored, scanned, skipped = [], [], 0, 0

    for proj in sorted(glob.glob(os.path.join(root, "*"))):
        if not os.path.isdir(proj):
            continue
        for path in sorted(glob.glob(os.path.join(proj, "*.jsonl"))):
            scanned += 1
            try:
                entry = _broken_entry(path)
            except OSError:
                skipped += 1
                continue
            if entry is None:
                continue
            # Dismissed outranks stale: it is the reason a person gave.
            if entry["sessionId"] in dismissed:
                entry["reason"] = "dismissed"
                entry["ignoredAt"] = dismissed[entry["sessionId"]].get("ignoredAt", "")
                ignored.append(entry)
            elif cutoff is not None and entry["mtime"] < cutoff:
                entry["reason"] = "stale"
                ignored.append(entry)
            else:
                broken.append(entry)

    print(json.dumps({"scanned": scanned, "skipped": skipped,
                      "broken": broken, "count": len(broken),
                      "ignored": ignored, "ignoredCount": len(ignored),
                      "maxAgeDays": max_age}))

COMMANDS = {
    "health": cmd_health,
    "preflight-json": cmd_preflight_json,
    "sessions-json": cmd_sessions_json,
    "repair-session": cmd_repair_session,
    "scan-sessions": cmd_scan_sessions,
    "ignore-session": cmd_ignore_session,
    "stale-models": cmd_stale_models,
    "strip-tags": cmd_strip_tags,
    "or-models": cmd_or_models,
    "lms-models": cmd_lms_models,
    "cache-models": cmd_cache_models,
    "apply": cmd_apply,
    "summary": cmd_summary,
    "models": cmd_models,
    "set-tier": cmd_set_tier,
    "set-url": cmd_set_url,
    "set-flag": cmd_set_flag,
    "set-all": cmd_set_all,
    "set-auth": cmd_set_auth,
    "scaffold": cmd_scaffold,
    "preset-rename": cmd_preset_rename,
    "auth-of": cmd_auth_of,
    "get": cmd_get,
    "presets": cmd_presets,
    "managed": cmd_managed,
    "settings-env": cmd_settings_env,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
        sys.exit("usage: cm-json.py <%s> ..." % "|".join(sorted(COMMANDS)))
    try:
        COMMANDS[sys.argv[1]](sys.argv[2:])
    except SystemExit:
        raise
    except Exception as exc:  # surface a clean message to the shell
        sys.exit("cm-json: %s" % exc)