✨ Gremlins MVP: store, launch, stream, call modules and templates

plugins/spire/skills/gremlins/grm/call.py

@@ -0,0 +1,86 @@
+"""The only subprocess code: run one `claude -p` call in the foreground and tee its stream."""
+import json
+import signal
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+POLL_S = 0.2
+READER_JOIN_S = 10.0   # an orphaned grandchild can hold stdout open after claude exits
+STOP_SIGNALS = tuple(s for s in (signal.SIGTERM, signal.SIGINT, getattr(signal, "SIGBREAK", None))
+                     if s is not None)
+
+
+def _feed(stdin, data: bytes) -> None:
+    """Write the prompt and close stdin; a child that exits early is not an error here."""
+    try:
+        stdin.write(data)
+    except (OSError, ValueError):
+        pass
+    try:
+        stdin.close()
+    except (OSError, ValueError):
+        pass
+
+
+def _read(stdout, raw, ss) -> None:
+    """Tee every stream line to the raw file and feed each JSON event to `ss`."""
+    try:
+        for line in iter(stdout.readline, b""):
+            raw.write(line)
+            raw.flush()
+            try:
+                ev = json.loads(line.decode("utf-8", "replace"))
+            except ValueError:
+                continue
+            ss.feed(ev, time.time())
+    except (OSError, ValueError):
+        pass   # the raw file or pipe was closed after READER_JOIN_S
+
+
+def run_claude(argv, prompt: str, env, cwd, raw_path, on_start, ss=None):
+    """Run one call to its end. Returns (ss, exit_code, stopped)."""
+    if ss is None:
+        from grm.stream import StreamState
+        ss = StreamState()
+    raw_path = Path(raw_path)
+    with open(raw_path, "wb") as raw, open(raw_path.with_suffix(".err"), "wb") as err:
+        child = subprocess.Popen(argv, cwd=cwd, env=env, stdin=subprocess.PIPE,
+                                 stdout=subprocess.PIPE, stderr=err)
+        stopped = False
+
+        def _stop(signum, frame):
+            nonlocal stopped
+            try:
+                if stopped:          # a second signal: stop waiting for a graceful exit
+                    child.kill()
+                else:
+                    child.terminate()
+            except OSError:
+                pass
+            stopped = True
+
+        old = {}
+        if threading.current_thread() is threading.main_thread():
+            old = {s: signal.signal(s, _stop) for s in STOP_SIGNALS}
+        try:
+            reader = threading.Thread(target=_read, args=(child.stdout, raw, ss), daemon=True)
+            reader.start()
+            threading.Thread(target=_feed, args=(child.stdin, prompt.encode("utf-8")),
+                             daemon=True).start()
+            on_start(child.pid)
+            while child.poll() is None:   # a bare wait() would block signal handlers on Windows
+                time.sleep(POLL_S)
+        except BaseException:
+            child.kill()
+            child.wait()
+            raise
+        finally:
+            for s, h in old.items():
+                if h is not None:
+                    signal.signal(s, h)
+        reader.join(READER_JOIN_S)
+        if not reader.is_alive():
+            child.stdout.close()
+    return ss, child.returncode, stopped

plugins/spire/skills/gremlins/grm/launch.py

@@ -0,0 +1,182 @@
+"""Launch fingerprint, env scrub, binary lookup and the `claude -p` argv (B1, B2, B7, B8)."""
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+
+MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$")
+EFFORTS = ("low", "medium", "high", "xhigh", "max")
+PERMISSION_MODE = "dontAsk"
+DEFAULT_TOOLS = ["Bash", "Read", "Edit", "Write", "Glob", "Grep", "Skill"]
+DEFAULT_DISALLOWED = ["Bash(git push:*)", "Bash(git rebase:*)", "Bash(git reset --hard:*)",
+                      "Bash(claude:*)"]
+
+# Env pinned into every launch; part of the fingerprint.
+ENV_PINNED_LAUNCH = {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"}
+
+# Host env that must not reach a child `claude -p`.
+SCRUB_DROP = frozenset({
+    # Desktop host vars
+    "CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_MESSAGING_TOKEN", "CLAUDE_CODE_HOST_SESSION_ID",
+    "CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_DESKTOP_APP_VERSION",
+    "CLAUDE_CODE_SDK_HAS_HOST_AUTH_REFRESH", "CLAUDE_CODE_TERMINAL_MCP_TOOLS",
+    "CLAUDE_CODE_REPORT_FINDINGS", "CLAUDE_CODE_SESSION_ID", "CLAUDE_AGENT_SDK_VERSION",
+    "CLAUDE_CODE_ENABLE_ASK_USER_QUESTION_TOOL", "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING",
+    # fingerprint and auth path
+    "CLAUDE_EFFORT", "ANTHROPIC_MODEL", "ANTHROPIC_SMALL_FAST_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL",
+    "ENABLE_PROMPT_CACHING_1H", "FORCE_PROMPT_CACHING_5M", "CLAUDE_CODE_SUBAGENT_PROMPT_CACHE_TTL",
+    "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX",
+    # uv
+    "VIRTUAL_ENV",
+})
+SCRUB_DROP_PATTERNS = (re.compile(r"^ANTHROPIC_DEFAULT_.+_MODEL$"), re.compile(r"^UV_"))
+SCRUB_KEEP = frozenset({
+    "CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH",
+    "PATH", "HOME", "USERPROFILE",
+    "ANTHROPIC_BASE_URL",      # kept: gateways need it
+})
+SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_GREMLIN_CHILD": "1"}
+SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH"   # set to the stored absolute binary path
+
+
+def canonical_json(obj) -> str:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+
+
+def _sha256_bytes(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+
+def _sha256_file(path: str) -> str:
+    with open(path, "rb") as f:
+        return _sha256_bytes(f.read())
+
+
+def scrub_class(name: str) -> str | None:
+    """'drop', 'pinned', 'keep' or None (on no list; kept)."""
+    n = name.upper()
+    if n in SCRUB_DROP or any(p.match(n) for p in SCRUB_DROP_PATTERNS):
+        return "drop"
+    if n in SCRUB_PINNED or n == SCRUB_PINNED_BINARY:
+        return "pinned"
+    if n in SCRUB_KEEP:
+        return "keep"
+    return None
+
+
+def _venv_prefixes(environ) -> list[str]:
+    out = []
+    venv = environ.get("VIRTUAL_ENV")
+    if venv:
+        out.append(venv)
+    if os.path.normcase(os.path.abspath(sys.prefix)) != os.path.normcase(os.path.abspath(sys.base_prefix)):
+        out.append(sys.prefix)
+    return [os.path.normcase(os.path.normpath(os.path.abspath(x))) for x in out]
+
+
+def _under(path: str, prefixes) -> bool:
+    try:
+        p = os.path.normcase(os.path.normpath(os.path.abspath(path)))
+    except ValueError:
+        return False
+    return any(p == px or p.startswith(px.rstrip("\\/") + os.sep) for px in prefixes)
+
+
+def scrubbed_env(binary: str, environ=None) -> dict:
+    """The child env: host env minus the drop list, venv PATH entries removed, pinned vars on
+    top, `CLAUDE_CODE_EXECPATH` = the stored binary path."""
+    if environ is None:
+        environ = os.environ
+    prefixes = _venv_prefixes(environ)
+    env = {}
+    for k, v in environ.items():
+        if scrub_class(k) in ("drop", "pinned"):
+            continue
+        if k.upper() == "PATH":
+            v = os.pathsep.join(e for e in v.split(os.pathsep) if e and not _under(e, prefixes))
+        env[k] = v
+    env.update(SCRUB_PINNED)
+    env[SCRUB_PINNED_BINARY] = binary
+    return env
+
+
+def resolve_binary(explicit: str | None = None) -> str | None:
+    """Absolute path of `claude`: `explicit`, else $CLAUDE_CODE_EXECPATH, else PATH."""
+    path = explicit or os.environ.get("CLAUDE_CODE_EXECPATH") or shutil.which("claude")
+    return os.path.abspath(path) if path else None
+
+
+def binary_version(path: str) -> str | None:
+    """First token of `<path> --version`, or None on any failure."""
+    try:
+        r = subprocess.run([path, "--version"], stdin=subprocess.DEVNULL, capture_output=True,
+                           timeout=60)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    toks = r.stdout.decode("utf-8", "replace").split()
+    return toks[0] if r.returncode == 0 and toks else None
+
+
+def _argv(fp: dict) -> list[str]:
+    """The fingerprinted command line; its order must never change (cache hit)."""
+    a = [fp["binary"], "-p", "--output-format", "stream-json", "--verbose",
+         "--model", fp["model"], "--effort", fp["effort"],
+         "--permission-mode", fp["permission_mode"], "--tools", ",".join(fp["tools"])]
+    if fp.get("allowed_tools"):
+        a += ["--allowedTools", ",".join(fp["allowed_tools"])]
+    if fp.get("disallowed_tools"):
+        a += ["--disallowedTools", ",".join(fp["disallowed_tools"])]
+    a += ["--strict-mcp-config", "--mcp-config", fp["mcp_config"],
+          "--append-system-prompt-file", fp["append_prompt"]["path"]]
+    if fp.get("agent"):
+        a += ["--agent", fp["agent"]]
+    return a
+
+
+def build_fingerprint(*, binary, model, effort, cwd, agent, preamble_path, mcp_config) -> dict:
+    """Validate the launch fields; return {"fingerprint", "argv", "fp_hash"}. Writes nothing."""
+    if not MODEL_RE.match(model or ""):
+        raise ValueError("model must be a full id (aliases are refused): %r" % model)
+    if effort not in EFFORTS:
+        raise ValueError("unknown effort %r (one of %s)" % (effort, ", ".join(EFFORTS)))
+    if not os.path.isdir(cwd):
+        raise ValueError("cwd is not a directory: %s" % cwd)
+    if not os.path.isfile(preamble_path):
+        raise ValueError("preamble not found: %s" % preamble_path)
+    if not binary or not os.path.isfile(binary):
+        raise ValueError("claude binary not found: %s" % binary)
+    preamble_path = os.path.abspath(preamble_path)
+    fp = {
+        "binary": os.path.abspath(binary),
+        "binary_version": binary_version(binary),
+        "model": model,
+        "effort": effort,
+        "permission_mode": PERMISSION_MODE,
+        "tools": list(DEFAULT_TOOLS),
+        # under dontAsk every tool not allowed is denied
+        "allowed_tools": list(DEFAULT_TOOLS),
+        "disallowed_tools": list(DEFAULT_DISALLOWED),
+        "agent": agent or None,
+        "cwd": os.path.abspath(cwd),
+        "append_prompt": {"path": preamble_path, "sha256": _sha256_file(preamble_path)},
+        "mcp_config": os.path.abspath(mcp_config),
+        "env_pinned": dict(ENV_PINNED_LAUNCH),
+    }
+    return {"fingerprint": fp, "argv": _argv(fp),
+            "fp_hash": _sha256_bytes(canonical_json(fp).encode("utf-8"))}
+
+
+def train_argv(run: dict, sid: str) -> list[str]:
+    return list(run["argv"]) + ["--session-id", sid]
+
+
+def fork_argv(run: dict, base_sid: str, sid: str) -> list[str]:
+    return list(run["argv"]) + ["--resume", base_sid, "--fork-session", "--session-id", sid]
+
+
+def resume_argv(run: dict, base_sid: str) -> list[str]:
+    """No `--session-id`: claude refuses it on a resume without a fork."""
+    return list(run["argv"]) + ["--resume", base_sid]

plugins/spire/skills/gremlins/grm/store.py

@@ -0,0 +1,116 @@
+"""State dir layout, atomic JSON files and the append-only ledger."""
+import contextlib
+import json
+import os
+import threading
+import time
+import uuid
+from pathlib import Path
+
+SUBDIRS = ("bases", "jobs", "raw", "notes", "logs", "briefs")
+RETRY_TRIES = 40
+RETRY_SLEEP_S = 0.05
+
+_ledger_lock = threading.Lock()
+
+
+def state_dir(run) -> Path:
+    """`<x>/run.md` -> `<x>/run-gremlins` (a sibling of the run log)."""
+    p = os.path.abspath(str(run))
+    if not p.lower().endswith(".md"):
+        raise ValueError("--run must name the run log .md file: %s" % p)
+    return Path(p[:-3] + "-gremlins")
+
+
+def init_state(run) -> Path:
+    """Create the state dir and its subdirs (idempotent)."""
+    state = state_dir(run)
+    for d in SUBDIRS:
+        (state / d).mkdir(parents=True, exist_ok=True)
+    return state
+
+
+def _retry(fn, tries: int = RETRY_TRIES, sleep_s: float = RETRY_SLEEP_S):
+    """Call fn(); retry PermissionError (Windows sharing violations)."""
+    for attempt in range(1, tries + 1):
+        try:
+            return fn()
+        except PermissionError:
+            if attempt == tries:
+                raise
+            time.sleep(sleep_s)
+
+
+def atomic_write_bytes(path, data: bytes) -> None:
+    """Write to a temp file in the same dir, then `os.replace` with retries."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.parent / (".%s.%d.%s.tmp" % (path.name, os.getpid(), uuid.uuid4().hex[:8]))
+    with open(tmp, "wb") as f:
+        f.write(data)
+        f.flush()
+        os.fsync(f.fileno())
+    try:
+        _retry(lambda: os.replace(tmp, path))
+    except BaseException:
+        with contextlib.suppress(OSError):
+            os.remove(tmp)
+        raise
+
+
+def read_bytes_retry(path) -> bytes:
+    def _read():
+        with open(path, "rb") as f:
+            return f.read()
+    return _retry(_read)
+
+
+def read_json(p) -> dict | None:
+    """The parsed file, or None if it does not exist. Corrupt content raises ValueError."""
+    try:
+        data = read_bytes_retry(p)
+    except FileNotFoundError:
+        return None
+    return json.loads(data.decode("utf-8"))
+
+
+def write_json(p, obj) -> None:
+    atomic_write_bytes(p, (json.dumps(obj, indent=1, sort_keys=True) + "\n").encode("utf-8"))
+
+
+def _ledger_path(run) -> Path:
+    return state_dir(run) / "ledger.jsonl"
+
+
+def ledger_append(run, line: dict) -> None:
+    """Append one JSON line with a single write; repairs a torn last line first."""
+    path = _ledger_path(run)
+    body = (json.dumps(line, sort_keys=True) + "\n").encode("utf-8")
+
+    def _append():
+        torn = False
+        with contextlib.suppress(FileNotFoundError), open(path, "rb") as r:
+            if r.seek(0, os.SEEK_END) > 0:
+                r.seek(-1, os.SEEK_END)
+                torn = r.read(1) != b"\n"
+        with open(path, "ab") as f:
+            f.write((b"\n" if torn else b"") + body)
+    with _ledger_lock:
+        _retry(_append)
+
+
+def ledger_lines(run) -> list[dict]:
+    """Every parsable line of the ledger; [] when there is none yet."""
+    try:
+        data = read_bytes_retry(_ledger_path(run))
+    except FileNotFoundError:
+        return []
+    out = []
+    for ln in data.splitlines():
+        try:
+            obj = json.loads(ln.decode("utf-8", "replace"))
+        except ValueError:
+            continue
+        if isinstance(obj, dict):
+            out.append(obj)
+    return out

plugins/spire/skills/gremlins/grm/stream.py

@@ -0,0 +1,215 @@
+"""Stream-json reader and accounting for one `claude -p` call: pure functions, no IO."""
+import datetime as _dt
+import json
+import re
+import time
+
+TTL_1H, TTL_5M = 3600, 300
+CACHE_HIT_RATIO = 0.9
+MU_FIELDS = ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens")
+TOKEN_KEYS = ("input", "output", "cache_creation", "cache_read", "cache_creation_1h", "cache_creation_5m")
+
+
+def iso(t: float | None = None) -> str:
+    """ISO 8601 with the local offset, from epoch seconds (now if not given)."""
+    if t is None:
+        t = time.time()
+    return _dt.datetime.fromtimestamp(t).astimezone().isoformat(timespec="seconds")
+
+
+def parse_iso(s) -> float | None:
+    """Epoch seconds of an ISO string written by `iso`, or None."""
+    if not s or not isinstance(s, str):
+        return None
+    try:
+        return _dt.datetime.fromisoformat(s).timestamp()
+    except ValueError:
+        return None
+
+
+class StreamState:
+    """What one `claude -p --output-format stream-json --verbose` stream told us so far."""
+
+    def __init__(self):
+        self.init = None              # the system/init event
+        self.session_id = None
+        self.msg_ids: list = []
+        self._ids: set = set()
+        self.first_usage = None       # usage of the first assistant event (cache verdict)
+        self.first_at = None          # its arrival time (the base prefix read)
+        self.last_usage = None        # usage of the last assistant event (base context)
+        self.last_assistant_at = None
+        self.result = None            # the result event
+        self.refused = None           # text of a model-restricted warning
+        self.events = 0
+
+    def feed(self, ev: dict, t: float | None) -> None:
+        if not isinstance(ev, dict):
+            return
+        self.events += 1
+        ty = ev.get("type")
+        if ty == "system":
+            st = ev.get("subtype")
+            if st == "init" and self.init is None:
+                self.init = ev
+                self.session_id = ev.get("session_id") or self.session_id
+            elif (st == "informational" and ev.get("level") == "warning"
+                  and "is restricted" in str(ev.get("content", ""))):
+                self.refused = str(ev.get("content"))
+        elif ty == "assistant":
+            m = ev.get("message") or {}
+            mid = m.get("id")
+            if mid not in self._ids:
+                self._ids.add(mid)
+                self.msg_ids.append(mid)
+            u = m.get("usage") or {}
+            if self.first_usage is None:
+                self.first_usage, self.first_at = u, t
+            self.last_usage, self.last_assistant_at = u, t
+        elif ty == "result":
+            self.result = ev
+            self.session_id = ev.get("session_id") or self.session_id
+
+    @property
+    def turns(self) -> int:
+        return len(self.msg_ids)
+
+
+def context_tokens(usage: dict | None) -> int:
+    """Prompt-side tokens of one request: input + cache creation + cache read."""
+    if not usage:
+        return 0
+    return (int(usage.get("input_tokens") or 0) + int(usage.get("cache_creation_input_tokens") or 0)
+            + int(usage.get("cache_read_input_tokens") or 0))
+
+
+_API_ERROR_TEXT = (   # checked in this order, after the status code
+    ("prompt_too_long", re.compile(r"prompt is too long|prompt too long|prompt_too_long", re.I)),
+    ("usage_limit", re.compile(r"usage limit|rate[ _-]?limit|hit your limit|api error:?\s*429", re.I)),
+    ("overloaded", re.compile(r"overloaded|api error:?\s*529", re.I)),
+    ("api_error", re.compile(r"api error", re.I)),
+)
+
+
+def api_error_reason(res: dict | None) -> tuple[str, str] | None:
+    """(reason, detail) for a result that carries an API error, else None. `reason` is
+    `usage_limit` (429), `overloaded` (529), `prompt_too_long` or `api_error`."""
+    if not res:
+        return None
+    status = res.get("api_error_status")
+    try:
+        status = int(status) if status not in (None, "") else None
+    except (TypeError, ValueError):
+        status = None
+    if not res.get("is_error") and status is None:
+        return None
+    parts = [res.get("result"), res.get("error"), res.get("errors")]
+    text = " | ".join(p if isinstance(p, str) else json.dumps(p) for p in parts if p)
+    detail = "api_error_status %s: %s" % (status, text[:200] if text else "(no text)")
+    if status == 429:
+        return "usage_limit", detail
+    if status == 529:
+        return "overloaded", detail
+    for reason, rx in _API_ERROR_TEXT:
+        if rx.search(text):
+            return reason, detail
+    if status is not None:
+        return "api_error", detail
+    return None
+
+
+def classify(ss: StreamState, exit_code, stopped: bool) -> tuple[str, str | None]:
+    """(status, reason) of a finished call; the first matching case wins."""
+    res = ss.result
+    match (bool(stopped), ss.refused, res, api_error_reason(res), exit_code):
+        case (True, _, _, _, _):
+            return "stopped", "stopped"
+        case (_, str(), _, _, _):
+            return "error", "model_refused"
+        case (_, _, None, _, code):
+            return "error", "exit_code" if code else "no_result"
+        case (_, _, _, (reason, _), _):
+            return "error", reason
+        case (_, _, dict() as r, _, _) if r.get("is_error"):
+            return "error", "is_error"
+        case (_, _, _, _, code) if code:
+            return "error", "exit_code"
+        case _:
+            return "ok", None
+
+
+def usage_from_result(result: dict, before_mu: dict, cost_before) -> dict:
+    """Tokens (with the 1h/5m split) from `result.usage`; `models_used` = modelUsage entries that
+    grew against the session's totals before the call, largest first; cost delta against the
+    session's cost before the call. Costs are rounded here, once, at 6 decimals."""
+    result = result or {}
+    usage = result.get("usage") or {}
+    cc = usage.get("cache_creation") or {}
+    tokens = {
+        "input": int(usage.get("input_tokens") or 0),
+        "output": int(usage.get("output_tokens") or 0),
+        "cache_creation": int(usage.get("cache_creation_input_tokens") or 0),
+        "cache_read": int(usage.get("cache_read_input_tokens") or 0),
+        "cache_creation_1h": int(cc.get("ephemeral_1h_input_tokens") or 0),
+        "cache_creation_5m": int(cc.get("ephemeral_5m_input_tokens") or 0),
+    }
+    after = result.get("modelUsage") or {}
+    before_mu = before_mu or {}
+    delta = {}
+    for m, v in after.items():
+        b = before_mu.get(m) or {}
+        delta[m] = {f: int((v or {}).get(f) or 0) - int(b.get(f) or 0) for f in MU_FIELDS}
+    grew = [m for m, d in delta.items() if any(x > 0 for x in d.values())]
+    models_used = sorted(grew, key=lambda m: (-sum(delta[m].values()), m))
+    sums = {f: sum(d[f] for d in delta.values()) for f in MU_FIELDS}
+    mismatch = (sums["inputTokens"], sums["outputTokens"], sums["cacheCreationInputTokens"],
+                sums["cacheReadInputTokens"]) != (tokens["input"], tokens["output"],
+                                                   tokens["cache_creation"], tokens["cache_read"])
+    total = result.get("total_cost_usd")
+    if total is None:
+        cost_delta = None
+    else:
+        cost_delta = round(float(total) - float(cost_before or 0.0), 6)
+        total = round(float(total), 6)
+    return {"tokens": tokens, "models_used": models_used, "mismatch": mismatch,
+            "cost_delta": cost_delta, "total": total, "model_usage": after}
+
+
+def cache_verdict(first_usage: dict | None, base_ctx) -> str | None:
+    """The first request's cache read against the base's context tokens."""
+    if first_usage is None:
+        return None
+    cr = int(first_usage.get("cache_read_input_tokens") or 0)
+    if base_ctx and cr >= CACHE_HIT_RATIO * float(base_ctx):
+        return "hit"
+    return "partial" if cr > 0 else "miss"
+
+
+def infer_ttl(tokens: dict | None, prev):
+    """The TTL of a base's cache entries from one call's tokens: 1h written -> 3600, only 5m
+    written -> 300, otherwise (read only, or nothing) `prev`."""
+    if not tokens:
+        return prev
+    if int(tokens.get("cache_creation_1h") or 0) > 0:
+        return TTL_1H
+    if int(tokens.get("cache_creation_5m") or 0) > 0:
+        return TTL_5M
+    return prev
+
+
+def wrote_only_5m(tokens: dict | None) -> bool:
+    """The call wrote 5-minute cache entries and no 1-hour ones."""
+    return (bool(tokens) and int(tokens.get("cache_creation_5m") or 0) > 0
+            and int(tokens.get("cache_creation_1h") or 0) == 0)
+
+
+def warmth(last_hit_ts: float | None, ttl_s, now: float) -> dict:
+    """warm until `margin` before last_hit + ttl, expiring until last_hit + ttl, then cold;
+    margin = max(120, ttl/10). No hit at all is cold."""
+    if last_hit_ts is None:
+        return {"base_state": "cold", "warm_until": None, "ttl_left_s": None}
+    ttl = float(ttl_s or TTL_1H)
+    margin = max(120.0, ttl / 10.0)
+    wu = float(last_hit_ts) + ttl
+    state = "warm" if now < wu - margin else ("expiring" if now < wu else "cold")
+    return {"base_state": state, "warm_until": iso(wu), "ttl_left_s": int(max(0.0, wu - now))}