diff --git a/integrations/antigravity/skills/lemoncrow/SKILL.md b/integrations/antigravity/skills/lemoncrow/SKILL.md
index b6d0cd27b..b0fcc308c 100644
--- a/integrations/antigravity/skills/lemoncrow/SKILL.md
+++ b/integrations/antigravity/skills/lemoncrow/SKILL.md
@@ -55,12 +55,18 @@ Global scope by default. Add `--workspace
` only if the user names a specif
Unknown key → run `lc settings show`, relay the valid keys.
-5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name, not through the `tool` broker (which refuses it) — it returns a markdown panel, relay it verbatim:
+5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name — it returns a markdown panel, relay it verbatim:
```json
{"name": "statusline_segment", "arguments": {"format": "markdown"}}
```
+ Host can only call the tools it lists? Go through the `tool` broker with `read_only: true`, which it requires; the totals can trail the newest session rows:
+
+ ```json
+ {"name": "tool", "arguments": {"action": "call", "name": "statusline_segment", "arguments": {"format": "markdown", "read_only": true}}}
+ ```
+
`format` accepts `markdown` (chat panel), `json` (raw report), `segment` (one statusline frame). Never recompute or restate the numbers.
6. **Any other verb** (e.g. "run benchmark X") → discover first, then run: `lc --help` (or `lc help `) to find the exact subcommand and flags, execute it, relay output. Never guess flags.
diff --git a/integrations/claude/plugin/hooks/mcp_read_allow.py b/integrations/claude/plugin/hooks/mcp_read_allow.py
index 035109370..4484c053c 100644
--- a/integrations/claude/plugin/hooks/mcp_read_allow.py
+++ b/integrations/claude/plugin/hooks/mcp_read_allow.py
@@ -14,7 +14,17 @@
``codemod``, ``memory``, ``compact``, ``verify``, ``agent``, ``workflow``) is
omitted and keeps prompting normally. ``tool`` is omitted too: it dispatches to
any rarely-used lc tool by name, including write-capable ones, so allowing it
-would launder the whole surface through one decision.
+would launder the whole surface through one decision. ``search`` (which caches
+every query in the workspace search cache) and ``context`` (which records the
+task on the session ledger) are writers under PRD-739 FR4, so they are omitted
+as well.
+
+What this list holds has to stay inside what the MCP ``tool`` broker will run,
+network tools aside -- the broker refuses those for reach, not for writes. That
+agreement is pinned by ``tests/integrations/test_mcp_read_allow_hook.py``, not
+by a shared import: Claude Code runs this file as a subprocess in plugin
+installs where ``lemoncrow`` is not importable, so it stays standard-library
+only.
Stays silent (no decision at all) for every other tool, so it never overrides a
user's own deny rule for something outside this list.
@@ -39,24 +49,63 @@
{
"blame",
"code_search",
- "context",
"graph",
"grep",
"orient",
"read",
"relations",
- "search",
"web_fetch",
}
)
+# ``graph`` is one name over many operations, so the tool alone does not say
+# whether a call reads: ``index_docs`` writes the design-doc store, ``pr_risk``
+# folds each changed file into the machine-wide semantic file index,
+# ``recall_docs`` embeds its query through the configured embedder, and
+# ``enable`` switches doc indexing on. Only the kinds that read the code index
+# or git history are auto-allowed; the rest keep prompting.
+_GRAPH_READ_ONLY_KINDS = frozenset(
+ {
+ "blast_radius",
+ "centrality",
+ "commit_provenance",
+ "coupling",
+ "cycles",
+ "dead_code",
+ "design_gaps",
+ "topology",
+ "verify_design",
+ }
+)
+_GRAPH_DEFAULT_KIND = "blast_radius"
+
+
+def _graph_reads_only(tool_input: Any) -> bool:
+ """True when this ``graph`` call names a kind that only reads.
-def _read_only_tool(tool_name: str) -> str | None:
- """Return the bare lc tool name when ``tool_name`` is an allowed read tool."""
+ Unreadable arguments count as not read-only: an omitted ``tool_input`` would
+ otherwise auto-allow whatever kind the call actually carried. ``kind`` is
+ raw model-supplied JSON, so a non-string one is judged rather than hashed --
+ the same shape the broker's own vetting uses -- because a ``list``/``dict``
+ would raise out of the membership test and out of the hook.
+ """
+ if not isinstance(tool_input, dict) or "enable" in tool_input:
+ return False
+ kind = tool_input.get("kind", _GRAPH_DEFAULT_KIND)
+ return isinstance(kind, str) and kind in _GRAPH_READ_ONLY_KINDS
+
+
+def _read_only_tool(tool_name: str, tool_input: Any = None) -> str | None:
+ """Return the bare lc tool name when this call is an allowed read call."""
parts = tool_name.split("__")
if len(parts) != 3 or parts[0] != "mcp" or parts[1] not in _SERVERS:
return None
- return parts[2] if parts[2] in _READ_ONLY_TOOLS else None
+ tool = parts[2]
+ if tool not in _READ_ONLY_TOOLS:
+ return None
+ if tool == "graph" and not _graph_reads_only(tool_input):
+ return None
+ return tool
def _allow(reason: str) -> None:
@@ -82,7 +131,7 @@ def main() -> int:
return 0
if not isinstance(payload, dict):
return 0
- tool = _read_only_tool(str(payload.get("tool_name") or ""))
+ tool = _read_only_tool(str(payload.get("tool_name") or ""), payload.get("tool_input"))
if tool is None:
return 0
_allow(f"lc {tool} is read-only (no writes, no shell); auto-allowed in every mode including Plan Mode.")
diff --git a/integrations/claude/plugin/skills/lemoncrow/SKILL.md b/integrations/claude/plugin/skills/lemoncrow/SKILL.md
index b6d0cd27b..b0fcc308c 100644
--- a/integrations/claude/plugin/skills/lemoncrow/SKILL.md
+++ b/integrations/claude/plugin/skills/lemoncrow/SKILL.md
@@ -55,12 +55,18 @@ Global scope by default. Add `--workspace ` only if the user names a specif
Unknown key → run `lc settings show`, relay the valid keys.
-5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name, not through the `tool` broker (which refuses it) — it returns a markdown panel, relay it verbatim:
+5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name — it returns a markdown panel, relay it verbatim:
```json
{"name": "statusline_segment", "arguments": {"format": "markdown"}}
```
+ Host can only call the tools it lists? Go through the `tool` broker with `read_only: true`, which it requires; the totals can trail the newest session rows:
+
+ ```json
+ {"name": "tool", "arguments": {"action": "call", "name": "statusline_segment", "arguments": {"format": "markdown", "read_only": true}}}
+ ```
+
`format` accepts `markdown` (chat panel), `json` (raw report), `segment` (one statusline frame). Never recompute or restate the numbers.
6. **Any other verb** (e.g. "run benchmark X") → discover first, then run: `lc --help` (or `lc help `) to find the exact subcommand and flags, execute it, relay output. Never guess flags.
diff --git a/integrations/codex/plugin/skills/lemoncrow/SKILL.md b/integrations/codex/plugin/skills/lemoncrow/SKILL.md
index b6d0cd27b..b0fcc308c 100644
--- a/integrations/codex/plugin/skills/lemoncrow/SKILL.md
+++ b/integrations/codex/plugin/skills/lemoncrow/SKILL.md
@@ -55,12 +55,18 @@ Global scope by default. Add `--workspace ` only if the user names a specif
Unknown key → run `lc settings show`, relay the valid keys.
-5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name, not through the `tool` broker (which refuses it) — it returns a markdown panel, relay it verbatim:
+5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name — it returns a markdown panel, relay it verbatim:
```json
{"name": "statusline_segment", "arguments": {"format": "markdown"}}
```
+ Host can only call the tools it lists? Go through the `tool` broker with `read_only: true`, which it requires; the totals can trail the newest session rows:
+
+ ```json
+ {"name": "tool", "arguments": {"action": "call", "name": "statusline_segment", "arguments": {"format": "markdown", "read_only": true}}}
+ ```
+
`format` accepts `markdown` (chat panel), `json` (raw report), `segment` (one statusline frame). Never recompute or restate the numbers.
6. **Any other verb** (e.g. "run benchmark X") → discover first, then run: `lc --help` (or `lc help `) to find the exact subcommand and flags, execute it, relay output. Never guess flags.
diff --git a/integrations/skills/lemoncrow/SKILL.md b/integrations/skills/lemoncrow/SKILL.md
index 6a949c61e..7b981eb20 100644
--- a/integrations/skills/lemoncrow/SKILL.md
+++ b/integrations/skills/lemoncrow/SKILL.md
@@ -53,12 +53,18 @@ Global scope by default. Add `--workspace ` only if the user names a specif
Unknown key → run `lc settings show`, relay the valid keys.
-5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name, not through the `tool` broker (which refuses it) — it returns a markdown panel, relay it verbatim:
+5. **"what are my savings?" / cost questions** — shell available (Claude Code, Codex CLI): run `lc usage` (add `optimize` for savings analysis, `optimize detail` for the per-operation breakdown) and relay it. No shell (chat-only host): call the `statusline_segment` tool by exact name — it returns a markdown panel, relay it verbatim:
```json
{"name": "statusline_segment", "arguments": {"format": "markdown"}}
```
+ Host can only call the tools it lists? Go through the `tool` broker with `read_only: true`, which it requires; the totals can trail the newest session rows:
+
+ ```json
+ {"name": "tool", "arguments": {"action": "call", "name": "statusline_segment", "arguments": {"format": "markdown", "read_only": true}}}
+ ```
+
`format` accepts `markdown` (chat panel), `json` (raw report), `segment` (one statusline frame). Never recompute or restate the numbers.
6. **Any other verb** (e.g. "run benchmark X") → discover first, then run: `lc --help` (or `lc help `) to find the exact subcommand and flags, execute it, relay output. Never guess flags.
diff --git a/src/lemoncrow/core/capabilities/code_context_contract.py b/src/lemoncrow/core/capabilities/code_context_contract.py
index 2a19fc20d..a497de625 100644
--- a/src/lemoncrow/core/capabilities/code_context_contract.py
+++ b/src/lemoncrow/core/capabilities/code_context_contract.py
@@ -76,7 +76,7 @@ class IndexStats(BaseModel):
imports_indexed: int
index_version: int = 0
# True when the Free-tier repo-size cap truncated indexing (see
- # code_context/engine.py's _FREE_TIER_MAX_FILES). Always False on Pro.
+ # infra/code_intel/inclusion.py's FREE_TIER_MAX_FILES). Always False on Pro.
capped: bool = False
diff --git a/src/lemoncrow/core/capabilities/plugin_runtime.py b/src/lemoncrow/core/capabilities/plugin_runtime.py
index 2d102d439..27e48c3e6 100644
--- a/src/lemoncrow/core/capabilities/plugin_runtime.py
+++ b/src/lemoncrow/core/capabilities/plugin_runtime.py
@@ -615,7 +615,9 @@ def _savings_cap_usd(subscription: dict[str, Any]) -> float | None:
return None
-def compute_usage_meter(root: str | Path, *, subscription: dict[str, Any] | None = None) -> dict[str, Any]:
+def compute_usage_meter(
+ root: str | Path, *, subscription: dict[str, Any] | None = None, fold: bool = True
+) -> dict[str, Any]:
"""Price trailing-window usage against the plan's monthly limit.
Realized spend and savings come from :func:`aggregate_window_savings` (the
@@ -628,6 +630,8 @@ def compute_usage_meter(root: str | Path, *, subscription: dict[str, Any] | None
``monthlyLimitInUsd <= 0`` (or absent) means "no local limit": spend and
savings are still reported, but ``warning``/``overLimit`` stay False.
+
+ ``fold=False`` leaves the savings aggregate alone (see :func:`aggregate_window_savings`).
"""
root_path = Path(root)
if subscription is None:
@@ -651,7 +655,7 @@ def compute_usage_meter(root: str | Path, *, subscription: dict[str, Any] | None
try:
from lemoncrow.core.capabilities.savings_summary import aggregate_window_savings
- window = aggregate_window_savings(root_path, days=BILLING_WINDOW_DAYS)
+ window = aggregate_window_savings(root_path, days=BILLING_WINDOW_DAYS, fold=fold)
spend_usd = round(max(0.0, float(window.spend_usd)), 4)
savings_usd = round(max(0.0, float(window.saved_usd)), 4)
except Exception:
@@ -5094,6 +5098,7 @@ def build_savings_report(
root: str | Path,
*,
session_id: str | None = None,
+ fold: bool = True,
) -> dict[str, Any]:
"""Compose the savings/cost report.
@@ -5101,6 +5106,10 @@ def build_savings_report(
transcript JSONL (tool_result.content[].saved entries).
- Without ``session_id``: all-session analytics aggregate from the
routing/compaction event log.
+
+ ``fold=False`` leaves the savings aggregate alone: the windowed totals are
+ read from it as it stands, without folding session ledgers it has not seen
+ or persisting it, so they can trail rows written since the last fold.
"""
root_path = Path(root)
session = aggregate_session_stats(root_path, session_id=session_id)
@@ -5139,7 +5148,7 @@ def build_savings_report(
# (sessions/*/savings.jsonl) so the CLI agrees with the statusline and
# web Savings page. Routing credit stays sourced from the analytics log.
analytics = load_live_savings_summary(root_path)
- lifetime_w = aggregate_window_savings(root_path, days=36500)
+ lifetime_w = aggregate_window_savings(root_path, days=36500, fold=fold)
tokens_saved = lifetime_w.tokens_saved
calls_avoided = lifetime_w.calls_saved
saved_usd = lifetime_w.saved_usd
@@ -5186,14 +5195,14 @@ def build_savings_report(
lifetime.setdefault("tokens_saved", tokens_saved)
lifetime.setdefault("saved_usd", saved_usd)
subscription = resolve_subscription(root_path)
- subscription = compute_usage_meter(root_path, subscription=subscription)
+ subscription = compute_usage_meter(root_path, subscription=subscription, fold=fold)
ab_calibration = _summarize_ab_calibration(root_path)
# --- Summary breakdown (1D, 7D, 30D) ---
# Realized savings from the per-session ledger (sessions/*/savings.jsonl) —
# the same source as the statusline and stop hook.
def _window(d: int) -> dict[str, Any]:
- w = aggregate_window_savings(root_path, days=d)
+ w = aggregate_window_savings(root_path, days=d, fold=fold)
return {
"calls": w.calls_saved,
"usd": round(w.saved_usd, 2),
diff --git a/src/lemoncrow/core/capabilities/savings_summary.py b/src/lemoncrow/core/capabilities/savings_summary.py
index 3bf640773..b304ef415 100644
--- a/src/lemoncrow/core/capabilities/savings_summary.py
+++ b/src/lemoncrow/core/capabilities/savings_summary.py
@@ -2551,7 +2551,7 @@ def _bump_historical_savings_cache(row: dict[str, Any]) -> None:
def _read_historical_savings(
- days: int, root: Path
+ days: int, root: Path, *, fold: bool = True
) -> tuple[float, int, int, int, float, float, float, float, int, int]:
"""Windowed savings/spend for ONE trailing window — blocking surface.
@@ -2559,12 +2559,13 @@ def _read_historical_savings(
reports): reconciles any session ledgers the persisted aggregate has not
folded yet before answering, so explicit surfaces always reflect the
on-disk ledger. The statusline path uses the non-blocking
- :func:`_read_historical_savings_many` directly.
+ :func:`_read_historical_savings_many` directly. ``fold=False`` skips that
+ reconcile and writes nothing.
Returns (savings_usd, tokens_saved, calls_saved, turns_saved, spend_usd,
carry_usd, routing_usd, read_saved_usd, read_saved_tokens, carry_tokens).
"""
- return _read_historical_savings_many((int(days),), root, block=True)[int(days)]
+ return _read_historical_savings_many((int(days),), root, block=True, fold=fold)[int(days)]
# ---------------------------------------------------------------------------
@@ -2963,6 +2964,19 @@ def _get_aggregate_state(root: Path) -> dict[str, Any]:
return agg
+def _aggregate_as_it_stands(root: Path) -> dict[str, Any]:
+ """This process's aggregate for *root*, else the persisted one, else empty.
+
+ The read-only counterpart of :func:`_get_aggregate_state`, whose bootstrap
+ folds and persists: this never does either.
+ """
+ with _aggregate_lock:
+ cached = _aggregate_state.get(str(root))
+ if cached is not None:
+ return cached
+ return _load_persisted_aggregate(root) or _empty_savings_aggregate()
+
+
def _refresh_aggregate_state(root: Path) -> None:
"""Reconcile, swap the in-memory aggregate, and rewrite cached window totals."""
root_str = str(root)
@@ -3000,7 +3014,7 @@ def _maybe_refresh_aggregate(root: Path, *, block: bool) -> None:
def _read_historical_savings_many(
- days_list: tuple[int, ...], root: Path, *, block: bool = False
+ days_list: tuple[int, ...], root: Path, *, block: bool = False, fold: bool = True
) -> dict[int, tuple[float, int, int, int, float, float, float, float, int, int]]:
"""Windowed savings for SEVERAL trailing windows from the day-bucketed
aggregate — never a sessions/** scan on the caller's thread (except a
@@ -3012,6 +3026,11 @@ def _read_historical_savings_many(
current totals (live rows already folded in O(1) by
:func:`_bump_historical_savings_cache`) and refresh in a background thread
that rewrites the cache for the next read.
+ ``fold=False`` (read-only surfaces): writes nothing -- no reconcile, no
+ bootstrap, no background refresh. It answers from the aggregate as it
+ stands, which trails any ledger rows nothing has folded yet, and it caches
+ nothing, so that trailing answer is never what a blocking surface is served
+ for the next TTL.
"""
now = time.time()
root_str = str(root)
@@ -3025,6 +3044,11 @@ def _read_historical_savings_many(
missing.append(days)
if not missing:
return results
+ if not fold:
+ agg = _aggregate_as_it_stands(root)
+ for days in missing:
+ results[days] = _window_from_aggregate(agg, days, now)
+ return results
_get_aggregate_state(root)
_maybe_refresh_aggregate(root, block=block)
with _aggregate_lock:
@@ -3093,7 +3117,7 @@ def total_saved_usd(self) -> float:
return self.saved_usd + self.carry_usd
-def aggregate_window_savings(root: str | Path, *, days: int) -> WindowSavings:
+def aggregate_window_savings(root: str | Path, *, days: int, fold: bool = True) -> WindowSavings:
"""Realized savings over the last *days* from the canonical per-session ledger.
Single source of truth for every windowed savings surface (CLI breakdown,
@@ -3103,9 +3127,12 @@ def aggregate_window_savings(root: str | Path, *, days: int) -> WindowSavings:
(composition time), not inside the per-row day buckets, mirroring how
:func:`compute_savings_summary` folds ``routing_saved_usd`` into its
``saved_usd`` — both still expose the routing figure separately too.
+
+ ``fold=False`` writes nothing, and trails ledger rows no fold has seen yet
+ (see :func:`_read_historical_savings_many`).
"""
usd, tok, calls, turns, spend, carry, routing, read_usd, read_tok, carry_tok = _read_historical_savings(
- int(days), Path(root)
+ int(days), Path(root), fold=fold
)
return WindowSavings(
saved_usd=round(usd + routing, 6),
diff --git a/src/lemoncrow/core/foundation/paths.py b/src/lemoncrow/core/foundation/paths.py
index 019558683..66ab43f0f 100644
--- a/src/lemoncrow/core/foundation/paths.py
+++ b/src/lemoncrow/core/foundation/paths.py
@@ -572,9 +572,20 @@ def resolve_workspace_store_dir(root: Path | str | None = None, workspace_root:
"LemonCrow wrote here" and "git does not see it" are kept inseparable.
"""
ws = Path(workspace_root).expanduser().resolve() if workspace_root is not None else resolve_workspace_root(root)
- store_root = ws / DEFAULT_STORE_DIRNAME
- _ensure_store_self_ignored(store_root)
- return store_root / "workspace"
+ _ensure_store_self_ignored(ws / DEFAULT_STORE_DIRNAME)
+ return workspace_store_dir(ws)
+
+
+def workspace_store_dir(workspace_root: Path | str) -> Path:
+ """``/.lemoncrow/workspace/``, creating nothing.
+
+ The same layout as :func:`resolve_workspace_store_dir` without its
+ self-ignore side effect, for readers that must leave no trace in the
+ checkout -- a tool the MCP broker runs read-only, say. Every writer of
+ project-local runtime data keeps using ``resolve_workspace_store_dir``, so
+ the courtesy stays attached to the moment something is written.
+ """
+ return Path(workspace_root).expanduser().resolve() / DEFAULT_STORE_DIRNAME / "workspace"
def resolve_store_root_for_workspace(workspace_root: Path | str | None = None) -> Path:
@@ -629,4 +640,5 @@ def resolve_store_root_for_workspace(workspace_root: Path | str | None = None) -
"safe_segment",
"session_dir",
"workspace_key",
+ "workspace_store_dir",
]
diff --git a/src/lemoncrow/gateway/adapters/mcp/broker_policy.py b/src/lemoncrow/gateway/adapters/mcp/broker_policy.py
index 4f2a02b12..40a68fa02 100644
--- a/src/lemoncrow/gateway/adapters/mcp/broker_policy.py
+++ b/src/lemoncrow/gateway/adapters/mcp/broker_policy.py
@@ -10,10 +10,11 @@
* ``scan`` runs the ast-grep binary in a subprocess.
* ``context`` records the task on the session ledger.
-* ``statusline_segment`` rewrites the session's statusline sidecar in its
- default ``segment`` format; ``markdown`` and ``json`` fold unfolded session
- ledgers into the persisted savings aggregate. The lemoncrow skill calls it
- directly by name instead.
+* ``statusline_segment`` without ``read_only=true``: it rewrites the session's
+ statusline sidecar in its default ``segment`` format, and ``markdown`` and
+ ``json`` fold unfolded session ledgers into the persisted savings aggregate.
+ With ``read_only=true`` it writes nothing, which is the call a host that can
+ only reach advertised tools makes for the savings panel.
* ``search`` stores each query's results in the workspace search cache
(``smart_state.json``).
* ``graph kind=index_docs`` writes the design-doc store; ``recall_docs`` embeds
@@ -45,9 +46,15 @@
"orient",
"read",
"relations",
+ "statusline_segment",
}
)
+#: ``graph``'s default kind, declared once. ``tool_graph`` and ``_op_graph``
+#: take their signature default from here, so an omitted ``kind`` means the same
+#: operation to the broker as it does to a direct call.
+GRAPH_DEFAULT_KIND: str = "blast_radius"
+
# `graph` runs only these kinds, which read the index or git history.
GRAPH_READ_ONLY_KINDS: frozenset[str] = frozenset(
{
@@ -70,10 +77,16 @@ def broker_refusal(name: str, arguments: Mapping[str, Any]) -> str | None:
"""Why the broker must not run *name* with *arguments*; ``None`` when it may."""
if name not in BROKER_READ_ONLY:
return f"{name!r} is not reachable through the broker, which runs read-only tools only. {_ALTERNATIVES}"
+ if name == "statusline_segment" and arguments.get("read_only") is not True:
+ return (
+ "statusline_segment is not reachable through the broker without read_only=true: otherwise it "
+ "rewrites the statusline sidecar or folds session ledgers into the savings aggregate. "
+ 'Call it with {"read_only": true}.'
+ )
if name == "graph":
if "enable" in arguments:
return f"graph `enable` is not reachable through the broker: it switches on indexing. {_ALTERNATIVES}"
- kind = arguments.get("kind", "blast_radius")
+ kind = arguments.get("kind", GRAPH_DEFAULT_KIND)
if not isinstance(kind, str) or kind not in GRAPH_READ_ONLY_KINDS:
return (
f"graph kind={kind!r} is not reachable through the broker, which runs only the kinds "
diff --git a/src/lemoncrow/gateway/adapters/mcp_server.py b/src/lemoncrow/gateway/adapters/mcp_server.py
index 7a35ca6c8..6dec547b7 100644
--- a/src/lemoncrow/gateway/adapters/mcp_server.py
+++ b/src/lemoncrow/gateway/adapters/mcp_server.py
@@ -81,6 +81,7 @@
tool_bash as tool_bash,
)
from lemoncrow.gateway.adapters.mcp.broker_policy import BROKER_READ_ONLY as _BROKER_READ_ONLY
+from lemoncrow.gateway.adapters.mcp.broker_policy import GRAPH_DEFAULT_KIND as _GRAPH_DEFAULT_KIND
from lemoncrow.gateway.adapters.mcp.broker_policy import broker_refusal as _broker_refusal
from lemoncrow.gateway.adapters.mcp.deferral import ( # noqa: F401 (re-exported for back-compat)
_defer_bash_enabled,
@@ -2561,10 +2562,12 @@ def _workspace_bridge_session_id() -> str:
# Claude resolves via the window-anchored resolver; a workspace-shared
# slot would cross-contaminate concurrent windows in one repo.
return ""
- from lemoncrow.core.foundation.paths import resolve_workspace_store_dir
+ from lemoncrow.core.foundation.paths import workspace_store_dir
ws = os.environ.get("CLAUDE_WORKSPACE_ROOT") or os.getcwd()
- path = resolve_workspace_store_dir(workspace_root=Path(ws)) / "session_state.json"
+ # Non-creating: this reader also runs on the read-only statusline route,
+ # which must leave no trace in the checkout.
+ path = workspace_store_dir(ws) / "session_state.json"
if not path.is_file():
return ""
data = json.loads(path.read_text(encoding="utf-8"))
@@ -9481,7 +9484,7 @@ def _synthesize_edges_for_paths(paths: list[str]) -> list[dict[str, Any]]:
def _op_graph(
*,
- kind: str = "blast_radius",
+ kind: str = _GRAPH_DEFAULT_KIND,
path: str | None = None,
paths: list[str] | None = None,
limit: int = 50,
@@ -10079,7 +10082,7 @@ def _parse_symbol(symbol: str) -> dict[str, Any]:
@mcp_tool(name="graph")
def tool_graph(
- kind: str = "blast_radius",
+ kind: str = _GRAPH_DEFAULT_KIND,
path: str | None = None,
paths: list[str] | None = None,
limit: int = 50,
@@ -10306,7 +10309,7 @@ def tool_blame(
@mcp_tool(name="statusline_segment")
-def tool_statusline_segment(format: str = "segment") -> str:
+def tool_statusline_segment(format: str = "segment", read_only: bool = False) -> str:
"""Savings surface for the active session.
- ``format="segment"`` (default): the pre-computed rotating statusline
@@ -10316,27 +10319,36 @@ def tool_statusline_segment(format: str = "segment") -> str:
that render chat markdown and have no shell to run the CLI.
- ``format="json"``: the raw savings report payload, JSON-encoded.
+ ``read_only=true`` writes nothing: ``segment`` returns the sidecar as it
+ stands -- empty when no host session names one -- and ``markdown``/``json``
+ read the savings aggregate without folding new session ledgers into it, so
+ their totals can trail the newest rows.
+
Hidden from tools/list (see HIDDEN_LLM_TOOLS) but callable by exact name,
which is how the lemoncrow skill answers "what are my savings?" without a
- shell. The `tool` broker refuses it: every format writes (the sidecar, or the
- savings aggregate).
+ shell. The `tool` broker runs it only with ``read_only=true``.
"""
fmt = (format or "segment").strip().lower()
if fmt in {"markdown", "md", "json"}:
from lemoncrow.core.capabilities.plugin_runtime import build_savings_report
from lemoncrow.core.capabilities.savings_summary import render_savings_markdown
- payload = build_savings_report(_lemoncrow_root())
+ payload = build_savings_report(_lemoncrow_root(), fold=not read_only)
if fmt == "json":
return json.dumps(payload, indent=2, sort_keys=True, default=str)
return render_savings_markdown(payload)
try:
+ # Fail closed like _write_statusline_sidecar_now: with no resolvable
+ # session id the sidecar falls back to the workspace store dir, and
+ # resolving that creates /.lemoncrow/ and its .gitignore.
+ if read_only and not _resolved_host_session_id():
+ return ""
sidecar = _get_host_session_sidecar_path()
seg_path = sidecar.parent / "statusline_segment"
sid = sidecar.parent.name
from lemoncrow.core.capabilities.savings_summary import savings_segment
- seg = savings_segment(session_id=sid)
+ seg = "" if read_only else savings_segment(session_id=sid)
if seg:
seg_path.write_text(seg, encoding="utf-8")
return seg
diff --git a/src/lemoncrow/infra/code_intel/coverage.py b/src/lemoncrow/infra/code_intel/coverage.py
index c9291fae1..b5e72f1b9 100644
--- a/src/lemoncrow/infra/code_intel/coverage.py
+++ b/src/lemoncrow/infra/code_intel/coverage.py
@@ -44,6 +44,7 @@
from pathlib import Path
from typing import Any
+from lemoncrow.infra.code_intel import inclusion
from lemoncrow.infra.code_intel.completeness import OBJECTIVE_EXHAUSTIVE
from lemoncrow.infra.code_intel.freshness import require_ready
from lemoncrow.infra.code_intel.inclusion import (
@@ -71,11 +72,6 @@
STATES: tuple[str, ...] = ("indexed", "stale", "missing", "excluded", "unparsed")
-# Kept verbatim for consumers that captured it (PLN-1677 N0). It predates the
-# indexer's rules being readable here; `exclusion_rules` lists the rules this
-# check applies now, and each excluded verdict names its own.
-_EXCLUSION_SOURCE = "git-ignore + unrecognised-file-type"
-
# `exclude_globs` given to an index run are not persisted, so a path the scan
# selects but the index does not hold may be excluded that way or not yet indexed.
_NOT_IN_LAST_RUN = "not in the last index run (an index-time exclude may apply)"
@@ -114,6 +110,10 @@ class CoverageReport:
repo_root: str
engine_index_version: int
+ #: The rules behind this report's ``excluded`` verdicts, in
+ #: :data:`EXCLUSION_RULES` order and joined with `` + ``; empty when no path
+ #: was excluded. A summary of the per-path ``rule`` fields, never a claim
+ #: about rules no returned path used.
exclusion_source: str
totals: dict[str, int]
paths: tuple[PathCoverage, ...]
@@ -202,9 +202,7 @@ def _index_selection(root: Path) -> _IndexSelection:
files = iter_source_files(root)
kept = files
if not licensing.has_feature("context_engine"):
- from lemoncrow.pro.capabilities.code_context.engine import _FREE_TIER_MAX_FILES
-
- kept = free_tier_selection(files, cap=_FREE_TIER_MAX_FILES)
+ kept = free_tier_selection(files, cap=inclusion.FREE_TIER_MAX_FILES)
return _IndexSelection(
selected=_relative_set(root, files),
kept=_relative_set(root, kept),
@@ -231,6 +229,11 @@ def _exclusion(
)
+def _exclusion_source(entries: Sequence[PathCoverage]) -> str:
+ fired = {entry.rule for entry in entries if entry.rule is not None}
+ return " + ".join(rule for rule in EXCLUSION_RULES if rule in fired)
+
+
def _disk_matches(root: Path, rel: str, row: FileRow) -> bool:
"""True when the indexed row still describes what is on disk.
@@ -331,7 +334,7 @@ def check_coverage(paths: list[str] | None = None, repo_root: Path | str = ".")
return CoverageReport(
repo_root=str(root),
engine_index_version=snapshot.index_version,
- exclusion_source=_EXCLUSION_SOURCE,
+ exclusion_source=_exclusion_source(entries),
totals=totals,
paths=tuple(entries),
)
diff --git a/src/lemoncrow/infra/code_intel/inclusion.py b/src/lemoncrow/infra/code_intel/inclusion.py
index 8b8955341..cb24ce0c2 100644
--- a/src/lemoncrow/infra/code_intel/inclusion.py
+++ b/src/lemoncrow/infra/code_intel/inclusion.py
@@ -32,6 +32,7 @@
__all__ = [
"EXCLUSION_RULES",
+ "FREE_TIER_MAX_FILES",
"REASON_SOURCE_FILE_SCAN",
"RULE_FREE_TIER_FILE_CAP",
"RULE_GIT_IGNORE",
@@ -157,12 +158,23 @@ def scan_selects(rel: str, patterns: Sequence[str]) -> bool:
)
+#: Free-tier repo-size cap for the context engine (``context_engine`` is a Pro
+#: feature at scale -- see licensing/features.py). Generous on purpose: this is
+#: well past a typical solo/small-team repo, so Free stays "genuinely useful";
+#: it's a real ceiling only for large monorepos, which is exactly what Pro's
+#: uncapped large-repo indexing is for. The engine's index run and the coverage
+#: verdict that predicts it both read it off this module at call time, so they
+#: always cap at the same number.
+FREE_TIER_MAX_FILES = 2_500
+
+
def free_tier_selection(files: Sequence[Path], *, cap: int) -> list[Path]:
"""The first *cap* paths of the sorted scan, or all of *files* when under it.
One definition of the Free-tier cap, so an index run and the coverage
- verdict that has to predict it cannot drift apart. The caller owns the cap
- value and the licensing check that decides whether it applies at all.
+ verdict that has to predict it cannot drift apart. Callers pass
+ :data:`FREE_TIER_MAX_FILES` and own the licensing check that decides whether
+ it applies at all.
"""
if len(files) <= cap:
return list(files)
diff --git a/src/lemoncrow/pro/capabilities/code_context/engine.py b/src/lemoncrow/pro/capabilities/code_context/engine.py
index fce8e1f10..cb5884256 100644
--- a/src/lemoncrow/pro/capabilities/code_context/engine.py
+++ b/src/lemoncrow/pro/capabilities/code_context/engine.py
@@ -173,12 +173,6 @@ def _query_is_natural_language(query: str) -> bool:
_MAX_FILE_BYTES = 1_000_000
-# Free-tier repo-size cap for the context engine (context_engine is a Pro
-# feature at scale -- see licensing/features.py). Generous on purpose: this is
-# well past a typical solo/small-team repo, so Free stays "genuinely useful";
-# it's a real ceiling only for large monorepos, which is exactly what Pro's
-# uncapped large-repo indexing is for.
-_FREE_TIER_MAX_FILES = 2_500
logger = logging.getLogger(__name__)
@@ -4219,19 +4213,19 @@ def _index_repo_unsafe(
if not self._excluded(path, exclude_globs or [])
]
from lemoncrow.core.capabilities import licensing
- from lemoncrow.infra.code_intel.inclusion import free_tier_selection
+ from lemoncrow.infra.code_intel import inclusion
capped = False
if not licensing.has_feature("context_engine"):
- kept = free_tier_selection(all_files, cap=_FREE_TIER_MAX_FILES)
+ kept = inclusion.free_tier_selection(all_files, cap=inclusion.FREE_TIER_MAX_FILES)
capped = len(kept) != len(all_files)
all_files = kept
if capped:
logger.warning(
"context_engine: repo exceeds the Free-tier cap of %d files; indexing the first %d only "
"(LemonCrow Pro removes this cap)",
- _FREE_TIER_MAX_FILES,
- _FREE_TIER_MAX_FILES,
+ inclusion.FREE_TIER_MAX_FILES,
+ inclusion.FREE_TIER_MAX_FILES,
)
total = len(all_files)
if progress_callback is not None:
diff --git a/tests/core/test_savings_aggregate.py b/tests/core/test_savings_aggregate.py
index 0e63dc101..2633f1c02 100644
--- a/tests/core/test_savings_aggregate.py
+++ b/tests/core/test_savings_aggregate.py
@@ -74,6 +74,23 @@ def _reset_process_state(root: Path) -> None:
del ss._historical_savings_cache[key]
+def test_an_unfolded_read_writes_nothing_and_caches_nothing(tmp_path: Path) -> None:
+ """``fold=False`` answers from the aggregate as it stands and leaves the window cache alone.
+
+ Caching that answer would serve the next blocking surface a total that
+ predates the ledger rows it exists to fold, for a whole TTL.
+ """
+ root = tmp_path / ".lemoncrow"
+ _append(root, "unfolded-s1", [_row(NOW - 3600, 700, 0.7)])
+ _reset_process_state(root)
+
+ unfolded = ss.aggregate_window_savings(root, days=30, fold=False)
+
+ assert (unfolded.tokens_saved, unfolded.saved_usd) == (0, 0.0)
+ assert not (root / "savings_aggregate.json").exists()
+ assert ss.aggregate_window_savings(root, days=30).tokens_saved == 700
+
+
def test_incremental_equals_full_recompute_multi_day(tmp_path: Path) -> None:
root = tmp_path / ".lemoncrow"
# s1: rows on both sides of the 30d boundary (40d ago ages out, 20d ago in).
diff --git a/tests/gateway/test_cap_tools_list_gate.py b/tests/gateway/test_cap_tools_list_gate.py
index 2337f1fd1..792b76eae 100644
--- a/tests/gateway/test_cap_tools_list_gate.py
+++ b/tests/gateway/test_cap_tools_list_gate.py
@@ -125,7 +125,9 @@ def _stub_handler(monkeypatch: pytest.MonkeyPatch, name: str) -> None:
def test_broker_calls_tools_that_are_hidden_under_the_core_profile(monkeypatch: pytest.MonkeyPatch, name: str) -> None:
"""A read-only tool hidden from tools/list must still be reachable through the broker.
- The parametrization is every allow-listed tool the core profile hides.
+ The parametrization is every allow-listed tool the core profile hides, except
+ `statusline_segment`, which runs only with read_only=true (see
+ :func:`test_broker_refuses_the_savings_panel_unless_read_only`).
The old guard refused a tool as "already exposed" whenever it sat in
_CORE_MCP_TOOLS, even when HIDDEN_LLM_TOOLS meant nothing ever advertised
@@ -266,7 +268,6 @@ def test_broker_note_is_absent_when_the_tool_is_genuinely_hidden(monkeypatch: py
"review_rationale",
"search",
"sql",
- "statusline_segment",
"tool",
"trace",
"verify",
@@ -337,6 +338,186 @@ def test_broker_refuses_graph_write_kinds(monkeypatch: pytest.MonkeyPatch, argum
assert _broker({"action": "call", "name": "graph", "arguments": {"kind": "dead_code"}})["called"] == "graph"
+def test_broker_and_graph_resolve_the_same_default_kind(monkeypatch: pytest.MonkeyPatch) -> None:
+ """An omitted `kind` is the same operation to the broker as to a direct call.
+
+ The default was declared three times -- broker_refusal, _op_graph and
+ tool_graph -- so changing one left the broker vetting one kind while the
+ server ran another.
+ """
+ import inspect
+
+ from lemoncrow.gateway.adapters import mcp_server
+ from lemoncrow.gateway.adapters.mcp import broker_policy
+
+ handlers = (mcp_server.tool_graph, mcp_server._op_graph)
+ assert {inspect.signature(handler).parameters["kind"].default for handler in handlers} == {
+ broker_policy.GRAPH_DEFAULT_KIND
+ }
+ assert broker_policy.GRAPH_DEFAULT_KIND in broker_policy.GRAPH_READ_ONLY_KINDS
+
+ # Refuse every kind, so the refusal names the kind the broker resolved an omitted one to.
+ monkeypatch.setattr(broker_policy, "GRAPH_READ_ONLY_KINDS", frozenset())
+ refusal = broker_policy.broker_refusal("graph", {})
+ assert refusal is not None
+ assert f"graph kind={broker_policy.GRAPH_DEFAULT_KIND!r} " in refusal
+
+
+@pytest.mark.parametrize(
+ "arguments",
+ [
+ {},
+ {"format": "markdown"},
+ {"format": "json", "read_only": False},
+ # The handler's validator coerces "false" to False, so a truthiness check
+ # here would let a folding call through.
+ {"format": "markdown", "read_only": "false"},
+ ],
+ ids=["segment", "markdown", "read_only-false", "read_only-string"],
+)
+def test_broker_refuses_the_savings_panel_unless_read_only(monkeypatch: pytest.MonkeyPatch, arguments: dict) -> None:
+ """Without read_only=true the panel rewrites the sidecar or folds ledgers, so the broker refuses it."""
+ from lemoncrow.gateway.adapters import mcp_server
+
+ monkeypatch.setenv("LEMONCROW_MCP_TOOL_PROFILE", "core")
+ _stub_handler(monkeypatch, "statusline_segment")
+ with pytest.raises(mcp_server._ToolArgumentError, match="not reachable through the broker without read_only=true"):
+ _broker({"action": "call", "name": "statusline_segment", "arguments": arguments})
+
+ ran = _broker({"action": "call", "name": "statusline_segment", "arguments": {"read_only": True}})
+ assert ran["called"] == "statusline_segment"
+
+
+def _canary(*roots: Path) -> dict[str, tuple[bytes, int]]:
+ """Every file under each root, by content and mtime."""
+ return {
+ f"{root.name}/{path.relative_to(root).as_posix()}": (path.read_bytes(), path.stat().st_mtime_ns)
+ for root in roots
+ for path in sorted(root.rglob("*"))
+ if path.is_file()
+ }
+
+
+def _seed_savings_ledger(root: Path, session_id: str) -> None:
+ """One session ledger nothing has folded yet, so any fold writes the savings aggregate."""
+ import json
+ from datetime import UTC, datetime
+
+ now = datetime.now(UTC).replace(tzinfo=None)
+ session = root / "sessions" / now.strftime("%Y") / now.strftime("%m") / now.strftime("%d") / "claude" / session_id
+ session.mkdir(parents=True)
+ row = {"tool": "read", "tokens": 1234, "calls": 1, "ts": now.isoformat(), "cost_saved_usd": 0.01}
+ (session / "savings.jsonl").write_text(json.dumps(row) + "\n", encoding="utf-8")
+
+
+def _clear_host_session_ids(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Make the resolved host session id come from this test's env alone.
+
+ Unsetting every host session-id env var is what CI, and any host that names
+ no session, looks like. The window resolver memoizes on the window file's
+ mtime, which is 0.0 for every test (there is no window file), so without
+ dropping that cache the first test in the process to resolve answers for
+ all of them.
+ """
+ from lemoncrow.gateway.adapters import mcp_server
+ from lemoncrow.gateway.adapters.mcp import ledger
+
+ monkeypatch.delenv("CLAUDE_CODE_SESSION_ID", raising=False)
+ for env_var, _host in mcp_server._HOST_SESSION_ENVS:
+ monkeypatch.delenv(env_var, raising=False)
+ monkeypatch.setattr(ledger, "_WINDOW_SID_CACHE", None)
+
+
+_CANARY_SESSION = "canary"
+
+
+@pytest.mark.parametrize("fmt", ["markdown", "json", "segment"])
+@pytest.mark.parametrize("session_id", [_CANARY_SESSION, ""], ids=["host-session", "no-host-session"])
+def test_the_savings_panel_is_reachable_through_the_broker_without_a_write(
+ monkeypatch: pytest.MonkeyPatch, tmp_path: Path, fmt: str, session_id: str
+) -> None:
+ """A host that can only call advertised tools still reaches the panel, and nothing on disk moves.
+
+ Both roots are canaries. The LemonCrow root is seeded with an unfolded
+ ledger and a subscription state so a fold or a refresh has something to
+ write; the workspace root is watched separately, and sits outside the
+ LemonCrow root, because with no session id the segment route falls back to
+ the workspace store dir -- and resolving that creates
+ ``/.lemoncrow/`` and its ``.gitignore`` in the checkout.
+ """
+ import threading
+
+ from lemoncrow.gateway.adapters import mcp_server
+
+ root = tmp_path / "root"
+ workspace = tmp_path / "workspace"
+ root.mkdir()
+ workspace.mkdir()
+ monkeypatch.setenv("LEMONCROW_ROOT", str(root))
+ monkeypatch.setenv("LEMONCROW_MCP_TOOL_PROFILE", "core")
+ monkeypatch.setenv("LEMONCROW_WORKSPACE_ROOT", str(workspace))
+ _clear_host_session_ids(monkeypatch)
+ if session_id:
+ monkeypatch.setenv("CLAUDE_CODE_SESSION_ID", session_id)
+ _seed_legacy_over_cap(root)
+ _seed_savings_ledger(root, session_id or _CANARY_SESSION)
+ # Without this the no-session case is vacuous on a developer machine, where
+ # the ambient CLAUDE_CODE_SESSION_ID resolves and the fallback is never taken.
+ assert mcp_server._resolved_host_session_id() == session_id
+ before = _canary(root, workspace)
+
+ panel = mcp_server._TOOL_BROKER_SPEC["handler"](
+ {"action": "call", "name": "statusline_segment", "arguments": {"format": fmt, "read_only": True}}
+ )
+ # A background fold would land after the call returns.
+ for thread in threading.enumerate():
+ if thread.name == "lemoncrow-savings-aggregate":
+ thread.join(timeout=10)
+
+ assert _canary(root, workspace) == before
+ assert not (workspace / ".lemoncrow").exists()
+ # The canary does see the writes this route avoids: without read_only the
+ # panel folds the ledger, and the segment refreshes its sidecar.
+ if fmt != "segment":
+ assert panel
+ mcp_server.TOOLS["statusline_segment"]["handler"]({"format": fmt})
+ assert (root / "savings_aggregate.json").is_file()
+ elif session_id:
+ mcp_server.TOOLS["statusline_segment"]["handler"]({})
+ assert _canary(root, workspace) != before
+
+
+def test_the_read_only_panel_leaves_no_trace_for_a_host_resolved_through_the_workspace_bridge(
+ monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+ """Codex/OpenCode name their session in a file inside the workspace store dir, and reading it is all this may do.
+
+ That read went through resolve_workspace_store_dir, whose self-ignore
+ courtesy creates ``/.lemoncrow/`` and its ``.gitignore``.
+ """
+ from lemoncrow.gateway.adapters import mcp_server
+
+ root = tmp_path / "root"
+ workspace = tmp_path / "workspace"
+ root.mkdir()
+ workspace.mkdir()
+ monkeypatch.setenv("LEMONCROW_ROOT", str(root))
+ monkeypatch.setenv("LEMONCROW_MCP_TOOL_PROFILE", "core")
+ monkeypatch.setenv("LEMONCROW_WORKSPACE_ROOT", str(workspace))
+ monkeypatch.setenv("CLAUDE_WORKSPACE_ROOT", str(workspace))
+ monkeypatch.setenv("LEMONCROW_AGENT", "codex")
+ _clear_host_session_ids(monkeypatch)
+ assert mcp_server._detect_agent() == "codex"
+ before = _canary(root, workspace)
+
+ mcp_server._TOOL_BROKER_SPEC["handler"](
+ {"action": "call", "name": "statusline_segment", "arguments": {"read_only": True}}
+ )
+
+ assert _canary(root, workspace) == before
+ assert not (workspace / ".lemoncrow").exists()
+
+
def test_broker_refusal_names_read_only_alternatives(monkeypatch: pytest.MonkeyPatch) -> None:
"""Over JSON-RPC, a refusal is an argument error that says where to go instead; the session carries on."""
from lemoncrow.gateway.adapters import mcp_server
diff --git a/tests/infra/code_intel/test_coverage.py b/tests/infra/code_intel/test_coverage.py
index 06d29051c..7edd98f07 100644
--- a/tests/infra/code_intel/test_coverage.py
+++ b/tests/infra/code_intel/test_coverage.py
@@ -162,14 +162,38 @@ def test_whole_repo_mode_covers_tracked_and_indexed_files(
def test_report_states_which_exclusion_rules_it_applied(make_workspace: WorkspaceFactory) -> None:
- """The rules are listed beside ``exclusion_source``, which keeps the value consumers captured."""
+ """Every rule the check applies is listed; none fired, so the summary names none."""
root = make_workspace(files=[{"file_path": "src/a.py"}])
report = check_coverage(paths=["src/a.py"], repo_root=root)
- assert report.exclusion_source == "git-ignore + unrecognised-file-type"
+ assert report.exclusion_source == ""
assert report.to_dict()["exclusion_rules"] == list(EXCLUSION_RULES)
assert report.repo_root == str(root)
+def test_exclusion_source_names_only_the_rules_that_fired(
+ workspace_root: Path, make_workspace: WorkspaceFactory
+) -> None:
+ """The summary agrees with the per-path ``rule`` fields.
+
+ It was a fixed string, so a report whose only exclusion was a skipped
+ directory still named git-ignore, which no path used, and never named the
+ skipped directory, which did.
+ """
+ _write(workspace_root, "data/fixture.py", "def rows():\n return []\n")
+ _write(workspace_root, "prompts/prompt.txt", "Review this diff.\n")
+ root = _index_one_file(workspace_root, make_workspace)
+ _git_init(root, "data/fixture.py", "prompts/prompt.txt")
+
+ report = check_coverage(paths=["prompts/prompt.txt", "src/a.py", "data/fixture.py"], repo_root=root)
+
+ assert sorted(entry.rule for entry in report.paths if entry.rule is not None) == [
+ "skipped-directory",
+ "unrecognised-file-type",
+ ]
+ assert report.exclusion_source == "skipped-directory + unrecognised-file-type"
+ assert report.to_dict()["exclusion_source"] == report.exclusion_source
+
+
def test_rebuilding_index_raises(make_workspace: WorkspaceFactory, tear_index: Callable[[Path], None]) -> None:
"""Verdicts judged against a torn index would call real, indexed files missing."""
root = make_workspace(
@@ -219,7 +243,7 @@ def test_free_tier_cap_reports_excluded(workspace_root: Path, monkeypatch: pytes
for name in ("c", "a", "b"):
_write(workspace_root, f"src/{name}.py", f"def {name}_fn():\n return 1\n")
monkeypatch.setattr("lemoncrow.core.capabilities.licensing.has_feature", lambda _feature: False)
- monkeypatch.setattr("lemoncrow.pro.capabilities.code_context.engine._FREE_TIER_MAX_FILES", 2)
+ monkeypatch.setattr("lemoncrow.infra.code_intel.inclusion.FREE_TIER_MAX_FILES", 2)
CodeContextEngine(workspace_root).index_repo()
report = check_coverage(paths=["src/a.py", "src/b.py", "src/c.py"], repo_root=workspace_root)
diff --git a/tests/infra/code_intel/test_inclusion_layering.py b/tests/infra/code_intel/test_inclusion_layering.py
index 3cc48eebf..92e5d0d87 100644
--- a/tests/infra/code_intel/test_inclusion_layering.py
+++ b/tests/infra/code_intel/test_inclusion_layering.py
@@ -3,9 +3,9 @@
``infra/code_intel/inclusion.py`` owns the indexer's file-selection rules and
``pro`` calls down into them -- ``repo_map/graph.py`` for the whole-repo scan,
``code_context/engine.py`` for the incremental one. An import back up into
-``pro`` at module scope would turn that into a real cycle; the reaches that
-remain (``coverage.py`` reading the engine's scan and its Free-tier cap) are
-deferred to call time on purpose, and this guard keeps them that way.
+``pro`` at module scope would turn that into a real cycle; the reach that
+remains (``coverage.py`` running the engine's own scan) is deferred to call
+time on purpose, and this guard keeps it that way.
"""
from __future__ import annotations
@@ -64,6 +64,28 @@ def test_code_intel_never_imports_pro_at_module_scope() -> None:
)
+def test_file_selection_imports_no_private_pro_symbol() -> None:
+ """The coverage verdict's call-time reach into ``pro`` names only public symbols.
+
+ ``coverage.py`` used to import ``_FREE_TIER_MAX_FILES`` from the engine, so a
+ rename inside ``pro`` broke ``infra`` with nothing in ``pro`` to say why. The
+ cap now lives in ``inclusion.py`` as ``FREE_TIER_MAX_FILES``.
+ """
+ offenders: dict[str, list[str]] = {}
+ for path in (CODE_INTEL / "coverage.py", INCLUSION):
+ tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+ names = [
+ f"{node.module}.{alias.name}"
+ for node in ast.walk(tree)
+ if isinstance(node, ast.ImportFrom) and _imports_pro(node)
+ for alias in node.names
+ if alias.name.startswith("_")
+ ]
+ if names:
+ offenders[path.relative_to(REPO_ROOT).as_posix()] = names
+ assert not offenders, f"lemoncrow.infra.code_intel imports private lemoncrow.pro symbols: {offenders}"
+
+
def test_inclusion_never_imports_pro_at_any_scope() -> None:
tree = ast.parse(INCLUSION.read_text(encoding="utf-8"), filename=str(INCLUSION))
lines = sorted(
diff --git a/tests/integrations/test_mcp_read_allow_hook.py b/tests/integrations/test_mcp_read_allow_hook.py
index 1038ae3ce..555f03aeb 100644
--- a/tests/integrations/test_mcp_read_allow_hook.py
+++ b/tests/integrations/test_mcp_read_allow_hook.py
@@ -7,10 +7,12 @@
from __future__ import annotations
+import importlib.util
import json
import os
import subprocess
import sys
+import types
from pathlib import Path
import pytest
@@ -18,6 +20,15 @@
HOOK = Path(__file__).resolve().parents[2] / "integrations" / "claude" / "plugin" / "hooks" / "mcp_read_allow.py"
+def _hook_module() -> types.ModuleType:
+ """Import the hook by path: it is a plugin script, not part of the package."""
+ spec = importlib.util.spec_from_file_location("lemoncrow_mcp_read_allow_under_test", HOOK)
+ assert spec is not None and spec.loader is not None
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
def _run(
payload: dict, env_extra: dict | None = None, stdin_text: str | None = None
) -> subprocess.CompletedProcess[str]:
@@ -36,10 +47,8 @@ def _run(
[
"mcp__lc__read",
"mcp__lc__code_search",
- "mcp__lc__search",
"mcp__lc__grep",
"mcp__lc__relations",
- "mcp__lc__context",
"mcp__lc__web_fetch",
"mcp__lemoncrow__read",
],
@@ -61,6 +70,11 @@ def test_allows_read_only_tools(tool_name: str) -> None:
"mcp__lc__codemod",
"mcp__lc__memory",
"mcp__lc__verify",
+ # `search` caches every query in the workspace search cache and
+ # `context` records the task on the session ledger, so PRD-739 FR4
+ # classifies both as writers -- Plan Mode must not auto-pass them.
+ "mcp__lc__search",
+ "mcp__lc__context",
# The op-dispatcher can reach write-capable tools by name, so it is not
# laundered through one allow decision.
"mcp__lc__tool",
@@ -94,6 +108,87 @@ def test_malformed_stdin_exits_zero_with_no_output() -> None:
assert proc.stdout == ""
+@pytest.mark.parametrize(
+ "tool_input",
+ [{}, {"kind": "dead_code"}],
+ ids=["default-kind", "dead_code"],
+)
+def test_allows_graph_kinds_that_only_read(tool_input: dict) -> None:
+ proc = _run({"tool_name": "mcp__lc__graph", "tool_input": tool_input})
+ assert proc.returncode == 0, proc.stderr
+ assert json.loads(proc.stdout)["hookSpecificOutput"]["permissionDecision"] == "allow"
+
+
+@pytest.mark.parametrize(
+ "tool_input",
+ [
+ {"kind": "pr_risk", "paths": ["a.py"]},
+ {"kind": "index_docs"},
+ {"kind": "recall_docs", "query": "design"},
+ {"kind": "dead_code", "enable": True},
+ ],
+ ids=["pr_risk", "index_docs", "recall_docs", "enable"],
+)
+def test_stays_silent_for_graph_kinds_that_write(tool_input: dict) -> None:
+ """`graph` is one name over many operations; the writing ones keep prompting."""
+ proc = _run({"tool_name": "mcp__lc__graph", "tool_input": tool_input})
+ assert proc.returncode == 0, proc.stderr
+ assert proc.stdout == ""
+
+
+def test_stays_silent_for_graph_when_the_arguments_are_unreadable() -> None:
+ """No arguments means no kind to judge, and the writing kinds look just like this."""
+ proc = _run({"tool_name": "mcp__lc__graph"})
+ assert proc.returncode == 0, proc.stderr
+ assert proc.stdout == ""
+
+
+@pytest.mark.parametrize("kind", [[], {"a": 1}], ids=["list", "dict"])
+def test_stays_silent_for_a_graph_kind_that_is_not_a_string(kind: object) -> None:
+ """`kind` is raw model input: a list or dict is unhashable, so testing it against a
+ frozenset raised TypeError out of the hook -- a traceback and exit 1 where the
+ contract is exit 0, no output, host prompts.
+ """
+ proc = _run({"tool_name": "mcp__lc__graph", "tool_input": {"kind": kind}})
+ assert proc.returncode == 0, proc.stderr
+ assert proc.stdout == ""
+
+
+# --------------------------------------------------------------------------- #
+# PRD-739 FR4 -- the hook and the broker do not drift #
+# --------------------------------------------------------------------------- #
+
+# The hook is a standalone script Claude Code runs as a subprocess, in installs
+# where `lemoncrow` is not importable, so it cannot import the broker's policy.
+# These tests pin the DIRECTION instead of equality: the broker legitimately
+# runs more than Plan Mode auto-passes (code_query, code_changes,
+# code_coverage_check), and `web_fetch` is auto-passed while the broker refuses
+# it -- for outbound network reach, not for writing anything, and PRD-739 Open
+# Question 2 leaves that undecided. Equality would fail on both.
+_ALLOWED_BUT_NOT_BROKERED = frozenset({"web_fetch"})
+
+
+def test_no_tool_the_broker_calls_a_writer_is_auto_allowed() -> None:
+ from lemoncrow.gateway.adapters.mcp.broker_policy import BROKER_READ_ONLY
+
+ drifted = _hook_module()._READ_ONLY_TOOLS - BROKER_READ_ONLY - _ALLOWED_BUT_NOT_BROKERED
+ assert not drifted, (
+ "Plan Mode auto-allows lc tools the broker will not run: "
+ f"{sorted(drifted)}. Remove them from the hook, or classify them read-only in broker_policy."
+ )
+
+
+def test_no_graph_kind_the_broker_refuses_is_auto_allowed() -> None:
+ from lemoncrow.gateway.adapters.mcp.broker_policy import GRAPH_DEFAULT_KIND, GRAPH_READ_ONLY_KINDS
+
+ hook = _hook_module()
+ drifted = hook._GRAPH_READ_ONLY_KINDS - GRAPH_READ_ONLY_KINDS
+ assert not drifted, f"Plan Mode auto-allows graph kinds the broker refuses: {sorted(drifted)}"
+ # An omitted kind has to mean the same operation on both sides, or the hook
+ # judges one call and the server runs another.
+ assert hook._GRAPH_DEFAULT_KIND == GRAPH_DEFAULT_KIND
+
+
def test_registered_in_plugin_hooks_json() -> None:
hooks = json.loads((HOOK.parent / "hooks.json").read_text(encoding="utf-8"))
commands = [hook["command"] for entry in hooks["hooks"]["PreToolUse"] for hook in entry["hooks"]]