fix(extract): persist CLI token usage in cost.json (#1769)

CHANGELOG.md

@@ -4,6 +4,8 @@ Full release notes with details on each version: [GitHub Releases](https://githu
 
 ## 0.9.12 (unreleased)
 
+- Fix: successful `graphify extract` runs now append their token usage to `graphify-out/cost.json` instead of only printing it (#1769, thanks @jsilrecavo). Both clustered and `--no-cluster` runs use the existing ledger schema and update cumulative totals; `cluster-only`/`label` also persist the real labeling usage added in #1694. A malformed or unwritable ledger warns without turning an otherwise successful graph build into a failure.
+
 - Fix: Java member calls resolve against the receiver's declared type instead of a bare method-name match (#1696/#1697, thanks @oleksii-tumanov). `gw.charge()` where `gw: PaymentGateway` now binds to `PaymentGateway.charge`, not a same-named `AuditLog.charge` in another file. Explicit-type receivers and `this` are exact; current-class fields, method parameters, and explicitly-typed locals resolve via a method-scoped type table; a missing, ambiguous, inherited, or chained receiver is skipped rather than guessed (same god-node guard as the C#/Swift/Ruby resolvers). Fully-qualified and nested-type receivers are deferred (they need package/nesting-aware type identity).
 
 - Fix: output/cache artifacts no longer land in the scanned corpus or CWD when `--out`/`--graph` point elsewhere (#1747, thanks @bbqboogiedwonsen). `extract <corpus> --out <dir>` correctly wrote the graph to `<dir>` but `detect()`'s word-count/stat-index cache still created a stray `graphify-out/cache/` inside the corpus (it uses the scan root); it now honors the `--out` dir via a threaded `cache_root`. And `cluster-only --graph <elsewhere>/graphify-out/graph.json` wrote `GRAPH_REPORT.md`/labels/analysis/re-clustered graph to the CWD instead of beside the input; it now writes beside `--graph` when that graph lives in a `graphify-out/` dir, while still restoring into the CWD for an archived `backup/graph.json` (#934).

graphify/cli.py

@@ -52,6 +52,29 @@
 
 def _default_graph_path() -> str:
     return str(Path(_GRAPHIFY_OUT) / "graph.json")
+
+
+def _record_cost_run(
+    output_dir: Path,
+    tokens: dict,
+    *,
+    files: int,
+    prefix: str,
+) -> None:
+    """Persist one successful CLI run without turning telemetry into a failure."""
+    from graphify.cost import record_cost_run
+
+    try:
+        record_cost_run(
+            output_dir,
+            input_tokens=tokens.get("input", 0),
+            output_tokens=tokens.get("output", 0),
+            files=files,
+        )
+    except Exception as exc:
+        print(f"{prefix} warning: could not update cost.json: {exc}", file=sys.stderr)
+
+
 class _StageTimer:
     """Print per-stage wall-clock timings to stderr when --timing is set (#1490).
 
@@ -1226,6 +1249,16 @@ def dispatch_command(cmd: str) -> None:
                 print(f"Skipped graph.html: {viz_err}")
                 stages.mark("export"); stages.total()
                 print(f"Done - {len(communities)} communities. GRAPH_REPORT.md and graph.json updated.")
+        _record_cost_run(
+            out,
+            tokens,
+            files=len({
+                str(data.get("source_file"))
+                for _, data in G.nodes(data=True)
+                if data.get("source_file")
+            }),
+            prefix="[graphify]",
+        )
 
     elif cmd == "update":
         force = os.environ.get("GRAPHIFY_FORCE", "").lower() in ("1", "true", "yes")
@@ -2138,6 +2171,7 @@ def _parse_float(name: str, raw: str) -> float:
                 f"{len(doc_files)} docs, {len(paper_files)} papers, "
                 f"{len(image_files)} images"
             )
+        corpus_file_count = sum(len(file_list) for file_list in files_by_type.values())
         # Surface files that were seen but not classified (extensionless non-shebang
         # project files like Dockerfile/Makefile, or unsupported extensions), so they
         # are no longer invisible in graphify's own output (#1692).
@@ -2459,12 +2493,18 @@ def _progress(idx: int, total: int, _result: dict) -> None:
             ):
                 print(
                     "[graphify extract] no incremental changes detected "
-                    "(--no-cluster); outputs left untouched."
+                    "(--no-cluster); graph outputs left untouched."
                 )
                 try:
                     _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target)
                 except Exception as exc:
                     print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr)
+                _record_cost_run(
+                    graphify_out,
+                    {"input": 0, "output": 0},
+                    files=corpus_file_count,
+                    prefix="[graphify extract]",
+                )
                 stages.total()
                 sys.exit(0)
 
@@ -2502,6 +2542,12 @@ def _progress(idx: int, total: int, _result: dict) -> None:
                 _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target)
             except Exception as exc:
                 print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr)
+            _record_cost_run(
+                graphify_out,
+                {"input": merged["input_tokens"], "output": merged["output_tokens"]},
+                files=corpus_file_count,
+                prefix="[graphify extract]",
+            )
             if global_merge:
                 from graphify.global_graph import global_add as _global_add
                 _tag = global_repo_tag or target.name
@@ -2597,6 +2643,13 @@ def _progress(idx: int, total: int, _result: dict) -> None:
         except Exception as exc:
             print(f"[graphify extract] warning: could not write manifest: {exc}", file=sys.stderr)
 
+        _record_cost_run(
+            graphify_out,
+            {"input": merged["input_tokens"], "output": merged["output_tokens"]},
+            files=corpus_file_count,
+            prefix="[graphify extract]",
+        )
+
         cost = _estimate_cost(backend, merged["input_tokens"], merged["output_tokens"])
         print(
             f"[graphify extract] wrote {graph_json_path}: "

graphify/cost.py

@@ -0,0 +1,118 @@
+"""Persistent token-usage ledger shared by CLI pipeline commands."""
+from __future__ import annotations
+
+import contextlib
+import json
+import os
+import secrets
+import stat
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+@contextmanager
+def _cost_lock(output_dir: Path) -> Iterator[None]:
+    """Serialize ledger updates across processes without stale lock cleanup."""
+    output_dir.mkdir(parents=True, exist_ok=True)
+    lock_path = output_dir / ".cost.lock"
+    with lock_path.open("a+b") as lock_file:
+        if os.name == "nt":
+            import msvcrt
+
+            lock_file.seek(0, os.SEEK_END)
+            if lock_file.tell() == 0:
+                lock_file.write(b"\0")
+                lock_file.flush()
+            lock_file.seek(0)
+            msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
+            try:
+                yield
+            finally:
+                lock_file.seek(0)
+                msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
+            try:
+                yield
+            finally:
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+
+
+def _total_from_runs(runs: list, key: str) -> int:
+    total = 0
+    for run in runs:
+        if not isinstance(run, dict):
+            raise ValueError("cost.json 'runs' entries must be JSON objects")
+        value = run.get(key, 0)
+        if isinstance(value, bool) or not isinstance(value, (int, float)):
+            raise ValueError(f"cost.json run '{key}' must be numeric")
+        total += int(value)
+    return total
+
+
+def _write_atomic(path: Path, payload: str) -> None:
+    existing_mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else None
+    while True:
+        tmp_name = path.parent / f".cost.{secrets.token_hex(8)}.tmp"
+        try:
+            fd = os.open(tmp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o666)
+            break
+        except FileExistsError:
+            continue
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as tmp_file:
+            tmp_file.write(payload)
+            tmp_file.flush()
+            os.fsync(tmp_file.fileno())
+        if existing_mode is not None:
+            os.chmod(tmp_name, existing_mode)
+        os.replace(tmp_name, path)
+    except Exception:
+        with contextlib.suppress(OSError):
+            os.unlink(tmp_name)
+        raise
+
+
+def record_cost_run(
+    output_dir: Path,
+    *,
+    input_tokens: int,
+    output_tokens: int,
+    files: int,
+) -> dict:
+    """Append one run to ``cost.json`` and return the updated ledger."""
+    with _cost_lock(output_dir):
+        cost_path = output_dir / "cost.json"
+        cost: dict
+        if cost_path.exists():
+            cost = json.loads(cost_path.read_text(encoding="utf-8"))
+            if not isinstance(cost, dict):
+                raise ValueError("cost.json must contain a JSON object")
+        else:
+            cost = {"runs": []}
+
+        runs = cost.setdefault("runs", [])
+        if not isinstance(runs, list):
+            raise ValueError("cost.json 'runs' must be a JSON array")
+
+        input_tokens = int(input_tokens or 0)
+        output_tokens = int(output_tokens or 0)
+        files = int(files or 0)
+        runs.append({
+            "date": datetime.now(timezone.utc).isoformat(),
+            "input_tokens": input_tokens,
+            "output_tokens": output_tokens,
+            "files": files,
+        })
+        cost["total_input_tokens"] = _total_from_runs(runs, "input_tokens")
+        cost["total_output_tokens"] = _total_from_runs(runs, "output_tokens")
+
+        _write_atomic(
+            cost_path,
+            json.dumps(cost, indent=2, ensure_ascii=False),
+        )
+        return cost

tests/test_cost.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import json
+import os
+import stat
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from graphify.cost import record_cost_run
+
+
+def test_record_cost_run_appends_to_existing_ledger(tmp_path):
+    out = tmp_path / "graphify-out"
+    out.mkdir()
+    original_run = {
+        "date": "2026-07-01T00:00:00+00:00",
+        "input_tokens": 100,
+        "output_tokens": 25,
+        "files": 4,
+    }
+    (out / "cost.json").write_text(
+        json.dumps({
+            "runs": [original_run],
+            "total_input_tokens": 100,
+            "total_output_tokens": 25,
+        }),
+        encoding="utf-8",
+    )
+
+    cost = record_cost_run(
+        out,
+        input_tokens=40,
+        output_tokens=10,
+        files=2,
+    )
+
+    assert cost["runs"][0] == original_run
+    assert cost["runs"][1]["input_tokens"] == 40
+    assert cost["runs"][1]["output_tokens"] == 10
+    assert cost["runs"][1]["files"] == 2
+    assert cost["runs"][1]["date"].endswith("+00:00")
+    assert cost["total_input_tokens"] == 140
+    assert cost["total_output_tokens"] == 35
+    assert json.loads((out / "cost.json").read_text(encoding="utf-8")) == cost
+
+
+def test_record_cost_run_rebuilds_missing_totals_from_history(tmp_path):
+    out = tmp_path / "graphify-out"
+    out.mkdir()
+    (out / "cost.json").write_text(
+        json.dumps({
+            "runs": [{
+                "date": "2026-07-01T00:00:00+00:00",
+                "input_tokens": 100,
+                "output_tokens": 25,
+                "files": 4,
+            }],
+        }),
+        encoding="utf-8",
+    )
+
+    cost = record_cost_run(
+        out,
+        input_tokens=40,
+        output_tokens=10,
+        files=2,
+    )
+
+    assert cost["total_input_tokens"] == 140
+    assert cost["total_output_tokens"] == 35
+
+
+def test_record_cost_run_preserves_ledger_when_atomic_replace_fails(
+    tmp_path, monkeypatch
+):
+    out = tmp_path / "graphify-out"
+    out.mkdir()
+    cost_path = out / "cost.json"
+    original = json.dumps({
+        "runs": [],
+        "total_input_tokens": 0,
+        "total_output_tokens": 0,
+    })
+    cost_path.write_text(original, encoding="utf-8")
+
+    def _replace_fails(source, destination):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr("graphify.cost.os.replace", _replace_fails)
+
+    with pytest.raises(OSError, match="replace failed"):
+        record_cost_run(out, input_tokens=1, output_tokens=2, files=3)
+
+    assert cost_path.read_text(encoding="utf-8") == original
+    assert not list(out.glob(".cost.*.tmp"))
+
+
+def test_record_cost_run_serializes_concurrent_writers(tmp_path):
+    out = tmp_path / "graphify-out"
+
+    def _record(_index):
+        record_cost_run(out, input_tokens=3, output_tokens=2, files=1)
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(_record, range(20)))
+
+    cost = json.loads((out / "cost.json").read_text(encoding="utf-8"))
+    assert len(cost["runs"]) == 20
+    assert cost["total_input_tokens"] == 60
+    assert cost["total_output_tokens"] == 40
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits")
+def test_record_cost_run_preserves_existing_permissions(tmp_path):
+    out = tmp_path / "graphify-out"
+    out.mkdir()
+    cost_path = out / "cost.json"
+    cost_path.write_text(
+        json.dumps({
+            "runs": [],
+            "total_input_tokens": 0,
+            "total_output_tokens": 0,
+        }),
+        encoding="utf-8",
+    )
+    cost_path.chmod(0o640)
+
+    record_cost_run(out, input_tokens=1, output_tokens=2, files=3)
+
+    assert stat.S_IMODE(cost_path.stat().st_mode) == 0o640
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX umask semantics")
+def test_record_cost_run_respects_umask_for_new_ledger(tmp_path):
+    out = tmp_path / "graphify-out"
+    old_umask = os.umask(0o027)
+    try:
+        record_cost_run(out, input_tokens=1, output_tokens=2, files=3)
+    finally:
+        os.umask(old_umask)
+
+    assert stat.S_IMODE((out / "cost.json").stat().st_mode) == 0o640