bvandeusenandClaude Fable 5
3ff8803593
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 7s
CI & Build / integration (push) Successful in 18s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Successful in 47s
CI & Build / Build & push image (push) Successful in 28s
Claude Code injects only the first ~2,048 chars of an MCP server's instructions and silently cuts the rest mid-word (#2562, observed live): _INSTRUCTIONS was 20,002 chars, so ~90% — including all Systems tagging guidance — never reached any session. Rearchitect delivery around what each surface actually delivers: - _INSTRUCTIONS becomes a 1,997-char purpose-sorted map, with a header comment stating the budget and where detail belongs instead. - Tool docstrings keep the per-tool HOW (audit: nearly all displaced topics were already duplicated there); backfill the four gaps — enter_project session scoping + project bootstrap, create_rule entity-vs-rule test, create_design_system not-a-rulebook, create_system two-records test. - The plugin static context (the delivery floor) gains the tag-to-Systems reflex and a surfaces-layering statement; plugin 0.1.25 -> 0.1.26 so the executing cache refreshes (#2209). - create_task / create_note / create_snippet return a systems_hint when a record is created untagged in a project that has Systems — in-band at the exact write it applies to, fail-open like the dedup gate. - Guards: _INSTRUCTIONS length budget, floor-states-the-reflex, and a displaced-topics sweep asserting every cut topic still lives on a delivered surface. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
137 lines
6.7 KiB
Python
137 lines
6.7 KiB
Python
"""MCP system tools + task/note issue wiring (service layer mocked)."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
def _fake_system(sid=1, name="Reader", project_id=5):
|
|
s = MagicMock()
|
|
s.to_dict.return_value = {"id": sid, "name": name, "project_id": project_id}
|
|
s.project_id = project_id
|
|
return s
|
|
|
|
|
|
@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(sid=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_absent_without_systems_and_fails_open():
|
|
from scribe.mcp.tools.systems import untagged_systems_hint
|
|
with patch("scribe.mcp.tools.systems.systems_svc") as svc:
|
|
svc.list_systems = AsyncMock(return_value=[])
|
|
assert await untagged_systems_hint(1, 5) is None
|
|
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_create_task_untagged_in_project_with_systems_carries_hint():
|
|
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 hint_svc:
|
|
notes_svc.create_note = AsyncMock(return_value=note)
|
|
dedup_svc.find_duplicate_note = AsyncMock(return_value=None)
|
|
hint_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_or_projectless_carries_no_hint():
|
|
note = MagicMock(); note.id = 61; note.to_dict.return_value = {"id": 61}
|
|
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 hint_svc:
|
|
notes_svc.create_note = AsyncMock(return_value=note)
|
|
dedup_svc.find_duplicate_note = AsyncMock(return_value=None)
|
|
systems_svc.set_record_systems = AsyncMock()
|
|
systems_svc.list_record_systems = AsyncMock(return_value=[])
|
|
hint_svc.list_systems = AsyncMock(return_value=[MagicMock()])
|
|
from scribe.mcp.tools.tasks import create_task
|
|
tagged = await create_task(title="tagged", project_id=5, system_ids=[4])
|
|
orphan = await create_task(title="orphan", project_id=0)
|
|
assert "systems_hint" not in tagged
|
|
assert "systems_hint" not in orphan
|