fix(optimize): rank findings in core instead of importing the CLI

tests/unit/test_optimize_rank.py

@@ -0,0 +1,109 @@
+"""Core ranking used by both the CLI text view and GET /optimize."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from types import SimpleNamespace
+
+from tokenjam.core.optimize.rank import (
+    ALWAYS_FULL_FINDINGS,
+    CARD_FINDING_NAMES,
+    rank_findings,
+    reclaimable_share,
+)
+from tokenjam.core.optimize.types import OptimizeReport, WindowSummary
+
+UTC = timezone.utc
+
+
+def _report(**kwargs) -> OptimizeReport:
+    summary = WindowSummary(
+        since=datetime(2026, 1, 1, tzinfo=UTC),
+        until=datetime(2026, 1, 8, tzinfo=UTC),
+        days=7,
+        sessions=3,
+        spans=10,
+        total_tokens=1000,
+        total_cost_usd=1.0,
+        thin_data=False,
+    )
+    return OptimizeReport(window=summary, **kwargs)
+
+
+def test_reclaimable_share_none_without_estimate():
+    assert reclaimable_share(SimpleNamespace(), 1000) is None
+    assert reclaimable_share(SimpleNamespace(past_overspend_tokens=None), 1000) is None
+    assert reclaimable_share(SimpleNamespace(past_overspend_tokens=50), 0) is None
+
+
+def test_reclaimable_share_clamps_negative():
+    assert reclaimable_share(SimpleNamespace(past_overspend_tokens=-10), 1000) == 0.0
+
+
+def test_rank_orders_by_share_then_name_order():
+    report = _report(
+        downgrade=SimpleNamespace(past_overspend_tokens=100),
+        findings={
+            "resend": SimpleNamespace(past_overspend_tokens=400),
+            "cache": SimpleNamespace(past_overspend_tokens=400),
+            "trim": SimpleNamespace(past_overspend_tokens=50),
+        },
+    )
+    ranked = rank_findings(report, None)
+    names = [name for name, _ in ranked]
+    # cache is listed before resend in CARD_FINDING_NAMES; equal share keeps that order
+    assert names[:3] == ["cache", "resend", "downsize"]
+    assert names[3] == "trim"
+
+
+def test_rank_skips_downsize_when_not_requested():
+    report = _report(
+        downgrade=SimpleNamespace(past_overspend_tokens=999),
+        findings={"cache": SimpleNamespace(past_overspend_tokens=10)},
+    )
+    ranked = rank_findings(report, ["cache"])
+    assert [name for name, _ in ranked] == ["cache"]
+
+
+def test_rank_includes_empty_downsize_when_requested():
+    report = _report(downgrade=None, findings={})
+    ranked = rank_findings(report, None)
+    assert ranked == [("downsize", None)]
+
+
+def test_rank_drops_unknown_finding_names():
+    report = _report(
+        downgrade=None,
+        findings={"budget-projection": SimpleNamespace(past_overspend_tokens=500)},
+    )
+    ranked = rank_findings(report, ["cache"])
+    assert ranked == []
+
+
+def test_relearn_is_unranked_even_with_a_token_figure():
+    report = _report(
+        downgrade=None,
+        findings={"relearn": SimpleNamespace(past_overspend_tokens=800)},
+    )
+    ranked = rank_findings(report, ["relearn"])
+    assert ranked == [("relearn", None)]
+    assert "relearn" in ALWAYS_FULL_FINDINGS
+
+
+def test_card_finding_names_match_cli_renderers():
+    from tokenjam.cli.cmd_optimize import _ALWAYS_FULL_FINDINGS, _FINDING_RENDERERS
+
+    assert tuple(_FINDING_RENDERERS) == CARD_FINDING_NAMES
+    assert _ALWAYS_FULL_FINDINGS is ALWAYS_FULL_FINDINGS
+
+
+def test_api_optimize_route_does_not_import_cli():
+    source = (
+        Path(__file__).resolve().parents[2]
+        / "tokenjam"
+        / "api"
+        / "routes"
+        / "optimize.py"
+    ).read_text(encoding="utf-8")
+    assert "tokenjam.cli" not in source
+    assert "rank_findings" in source

tokenjam/api/routes/optimize.py

@@ -31,7 +31,6 @@
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
 
 from tokenjam.api.deps import require_api_key, require_relearn_write_auth
-from tokenjam.cli.cmd_optimize import _rank_findings
 from tokenjam.core.data_span import available_data_span
 from tokenjam.core.framing import (
     PERSONAS,
@@ -44,6 +43,7 @@
     ANALYZER_REGISTRY,
     disabled_analyzers_for_persona,
     findings_for_persona,
+    rank_findings,
     report_store,
 )
 from tokenjam.core.persona_scope import persona_scopes_population
@@ -246,13 +246,13 @@ def get_optimize(
     if "downsize" in persona_disabled:
         payload["downgrade"] = None
 
-    # Biggest-waste-first ranking — the same `_rank_findings` the CLI's text
+    # Biggest-waste-first ranking — the same `rank_findings` the CLI's text
     # view ranks by, so the web doesn't fall back to Object.keys() insertion
     # order. `share` of None means "no quantified estimate" (unranked), which
     # is NOT zero — the UI must not sort those away as de-minimis.
     payload["finding_rank"] = [
         {"name": name, "share": share}
-        for name, share in _rank_findings(report, None)
+        for name, share in rank_findings(report, None)
         if name not in persona_disabled
     ]
 

tokenjam/cli/cmd_optimize.py

@@ -25,13 +25,16 @@
     render_savings,
 )
 from tokenjam.core.optimize import (
+    ALWAYS_FULL_FINDINGS as _ALWAYS_FULL_FINDINGS,
     ANALYZER_REGISTRY,
     MODEL_DOWNGRADE_CAVEAT,
     BudgetProjection,
     DowngradeFinding,
     OptimizeReport,
     build_report,
     disabled_analyzers_for_persona as _disabled_analyzers,
+    rank_findings as _rank_findings_core,
+    reclaimable_share as _reclaimable_share,  # noqa: F401 — re-exported for tests
     report_from_dict,
     report_to_dict,
 )
@@ -587,7 +590,7 @@ def cmd_optimize(
 # "~0.0% of window tokens" pointer — the same "nothing found" failure as the
 # empty-state bug. These findings always render in full (the `unranked`
 # bucket): their own detail when populated, their own empty-state when not.
-_ALWAYS_FULL_FINDINGS = {"relearn"}
+# `_ALWAYS_FULL_FINDINGS` is imported from core so CLI and API cannot drift.
 
 # Display labels for the "Minor findings" collapsed pointer list — must match
 # the header text each renderer prints in its numbered form.
@@ -616,24 +619,6 @@ def _numbered_marker(n: int) -> str:
     return f"({n})"  # defensive — no report should ever have this many
 
 
-def _reclaimable_share(finding: Any, window_total_tokens: int) -> float | None:
-    """Estimated-recoverable-tokens share of the window, for ranking.
-
-    Returns ``None`` — not 0.0 — when the finding has no quantified estimate
-    at all (analyzer disabled, no candidates, or cache-recommend, which
-    recommends a cache_control placement rather than a token count). Those
-    findings still render in full (they're not "de-minimis", they're
-    "unranked") — only a finding with a real-but-tiny share collapses into
-    the Minor findings pointer list. Conflating the two would hide an
-    analyzer's own diagnostic empty-state message (e.g. "no tool spans in
-    this window") behind a generic pointer.
-    """
-    tokens = getattr(finding, "past_overspend_tokens", None)
-    if tokens is None or window_total_tokens <= 0:
-        return None
-    return max(float(tokens), 0.0) / window_total_tokens
-
-
 def _echo_scan_not_ready(payload: dict, output_json: bool) -> None:
     """Report that the daemon's analyzer scan has not produced a result yet.
 
@@ -676,51 +661,13 @@ def _echo_scan_not_ready(payload: dict, output_json: bool) -> None:
 def _rank_findings(
     report: OptimizeReport, requested: list[str] | None,
 ) -> list[tuple[str, float | None]]:
-    """Rank findings with something to show by reclaimable token share
-    (largest first; unranked findings — no quantified estimate — sort last).
-    Ties fall back to ANALYZER_ORDER for determinism.
-    """
-    window_tokens = report.window.total_tokens
-    order = ["downsize", *_FINDING_RENDERERS.keys()]
-    order_index = {name: i for i, name in enumerate(order)}
-
-    items: list[tuple[str, float | None]] = []
-    # Render an explicit "no candidates" empty state when the downsize
-    # analyzer ran but found nothing — the Optimize web tab does this
-    # (PR #130 / issue #126) and the CLI used to silently skip the section,
-    # which makes reviewers think the analyzer didn't run. Skip the section
-    # entirely (empty state or full) when the user asked for a different
-    # positional subset (`tj optimize cache` shouldn't mention downsize at
-    # all). This also covers `tj optimize placement`: that alias resolves to
-    # running the `downsize` analyzer (see `_resolve_analyzer_names`), so
-    # `report.downgrade` is populated even though the user never typed
-    # "downsize" — without this guard its card would leak into a report the
-    # user only asked to see `placement` in.
-    downsize_was_requested = (not requested) or ("downsize" in requested)
-    if downsize_was_requested:
-        if report.downgrade is not None:
-            items.append(("downsize", _reclaimable_share(report.downgrade, window_tokens)))
-        else:
-            items.append(("downsize", None))
-
-    for name, finding in (report.findings or {}).items():
-        if name not in _FINDING_RENDERERS:
-            continue
-        # A non-token finding (e.g. relearn) is forced into the unranked bucket
-        # so its clusters always render in full — a large window denominator
-        # must not collapse them into the de-minimis pointer list.
-        share = (
-            None if name in _ALWAYS_FULL_FINDINGS
-            else _reclaimable_share(finding, window_tokens)
-        )
-        items.append((name, share))
-
-    items.sort(key=lambda item: (
-        item[1] is None,
-        -(item[1] or 0.0),
-        order_index.get(item[0], len(order)),
-    ))
-    return items
+    """Rank findings with something to show by reclaimable token share."""
+    return _rank_findings_core(
+        report,
+        requested,
+        known_names=_FINDING_RENDERERS,
+        always_full=_ALWAYS_FULL_FINDINGS,
+    )
 
 
 # ---------------------------------------------------------------------------

tokenjam/core/optimize/__init__.py

@@ -26,6 +26,12 @@
 
 # Re-export the public surface used by cmd_optimize.py, mcp/server.py, tests.
 from tokenjam.core.optimize import report_store
+from tokenjam.core.optimize.rank import (
+    ALWAYS_FULL_FINDINGS,
+    CARD_FINDING_NAMES,
+    rank_findings,
+    reclaimable_share,
+)
 from tokenjam.core.optimize.registry import ANALYZER_REGISTRY, register
 from tokenjam.core.optimize.runner import (
     ANALYZER_ORDER,
@@ -69,8 +75,10 @@
 )
 
 __all__ = [
+    "ALWAYS_FULL_FINDINGS",
     "ANALYZER_ORDER",
     "ANALYZER_REGISTRY",
+    "CARD_FINDING_NAMES",
     "GATED_PERSONAS",
     "PERSONA_DISABLED_ANALYZERS",
     "AnalyzerContext",
@@ -96,6 +104,8 @@
     "disabled_analyzers_for_personas",
     "findings_for_persona",
     "project_budget",
+    "rank_findings",
+    "reclaimable_share",
     "register",
     "report_from_dict",
     "report_store",

tokenjam/core/optimize/rank.py

@@ -0,0 +1,100 @@
+"""Shared finding ranking for CLI text view and the /optimize payload.
+
+Lives in core so the API does not import `tokenjam.cli`. The name set is the
+card-bearing findings (not the full analyzer registry — budget-projection has
+no ranked card).
+"""
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from tokenjam.core.optimize.types import OptimizeReport
+
+# Insertion order matches `cmd_optimize._FINDING_RENDERERS`. A drift test
+# pins the two together so a new card cannot rank in one surface only.
+CARD_FINDING_NAMES: tuple[str, ...] = (
+    "cache",
+    "cache-recommend",
+    "resend",
+    "script",
+    "reuse",
+    "trim",
+    "subagent",
+    "relearn",
+    "verbosity",
+    "deadweight",
+    "placement",
+    "summarize",
+    "stream-usage",
+)
+
+# Findings that must never collapse into the "Minor findings" pointer by
+# token share. `relearn` is a cluster finding; its token figure is not a
+# real fraction of the window.
+ALWAYS_FULL_FINDINGS: frozenset[str] = frozenset({"relearn"})
+
+
+def reclaimable_share(finding: Any, window_total_tokens: int) -> float | None:
+    """Estimated-recoverable-tokens share of the window, for ranking.
+
+    Returns ``None`` — not 0.0 — when the finding has no quantified estimate
+    at all. Those findings still render in full (unranked), they are not
+    de-minimis.
+    """
+    tokens = getattr(finding, "past_overspend_tokens", None)
+    if tokens is None or window_total_tokens <= 0:
+        return None
+    return max(float(tokens), 0.0) / window_total_tokens
+
+
+def rank_findings(
+    report: OptimizeReport,
+    requested: list[str] | None,
+    *,
+    known_names: Mapping[str, Any] | Sequence[str] = CARD_FINDING_NAMES,
+    always_full: Sequence[str] | set[str] | frozenset[str] = ALWAYS_FULL_FINDINGS,
+) -> list[tuple[str, float | None]]:
+    """Rank findings with something to show by reclaimable token share.
+
+    Largest first; unranked findings (no quantified estimate) sort last.
+    Ties fall back to ``known_names`` insertion order, with ``downsize``
+    first — matching the CLI renderer table.
+    """
+    window_tokens = report.window.total_tokens
+    names = tuple(known_names)
+    known = set(names)
+    full = set(always_full)
+    order = ["downsize", *names]
+    order_index = {name: i for i, name in enumerate(order)}
+
+    items: list[tuple[str, float | None]] = []
+    # Render an explicit "no candidates" empty state when the downsize
+    # analyzer ran but found nothing. Skip the section entirely when the
+    # user asked for a different positional subset (`tj optimize cache`
+    # should not mention downsize). `tj optimize placement` resolves to
+    # running the downsize analyzer, so `report.downgrade` is populated
+    # even though the user never typed "downsize" — without this guard
+    # its card would leak into that report.
+    downsize_was_requested = (not requested) or ("downsize" in requested)
+    if downsize_was_requested:
+        if report.downgrade is not None:
+            items.append(("downsize", reclaimable_share(report.downgrade, window_tokens)))
+        else:
+            items.append(("downsize", None))
+
+    for name, finding in (report.findings or {}).items():
+        if name not in known:
+            continue
+        share = (
+            None if name in full
+            else reclaimable_share(finding, window_tokens)
+        )
+        items.append((name, share))
+
+    items.sort(key=lambda item: (
+        item[1] is None,
+        -(item[1] or 0.0),
+        order_index.get(item[0], len(order)),
+    ))
+    return items