feat(mcp): enter_project becomes a small primer: goal, recent work, open work, vocabulary (#4045)
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / integration (push) Successful in 46s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / Python tests (push) Successful in 1m44s
CI & Build / Build & push image (push) Successful in 23s

The handshake carried the whole project record, every milestone's plan, full
rule text, the notes most recently edited and ~9k of design guidance. For
project 2 that was ~222k characters, past what an MCP client accepts as a tool
result. Each category was walked through with the operator and sized to what a
session needs on arrival; each names the call that has the rest.

- project: id, title, status and the full goal (session start's "full goal"
  pointer still lands here). get_project keeps the whole record.
- milestone_summary: the 5 most recently touched milestones, any status, most
  recent first, without plans. Summaries gain last_touched_at: the later of
  the milestone's own edit and its newest step update, from the query that
  already counts steps. milestone_summary_omitted counts the rest and points
  to list_milestones. get_project and list_milestones list every milestone,
  also without plans.
- open_tasks: the 10 most recently touched open tasks, with or without a
  milestone, each naming its milestone. list_notes gains sort="touched"
  (the later of updated_at and the newest work-log), because a log doesn't
  bump updated_at.
- recent_notes: dropped. Retrieval surfaces notes by relevance, and
  get_recent covers recency.
- systems: id and name.
- design_system: summary plus guidance_call. get_design_system gains
  resolved_guidance, the chain-merged prose; its own guidance field is only
  the departures, so session start's old pointer to it led to a fragment.
  The session start pointer and using-scribe's "Building UI" section now
  name resolved_guidance.
- rules: rules_payload(brief=True) gives project_rules as id and title plus
  subscribed_rulebooks, and records only what it shows. Retrieval delivers
  rules in full and ignores subscriptions (#4052). Other callers unchanged.
- pattern_coverage, inception and systems_bootstrap: unchanged.

Clients: the plugin's using-scribe skill, the compaction notice and session
start are updated here; the REST project summary only gains last_touched_at.
Plugin version minted.

Tests: a size ceiling on the handshake for a large project; milestone and
task selection and naming; brief rules; resolved_guidance; the session
start pointer; and a real-Postgres test that a work-log touches its task and
a step update touches its milestone.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy
This commit is contained in:
2026-09-14 22:04:38 -04:00
co-authored by Claude Opus 5
parent 9b2de3552f
commit 7f974d9749
17 changed files with 445 additions and 182 deletions
+8
View File
@@ -108,14 +108,22 @@ async def get_design_system(design_system_id: int) -> dict:
For what it actually resolves to once inheritance is applied, use
`resolve_design_system`. The two answer different questions and a system
that overrides nothing has an empty token list but a full resolved set.
`resolved_guidance` is the prose to build UI from: the guidance of every
system in the inheritance chain, outermost ancestor first, so the house
style comes before this system's departures from it. `design_system`'s
own `guidance` field is only the departures. Read `resolved_guidance`
before writing UI; enter_project carries just the summary (#4045).
"""
uid = current_user_id()
system = await ds_svc.get_design_system(uid, design_system_id)
if system is None:
raise ValueError(f"design system {design_system_id} not found")
tokens = await ds_svc.list_tokens(uid, design_system_id)
context = await ds_svc.design_context(uid, design_system_id)
return {
"design_system": system.to_dict(),
"resolved_guidance": context["guidance"] if context else [],
"tokens": [t.to_dict() for t in tokens],
}
+84 -83
View File
@@ -42,41 +42,22 @@ async def list_projects() -> dict:
return {"projects": [p.to_dict() for p in rows]}
# Done milestones a project read still lists: the recent ones say what just
# finished, and older ones are a list_milestones call away (#4045).
_DONE_MILESTONES_KEPT = 5
async def _milestone_block(uid: int, project_id: int) -> dict:
"""`milestone_summary` for a project read, brief and bounded (#4045).
Every open milestone plus the most recent done ones, without plan bodies.
`milestone_summary_omitted` is attached only when older done milestones
were left out, and names the calls that reach them (#2483).
"""
rows = await milestones_svc.get_project_milestone_summary(uid, project_id)
brief, omitted = milestones_svc.brief_milestone_summary(
rows, done_kept=_DONE_MILESTONES_KEPT,
)
out: dict = {"milestone_summary": brief}
if omitted:
out["milestone_summary_omitted"] = (
f"{omitted} older done milestone(s) not listed. "
f"list_milestones({project_id}) lists every milestone; "
"get_milestone(id) has one milestone's plan and steps."
)
return out
# How many of each kind the handshake lists. Enough to say what was worked on
# lately and what's open; the rest is a list_milestones / list_tasks call away.
# The handshake once carried every milestone's plan and came to ~222k
# characters, past what a client accepts as a tool result (#4045).
_HANDSHAKE_MILESTONES = 5
_HANDSHAKE_OPEN_TASKS = 10
async def enter_project(project_id: int) -> dict:
"""Session-start handshake: load full context for working on a project.
"""Session-start handshake: a primer on the project before you work in it.
Call this FIRST whenever you're about to do project-scoped work
(start_planning, create_task, update_*, anything that takes a project_id).
One round-trip returns the project, its applicable rules (both rulebook-
subscribed and project-scoped), milestone progress, open tasks, and
recently-updated notes — everything you need to know the lay of the land
before mutating.
One round-trip returns what the project is for, what was worked on
lately, what's open, and the vocabulary to record against. It is kept
small on purpose: each part names the call that has the rest.
No persistent server state: this is a read snapshot. Re-call if the
session goes idle long enough that the data feels stale.
@@ -84,15 +65,29 @@ async def enter_project(project_id: int) -> dict:
Args:
project_id: The project to enter.
Returns a dict with keys: project, milestone_summary, applicable_rules,
project_rules, subscribed_rulebooks, applicable_rules_truncated,
open_tasks, recent_notes, design_system, systems, pattern_coverage —
plus systems_bootstrap, present only when it applies (see below).
Returns a dict with keys: project, milestone_summary, open_tasks, systems,
design_system, project_rules, subscribed_rulebooks, pattern_coverage —
plus milestone_summary_omitted, inception and systems_bootstrap, each
present only when it applies (see below).
`milestone_summary` lists every open milestone and the most recently
finished done ones, each with its description and progress but NOT its
plan: get_milestone(id) reads a plan. `milestone_summary_omitted` appears
only when older done milestones were left out, and says how many.
`project` is id, title, status and the full goal. get_project has the
whole record.
`milestone_summary` is the 5 most recently touched milestones, any status,
most recent first. Touched counts a step changing, not only the milestone
itself. Each carries its description and progress but NOT its plan:
get_milestone(id) reads a plan and its steps. `milestone_summary_omitted`
says how many others exist; list_milestones lists them all.
`open_tasks` is the 10 most recently touched todo / in-progress tasks,
with or without a milestone. A work-log counts as touching its task. Each
names its milestone. list_tasks has the rest.
`project_rules` lists the project's own rules by id and title, and
`subscribed_rulebooks` the rulebooks it draws on. A rule reaches you in
full when your work matches it; get_rule(id) reads one, and
search(content_type="rule") asks whether one covers what you are about
to do.
`pattern_coverage` (usually null) is the shape-accounting line — how many
of the bound repo's extracted shapes carry a classification against canon
@@ -110,7 +105,7 @@ async def enter_project(project_id: int) -> dict:
updating a record, ask which of these areas it is about and pass their ids
as `system_ids`. If the area a record describes is missing from this list,
create it with create_system rather than leaving the area unmodelled. Read
a subsystem's accumulated records with list_system_records.
a subsystem's accumulated records with list_system_records. Each is id and name; get_system has the charter.
`inception` (milestone 297) appears ONLY when the project is yours and
nobody has decided what it inherits: it carries the current defaults
@@ -125,10 +120,12 @@ async def enter_project(project_id: int) -> dict:
lists. It stops appearing the moment the first System exists.
`design_system` is null unless the project points at one. When present it
carries the chain-merged guidance (the house style AND this project's
departures from it) plus a summary of the token set — treat it as binding
for any UI you write, and pull the values with resolve_design_system or
get_design_system_stylesheet before reaching for a literal.
is a summary (title, what it inherits, token count and groups) and
`guidance_call`. The guidance binds any UI you write the way a rule does:
before writing UI, read `resolved_guidance` from get_design_system (the
house style AND this project's departures from it), and pull values with
resolve_design_system or get_design_system_stylesheet before reaching for
a literal.
Entering a project also SCOPES the session: reference and offer work on
this project only, and pass its id to search / list_* so results stay
@@ -151,19 +148,22 @@ async def enter_project(project_id: int) -> dict:
applicable = await rulebooks_svc.get_applicable_rules(
project_id=project_id, user_id=uid,
)
milestones = await _milestone_block(uid, project_id)
milestone_rows = await milestones_svc.get_project_milestone_summary(
uid, project_id,
)
milestone_summary, omitted = milestones_svc.brief_milestone_summary(
milestone_rows, limit=_HANDSHAKE_MILESTONES,
)
milestone_titles = {m["id"]: m.get("title") for m in milestone_rows}
open_tasks, _ = await notes_svc.list_notes(
uid, is_task=True, project_id=project_id,
status=["todo", "in_progress"], sort="updated_at", limit=10,
)
recent_notes, _ = await notes_svc.list_notes(
uid, is_task=False, project_id=project_id,
sort="updated_at", limit=5,
status=["todo", "in_progress"], sort="touched", limit=_HANDSHAKE_OPEN_TASKS,
)
# The tagging vocabulary. Surfaced HERE because an instruction to "tag
# records to Systems" is only executable if the list is in front of the
# agent when it writes — which it never was, and tagging stopped within
# three days of the feature landing (#2546's audit).
# three days of the feature landing (#2546's audit). Untagged writes now
# also ask with the vocabulary listed; this copy lets the first write tag.
systems = await systems_svc.list_systems(uid, project_id)
# The arrival-moment half of the bootstrap ask (#2683): session start is
@@ -185,24 +185,30 @@ async def enter_project(project_id: int) -> dict:
if project.user_id == uid and not inception_svc.is_decided(project):
inception_ask = await inception_svc.inception_ask(uid, project_id)
# Probably the largest surfacing by volume, and it emitted nothing — so
# the pulls it caused floated unattributed and the surfaced:pulled ratio
# ran against a denominator missing its biggest contributor (#2477). An
# AMBIENT source: these are top-N-by-recency, not a ranked choice, and the
# readout counts them apart so dead-weight detection isn't poisoned by
# An AMBIENT source (#2477): top-N-by-recency, not a ranked choice, and
# the readout counts it apart so dead-weight detection isn't poisoned by
# "recently updated in a project you opened".
record_surfaced(
user_id=uid,
note_ids=[int(t.id) for t in open_tasks] + [int(n.id) for n in recent_notes],
note_ids=[int(t.id) for t in open_tasks],
source="enter_project",
)
# A project need not have one, and most installs won't — null is ordinary
# here, not a missing prerequisite.
# here, not a missing prerequisite. Summary only: the guidance is ~9k of
# prose most sessions never use, so it's one call away (#4045).
design_system = None
if project.design_system_id:
design_system = await design_systems_svc.design_context(
design = await design_systems_svc.design_context(
uid, project.design_system_id,
)
if design:
design_system = {
k: design[k]
for k in ("id", "title", "inherits_from", "token_count", "token_groups")
}
design_system["guidance_call"] = (
f"get_design_system({design['id']}) → resolved_guidance"
)
# Cache read ONLY — computing coverage moves a repo tarball and never
# belongs in this request path. Null is the ordinary state (no forge, or
@@ -223,38 +229,33 @@ async def enter_project(project_id: int) -> dict:
)
out = {
"project": project.to_dict(),
"project": {
"id": project.id, "title": project.title,
"status": project.status, "goal": project.goal,
},
"pattern_coverage": coverage_svc.coverage_line(coverage) if coverage else None,
# Trimmed to what tagging needs. The full charter is get_system's job —
# this list rides along on every session start, so it stays lean.
"systems": [
{
"id": s.id, "name": s.name,
"description": (s.description or "").split("\n")[0][:200],
}
for s in systems
],
"systems": [{"id": s.id, "name": s.name} for s in systems],
"design_system": design_system,
**milestones,
**rulebooks_svc.rules_payload(applicable, user_id=uid, source="enter_project"),
"milestone_summary": milestone_summary,
**rulebooks_svc.rules_payload(
applicable, user_id=uid, source="enter_project", brief=True,
),
"open_tasks": [
{
"id": t.id, "title": t.title, "status": t.status,
"priority": t.priority, "task_kind": t.task_kind,
"milestone_id": t.milestone_id,
"milestone_title": milestone_titles.get(t.milestone_id),
}
for t in open_tasks
],
"recent_notes": [
{
"id": n.id, "title": n.title,
"updated_at": n.updated_at.isoformat() if n.updated_at else None,
}
for n in recent_notes
],
}
# Attached only when it applies — a key that usually says null trains
# readers to skip it (#2483), and this one exists to be acted on.
if omitted:
out["milestone_summary_omitted"] = (
f"{omitted} other milestone(s) not listed. "
f"list_milestones({project_id}) lists every milestone."
)
if systems_bootstrap:
out["systems_bootstrap"] = systems_bootstrap
if inception_ask:
@@ -265,18 +266,18 @@ async def enter_project(project_id: int) -> dict:
async def get_project(project_id: int) -> dict:
"""Fetch a Scribe project by ID.
Returns full project fields, a milestone_summary list (shaped as in
enter_project: open milestones and the recent done ones, no plan bodies,
with milestone_summary_omitted when older done ones were left out), and
the rulebook-applicable_rules / subscribed_rulebooks pair the assistant
should consult when working on this project.
Returns full project fields, a milestone_summary list (every milestone,
with description and progress but no plan body; get_milestone reads a
plan), and the rulebook-applicable_rules / subscribed_rulebooks pair the
assistant should consult when working on this project.
"""
uid = current_user_id()
project = await projects_svc.get_project(uid, project_id)
if project is None:
raise ValueError(f"project {project_id} not found")
data = project.to_dict()
data.update(await _milestone_block(uid, project_id))
rows = await milestones_svc.get_project_milestone_summary(uid, project_id)
data["milestone_summary"], _ = milestones_svc.brief_milestone_summary(rows)
applicable = await rulebooks_svc.get_applicable_rules(
project_id=project_id, user_id=uid,
)
+26 -16
View File
@@ -213,9 +213,13 @@ async def get_project_milestone_summaries(
)).scalars().all())
counts: dict[int, dict[str, int]] = {}
step_touched: dict[int, datetime] = {}
if milestones:
rows = await session.execute(
select(Note.milestone_id, Note.status, func.count(Note.id))
select(
Note.milestone_id, Note.status, func.count(Note.id),
func.max(Note.updated_at),
)
.where(
Note.milestone_id.in_([m.id for m in milestones]),
Note.status.isnot(None),
@@ -223,13 +227,21 @@ async def get_project_milestone_summaries(
)
.group_by(Note.milestone_id, Note.status)
)
for milestone_id, status, count in rows.fetchall():
for milestone_id, status, count, latest in rows.fetchall():
counts.setdefault(milestone_id, {})[status] = count
if latest and (milestone_id not in step_touched
or latest > step_touched[milestone_id]):
step_touched[milestone_id] = latest
out: dict[int, list[dict]] = {pid: [] for pid in project_ids}
for m in milestones:
entry = m.to_dict()
entry.update(_progress_from_counts(counts.get(m.id, {})))
# A milestone's own updated_at doesn't move when its steps do, so a
# plan whose steps closed today would read as untouched since it was
# written (#4045). Touched is the later of the two.
touched = [t for t in (m.updated_at, step_touched.get(m.id)) if t]
entry["last_touched_at"] = max(touched).isoformat() if touched else None
out.setdefault(m.project_id, []).append(entry)
return out
@@ -254,22 +266,20 @@ _BRIEF_FIELDS = (
def brief_milestone_summary(
rows: list[dict], *, done_kept: int | None = None,
rows: list[dict], *, limit: int | None = None,
) -> tuple[list[dict], int]:
"""Trim summary rows to the listing fields, optionally capping done ones.
"""Trim summary rows to the listing fields, optionally keeping only the
most recently touched.
`done_kept` keeps only the N most recently updated done milestones (every
open one stays), in the original order_index order. A done plan is
history, and a handshake that grows with a project's whole history grows
past what a client accepts. None keeps every row. Returns (rows, omitted),
where `omitted` counts the done milestones left out.
With `limit`, keeps the N rows with the latest `last_touched_at`, whatever
their status, most recent first: a handshake says what was worked on
lately, and "active" alone doesn't (a plan can sit active for months).
Without it, every row stays in its original order. Returns (rows, omitted).
"""
if done_kept is None:
kept = rows
else:
done = [r for r in rows if r.get("status") == "done"]
recent = sorted(done, key=lambda r: r.get("updated_at") or "", reverse=True)
keep_ids = {r.get("id") for r in recent[:done_kept]}
kept = [r for r in rows if r.get("status") != "done" or r.get("id") in keep_ids]
kept = rows
if limit is not None:
kept = sorted(
rows, key=lambda r: r.get("last_touched_at") or "", reverse=True,
)[:limit]
brief = [{k: r[k] for k in _BRIEF_FIELDS if k in r} for r in kept]
return brief, len(rows) - len(kept)
+16
View File
@@ -318,6 +318,9 @@ async def list_notes(
degraded but never empty. Superseded records are not demoted here: the
penalty reorders a top-k, and reordering a paginated, counted list would
make page boundaries lie. The search surfaces carry the demotion.
`sort` is a Note column name, or "touched": the later of updated_at and
the record's newest work-log.
"""
from scribe.services.access import notes_visibility_clause
@@ -442,6 +445,19 @@ async def list_notes(
if semantic_order is not None:
# A query is a relevance claim — see the docstring.
query = query.order_by(semantic_order)
elif sort == "touched":
# Last touched: the later of the task's own edit and its newest
# work-log. Logging doesn't bump updated_at, so a task worked
# through its logs would otherwise sort as untouched (#4045).
# GREATEST skips a NULL, so a task with no logs sorts by its edit.
from scribe.models.task_log import TaskLog
last_log = (
select(func.max(TaskLog.created_at))
.where(TaskLog.task_id == Note.id)
.scalar_subquery()
)
touched = func.greatest(Note.updated_at, last_log)
query = query.order_by(touched.asc() if order == "asc" else touched.desc())
else:
sort_col = getattr(Note, sort, Note.updated_at)
if order == "asc":
+3 -2
View File
@@ -2236,8 +2236,9 @@ async def build_session_context(
f"Values: `resolve_design_system({design['id']})` · "
f"stylesheet: `get_design_system_stylesheet({design['id']})` "
f"· the prose (aesthetic, voice, where the accent may "
f"appear): `enter_project` returns it, or "
f"`get_design_system({design['id']})`.",
f"appear), inherited house style included: "
f"`get_design_system({design['id']})`"
f"`resolved_guidance`.",
]
elif unbound_repo:
lines += [
+21 -1
View File
@@ -1306,7 +1306,9 @@ async def get_applicable_rules(
}
def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict:
def rules_payload(
applicable: dict, *, user_id: int | None, source: str, brief: bool = False,
) -> dict:
"""The caller-facing shape of a get_applicable_rules() result.
Every surface that hands rules to an agent (enter_project, get_project,
@@ -1333,7 +1335,25 @@ def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict
(`plugin_context`) — computed a marker and showed nobody
anything, and counting those would put rules in the denominator that no
agent ever saw.
`brief` is the session handshake's form (#4045): the project's own rules
as id and title, and the subscribed rulebooks, nothing else. Rules reach a
session in full by retrieval, which ignores subscriptions, so the handshake
lists which constraints exist rather than restating them; get_rule reads
one. Only what is shown is recorded as surfaced.
"""
if brief:
project_rules = [
{"id": r["id"], "title": r["title"]}
for r in applicable.get("project_rules", [])
]
record_rule_surfaced(
user_id=user_id, rule_ids=[r["id"] for r in project_rules], source=source,
)
return {
"project_rules": project_rules,
"subscribed_rulebooks": applicable["subscribed_rulebooks"],
}
record_rule_surfaced(
user_id=user_id,
rule_ids=(