bvandeusenandClaude Opus 5
c58529718b
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 14s
CI & Build / TypeScript typecheck (push) Successful in 48s
CI & Build / integration (push) Successful in 38s
CI & Build / Python tests (push) Failing after 57s
CI & Build / Build & push image (push) Skipped
Step 1 found the reason the standard names never held, and it is sharper than "prose doesn't fire": the list WAS real and it WAS seeded — but only on the inception path, for a project with zero Systems. Ad-hoc create_system never consulted it, which is how Forge minted "CI and Release" and Portal minted "CI & release" after the constant already existed. This wires the vocabulary to the moment that mints a name. - services/systems.assess_system_name: the local duplicate gate AND the catalog lookup, in ONE service function both doors call. The gate lived only in the MCP tool, which is exactly how the web UI shipped without a check the agent surface enforced (#2482). REST now answers 409 with the System that already covers the area. - An `exact` catalog hit is APPLIED (mechanical — the names differ only in spelling). An `overlap` is only OFFERED, on both doors: applying a judgment call silently is how a cross-project rule surfaces in the wrong project. - canonical_systems.best_overlap is the ONE scorer behind the create-time offer and the review sweep, so the two surfaces can never name different areas for one System. It also takes the catalog the caller already holds, so the review is not an N+1. UI (folded in from step 1 — rule 27, that step shipped with no human surface): - SystemsSection: a Shared area picker on create and edit, the area on each card, and a collapsed review of proposals that appears only when there is something to decide. `exact` and `overlap` never share a style — one is mechanical, the other is the reviewer's judgment, and presenting them alike is how a wrong mapping gets waved through. - Settings → Admin → Areas: the catalog itself, showing each entry's slug, because the slug is what decides whether two names are the same area and a rename moves it. - A picker rather than a live matcher: reproducing the slug rule in TypeScript would give this feature two matchers to keep in step — the exact drift the catalog exists to end. The server stays authoritative. tests/helpers.fake_system gains canonical_id=None: an unnamed attribute is an auto-MagicMock and therefore truthy, which is the trap that helper exists for (note 2109) and a nullable FK walks straight into it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
359 lines
18 KiB
Python
359 lines
18 KiB
Python
"""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
|
|
|
|
|
|
# The name assessment both doors run before minting (milestone 307). A test
|
|
# that patches systems_svc wholesale must stub it, or the awaited MagicMock
|
|
# raises — this shape is the "nothing matched" answer.
|
|
_NO_MATCH = {"duplicate": None, "canonical": None}
|
|
|
|
|
|
@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.assess_system_name = AsyncMock(return_value=_NO_MATCH)
|
|
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"
|
|
# An unmatched name is a project-specific area: created, no offer, no fuss.
|
|
assert "canonical_suggestion" not in result and "canonical_note" not in result
|
|
|
|
|
|
@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.assess_system_name = AsyncMock(return_value=_NO_MATCH)
|
|
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_create_system_applies_an_exact_area_and_offers_a_similar_one():
|
|
"""The two bases must behave differently, and this is where it is decided.
|
|
|
|
`exact` differs from the catalog name only in spelling, so it is APPLIED —
|
|
that is the mechanical case the catalog exists to collapse. `overlap` is a
|
|
judgment call, so it is only OFFERED: applying it silently is how a
|
|
cross-project rule ends up surfacing in the wrong project.
|
|
"""
|
|
from scribe.mcp.tools.systems import create_system
|
|
|
|
with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("scribe.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value={
|
|
"duplicate": None,
|
|
"canonical": {"id": 3, "name": "CI & Release", "basis": "exact"},
|
|
})
|
|
svc.create_system = AsyncMock(return_value=fake_system(name="CI and Release"))
|
|
exact = await create_system(project_id=5, name="CI and Release")
|
|
assert svc.create_system.await_args.kwargs["canonical_id"] == 3
|
|
assert "canonical_note" in exact and "canonical_suggestion" not in exact
|
|
|
|
with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("scribe.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value={
|
|
"duplicate": None,
|
|
"canonical": {"id": 3, "name": "CI & Release", "basis": "overlap", "score": 0.33},
|
|
})
|
|
svc.create_system = AsyncMock(return_value=fake_system(id=9, name="CI & runners"))
|
|
similar = await create_system(project_id=5, name="CI & runners")
|
|
assert svc.create_system.await_args.kwargs["canonical_id"] is None
|
|
assert similar["canonical_suggestion"]["id"] == 3
|
|
assert "map_system_to_canonical(9, 3)" in similar["canonical_suggestion"]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_system_duplicate_names_the_existing_one_and_creates_nothing():
|
|
from scribe.mcp.tools.systems import create_system
|
|
with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
|
|
patch("scribe.mcp.tools.systems.systems_svc") as svc:
|
|
svc.assess_system_name = AsyncMock(return_value={
|
|
"duplicate": {"id": 4, "name": "Reader"}, "canonical": None,
|
|
})
|
|
svc.create_system = AsyncMock()
|
|
result = await create_system(project_id=5, name="reader")
|
|
assert result["duplicate"] is True and result["existing_id"] == 4
|
|
assert "Reader" in result["message"]
|
|
svc.create_system.assert_not_awaited()
|
|
|
|
|
|
@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"]
|