from unittest.mock import AsyncMock, MagicMock, patch import pytest def _rule(rid, title, topic_id): r = MagicMock() r.id, r.title, r.topic_id = rid, title, topic_id r.statement = "FULL STATEMENT SHOULD NOT BE INJECTED" return r def _note(nid, title, user_id=1, note_type="note", is_task=False, task_kind="work"): n = MagicMock() n.id, n.title = nid, title # Real values, defaulting to the caller used in these tests: the injected menu # compares user_id to decide whether the line needs a "shared by …" # attribution, and reads is_task/task_kind/note_type for the kind marker. An # auto-MagicMock is truthy, so every line would read as another user's task. n.user_id = user_id n.note_type, n.is_task, n.task_kind = note_type, is_task, task_kind return n # ─── knowledge auto-inject (Path A) ────────────────────────────────────────── @pytest.mark.asyncio async def test_get_autoinject_config_defaults_and_clamps(): from scribe.services import plugin_context as pc # No settings stored → defaults. with patch.object(pc, "get_setting", AsyncMock(side_effect=lambda uid, k, d: d)): cfg = await pc.get_autoinject_config(1) assert cfg == { "enabled": pc.AUTOINJECT_DEFAULT_ENABLED, "threshold": pc.AUTOINJECT_DEFAULT_THRESHOLD, "top_k": pc.AUTOINJECT_DEFAULT_TOP_K, } # Out-of-range values are clamped; top_k capped at the hard ceiling. stored = { pc.AUTOINJECT_ENABLED_KEY: "false", pc.AUTOINJECT_THRESHOLD_KEY: "5", pc.AUTOINJECT_TOP_K_KEY: "999", } with patch.object(pc, "get_setting", AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))): cfg = await pc.get_autoinject_config(1) assert cfg["enabled"] is False assert cfg["threshold"] == 1.0 assert cfg["top_k"] == pc._AUTOINJECT_MAX_TOP_K @pytest.mark.asyncio async def test_build_autoinject_hint_disabled_returns_empty_and_skips_search(): from scribe.services import plugin_context as pc search = AsyncMock() with patch.object(pc, "get_autoinject_config", AsyncMock(return_value={"enabled": False, "threshold": 0.55, "top_k": 3})), \ patch.object(pc, "semantic_search_notes", search), \ patch.object(pc, "record_retrieval", MagicMock()): out = await pc.build_autoinject_hint(1, "anything") assert out["context"] == "" and out["note_ids"] == [] search.assert_not_called() # disabled → no retrieval at all @pytest.mark.asyncio async def test_build_autoinject_hint_titles_only_with_margin_gate(): from scribe.services import plugin_context as pc # top=0.80; 0.74 within band (0.10), 0.61 outside → dropped. hits = [(0.80, _note(11, "Pool sizing decision")), (0.74, _note(22, "run_maintenance thresholds")), (0.61, _note(33, "unrelated-ish"))] rec = MagicMock() with patch.object(pc, "get_autoinject_config", AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \ patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \ patch.object(pc, "record_retrieval", rec): out = await pc.build_autoinject_hint(1, "postgres pool", project_id=2, exclude_ids=[99]) # Margin gate kept the top two, dropped the straggler. assert out["note_ids"] == [11, 22] assert '#11 [note] "Pool sizing decision" (0.80)' in out["context"] assert "#33" not in out["context"] # Title-first: no body text, ever. assert "get_note(id)" in out["context"] # Telemetry fired with the auto_inject source and the full candidate set. rec.assert_called_once() assert rec.call_args.kwargs["source"] == "auto_inject" @pytest.mark.asyncio async def test_build_autoinject_hint_blank_query_returns_empty(): from scribe.services import plugin_context as pc search = AsyncMock() with patch.object(pc, "get_autoinject_config", AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \ patch.object(pc, "semantic_search_notes", search): out = await pc.build_autoinject_hint(1, " ") assert out["context"] == "" search.assert_not_called() @pytest.mark.asyncio async def test_build_session_context_renders_titles_grouped_by_topic(): rules = [ _rule(1, "`dev` is home", 1), _rule(2, "Release — never without explicit request", 1), _rule(3, "No GitHub — Fabled-Git only", 2), ] with patch("scribe.services.plugin_context.rulebooks_svc.list_always_on_rules", AsyncMock(return_value=rules)), \ patch("scribe.services.plugin_context._topic_titles", AsyncMock(return_value={1: "git-workflow", 2: "fabled-git"})): from scribe.services.plugin_context import build_session_context out = await build_session_context(user_id=7, project_id=0) ctx = out["context"] assert out["rule_count"] == 3 assert out["project"] is None # Titles present, grouped under topic headings assert "### git-workflow" in ctx assert "### fabled-git" in ctx assert "- [1] `dev` is home" in ctx assert "- [3] No GitHub — Fabled-Git only" in ctx # Full statements must NOT be dumped (push channel injects titles only) assert "FULL STATEMENT SHOULD NOT BE INJECTED" not in ctx @pytest.mark.asyncio async def test_build_session_context_includes_project_when_scoped(): # design_system_id explicitly None: a bare MagicMock would hand back a truthy # auto-attribute and send this through the design branch, which is the # opposite of what this test is about. project = MagicMock(id=2, title="FabledScribe", goal="ship it", design_system_id=None) with patch("scribe.services.plugin_context.rulebooks_svc.list_always_on_rules", AsyncMock(return_value=[_rule(1, "rule", 1)])), \ patch("scribe.services.plugin_context._topic_titles", AsyncMock(return_value={1: "git-workflow"})), \ patch("scribe.services.plugin_context.projects_svc.get_project", AsyncMock(return_value=project)), \ patch("scribe.services.plugin_context.notes_svc.list_notes", AsyncMock(return_value=([], 4))): from scribe.services.plugin_context import build_session_context out = await build_session_context(user_id=7, project_id=2) assert out["project"] == {"id": 2, "title": "FabledScribe"} assert "## Active project: FabledScribe (id 2)" in out["context"] assert "Open todo tasks: 4" in out["context"] # No design system on the project -> no design block at all. An install with # none is the ordinary case, not a degraded one. assert "## Design system" not in out["context"] @pytest.mark.asyncio async def test_build_session_context_pushes_the_projects_design_system(): """The gap this closes: a design system had no push channel, so its standards reached a session only if the agent already knew to go looking — the same silent failure as a token nobody declares.""" project = MagicMock(id=2, title="App", goal="", design_system_id=9) design = { "id": 9, "title": "App kit", "description": "", "inherits_from": ["House"], "guidance": [], "token_count": 95, "token_groups": ["accent", "surface", "type"], } with patch("scribe.services.plugin_context.rulebooks_svc.list_always_on_rules", AsyncMock(return_value=[_rule(1, "rule", 1)])), \ patch("scribe.services.plugin_context._topic_titles", AsyncMock(return_value={1: "git-workflow"})), \ patch("scribe.services.plugin_context.projects_svc.get_project", AsyncMock(return_value=project)), \ patch("scribe.services.plugin_context.notes_svc.list_notes", AsyncMock(return_value=([], 0))), \ patch("scribe.services.plugin_context.design_systems_svc.design_context", AsyncMock(return_value=design)): from scribe.services.plugin_context import build_session_context out = await build_session_context(user_id=7, project_id=2) ctx = out["context"] assert "## Design system: App kit (id 9) (inherits House)" in ctx assert "95 tokens across accent, surface, type" in ctx # A pointer to the values, never the values themselves — a hundred token # declarations would crowd out the context they are meant to inform. The # summary carries counts and group NAMES only, which is why design_context # returns those rather than the resolved tokens. assert "resolve_design_system(9)" in ctx assert "get_design_system_stylesheet(9)" in ctx @pytest.mark.asyncio async def test_build_session_context_survives_an_unreadable_design_system(): """design_context returns None when the caller may not read the system. That must degrade to "no design block", not to a crash that costs the session its rules too.""" project = MagicMock(id=2, title="App", goal="", design_system_id=9) with patch("scribe.services.plugin_context.rulebooks_svc.list_always_on_rules", AsyncMock(return_value=[_rule(1, "rule", 1)])), \ patch("scribe.services.plugin_context._topic_titles", AsyncMock(return_value={1: "git-workflow"})), \ patch("scribe.services.plugin_context.projects_svc.get_project", AsyncMock(return_value=project)), \ patch("scribe.services.plugin_context.notes_svc.list_notes", AsyncMock(return_value=([], 0))), \ patch("scribe.services.plugin_context.design_systems_svc.design_context", AsyncMock(return_value=None)): from scribe.services.plugin_context import build_session_context out = await build_session_context(user_id=7, project_id=2) assert "## Design system" not in out["context"] assert "## Active project: App (id 2)" in out["context"] @pytest.mark.asyncio async def test_build_session_context_unbound_repo_emits_bind_hint(): with patch("scribe.services.plugin_context.rulebooks_svc.list_always_on_rules", AsyncMock(return_value=[_rule(1, "rule", 1)])), \ patch("scribe.services.plugin_context._topic_titles", AsyncMock(return_value={1: "git-workflow"})): from scribe.services.plugin_context import build_session_context out = await build_session_context( user_id=7, project_id=0, unbound_repo="host/owner/repo", ) ctx = out["context"] assert out["project"] is None assert "## Repository not yet bound" in ctx assert 'bind_repo(repo_url="host/owner/repo"' in ctx # No project block when unbound. assert "## Active project" not in ctx @pytest.mark.asyncio async def test_build_process_manifest_renders_stub_specs(): items = [ {"id": 5, "title": "Drift Audit", "tags": [], "snippet": "Find drifted docs."}, {"id": 9, "title": "DRY Pass", "tags": [], "snippet": ""}, ] with patch("scribe.services.plugin_context.knowledge_svc.query_knowledge", AsyncMock(return_value=(items, 2))): from scribe.services.plugin_context import build_process_manifest out = await build_process_manifest(user_id=7) assert out["total"] == 2 drift = out["processes"][0] assert drift["id"] == 5 assert drift["name"] == "Drift Audit" assert drift["slug"] == "drift-audit" # kebab-cased assert "Drift Audit" in drift["description"] # auto-surface trigger assert "Find drifted docs." in drift["description"] # preview folded in # No-snippet process still gets a usable description. assert "DRY Pass" in out["processes"][1]["description"] @pytest.mark.asyncio async def test_build_process_manifest_dedupes_slugs_and_skips_blank_titles(): items = [ {"id": 1, "title": "My Process", "tags": [], "snippet": "a"}, {"id": 2, "title": "my process", "tags": [], "snippet": "b"}, # same slug {"id": 3, "title": " ", "tags": [], "snippet": "skip me"}, # blank title ] with patch("scribe.services.plugin_context.knowledge_svc.query_knowledge", AsyncMock(return_value=(items, 3))): from scribe.services.plugin_context import build_process_manifest out = await build_process_manifest(user_id=7) slugs = [p["slug"] for p in out["processes"]] assert slugs == ["my-process", "my-process-2"] # collision suffixed with id assert out["total"] == 2 # blank-title entry dropped @pytest.mark.asyncio async def test_build_process_manifest_truncates_long_preview(): items = [{"id": 1, "title": "Big", "tags": [], "snippet": "x" * 500}] with patch("scribe.services.plugin_context.knowledge_svc.query_knowledge", AsyncMock(return_value=(items, 1))): from scribe.services.plugin_context import build_process_manifest out = await build_process_manifest(user_id=7) assert "…" in out["processes"][0]["description"] assert "x" * 500 not in out["processes"][0]["description"] @pytest.mark.asyncio async def test_build_session_context_caps_length(): many = [_rule(i, "x" * 200, 1) for i in range(200)] with patch("scribe.services.plugin_context.rulebooks_svc.list_always_on_rules", AsyncMock(return_value=many)), \ patch("scribe.services.plugin_context._topic_titles", AsyncMock(return_value={1: "git-workflow"})): from scribe.services.plugin_context import build_session_context out = await build_session_context(user_id=7) assert len(out["context"]) <= 9000 + 60 # cap + truncation note assert "truncated" in out["context"]