From 3455f9cb9a662d413579ba4a4002cc5a924549b1 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 9 Aug 2026 15:48:41 -0400 Subject: [PATCH] =?UTF-8?q?refactor(systems):=20one=20seam=20=E2=80=94=20t?= =?UTF-8?q?he=20Systems=20question=20rides=20every=20read=20and=20write=20?= =?UTF-8?q?of=20a=20record?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scribe issue #2570, from the operator's challenge: the invariant is "always be asking whether what you're touching is a System's territory and whether the work is filed there" — not a nudge in one corner. The prior shape failed it twice: the hint fired only on creates (with a special-cased zero-Systems branch), and get_task/get_note returned records WITHOUT their Systems, so the read-side reflex had nothing to fire on (same per-kind asymmetry as #2481). - attach_systems(): single helper used by get/create/update for tasks, notes, and snippets, plus add_task_log. Tagged records always show `systems`; an untagged project record carries the `systems_hint` question instead. Neither field attaches empty (#2483). Hint is owner-only; everything fail-open (#2109). - untagged_systems_hint unified to ONE question — the vocabulary listing varies, the question doesn't; the zero-Systems branch stops being special text. - Docstrings state the uniform contract; floor prose now names the read-side reflex (systems visible -> list_system_records the pile). - Plugin 0.1.27 -> 0.1.28. Co-Authored-By: Claude Fable 5 --- plugin/.claude-plugin/plugin.json | 2 +- plugin/hooks/scribe_static_context.md | 9 ++-- src/scribe/mcp/tools/notes.py | 31 +++++------ src/scribe/mcp/tools/snippets.py | 27 ++++------ src/scribe/mcp/tools/systems.py | 76 ++++++++++++++++++--------- src/scribe/mcp/tools/tasks.py | 51 ++++++++++++------ tests/test_mcp_tool_systems.py | 59 ++++++++++++++++++--- 7 files changed, 169 insertions(+), 86 deletions(-) diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json index 8700a91..e6751cf 100644 --- a/plugin/.claude-plugin/plugin.json +++ b/plugin/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "scribe", "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.", - "version": "0.1.27", + "version": "0.1.28", "author": { "name": "Bryan Van Deusen" }, "mcpServers": { "scribe": { diff --git a/plugin/hooks/scribe_static_context.md b/plugin/hooks/scribe_static_context.md index c45d996..9bd8493 100644 --- a/plugin/hooks/scribe_static_context.md +++ b/plugin/hooks/scribe_static_context.md @@ -46,9 +46,12 @@ for the operator's work, and as your own working memory across sessions. missing Systems: a pass that walks the subsystems has just enumerated the vocabulary, so mint what it names. Create liberally; the duplicate gate on `create_system` (and reviewing the existing list) is the guardrail against - sprawl, not restraint. Untagged writes come back with a `systems_hint` — - treat it as the tagging question asked at exactly the right moment, not as - noise to skip past. + sprawl, not restraint. Every read and write of a project record shows its + `systems` — that is the "am I in a System's territory?" signal, and + `list_system_records` reads that territory's whole pile before you work in + it. An untagged project record carries the `systems_hint` question instead, + on creates, updates, and work-logs alike — treat it as the tagging question + asked at the moment of work, not as noise to skip past. - **Reuse before rebuilding** — before writing a new helper/utility/component, search recorded **snippets** (reusable code recorded once for recall) and reuse the prior art instead of re-solving it; when you build something diff --git a/src/scribe/mcp/tools/notes.py b/src/scribe/mcp/tools/notes.py index c4d2d0b..f4e2dc9 100644 --- a/src/scribe/mcp/tools/notes.py +++ b/src/scribe/mcp/tools/notes.py @@ -90,7 +90,9 @@ async def _attach_supersession(uid: int, note_id: int, data: dict) -> None: async def get_note(note_id: int) -> dict: """Fetch the full content of a single Scribe note by its ID. - Returns id, title, body (markdown), tags, project_id, created_at, updated_at. + Returns id, title, body (markdown), tags, project_id, created_at, + updated_at — plus `systems` (the areas this note is filed under) or, for + an untagged project note, the `systems_hint` question. A note another user shared with you also carries `shared`, `owner` and `permission` — read it as their suggestion, not as settled practice you set. @@ -112,6 +114,9 @@ async def get_note(note_id: int) -> dict: # like dead weight next to snippets that merely had a counter (#2085). record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_note") await _attach_supersession(uid, note_id, out) + await systems_tools.attach_systems( + uid, getattr(note, "user_id", uid) or uid, out, note.id, note.project_id + ) return out @@ -150,10 +155,10 @@ async def create_note( Returns the created note object including its assigned id, OR — when a near-duplicate is found and force is false — {"duplicate": true, - "existing_id": ..., "message": ...} and nothing is created. Created - untagged in a project that has Systems, the response carries a - `systems_hint` naming them — answer it: tag the record, create the missing - System, or deliberately leave it untagged. + "existing_id": ..., "message": ...} and nothing is created. A tagged + record shows its `systems`; created untagged in a project, the response + carries the `systems_hint` question instead — answer it: tag the record, + create the missing System, or deliberately leave it untagged. """ uid = current_user_id() if not force: @@ -180,14 +185,7 @@ async def create_note( # not-found, and leave the note rather than silently rolling it back. raise ValueError(str(exc)) from exc data = note.to_dict() - if system_ids: - data["systems"] = [ - s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id) - ] - elif project_id: - hint = await systems_tools.untagged_systems_hint(uid, project_id) - if hint: - data["systems_hint"] = hint + await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None) await _attach_supersession(uid, note.id, data) return data @@ -236,10 +234,9 @@ async def update_note( except PermissionError as exc: raise ValueError(str(exc)) from exc data = note.to_dict() - if system_ids is not None: - data["systems"] = [ - s.to_dict() for s in await systems_svc.list_record_systems(uid, note_id) - ] + await systems_tools.attach_systems( + uid, getattr(note, "user_id", uid) or uid, data, note_id, note.project_id + ) await _attach_supersession(uid, note_id, data) return data diff --git a/src/scribe/mcp/tools/snippets.py b/src/scribe/mcp/tools/snippets.py index f4aec0c..7cd09ed 100644 --- a/src/scribe/mcp/tools/snippets.py +++ b/src/scribe/mcp/tools/snippets.py @@ -133,9 +133,10 @@ async def create_snippet( and it really is the same reusable thing found in another place, prefer merge_snippets(existing_id, [new...]) — or record then merge — to unify them into ONE canonical record (which then carries every call site as a location), - rather than forcing a second copy with force=true. Created untagged in a - project that has Systems, the response carries a `systems_hint` naming - them — answer it: tag, create the missing System, or deliberately skip. + rather than forcing a second copy with force=true. A tagged record shows + its `systems`; created untagged in a project, the response carries the + `systems_hint` question instead — answer it: tag, create the missing + System, or deliberately skip. WHAT THE GATE MATCHES ON. Exact identity first — an existing snippet at the same repo · path · symbol, or holding byte-identical code. Those are certain, @@ -174,14 +175,7 @@ async def create_snippet( if system_ids: await systems_svc.set_record_systems(uid, note.id, system_ids) data = snippets_svc.snippet_to_dict(note) - if system_ids: - data["systems"] = [ - s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id) - ] - elif project_id: - hint = await systems_tools.untagged_systems_hint(uid, project_id) - if hint: - data["systems_hint"] = hint + await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None) return data @@ -205,6 +199,9 @@ async def get_snippet(snippet_id: int) -> dict: # paths, and counting those would inflate exactly the number that is # supposed to mean "someone chose to look at this" (#2085). record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_snippet") + await systems_tools.attach_systems( + uid, note.user_id, data, note.id, note.project_id + ) return data @@ -397,11 +394,9 @@ async def update_snippet( await systems_svc.set_record_systems(note.user_id, snippet_id, system_ids) data = snippets_svc.snippet_to_dict(note) data.update(await access_svc.describe_provenance(uid, note)) - if system_ids is not None: - data["systems"] = [ - s.to_dict() - for s in await systems_svc.list_record_systems(note.user_id, snippet_id) - ] + await systems_tools.attach_systems( + uid, note.user_id, data, snippet_id, note.project_id + ) return data diff --git a/src/scribe/mcp/tools/systems.py b/src/scribe/mcp/tools/systems.py index 60ebb90..aa39bc7 100644 --- a/src/scribe/mcp/tools/systems.py +++ b/src/scribe/mcp/tools/systems.py @@ -17,40 +17,68 @@ from scribe.services import systems as systems_svc async def untagged_systems_hint(user_id: int, project_id: int) -> str | None: - """Nudge text for a record created untagged in a project that has Systems. + """The Systems question, for an untagged project record. - Not a tool. create_task / create_note / create_snippet attach this to - their responses so the tagging question arrives in-band at the exact write - it applies to — instruction prose alone demonstrably doesn't fire at write - time, while in-band behavior (the duplicate gate) does (#2562). + Not a tool — attach_systems() rides this on tool responses so the question + arrives in-band at the exact moment a record is touched. It is ONE + question regardless of vocabulary state ("which area is this about, and is + it modelled?"); only the vocabulary listing varies, because an empty + vocabulary is not an exemption — it is the question at its most urgent + (#2562, #2569). Instruction prose alone demonstrably doesn't fire at write + time; in-band behavior (the duplicate gate) does. """ - # Fail-open like the dedup gate: a hint must never break a create. + # Fail-open like the dedup gate: a hint must never break the call. try: systems = await systems_svc.list_systems(user_id, project_id) except Exception: return None - if not systems: - # Zero Systems is the one state nothing else nudges out of: the hint - # below needs a vocabulary to name, so without this branch the FIRST - # create_system depends entirely on prose that demonstrably doesn't - # fire at write time (#2562). - return ( - "Created untagged — this project has no Systems yet. If this " - "record is about a code subsystem/area, create_system it (name + " - "a one-paragraph charter) and tag the record. An audit or sweep " - "that names areas is exactly the moment to mint them; the " - "duplicate gate is what guards against sprawl, not holding back." - ) - names = ", ".join(f"#{s.id} {s.name}" for s in systems) + if systems: + vocab = "Existing Systems: " + ", ".join( + f"#{s.id} {s.name}" for s in systems + ) + "." + else: + vocab = "This project has no Systems yet." return ( - f"Created untagged. This project's Systems: {names}. If this record is " - "about one or more of those areas, tag it (update it with " - "system_ids=[...]) — cross-cutting records like audits take several; " - "if an area it names is missing, create_system it and tag; only leave " - "it untagged if it is about no particular area." + "This record is untagged — which area(s) of the project is it about? " + f"{vocab} Tag it (system_ids=[...]) — cross-cutting records like " + "audits take several; create_system any area it concerns that isn't " + "modelled yet (a sweep that names areas is the moment to mint them, " + "and the duplicate gate guards against sprawl). Leave it untagged " + "only if it is genuinely about no particular area." ) +async def attach_systems( + caller_id: int, + owner_id: int, + data: dict, + note_id: int, + project_id: int | None, +) -> None: + """Attach a record's Systems to its payload — or the Systems question. + + ONE seam for every tool that returns a project record, read or write: a + tagged record shows its areas (the touching-a-System reflex needs the + affiliation visible on read, not just settable on write), an untagged + project record carries the question instead. Neither field is ever + attached empty (same reasoning as notes._attach_supersession / #2483 — a + field that always says nothing trains readers to skip fields). The hint + goes only to the record's owner: tagging someone else's record in someone + else's project is not the caller's call to make. Fail-open — decoration + must never break the call it rides on. + """ + try: + systems = await systems_svc.list_record_systems(owner_id, note_id) + if systems: + data["systems"] = [s.to_dict() for s in systems] + elif project_id and caller_id == owner_id: + hint = await untagged_systems_hint(owner_id, project_id) + if hint: + data["systems_hint"] = hint + except Exception: + pass + + async def create_system( project_id: int, name: str, diff --git a/src/scribe/mcp/tools/tasks.py b/src/scribe/mcp/tools/tasks.py index b0ec78d..bc1e6fa 100644 --- a/src/scribe/mcp/tools/tasks.py +++ b/src/scribe/mcp/tools/tasks.py @@ -67,7 +67,10 @@ async def get_task(task_id: int) -> dict: """Fetch a single Scribe task by ID. Returns id, title, body, status, priority, tags, project_id, milestone_id, - parent_id, parent_title, due_date, created_at, updated_at. For legacy + parent_id, parent_title, due_date, created_at, updated_at — plus `systems` + (the areas this task is filed under; read a subsystem's whole pile with + list_system_records) or, for an untagged project task, the `systems_hint` + question. For legacy kind=plan tasks, the response also includes applicable_rules + subscribed_rulebooks from the task's project's rulebook subscriptions (new plans are milestones — use get_milestone for those). @@ -108,6 +111,9 @@ async def get_task(task_id: int) -> dict: # surfaced task counted as never-pulled because the tool that opens one didn't # say so, driving auto-inject's measured pull-through toward zero for its own # dominant kind. #1038 and #2085 are explicitly gated on that number (#2245). + await systems_tools.attach_systems( + uid, getattr(note, "user_id", uid) or uid, data, note.id, note.project_id + ) record_pulled(user_id=uid, note_id=int(note.id), source="mcp_get_task") return data @@ -152,9 +158,10 @@ async def create_task( Returns the created task, OR — when a near-duplicate is found and force is false — {"duplicate": true, "existing_id": ..., "message": ...} (nothing - created). Created untagged in a project that has Systems, the response - carries a `systems_hint` naming them — answer it: tag the record, create - the missing System, or deliberately leave it untagged. + created). A tagged record shows its `systems`; created untagged in a + project, the response carries the `systems_hint` question instead — + answer it: tag the record, create the missing System, or deliberately + leave it untagged. """ uid = current_user_id() if kind == "plan": @@ -187,14 +194,7 @@ async def create_task( if system_ids: await systems_svc.set_record_systems(uid, note.id, system_ids) data = note.to_dict() - if system_ids: - data["systems"] = [ - s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id) - ] - elif project_id: - hint = await systems_tools.untagged_systems_hint(uid, project_id) - if hint: - data["systems_hint"] = hint + await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None) return data @@ -258,10 +258,9 @@ async def update_task( if system_ids is not None: await systems_svc.set_record_systems(uid, task_id, system_ids) data = note.to_dict() - if system_ids is not None: - data["systems"] = [ - s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id) - ] + await systems_tools.attach_systems( + uid, getattr(note, "user_id", uid) or uid, data, task_id, note.project_id + ) return data @@ -271,13 +270,31 @@ async def add_task_log(task_id: int, content: str) -> dict: Use this to record work sessions, decisions, or status updates over time without overwriting the task's main body. Each entry is stored separately and shown chronologically in the task view. + + The response shows the task's `systems` — or, if the task is an untagged + project record, the `systems_hint` question: logging work IS working in + some area, so answer it (update_task with system_ids, or create_system + the missing area) rather than logging past it. """ uid = current_user_id() log = await task_logs_svc.create_log(uid, task_id, content) - return log.to_dict() if hasattr(log, "to_dict") else { + data = log.to_dict() if hasattr(log, "to_dict") else { "id": log.id, "task_id": log.task_id, "content": log.content, "created_at": log.created_at.isoformat() if log.created_at else None, } + # A work-log is work happening on the task NOW — the strongest moment to + # ask which System's territory that work is in. Fail-open decoration. + try: + loaded = await notes_svc.get_note_for_user(uid, task_id) + if loaded: + task = loaded[0] + await systems_tools.attach_systems( + uid, getattr(task, "user_id", uid) or uid, + data, task_id, task.project_id, + ) + except Exception: + pass + return data async def start_planning(project_id: int, title: str) -> dict: diff --git a/tests/test_mcp_tool_systems.py b/tests/test_mcp_tool_systems.py index 3b338b9..3656b8a 100644 --- a/tests/test_mcp_tool_systems.py +++ b/tests/test_mcp_tool_systems.py @@ -132,36 +132,79 @@ async def test_create_system_distinct_name_passes_the_gate(): @pytest.mark.asyncio -async def test_create_task_untagged_in_project_with_systems_carries_hint(): +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 hint_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) - hint_svc.list_systems = AsyncMock(return_value=[sys_a]) + 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_or_projectless_carries_no_hint(): +async def test_create_task_tagged_shows_systems_and_projectless_gets_neither(): note = MagicMock(); note.id = 61; note.to_dict.return_value = {"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 hint_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() - systems_svc.list_record_systems = AsyncMock(return_value=[]) - hint_svc.list_systems = AsyncMock(return_value=[MagicMock()]) + 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_hint" not in orphan + 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: + 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=[]) + 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"]