diff --git a/src/lore/cli/commands/capture.py b/src/lore/cli/commands/capture.py index 2806541..b31842d 100644 --- a/src/lore/cli/commands/capture.py +++ b/src/lore/cli/commands/capture.py @@ -76,6 +76,7 @@ from typing import Any, Optional from lore.cli.commands._project import resolve_project +from lore.subagent_config import subagent_config logger = logging.getLogger(__name__) @@ -581,6 +582,7 @@ def _spawn_subagent( return None extract_log.parent.mkdir(parents=True, exist_ok=True) log_fh = extract_log.open("a", encoding="utf-8") + cfg = subagent_config(role="capture", with_lore_mcp=True) try: proc = subprocess.Popen( # noqa: S603 — input is internal, not user-supplied # Claude Code 2.1.x rejects `--print --output-format stream-json` @@ -598,11 +600,17 @@ def _spawn_subagent( # The capture prompt is internally generated and only invokes # mcp__lore__* tools (read + write own memory store), so # bypassing permission prompts is the correct trust posture. + # + # ``cfg.claude_flags()`` adds ``--model``, ``--strict-mcp-config``, + # ``--mcp-config``, and ``--settings`` so the subagent runs on + # Haiku with only ``mcp__lore__*`` exposed and no inherited + # plugins / thinking — see lore.subagent_config. [ "claude", "-p", prompt, "--output-format", "stream-json", "--verbose", "--permission-mode", "bypassPermissions", + *cfg.claude_flags(), ], stdin=subprocess.DEVNULL, stdout=log_fh, diff --git a/src/lore/cli/commands/dream.py b/src/lore/cli/commands/dream.py index 7f68b70..cdd7cd9 100644 --- a/src/lore/cli/commands/dream.py +++ b/src/lore/cli/commands/dream.py @@ -41,6 +41,8 @@ from pathlib import Path from typing import Any, Optional +from lore.subagent_config import subagent_config + logger = logging.getLogger(__name__) DEFAULT_ORG_ID = "solo" @@ -413,6 +415,7 @@ def _spawn_subagent( return None extract_log.parent.mkdir(parents=True, exist_ok=True) log_fh = extract_log.open("a", encoding="utf-8") + cfg = subagent_config(role="dream", with_lore_mcp=True) try: return subprocess.Popen( # noqa: S603 — internal prompt # See cli/commands/capture.py for why both flags are required: @@ -422,11 +425,16 @@ def _spawn_subagent( # call from being denied with "you haven't granted it yet". # Dream is a trusted internal subagent; bypassing prompts is # the correct trust posture. + # + # ``cfg.claude_flags()`` pins the dream subagent to Sonnet + # (overridable via LORE_DREAM_MODEL / LORE_SUBAGENT_MODEL) + # with only ``mcp__lore__*`` exposed and no plugins / thinking. [ "claude", "-p", prompt, "--output-format", "stream-json", "--verbose", "--permission-mode", "bypassPermissions", + *cfg.claude_flags(), ], stdin=subprocess.DEVNULL, stdout=log_fh, diff --git a/src/lore/services/graph_extraction.py b/src/lore/services/graph_extraction.py index 02c0d8f..1ee6a7a 100644 --- a/src/lore/services/graph_extraction.py +++ b/src/lore/services/graph_extraction.py @@ -45,6 +45,7 @@ NewRelationship, Store, ) +from lore.subagent_config import subagent_config logger = logging.getLogger(__name__) @@ -238,12 +239,14 @@ def _spawn_claude(prompt: str) -> "subprocess.Popen[bytes]": Caller is responsible for ``.wait()`` and reading stdout. """ + cfg = subagent_config(role="graph", with_lore_mcp=False) return subprocess.Popen( # noqa: S603 — internal prompt [ "claude", "-p", prompt, "--output-format", "stream-json", "--verbose", "--permission-mode", "default", + *cfg.claude_flags(), ], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, diff --git a/src/lore/subagent_config.py b/src/lore/subagent_config.py new file mode 100644 index 0000000..737aa29 --- /dev/null +++ b/src/lore/subagent_config.py @@ -0,0 +1,153 @@ +"""Cheap-subagent config materialization. + +Lore spawns ``claude -p`` subagents from three places (capture-extract, +dream, graph_extraction). Without flags, those subagents inherit the +parent user's full Claude Code stack — default model (often Opus), +``alwaysThinkingEnabled``, ``effortLevel``, every enabled plugin, and +every MCP server (including lore itself, recursively). On Opus this +costs roughly $0.35 / spawn and ~46k cache-creation tokens just to load +the system prompt before the subagent does its real work. + +This module materializes two artifacts under ``~/.lore/subagent/`` — +a minimal MCP config (lore-only or empty) and a minimal settings +override (no plugins, no thinking, low effort) — and returns the +paths plus the chosen model. Spawn sites add ``--model``, +``--strict-mcp-config``, ``--mcp-config``, and ``--settings``. + +Environment overrides: + * ``LORE_SUBAGENT_MODEL`` — fallback default for all roles + * ``LORE_DREAM_MODEL`` — dream-specific override + * ``LORE_GRAPH_MODEL`` — graph-extraction-specific override + +Defaults: capture and graph_extraction use Haiku 4.5 (one-shot +extraction is well within Haiku's range); dream uses Sonnet 4.6 +because it does multi-step reflection and runs at most once per 24h. +""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from pathlib import Path + +# ── Model defaults ──────────────────────────────────────────────── + +_DEFAULT_CHEAP_MODEL = "claude-haiku-4-5" +_DEFAULT_DREAM_MODEL = "claude-sonnet-4-6" + + +def _resolve_model(role: str) -> str: + base = os.environ.get("LORE_SUBAGENT_MODEL") + if role == "dream": + return os.environ.get("LORE_DREAM_MODEL") or base or _DEFAULT_DREAM_MODEL + if role == "graph": + return os.environ.get("LORE_GRAPH_MODEL") or base or _DEFAULT_CHEAP_MODEL + return base or _DEFAULT_CHEAP_MODEL + + +# ── Path layout ─────────────────────────────────────────────────── + + +def _subagent_dir() -> Path: + home = os.environ.get("LORE_HOME") or os.path.expanduser("~/.lore") + return Path(home) / "subagent" + + +def _mcp_with_lore_path() -> Path: + return _subagent_dir() / "mcp-with-lore.json" + + +def _mcp_empty_path() -> Path: + return _subagent_dir() / "mcp-empty.json" + + +def _settings_path() -> Path: + return _subagent_dir() / "settings.json" + + +# ── Config bodies ───────────────────────────────────────────────── + + +def _mcp_with_lore_body() -> dict: + # No ``env`` block: the spawned ``lore mcp`` server inherits env + # from ``claude -p`` which inherits from this Python process. That + # lets LORE_API_URL / LORE_API_KEY / LORE_STORE flow through + # naturally without baking secrets into a file under ~/.lore. + return { + "mcpServers": { + "lore": { + "command": "lore", + "args": ["mcp"], + } + } + } + + +def _mcp_empty_body() -> dict: + return {"mcpServers": {}} + + +def _settings_body() -> dict: + # ``--settings`` merges with the user's settings.json. Setting these + # keys explicitly overrides any inherited values. + return { + "enabledPlugins": {}, + "alwaysThinkingEnabled": False, + "effortLevel": "low", + } + + +# ── Materialization ─────────────────────────────────────────────── + + +def _write_if_changed(path: Path, body: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + new = json.dumps(body, indent=2, sort_keys=True) + "\n" + try: + if path.exists() and path.read_text(encoding="utf-8") == new: + return + except OSError: + pass + path.write_text(new, encoding="utf-8") + + +@dataclass(frozen=True, slots=True) +class SubagentConfig: + model: str + mcp_config_path: Path + settings_path: Path + + def claude_flags(self) -> list[str]: + return [ + "--model", self.model, + "--strict-mcp-config", + "--mcp-config", str(self.mcp_config_path), + "--settings", str(self.settings_path), + ] + + +def subagent_config(*, role: str, with_lore_mcp: bool) -> SubagentConfig: + """Return paths + model for a subagent spawn. + + ``role`` is ``"capture"``, ``"dream"``, or ``"graph"`` and selects + the env-var override chain. ``with_lore_mcp=True`` materializes a + config exposing only ``mcp__lore__*`` to the subagent; + ``False`` materializes an empty MCP config (graph_extraction + needs no tool calls). + """ + settings_path = _settings_path() + _write_if_changed(settings_path, _settings_body()) + + if with_lore_mcp: + mcp_path = _mcp_with_lore_path() + _write_if_changed(mcp_path, _mcp_with_lore_body()) + else: + mcp_path = _mcp_empty_path() + _write_if_changed(mcp_path, _mcp_empty_body()) + + return SubagentConfig( + model=_resolve_model(role), + mcp_config_path=mcp_path, + settings_path=settings_path, + ) diff --git a/tests/services/test_graph_extraction.py b/tests/services/test_graph_extraction.py index e65d775..667aac3 100644 --- a/tests/services/test_graph_extraction.py +++ b/tests/services/test_graph_extraction.py @@ -118,7 +118,7 @@ def test_skips_non_assistant_events(self): class TestSpawnClaudeArgs: - def test_passes_required_flags(self, monkeypatch): + def test_passes_required_flags(self, monkeypatch, tmp_path): captured = {} class FakePopen: @@ -127,6 +127,7 @@ def __init__(self, cmd, **kwargs): captured["kwargs"] = kwargs monkeypatch.setattr(subprocess, "Popen", FakePopen) + monkeypatch.setenv("LORE_HOME", str(tmp_path)) gx._spawn_claude("hello prompt") cmd = captured["cmd"] assert cmd[0] == "claude" @@ -142,6 +143,13 @@ def __init__(self, cmd, **kwargs): assert "--verbose" in flags assert "--permission-mode" in flags assert "default" in flags # not bypassPermissions + # Cheap-subagent flags (lore.subagent_config) — guards against + # silently regressing back to inheriting the user's full Claude + # Code stack on every spawn. + assert "--model" in flags + assert "--strict-mcp-config" in flags + assert "--mcp-config" in flags + assert "--settings" in flags # Stdin/stdout hygiene. assert captured["kwargs"]["stdin"] is subprocess.DEVNULL diff --git a/tests/test_subagent_config.py b/tests/test_subagent_config.py new file mode 100644 index 0000000..01a72ff --- /dev/null +++ b/tests/test_subagent_config.py @@ -0,0 +1,126 @@ +"""Tests for ``lore.subagent_config``. + +Guards the model defaults, env-var override chain, MCP config bodies, +and lazy materialization. Without this, the cheap-subagent flag set +could silently regress back to inheriting the user's full Claude Code +stack (Opus + plugins + thinking) on every spawn. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from lore import subagent_config as sc + + +@pytest.fixture(autouse=True) +def _isolated_lore_home(monkeypatch, tmp_path): + monkeypatch.setenv("LORE_HOME", str(tmp_path)) + for var in ("LORE_SUBAGENT_MODEL", "LORE_DREAM_MODEL", "LORE_GRAPH_MODEL"): + monkeypatch.delenv(var, raising=False) + return tmp_path + + +# ── model resolution ────────────────────────────────────────────── + + +class TestModelResolution: + def test_capture_defaults_to_haiku(self): + cfg = sc.subagent_config(role="capture", with_lore_mcp=True) + assert "haiku" in cfg.model + + def test_graph_defaults_to_haiku(self): + cfg = sc.subagent_config(role="graph", with_lore_mcp=False) + assert "haiku" in cfg.model + + def test_dream_defaults_to_sonnet(self): + cfg = sc.subagent_config(role="dream", with_lore_mcp=True) + assert "sonnet" in cfg.model + + def test_lore_subagent_model_overrides_all_roles(self, monkeypatch): + monkeypatch.setenv("LORE_SUBAGENT_MODEL", "claude-opus-4-7") + for role in ("capture", "dream", "graph"): + assert sc.subagent_config(role=role, with_lore_mcp=False).model == "claude-opus-4-7" + + def test_role_specific_override_wins(self, monkeypatch): + monkeypatch.setenv("LORE_SUBAGENT_MODEL", "claude-opus-4-7") + monkeypatch.setenv("LORE_DREAM_MODEL", "claude-sonnet-4-6") + monkeypatch.setenv("LORE_GRAPH_MODEL", "claude-haiku-4-5") + assert sc.subagent_config(role="dream", with_lore_mcp=True).model == "claude-sonnet-4-6" + assert sc.subagent_config(role="graph", with_lore_mcp=False).model == "claude-haiku-4-5" + assert sc.subagent_config(role="capture", with_lore_mcp=True).model == "claude-opus-4-7" + + +# ── MCP config body ─────────────────────────────────────────────── + + +class TestMcpConfig: + def test_with_lore_mcp_writes_lore_only(self, _isolated_lore_home): + cfg = sc.subagent_config(role="capture", with_lore_mcp=True) + body = json.loads(Path(cfg.mcp_config_path).read_text()) + assert list(body["mcpServers"].keys()) == ["lore"] + # No baked-in env: the spawned `lore mcp` inherits credentials + # from the parent process, so secrets stay out of ~/.lore. + assert "env" not in body["mcpServers"]["lore"] + + def test_without_lore_mcp_writes_empty(self, _isolated_lore_home): + cfg = sc.subagent_config(role="graph", with_lore_mcp=False) + body = json.loads(Path(cfg.mcp_config_path).read_text()) + assert body == {"mcpServers": {}} + + def test_paths_differ_for_with_vs_without(self, _isolated_lore_home): + with_lore = sc.subagent_config(role="capture", with_lore_mcp=True) + without = sc.subagent_config(role="graph", with_lore_mcp=False) + assert with_lore.mcp_config_path != without.mcp_config_path + + +# ── settings body ───────────────────────────────────────────────── + + +class TestSettings: + def test_disables_plugins_thinking_and_high_effort(self, _isolated_lore_home): + cfg = sc.subagent_config(role="capture", with_lore_mcp=True) + body = json.loads(Path(cfg.settings_path).read_text()) + assert body["enabledPlugins"] == {} + assert body["alwaysThinkingEnabled"] is False + assert body["effortLevel"] == "low" + + +# ── claude_flags() shape ────────────────────────────────────────── + + +class TestClaudeFlags: + def test_flag_set_complete_and_ordered(self, _isolated_lore_home): + cfg = sc.subagent_config(role="capture", with_lore_mcp=True) + flags = cfg.claude_flags() + # Each flag carries one positional value; pair them up. + assert flags[0] == "--model" + assert flags[1] == cfg.model + assert "--strict-mcp-config" in flags + # mcp-config / settings paths must round-trip. + i_mcp = flags.index("--mcp-config") + assert flags[i_mcp + 1] == str(cfg.mcp_config_path) + i_set = flags.index("--settings") + assert flags[i_set + 1] == str(cfg.settings_path) + + +# ── materialization ─────────────────────────────────────────────── + + +class TestMaterialization: + def test_creates_subagent_dir_lazily(self, _isolated_lore_home): + subagent_dir = _isolated_lore_home / "subagent" + assert not subagent_dir.exists() + sc.subagent_config(role="capture", with_lore_mcp=True) + assert subagent_dir.is_dir() + + def test_repeated_calls_do_not_rewrite_unchanged_files(self, _isolated_lore_home): + cfg = sc.subagent_config(role="capture", with_lore_mcp=True) + mtime_before = Path(cfg.mcp_config_path).stat().st_mtime_ns + # Second call with same inputs should be a no-op write. + sc.subagent_config(role="capture", with_lore_mcp=True) + mtime_after = Path(cfg.mcp_config_path).stat().st_mtime_ns + assert mtime_before == mtime_after