Merge pull request 'The Systems question rides every read and write — unified seam, create_system dedup gate, sweeps-are-discovery prose' (#107) from dev into main
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Successful in 16s
CI & Build / TypeScript typecheck (push) Successful in 33s
CI & Build / Python tests (push) Successful in 48s
CI & Build / Build & push image (push) Successful in 17s

This commit was merged in pull request #107.
This commit is contained in:
2026-08-09 16:04:58 -04:00
8 changed files with 252 additions and 84 deletions
+1 -1
View File
@@ -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.26",
"version": "0.1.28",
"author": { "name": "Bryan Van Deusen" },
"mcpServers": {
"scribe": {
+12 -5
View File
@@ -39,12 +39,19 @@ for the operator's work, and as your own working memory across sessions.
unrelated open task.
- **Tag to Systems as you write** — `enter_project` lists the project's
Systems (its named subsystems/areas). When you create or meaningfully update
a record, ask which area it is about and pass `system_ids`; if the area has
a record, ask which areas it is about and pass `system_ids`; if an area has
no System yet, create it with `create_system` (name + a one-paragraph
charter) rather than leaving it unmodelled. A record about no particular
area takes none — don't force it. Untagged writes in a project that has
Systems come back with a `systems_hint` naming them — treat it as the tagging
question asked at exactly the right moment, not as noise to skip past.
charter) rather than leaving it unmodelled. Cross-cutting records — audits,
sweeps, reviews — take SEVERAL tags, and are the best moment to DISCOVER
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. 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
+7 -1
View File
@@ -88,7 +88,13 @@ Two constraints on *how* that's achieved:
`list_system_records` returns? If the area has no System yet, create one
(`create_system`: name + a one-paragraph charter) — an area that plainly
exists deserves naming the moment two records would share it; don't wait to
be asked. A record about no particular area takes none.
be asked. Cross-cutting records — audits, sweeps, reviews — take *several*
tags and are the prime discovery moment: a pass that walks the subsystems
has just enumerated the vocabulary, so mint the Systems it names as it
names them. Create liberally — `create_system` is duplicate-gated, and that
gate (plus reviewing the existing list) is the guardrail against sprawl,
not restraint. Only a record genuinely about no particular area goes
untagged.
8. **State updates in place; chronicles don't.** A dev-log records what
*happened* — write it once, never rewrite it. A durable finding (how a
+14 -17
View File
@@ -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
+11 -16
View File
@@ -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
+84 -16
View File
@@ -17,29 +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:
return None
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 of those areas, tag it (update it with system_ids=[...]); if "
"its area is missing, create_system it and tag; if it is about no "
"particular area, leave it untagged."
"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,
@@ -54,17 +93,46 @@ async def create_system(
Create one the moment two records would share an area that has no System
yet — the same two-or-more test snippets use. Don't wait to be asked to
name an area that plainly exists in the code; an unmodelled area means
every record about it stays untaggable. Give it a one-paragraph charter,
not just a label: the description is what tells a later session whether a
record belongs here.
every record about it stays untaggable. An audit or sweep that walks the
codebase is a DISCOVERY moment: mint the Systems it names as it names
them — the duplicate gate below, plus reviewing the existing list, is what
guards against sprawl, not holding back. Give each one a one-paragraph
charter, not just a label: the description is what tells a later session
whether a record belongs here.
Args:
project_id: The project this system belongs to (required).
name: Short label (required).
description: What the system is and how it's used — a name is rarely enough.
color: Optional UI accent (hex), or empty.
Duplicate-gated like the other creates: if a System with the same
normalized name already exists in this project (archived included), the
call returns {"duplicate": true, "existing_id": ...} instead of creating —
tag records to that one, or update_system it if its charter needs work.
"""
uid = current_user_id()
norm = " ".join(name.split()).lower()
if norm:
try:
existing = await systems_svc.list_systems(
uid, project_id, include_archived=True
)
except Exception:
existing = []
for s in existing:
if " ".join(s.name.split()).lower() == norm:
return {
"duplicate": True,
"existing_id": s.id,
"message": (
f"System '{s.name}' (#{s.id}) already covers this area "
"in this project. Tag records to it with system_ids, "
"or update_system it if the charter needs revising — "
"a second System with the same name would split the "
"area's records across two piles."
),
}
system = await systems_svc.create_system(
uid, project_id=project_id, name=name,
description=description or None, color=color or None,
+34 -17
View File
@@ -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:
+89 -11
View File
@@ -89,11 +89,14 @@ async def test_untagged_hint_names_the_projects_systems():
@pytest.mark.asyncio
async def test_untagged_hint_absent_without_systems_and_fails_open():
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:
# Zero Systems is the state nothing else nudges out of — the hint must
# prompt the FIRST create_system, not go silent.
svc.list_systems = AsyncMock(return_value=[])
assert await untagged_systems_hint(1, 5) is None
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"))
@@ -101,36 +104,111 @@ async def test_untagged_hint_absent_without_systems_and_fails_open():
@pytest.mark.asyncio
async def test_create_task_untagged_in_project_with_systems_carries_hint():
async def test_create_system_same_normalized_name_is_duplicate_gated():
existing = _fake_system(sid=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(sid=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(sid=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 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():
note = MagicMock(); note.id = 61; note.to_dict.return_value = {"id": 61}
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 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"]