"""MCP system tools + task/note issue wiring (service layer mocked).""" from unittest.mock import AsyncMock, MagicMock, patch import pytest from tests.helpers import fake_note, fake_system @pytest.mark.asyncio async def test_create_system_returns_dict(): with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.create_system = AsyncMock(return_value=fake_system(name="Reader")) from scribe.mcp.tools.systems import create_system result = await create_system(project_id=5, name="Reader", description="pdf reader") assert result["name"] == "Reader" @pytest.mark.asyncio async def test_create_system_no_access_raises(): with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.create_system = AsyncMock(return_value=None) from scribe.mcp.tools.systems import create_system with pytest.raises(ValueError): await create_system(project_id=5, name="Reader") @pytest.mark.asyncio async def test_get_system_splits_records_by_kind(): issue = MagicMock(); issue.to_dict.return_value = {"id": 10}; issue.task_kind = "issue"; issue.status = "todo" work = MagicMock(); work.to_dict.return_value = {"id": 11}; work.task_kind = "work"; work.status = "todo" note = MagicMock(); note.to_dict.return_value = {"id": 12}; note.task_kind = "work"; note.status = None with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.get_system = AsyncMock(return_value=fake_system(id=3)) svc.list_records_for_system = AsyncMock(return_value=[issue, work, note]) from scribe.mcp.tools.systems import get_system result = await get_system(system_id=3) assert [r["id"] for r in result["issues"]] == [10] assert [r["id"] for r in result["tasks"]] == [11] assert [r["id"] for r in result["notes"]] == [12] @pytest.mark.asyncio async def test_update_system_not_found_raises(): with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.update_system = AsyncMock(return_value=None) from scribe.mcp.tools.systems import update_system with pytest.raises(ValueError): await update_system(system_id=99, name="x") @pytest.mark.asyncio async def test_create_task_issue_sets_kind_provenance_and_systems(): note = MagicMock(); note.id = 50; note.to_dict.return_value = {"id": 50, "task_kind": "issue"} with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.systems_svc") as systems_svc: notes_svc.create_note = AsyncMock(return_value=note) systems_svc.set_record_systems = AsyncMock(return_value=[2, 3]) systems_svc.list_record_systems = AsyncMock(return_value=[]) from scribe.mcp.tools.tasks import create_task result = await create_task(title="bug", kind="issue", system_ids=[2, 3], arose_from_id=9) _, kwargs = notes_svc.create_note.call_args assert kwargs["task_kind"] == "issue" assert kwargs["arose_from_id"] == 9 systems_svc.set_record_systems.assert_awaited_once_with(1, 50, [2, 3]) assert result["id"] == 50 @pytest.mark.asyncio async def test_untagged_hint_names_the_projects_systems(): sys_a = MagicMock(); sys_a.id = 1; sys_a.name = "workers" sys_b = MagicMock(); sys_b.id = 2; sys_b.name = "scrape-pipeline" with patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.list_systems = AsyncMock(return_value=[sys_a, sys_b]) from scribe.mcp.tools.systems import untagged_systems_hint hint = await untagged_systems_hint(1, 5) assert "workers" in hint and "scrape-pipeline" in hint assert "system_ids" in hint assert "create_system" in hint @pytest.mark.asyncio async def test_untagged_hint_zero_systems_prompts_first_create_and_fails_open(): from scribe.mcp.tools.systems import untagged_systems_hint with patch("scribe.mcp.tools.systems.systems_svc") as svc, \ patch("scribe.mcp.tools.systems.notes_svc") as notes: # Zero Systems in a YOUNG project (below the bootstrap threshold): # the mild question stays proportionate — it must prompt the FIRST # create_system, not go silent. svc.list_systems = AsyncMock(return_value=[]) notes.list_notes = AsyncMock(return_value=([], 3)) hint = await untagged_systems_hint(1, 5) assert "no Systems yet" in hint and "create_system" in hint with patch("scribe.mcp.tools.systems.systems_svc") as svc: # A hint must never break a create — DB failure degrades to no hint. svc.list_systems = AsyncMock(side_effect=RuntimeError("db down")) assert await untagged_systems_hint(1, 5) is None @pytest.mark.asyncio async def test_untagged_hint_escalates_in_a_mature_zero_systems_project(): """#2683: at 282 records and zero Systems (Minstrel), the generic question had demonstrably never converted. The escalated ask must carry the project's OWN evidence — record count and recent titles — and demand a concrete deliverable, because that is the property separating the nudges that convert from the prose that doesn't.""" from scribe.mcp.tools.systems import untagged_systems_hint recent = [fake_note(title="Fix scrape retry backoff"), fake_note(title="Worker pool sizing")] with patch("scribe.mcp.tools.systems.systems_svc") as svc, \ patch("scribe.mcp.tools.systems.notes_svc") as notes: svc.list_systems = AsyncMock(return_value=[]) notes.list_notes = AsyncMock(return_value=(recent, 282)) # The standard names come from the GLOBAL catalog now (milestone 307), # not a module constant — so the ask reads them through the service. # That the SEEDED vocabulary is these eight is migration 0087's # business, asserted against a real database in the inception # integration test; what belongs here is that whatever the catalog # holds reaches the ask verbatim. svc.standard_systems = AsyncMock(return_value=[ ("CI & Release", "..."), ("Auth & Access", "..."), ]) hint = await untagged_systems_hint(1, 5) assert "282 records" in hint assert "Fix scrape retry backoff" in hint # the project's own evidence assert "3-6" in hint and "create_system" in hint # NOT an approval flow (#2798): the agent mints directly, and the # standard cross-project names carry the consistency instead. assert "without asking permission" in hint assert "propose" not in hint and "confirmed" not in hint assert "CI & Release" in hint and "Auth & Access" in hint # The generic wording is REPLACED, not appended — two questions is noise. assert "no Systems yet" not in hint @pytest.mark.asyncio async def test_bootstrap_ask_still_asks_when_the_catalog_is_unreachable(): """The standard names are an AID to the question, not the question. If the catalog read fails (or an install has an empty one), the ask must still carry the project's evidence and demand the same deliverable — degrading to a weaker nudge is acceptable, going silent is not.""" from scribe.mcp.tools.systems import untagged_systems_hint recent = [fake_note(title="Fix scrape retry backoff")] with patch("scribe.mcp.tools.systems.systems_svc") as svc, \ patch("scribe.mcp.tools.systems.notes_svc") as notes: svc.list_systems = AsyncMock(return_value=[]) notes.list_notes = AsyncMock(return_value=(recent, 282)) svc.standard_systems = AsyncMock(side_effect=RuntimeError("db down")) hint = await untagged_systems_hint(1, 5) assert hint is not None assert "282 records" in hint and "create_system" in hint assert "3-6" in hint and "without asking permission" in hint @pytest.mark.asyncio async def test_bootstrap_ask_stays_quiet_below_threshold_and_fails_open(): from scribe.mcp.tools import systems as tools with patch("scribe.mcp.tools.systems.notes_svc") as notes: notes.list_notes = AsyncMock( return_value=([], tools._BOOTSTRAP_MIN_RECORDS - 1)) assert await tools.bootstrap_systems_ask(1, 5) is None with patch("scribe.mcp.tools.systems.notes_svc") as notes: # The record count is decoration on a hint on a create — three layers # deep, nothing there may break the call. notes.list_notes = AsyncMock(side_effect=RuntimeError("db down")) assert await tools.bootstrap_systems_ask(1, 5) is None @pytest.mark.asyncio async def test_populated_vocabulary_never_counts_records(): """The bootstrap query is zero-state-only: once Systems exist the hint must not spend a count query per untagged record.""" from scribe.mcp.tools.systems import untagged_systems_hint sys_a = MagicMock(); sys_a.id = 1; sys_a.name = "workers" with patch("scribe.mcp.tools.systems.systems_svc") as svc, \ patch("scribe.mcp.tools.systems.notes_svc") as notes: svc.list_systems = AsyncMock(return_value=[sys_a]) notes.list_notes = AsyncMock() hint = await untagged_systems_hint(1, 5) assert "workers" in hint notes.list_notes.assert_not_awaited() @pytest.mark.asyncio async def test_create_system_same_normalized_name_is_duplicate_gated(): existing = fake_system(id=7, name="Scrape Pipeline") existing.id = 7 existing.name = "Scrape Pipeline" with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.list_systems = AsyncMock(return_value=[existing]) svc.create_system = AsyncMock() from scribe.mcp.tools.systems import create_system result = await create_system(project_id=5, name=" scrape pipeline ") assert result["duplicate"] is True assert result["existing_id"] == 7 svc.create_system.assert_not_awaited() @pytest.mark.asyncio async def test_create_system_distinct_name_passes_the_gate(): other = fake_system(id=7, name="Workers") other.name = "Workers" with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.list_systems = AsyncMock(return_value=[other]) svc.create_system = AsyncMock(return_value=fake_system(id=8, name="Exporter")) from scribe.mcp.tools.systems import create_system result = await create_system(project_id=5, name="Exporter") assert result["name"] == "Exporter" @pytest.mark.asyncio async def test_create_task_untagged_in_project_carries_the_question(): note = MagicMock(); note.id = 60; note.to_dict.return_value = {"id": 60} sys_a = MagicMock(); sys_a.id = 4; sys_a.name = "plugin-hooks" with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.dedup_svc") as dedup_svc, \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: notes_svc.create_note = AsyncMock(return_value=note) dedup_svc.find_duplicate_note = AsyncMock(return_value=None) seam_svc.list_record_systems = AsyncMock(return_value=[]) seam_svc.list_systems = AsyncMock(return_value=[sys_a]) from scribe.mcp.tools.tasks import create_task result = await create_task(title="untagged", project_id=5) assert "plugin-hooks" in result["systems_hint"] @pytest.mark.asyncio async def test_create_task_tagged_shows_systems_and_projectless_gets_neither(): note = MagicMock(); note.id = 61 # Fresh dict per call — a shared return_value dict lets the first create's # mutation leak into the second create's response. note.to_dict.side_effect = lambda: {"id": 61} tagged_sys = MagicMock(); tagged_sys.to_dict.return_value = {"id": 4, "name": "plugin-hooks"} with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.dedup_svc") as dedup_svc, \ patch("scribe.mcp.tools.tasks.systems_svc") as systems_svc, \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: notes_svc.create_note = AsyncMock(return_value=note) dedup_svc.find_duplicate_note = AsyncMock(return_value=None) systems_svc.set_record_systems = AsyncMock() seam_svc.list_record_systems = AsyncMock(return_value=[tagged_sys]) from scribe.mcp.tools.tasks import create_task tagged = await create_task(title="tagged", project_id=5, system_ids=[4]) seam_svc.list_record_systems = AsyncMock(return_value=[]) orphan = await create_task(title="orphan", project_id=0) assert tagged["systems"] == [{"id": 4, "name": "plugin-hooks"}] assert "systems_hint" not in tagged assert "systems" not in orphan and "systems_hint" not in orphan @pytest.mark.asyncio async def test_get_task_shows_the_records_systems_on_read(): """The touching-a-System reflex needs the affiliation visible on READ — a session opening the task it is about to work must see its areas.""" note = MagicMock() note.id = 70; note.user_id = 1; note.project_id = 5 note.deleted_at = None; note.parent_id = None note.to_dict.return_value = {"id": 70, "task_kind": "work"} tagged_sys = MagicMock(); tagged_sys.to_dict.return_value = {"id": 2, "name": "exporter"} with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.access_svc") as access_svc, \ patch("scribe.mcp.tools.tasks.record_pulled"), \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: notes_svc.get_note_for_user = AsyncMock(return_value=(note, "owner")) access_svc.describe_provenance = AsyncMock(return_value={}) seam_svc.list_record_systems = AsyncMock(return_value=[tagged_sys]) from scribe.mcp.tools.tasks import get_task result = await get_task(task_id=70) assert result["systems"] == [{"id": 2, "name": "exporter"}] @pytest.mark.asyncio async def test_add_task_log_on_untagged_project_task_asks_the_question(): """Logging work IS working in some area — the strongest moment to ask.""" log = MagicMock(); log.to_dict.return_value = {"id": 9, "task_id": 70} task = MagicMock(); task.user_id = 1; task.project_id = 5 with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.task_logs_svc") as logs_svc, \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc, \ patch("scribe.mcp.tools.systems.notes_svc") as seam_notes: logs_svc.create_log = AsyncMock(return_value=log) notes_svc.get_note_for_user = AsyncMock(return_value=(task, "owner")) seam_svc.list_record_systems = AsyncMock(return_value=[]) seam_svc.list_systems = AsyncMock(return_value=[]) # Young project: below the #2683 bootstrap threshold, so the mild # question is the expected form here. seam_notes.list_notes = AsyncMock(return_value=([], 2)) from scribe.mcp.tools.tasks import add_task_log result = await add_task_log(task_id=70, content="progress") assert "no Systems yet" in result["systems_hint"]