Delete the eight functions nobody called

Measured, not guessed: every top-level def in every tracked .py file was
counted against every other tracked file, including markdown, scripts, Go
and TypeScript. Eight names had zero references anywhere:
cartridges_holding_atom and clear_mounts (memory.py reads each cartridge's
own sources.json, and set_mounts already clears by replacing the set),
gradient_checkpointing_enabled, group_hits_by_source, ingested_paths,
is_finite_number, parse_runtime_json (the runtime answers in prose now, and
the JSON path was never wired back), write_sources_json.

Typer commands and pytest cases were excluded from the count because a
framework calls them by name. The same scan now returns zero.

263 tests pass; repo-wide mypy is still 24 errors in 6 files, unchanged.
This commit is contained in:
ZacharyZhang-NY committed 2026-10-08 07:38:36 -04:00
1 parent 330e801e92
commit 56b94cd58e
7 files changed
-59

No files matched your search

-14
View File
@@ -44,15 +44,6 @@ def membership_count(database: Database, cartridge_id: str) -> int:
return _count(connection, cartridge_id)
def cartridges_holding_atom(database: Database, atom_id: str) -> list[str]:
with database.connect() as connection:
rows = connection.execute(
"SELECT cartridge_id FROM cartridge_atoms WHERE atom_id = ? ORDER BY cartridge_id",
(atom_id,),
).fetchall()
return [str(row["cartridge_id"]) for row in rows]
def atoms_of_cartridge(database: Database, cartridge_id: str) -> list[str]:
with database.connect() as connection:
rows = connection.execute(
@@ -94,11 +85,6 @@ def set_mounts(database: Database, cartridge_ids: list[str], mounted_at: datetim
)
def clear_mounts(database: Database) -> None:
with database.connect() as connection:
connection.execute("DELETE FROM mounts")
def _count(connection: sqlite3.Connection, cartridge_id: str) -> int:
row = connection.execute(
"SELECT COUNT(*) AS n FROM cartridge_atoms WHERE cartridge_id = ?", (cartridge_id,)
@@ -314,8 +314,3 @@ def extract_generic_calls(text: str) -> list[dict[str, str]]:
return calls
def write_sources_json(path: Path, sources: list[SourceRecord]) -> None:
path.write_text(
json.dumps([source.model_dump(mode="json") for source in sources], ensure_ascii=False, indent=2),
encoding="utf-8",
)
-4
View File
@@ -80,10 +80,6 @@ def format_scores(scores: dict[str, Any]) -> str:
return ", ".join(parts) or "none"
def is_finite_number(value: Any) -> bool:
return isinstance(value, (int, float)) and math.isfinite(float(value))
def compact_generation_context(context: Any, section_hint: str, max_chars: int = 2400) -> Any:
text_hits = [hit for hit in context.hits if hit.atom.modality.value == "text"]
hits = (text_hits or context.hits)[:1]
-7
View File
@@ -169,13 +169,6 @@ class RetrievalService:
return [hit.atom for hit in self.search(query, top_k, filters=filters).hits]
def group_hits_by_source(hits: list[RetrievalHit]) -> dict[str, list[RetrievalHit]]:
grouped: dict[str, list[RetrievalHit]] = defaultdict(list)
for hit in hits:
grouped[hit.atom.source].append(hit)
return dict(grouped)
def encode_cursor(offset: int) -> str:
return base64.urlsafe_b64encode(str(offset).encode()).decode()
-15
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
import base64
import json
import logging
import subprocess
from dataclasses import dataclass
@@ -429,17 +428,3 @@ def gpu_status() -> dict[str, Any]:
}
def parse_runtime_json(text: str) -> dict[str, Any]:
stripped = text.strip()
if stripped.startswith("```"):
stripped = stripped.strip("`")
if stripped.lower().startswith("json"):
stripped = stripped[4:].strip()
try:
return json.loads(stripped)
except json.JSONDecodeError:
start = stripped.find("{")
end = stripped.rfind("}")
if start >= 0 and end > start:
return json.loads(stripped[start : end + 1])
raise
-9
View File
@@ -1313,15 +1313,6 @@ def model_type_for_proof(model: Any) -> str | None:
return None
def gradient_checkpointing_enabled(model: Any) -> bool:
if bool(getattr(model, "is_gradient_checkpointing", False)):
return True
config = getattr(model, "config", None)
if bool(getattr(config, "gradient_checkpointing", False)):
return True
return bool(getattr(model, "_gradient_checkpointing_func", None))
def make_optimizer(model: Any, learning_rate: float) -> tuple[Any, str]:
params = [param for param in model.parameters() if param.requires_grad]
try:
-5
View File
@@ -236,11 +236,6 @@ def ledger_outcomes() -> tuple[set[str], set[str]]:
return ingested, failed
def ingested_paths() -> set[str]:
"""Every file a shard ledger recorded as ingested, excluding the files it recorded as failed."""
return ledger_outcomes()[0]
def retry_failed() -> int:
"""Drop failed ledger entries so those files are attempted again, archiving what was dropped."""
archive = RUNS / "failures.jsonl"