refactor(systems): one seam — the Systems question rides every read and write of a record
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / TypeScript typecheck (push) Successful in 10s
CI & Build / integration (push) Successful in 12s
CI & Build / Python tests (push) Failing after 29s
CI & Build / Build & push image (push) Skipped

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 <noreply@anthropic.com>
This commit is contained in:
2026-08-09 15:48:41 -04:00
co-authored by Claude Fable 5
parent 56b1952de5
commit 3455f9cb9a
7 changed files with 169 additions and 86 deletions
+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: