enter_project becomes a small primer instead of a 222k-character dump #157

Merged
bvandeusen merged 2 commits from dev into main 2026-09-14 22:12:18 -04:00
19 changed files with 559 additions and 88 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
{ {
"name": "scribe", "name": "scribe",
"description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).", "description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
"version": "2026.09.14.2254", "version": "2026.09.15.0204",
"author": { "author": {
"name": "Bryan Van Deusen" "name": "Bryan Van Deusen"
}, },
+1 -1
View File
@@ -180,7 +180,7 @@ fi
# Compaction re-grounding: lead with a reload banner when this fire is a compact. # Compaction re-grounding: lead with a reload banner when this fire is a compact.
if [ "$source" = "compact" ]; then if [ "$source" = "compact" ]; then
prepend "> ⟳ This session was just COMPACTED — earlier turns are now a summary, so in-flight detail may be lost. Any rules that had been retrieved went into that summary with everything else, so treat yourself as holding none: before the next consequential act, ask again with \`search(content_type=\"rule\")\` rather than trusting a half-remembered one. Re-run \`enter_project()\` for the active project, check its open tasks and recent notes, and reconcile what you are mid-way through against what Scribe records. Scribe is the record." prepend "> ⟳ This session was just COMPACTED — earlier turns are now a summary, so in-flight detail may be lost. Any rules that had been retrieved went into that summary with everything else, so treat yourself as holding none: before the next consequential act, ask again with \`search(content_type=\"rule\")\` rather than trusting a half-remembered one. Re-run \`enter_project()\` for the active project, check its recent milestones and open tasks, and reconcile what you are mid-way through against what Scribe records. Scribe is the record."
fi fi
# Nothing at all to inject → stay silent. # Nothing at all to inject → stay silent.
+6 -5
View File
@@ -15,8 +15,8 @@ asked for.
If the working repo maps to a Scribe project (you're in a known repo, or If the working repo maps to a Scribe project (you're in a known repo, or
`list_repo_bindings` shows a binding), call `enter_project(id)` — it returns the `list_repo_bindings` shows a binding), call `enter_project(id)` — it returns the
project plus the rules bound to the areas it works in, open tasks, and recent project's goal, the milestones and open tasks worked on most recently, its
notes in one shot. Systems and the titles of its own rules in one shot.
Then **ask before you act**: before anything hard to reverse or outward-facing, Then **ask before you act**: before anything hard to reverse or outward-facing,
search the rules for what you are about to do. Reflex 2 below is why asking, search the rules for what you are about to do. Reflex 2 below is why asking,
@@ -294,9 +294,10 @@ really is a standing instruction about how to work.
## Building UI: the project's design system binds ## Building UI: the project's design system binds
`enter_project` returns a `design_system` when the project has one, with the `enter_project` names the project's `design_system` when it has one. Before
guidance **chain-merged** — the house style it inherits plus its own departures writing UI, read its guidance: `get_design_system(id)` returns it as
from it. Treat it the way you treat a rule. `resolved_guidance`, **chain-merged** — the house style it inherits plus its own
departures from it. Treat it the way you treat a rule.
Before writing a colour, size, radius, weight or duration by hand, reach for a Before writing a colour, size, radius, weight or duration by hand, reach for a
token: `resolve_design_system(id)` for the values, or token: `resolve_design_system(id)` for the values, or
+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 For what it actually resolves to once inheritance is applied, use
`resolve_design_system`. The two answer different questions and a system `resolve_design_system`. The two answer different questions and a system
that overrides nothing has an empty token list but a full resolved set. 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() uid = current_user_id()
system = await ds_svc.get_design_system(uid, design_system_id) system = await ds_svc.get_design_system(uid, design_system_id)
if system is None: if system is None:
raise ValueError(f"design system {design_system_id} not found") raise ValueError(f"design system {design_system_id} not found")
tokens = await ds_svc.list_tokens(uid, design_system_id) tokens = await ds_svc.list_tokens(uid, design_system_id)
context = await ds_svc.design_context(uid, design_system_id)
return { return {
"design_system": system.to_dict(), "design_system": system.to_dict(),
"resolved_guidance": context["guidance"] if context else [],
"tokens": [t.to_dict() for t in tokens], "tokens": [t.to_dict() for t in tokens],
} }
+6 -3
View File
@@ -22,12 +22,15 @@ from scribe.services.record_refs import refuse_guessed_ids
async def list_milestones(project_id: int) -> dict: async def list_milestones(project_id: int) -> dict:
"""List milestones for a Scribe project, ordered by order_index. """List milestones for a Scribe project, ordered by order_index.
Returns id, title, description, body (the plan/design), status Returns every milestone, done ones included: id, title, description,
(active/done), order_index, and task counts. status (active/done), order_index and progress (total, completed, pct,
status_counts). The plan itself is not listed: get_milestone(id) returns a
milestone's body and its steps.
""" """
uid = current_user_id() uid = current_user_id()
rows = await milestones_svc.get_project_milestone_summary(uid, project_id) rows = await milestones_svc.get_project_milestone_summary(uid, project_id)
return {"milestones": rows} brief, _ = milestones_svc.brief_milestone_summary(rows)
return {"milestones": brief}
async def get_milestone(milestone_id: int) -> dict: async def get_milestone(milestone_id: int) -> dict:
+84 -54
View File
@@ -42,15 +42,22 @@ async def list_projects() -> dict:
return {"projects": [p.to_dict() for p in rows]} return {"projects": [p.to_dict() for p in rows]}
# 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: 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 Call this FIRST whenever you're about to do project-scoped work
(start_planning, create_task, update_*, anything that takes a project_id). (start_planning, create_task, update_*, anything that takes a project_id).
One round-trip returns the project, its applicable rules (both rulebook- One round-trip returns what the project is for, what was worked on
subscribed and project-scoped), milestone progress, open tasks, and lately, what's open, and the vocabulary to record against. It is kept
recently-updated notes — everything you need to know the lay of the land small on purpose: each part names the call that has the rest.
before mutating.
No persistent server state: this is a read snapshot. Re-call if the No persistent server state: this is a read snapshot. Re-call if the
session goes idle long enough that the data feels stale. session goes idle long enough that the data feels stale.
@@ -58,10 +65,29 @@ async def enter_project(project_id: int) -> dict:
Args: Args:
project_id: The project to enter. project_id: The project to enter.
Returns a dict with keys: project, milestone_summary, applicable_rules, Returns a dict with keys: project, milestone_summary, open_tasks, systems,
project_rules, subscribed_rulebooks, applicable_rules_truncated, design_system, project_rules, subscribed_rulebooks, pattern_coverage —
open_tasks, recent_notes, design_system, systems, pattern_coverage — plus milestone_summary_omitted, inception and systems_bootstrap, each
plus systems_bootstrap, present only when it applies (see below). present only when it applies (see below).
`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 `pattern_coverage` (usually null) is the shape-accounting line — how many
of the bound repo's extracted shapes carry a classification against canon of the bound repo's extracted shapes carry a classification against canon
@@ -79,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 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, 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 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 `inception` (milestone 297) appears ONLY when the project is yours and
nobody has decided what it inherits: it carries the current defaults nobody has decided what it inherits: it carries the current defaults
@@ -94,10 +120,12 @@ async def enter_project(project_id: int) -> dict:
lists. It stops appearing the moment the first System exists. lists. It stops appearing the moment the first System exists.
`design_system` is null unless the project points at one. When present it `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 is a summary (title, what it inherits, token count and groups) and
departures from it) plus a summary of the token set — treat it as binding `guidance_call`. The guidance binds any UI you write the way a rule does:
for any UI you write, and pull the values with resolve_design_system or before writing UI, read `resolved_guidance` from get_design_system (the
get_design_system_stylesheet before reaching for a literal. 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 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 this project only, and pass its id to search / list_* so results stay
@@ -120,21 +148,22 @@ async def enter_project(project_id: int) -> dict:
applicable = await rulebooks_svc.get_applicable_rules( applicable = await rulebooks_svc.get_applicable_rules(
project_id=project_id, user_id=uid, project_id=project_id, user_id=uid,
) )
milestone_summary = await milestones_svc.get_project_milestone_summary( milestone_rows = await milestones_svc.get_project_milestone_summary(
uid, project_id, 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( open_tasks, _ = await notes_svc.list_notes(
uid, is_task=True, project_id=project_id, uid, is_task=True, project_id=project_id,
status=["todo", "in_progress"], sort="updated_at", limit=10, status=["todo", "in_progress"], sort="touched", limit=_HANDSHAKE_OPEN_TASKS,
)
recent_notes, _ = await notes_svc.list_notes(
uid, is_task=False, project_id=project_id,
sort="updated_at", limit=5,
) )
# The tagging vocabulary. Surfaced HERE because an instruction to "tag # The tagging vocabulary. Surfaced HERE because an instruction to "tag
# records to Systems" is only executable if the list is in front of the # 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 # 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) systems = await systems_svc.list_systems(uid, project_id)
# The arrival-moment half of the bootstrap ask (#2683): session start is # The arrival-moment half of the bootstrap ask (#2683): session start is
@@ -156,24 +185,30 @@ async def enter_project(project_id: int) -> dict:
if project.user_id == uid and not inception_svc.is_decided(project): if project.user_id == uid and not inception_svc.is_decided(project):
inception_ask = await inception_svc.inception_ask(uid, project_id) inception_ask = await inception_svc.inception_ask(uid, project_id)
# Probably the largest surfacing by volume, and it emitted nothing — so # An AMBIENT source (#2477): top-N-by-recency, not a ranked choice, and
# the pulls it caused floated unattributed and the surfaced:pulled ratio # the readout counts it apart so dead-weight detection isn't poisoned by
# 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
# "recently updated in a project you opened". # "recently updated in a project you opened".
record_surfaced( record_surfaced(
user_id=uid, 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", source="enter_project",
) )
# A project need not have one, and most installs won't — null is ordinary # 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 design_system = None
if project.design_system_id: 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, 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 # Cache read ONLY — computing coverage moves a repo tarball and never
# belongs in this request path. Null is the ordinary state (no forge, or # belongs in this request path. Null is the ordinary state (no forge, or
@@ -194,38 +229,33 @@ async def enter_project(project_id: int) -> dict:
) )
out = { 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, "pattern_coverage": coverage_svc.coverage_line(coverage) if coverage else None,
# Trimmed to what tagging needs. The full charter is get_system's job — "systems": [{"id": s.id, "name": s.name} for s in systems],
# 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
],
"design_system": design_system, "design_system": design_system,
"milestone_summary": milestone_summary, "milestone_summary": milestone_summary,
**rulebooks_svc.rules_payload(applicable, user_id=uid, source="enter_project"), **rulebooks_svc.rules_payload(
applicable, user_id=uid, source="enter_project", brief=True,
),
"open_tasks": [ "open_tasks": [
{ {
"id": t.id, "title": t.title, "status": t.status, "id": t.id, "title": t.title, "status": t.status,
"priority": t.priority, "task_kind": t.task_kind,
"milestone_id": t.milestone_id, "milestone_id": t.milestone_id,
"milestone_title": milestone_titles.get(t.milestone_id),
} }
for t in open_tasks 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 # 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. # 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: if systems_bootstrap:
out["systems_bootstrap"] = systems_bootstrap out["systems_bootstrap"] = systems_bootstrap
if inception_ask: if inception_ask:
@@ -236,18 +266,18 @@ async def enter_project(project_id: int) -> dict:
async def get_project(project_id: int) -> dict: async def get_project(project_id: int) -> dict:
"""Fetch a Scribe project by ID. """Fetch a Scribe project by ID.
Returns full project fields, a milestone_summary list, and the Returns full project fields, a milestone_summary list (every milestone,
rulebook-applicable_rules / subscribed_rulebooks pair the assistant with description and progress but no plan body; get_milestone reads a
should consult when working on this project. plan), and the rulebook-applicable_rules / subscribed_rulebooks pair the
assistant should consult when working on this project.
""" """
uid = current_user_id() uid = current_user_id()
project = await projects_svc.get_project(uid, project_id) project = await projects_svc.get_project(uid, project_id)
if project is None: if project is None:
raise ValueError(f"project {project_id} not found") raise ValueError(f"project {project_id} not found")
data = project.to_dict() data = project.to_dict()
data["milestone_summary"] = await milestones_svc.get_project_milestone_summary( rows = await milestones_svc.get_project_milestone_summary(uid, project_id)
uid, project_id, data["milestone_summary"], _ = milestones_svc.brief_milestone_summary(rows)
)
applicable = await rulebooks_svc.get_applicable_rules( applicable = await rulebooks_svc.get_applicable_rules(
project_id=project_id, user_id=uid, project_id=project_id, user_id=uid,
) )
+47 -2
View File
@@ -213,9 +213,13 @@ async def get_project_milestone_summaries(
)).scalars().all()) )).scalars().all())
counts: dict[int, dict[str, int]] = {} counts: dict[int, dict[str, int]] = {}
step_touched: dict[int, datetime] = {}
if milestones: if milestones:
rows = await session.execute( 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( .where(
Note.milestone_id.in_([m.id for m in milestones]), Note.milestone_id.in_([m.id for m in milestones]),
Note.status.isnot(None), Note.status.isnot(None),
@@ -223,13 +227,21 @@ async def get_project_milestone_summaries(
) )
.group_by(Note.milestone_id, Note.status) .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 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} out: dict[int, list[dict]] = {pid: [] for pid in project_ids}
for m in milestones: for m in milestones:
entry = m.to_dict() entry = m.to_dict()
entry.update(_progress_from_counts(counts.get(m.id, {}))) 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) out.setdefault(m.project_id, []).append(entry)
return out return out
@@ -238,3 +250,36 @@ async def get_project_milestone_summary(user_id: int, project_id: int) -> list[d
"""Ordered milestones with progress — the one-project view of """Ordered milestones with progress — the one-project view of
get_project_milestone_summaries (two queries, not N+1).""" get_project_milestone_summaries (two queries, not N+1)."""
return (await get_project_milestone_summaries(user_id, [project_id])).get(project_id, []) return (await get_project_milestone_summaries(user_id, [project_id])).get(project_id, [])
# What a milestone LISTING needs: enough to say what each plan is and how far
# along it is. The plan itself (`body`) is get_milestone's job. Summaries once
# carried it, and on a project with 39 milestones enter_project came to ~222k
# characters, 168k of them plan bodies. That is past what an MCP client will
# accept as a tool result, so the session handshake arrived as a file to page
# through (#4045). user_id / project_id / timestamps repeat what the caller
# already knows.
_BRIEF_FIELDS = (
"id", "title", "description", "status", "order_index",
"total", "completed", "pct", "status_counts",
)
def brief_milestone_summary(
rows: list[dict], *, limit: int | None = None,
) -> tuple[list[dict], int]:
"""Trim summary rows to the listing fields, optionally keeping only the
most recently touched.
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).
"""
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 degraded but never empty. Superseded records are not demoted here: the
penalty reorders a top-k, and reordering a paginated, counted list would penalty reorders a top-k, and reordering a paginated, counted list would
make page boundaries lie. The search surfaces carry the demotion. 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 from scribe.services.access import notes_visibility_clause
@@ -442,6 +445,19 @@ async def list_notes(
if semantic_order is not None: if semantic_order is not None:
# A query is a relevance claim — see the docstring. # A query is a relevance claim — see the docstring.
query = query.order_by(semantic_order) 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: else:
sort_col = getattr(Note, sort, Note.updated_at) sort_col = getattr(Note, sort, Note.updated_at)
if order == "asc": if order == "asc":
+3 -2
View File
@@ -2236,8 +2236,9 @@ async def build_session_context(
f"Values: `resolve_design_system({design['id']})` · " f"Values: `resolve_design_system({design['id']})` · "
f"stylesheet: `get_design_system_stylesheet({design['id']})` " f"stylesheet: `get_design_system_stylesheet({design['id']})` "
f"· the prose (aesthetic, voice, where the accent may " f"· the prose (aesthetic, voice, where the accent may "
f"appear): `enter_project` returns it, or " f"appear), inherited house style included: "
f"`get_design_system({design['id']})`.", f"`get_design_system({design['id']})`"
f"`resolved_guidance`.",
] ]
elif unbound_repo: elif unbound_repo:
lines += [ 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. """The caller-facing shape of a get_applicable_rules() result.
Every surface that hands rules to an agent (enter_project, get_project, 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 (`plugin_context`) — computed a marker and showed nobody
anything, and counting those would put rules in the denominator that no anything, and counting those would put rules in the denominator that no
agent ever saw. 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( record_rule_surfaced(
user_id=user_id, user_id=user_id,
rule_ids=( rule_ids=(
+1 -1
View File
@@ -112,7 +112,7 @@ TOPICS: tuple[Topic, ...] = (
Topic("scribe is the system of record; keep one copy", U, ("one copy",), Topic("scribe is the system of record; keep one copy", U, ("one copy",),
"let any existing local memory shrink", index=("one copy",)), "let any existing local memory shrink", index=("one copy",)),
Topic("orient: enter the project, check repo bindings", U, ("enter_project", "list_repo_bindings"), Topic("orient: enter the project, check repo bindings", U, ("enter_project", "list_repo_bindings"),
"returns the project plus the rules bound to the areas it works in", "the milestones and open tasks worked on most recently",
index=("enter_project",)), index=("enter_project",)),
Topic("rules are retrieved; ask before a consequential act", U, ('content_type="rule"', "nothing matched"), Topic("rules are retrieved; ask before a consequential act", U, ('content_type="rule"', "nothing matched"),
"an empty session is not evidence of an empty rulebook", "an empty session is not evidence of an empty rulebook",
@@ -0,0 +1,81 @@
"""Real-Postgres tests for "recently touched" in the enter_project handshake (#4045).
What a mock cannot show: that a work-log moves its task up the list even
though logging never bumps the task's updated_at, and that a milestone whose
steps changed today reads as touched today even though its own row is old.
"""
from datetime import datetime, timedelta, timezone
import pytest
import pytest_asyncio
from scribe.models import async_session
from scribe.models.milestone import Milestone
from scribe.models.note import Note
from scribe.models.project import Project
from scribe.models.task_log import TaskLog
from scribe.services import milestones as milestones_svc
from scribe.services import notes as notes_svc
from tests.helpers import ensure_user
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")]
NOW = datetime.now(timezone.utc)
@pytest_asyncio.fixture
async def world():
"""Two open tasks and two milestones with deliberately staged timestamps."""
async with async_session() as s:
owner = await ensure_user(s, "handshake_owner")
project = Project(user_id=owner.id, title="Handshake target")
s.add(project)
await s.flush()
# Milestone "old plan": its own row is 30 days old, one step changed now.
old_plan = Milestone(user_id=owner.id, project_id=project.id, title="old plan",
updated_at=NOW - timedelta(days=30))
# Milestone "recent edit": its own row changed 2 days ago, no steps.
recent_edit = Milestone(user_id=owner.id, project_id=project.id, title="recent edit",
updated_at=NOW - timedelta(days=2))
s.add_all([old_plan, recent_edit])
await s.flush()
# "logged": last edited 10 days ago, but worked through a log just now.
logged = Note(user_id=owner.id, project_id=project.id, title="logged",
status="todo", task_kind="work",
updated_at=NOW - timedelta(days=10))
# "edited": last edited 1 day ago, no logs.
edited = Note(user_id=owner.id, project_id=project.id, title="edited",
status="todo", task_kind="work",
updated_at=NOW - timedelta(days=1))
step = Note(user_id=owner.id, project_id=project.id, milestone_id=old_plan.id,
title="step", status="done", task_kind="work", updated_at=NOW)
s.add_all([logged, edited, step])
await s.flush()
s.add(TaskLog(task_id=logged.id, user_id=owner.id, content="worked on it",
created_at=NOW))
ids = {"owner": owner.id, "pid": project.id, "logged": logged.id,
"edited": edited.id, "old_plan": old_plan.id, "recent_edit": recent_edit.id}
await s.commit()
return ids
async def test_a_work_log_counts_as_touching_its_task(world):
by_edit, _ = await notes_svc.list_notes(
world["owner"], is_task=True, project_id=world["pid"],
status=["todo", "in_progress"], sort="updated_at",
)
by_touch, _ = await notes_svc.list_notes(
world["owner"], is_task=True, project_id=world["pid"],
status=["todo", "in_progress"], sort="touched",
)
assert [t.id for t in by_edit] == [world["edited"], world["logged"]]
assert [t.id for t in by_touch] == [world["logged"], world["edited"]]
async def test_a_step_changing_counts_as_touching_its_milestone(world):
rows = await milestones_svc.get_project_milestone_summary(world["owner"], world["pid"])
brief, omitted = milestones_svc.brief_milestone_summary(rows, limit=1)
assert omitted == 1
assert [m["id"] for m in brief] == [world["old_plan"]]
+22
View File
@@ -197,3 +197,25 @@ async def test_update_token_leaves_supersedes_alone_when_omitted():
from scribe.mcp.tools.design_systems import update_design_token from scribe.mcp.tools.design_systems import update_design_token
await update_design_token(token_id=9, purpose="text on action surfaces") await update_design_token(token_id=9, purpose="text on action surfaces")
assert "supersedes" not in svc.update_token.await_args.kwargs assert "supersedes" not in svc.update_token.await_args.kwargs
# --- get: the guidance a UI session builds from -----------------------------
@pytest.mark.asyncio
async def test_get_design_system_carries_the_chain_merged_guidance():
"""enter_project carries only a summary and points here (#4045). The
system's own `guidance` field is only its departures; a session building
UI needs the house style too, so the merged form rides alongside."""
from scribe.mcp.tools.design_systems import get_design_system
merged = [{"design_system_id": 1, "title": "House", "guidance": "house style"},
{"design_system_id": 2, "title": "App", "guidance": "departures"}]
with patch("scribe.mcp.tools.design_systems.ds_svc.get_design_system",
AsyncMock(return_value=_fake_design_system())), \
patch("scribe.mcp.tools.design_systems.ds_svc.list_tokens",
AsyncMock(return_value=[])), \
patch("scribe.mcp.tools.design_systems.ds_svc.design_context",
AsyncMock(return_value={"guidance": merged})):
out = await get_design_system(2)
assert out["resolved_guidance"] == merged
+1 -1
View File
@@ -14,7 +14,7 @@ pytestmark = pytest.mark.usefixtures("_bind_user")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_list_milestones_returns_dict_with_progress(): async def test_list_milestones_returns_dict_with_progress():
rows = [{"id": 1, "title": "MS1", "status": "active", "task_count": 2}] rows = [{"id": 1, "title": "MS1", "status": "active", "total": 2}]
with patch( with patch(
"scribe.mcp.tools.milestones.milestones_svc.get_project_milestone_summary", "scribe.mcp.tools.milestones.milestones_svc.get_project_milestone_summary",
AsyncMock(return_value=rows), AsyncMock(return_value=rows),
+19 -14
View File
@@ -71,7 +71,7 @@ async def test_list_projects_wraps_in_dict():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_project_enriches_with_milestone_summary(): async def test_get_project_enriches_with_milestone_summary():
p = fake_project(id=5, title="found") p = fake_project(id=5, title="found")
milestone_summary = [{"id": 10, "title": "MS", "task_count": 3}] milestone_summary = [{"id": 10, "title": "MS", "status": "active", "total": 3}]
applicable_payload = { applicable_payload = {
"rules": [], "truncated": False, "subscribed_rulebooks": [], "rules": [], "truncated": False, "subscribed_rulebooks": [],
} }
@@ -165,8 +165,8 @@ async def test_update_project_raises_when_not_found():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_enter_project_composes_full_context(): async def test_enter_project_composes_full_context():
"""enter_project pulls project + rules + milestone summary + open tasks + """enter_project pulls project + rule titles + milestone summary + open
recent notes in one composed call.""" tasks in one composed call, each in its brief handshake form (#4045)."""
p = fake_project(id=5, title="P") p = fake_project(id=5, title="P")
applicable_payload = { applicable_payload = {
"rules": [{"id": 1, "title": "r1", "statement": "s", "rules": [{"id": 1, "title": "r1", "statement": "s",
@@ -175,7 +175,7 @@ async def test_enter_project_composes_full_context():
"truncated": False, "truncated": False,
"subscribed_rulebooks": [{"id": 2, "title": "rb"}], "subscribed_rulebooks": [{"id": 2, "title": "rb"}],
} }
milestone_summary = [{"id": 10, "title": "MS", "task_count": 3}] milestone_summary = [{"id": 10, "title": "MS", "status": "active", "total": 3}]
task1 = MagicMock() task1 = MagicMock()
task1.id = 100; task1.title = "T1"; task1.status = "in_progress" task1.id = 100; task1.title = "T1"; task1.status = "in_progress"
@@ -199,14 +199,19 @@ async def test_enter_project_composes_full_context():
): ):
out = await enter_project(project_id=5) out = await enter_project(project_id=5)
assert out["project"]["id"] == 5 assert out["project"] == {"id": 5, "title": "P", "status": "active", "goal": ""}
assert out["milestone_summary"] == milestone_summary assert out["milestone_summary"] == milestone_summary
assert out["applicable_rules"][0]["title"] == "r1" # Rules arrive in full by retrieval; the handshake lists the project's own
assert out["project_rules"][0]["id"] == 99 # by id and title and drops the subscription bookkeeping (#4045).
assert out["project_rules"] == [{"id": 99, "title": "pr1"}]
assert out["subscribed_rulebooks"] == [{"id": 2, "title": "rb"}] assert out["subscribed_rulebooks"] == [{"id": 2, "title": "rb"}]
assert out["open_tasks"][0]["id"] == 100 for gone in ("applicable_rules", "applicable_rules_truncated",
assert out["open_tasks"][0]["status"] == "in_progress" "suppressed_rules", "suppressed_topics", "recent_notes"):
assert out["recent_notes"][0]["id"] == 200 assert gone not in out, gone
assert out["open_tasks"] == [{
"id": 100, "title": "T1", "status": "in_progress",
"milestone_id": 10, "milestone_title": "MS",
}]
# No design system on this project -> the key is present and null, not # No design system on this project -> the key is present and null, not
# absent. A caller that has to distinguish "no key" from "no system" will # absent. A caller that has to distinguish "no key" from "no system" will
# eventually get it wrong. # eventually get it wrong.
@@ -252,10 +257,7 @@ async def test_enter_project_surfaces_the_systems_vocabulary():
): ):
out = await enter_project(project_id=5) out = await enter_project(project_id=5)
assert out["systems"] == [ assert out["systems"] == [{"id": 3, "name": "retrieval"}]
{"id": 3, "name": "retrieval",
"description": "Embeddings, ranking, auto-inject."}
]
def _enter_project_stubs(p): def _enter_project_stubs(p):
@@ -371,6 +373,9 @@ async def test_enter_project_hands_back_the_design_system_when_the_project_has_o
assert out["design_system"]["token_count"] == 95 assert out["design_system"]["token_count"] == 95
assert out["design_system"]["inherits_from"] == ["House"] assert out["design_system"]["inherits_from"] == ["House"]
# Summary only: the guidance is one call away (#4045).
assert "guidance" not in out["design_system"]
assert out["design_system"]["guidance_call"] == "get_design_system(9) → resolved_guidance"
assert ctx.await_args.args == (7, 9) # caller's id, the project's system assert ctx.await_args.args == (7, 9) # caller's id, the project's system
+198
View File
@@ -0,0 +1,198 @@
"""The enter_project handshake stays a small primer (#4045).
enter_project once carried every milestone's full plan body, the whole project
record, full rule text and ~9k of design guidance. On a project with 39
milestones that came to ~222k characters, past what an MCP client accepts as
a tool result, so the session handshake arrived as a file to page through.
"""
import contextlib
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from scribe.mcp.tools.milestones import list_milestones
from scribe.mcp.tools.projects import enter_project, get_project
from scribe.services.milestones import brief_milestone_summary
from tests.helpers import fake_project
pytestmark = pytest.mark.usefixtures("_bind_user")
PLAN = "A plan paragraph long enough to matter. " * 125 # ~5k chars
GOAL = "What the project is for. " * 50 # ~1.2k chars
def _milestone(mid: int, status: str, touched_day: int) -> dict:
"""A summary row as get_project_milestone_summary returns it."""
touched = f"2026-08-{touched_day:02d}T00:00:00+00:00"
return {
"id": mid, "user_id": 7, "project_id": 5, "title": f"M{mid}",
"description": f"what M{mid} is for", "body": PLAN, "status": status,
"order_index": mid, "created_at": "2026-01-01T00:00:00+00:00",
"updated_at": "2026-01-01T00:00:00+00:00", "last_touched_at": touched,
"total": 4, "completed": 2, "pct": 50.0,
"status_counts": {"todo": 2, "in_progress": 0, "done": 2, "cancelled": 0},
}
def _history(count: int) -> list[dict]:
"""`count` milestones, alternating done/active; higher id = touched later."""
return [
_milestone(i, "done" if i % 2 else "active", 1 + i % 28)
for i in range(count)
]
def test_brief_rows_leave_out_the_plan_and_what_the_caller_already_knows():
brief, omitted = brief_milestone_summary([_milestone(1, "active", 3)])
assert omitted == 0
assert brief == [{
"id": 1, "title": "M1", "description": "what M1 is for", "status": "active",
"order_index": 1, "total": 4, "completed": 2, "pct": 50.0,
"status_counts": {"todo": 2, "in_progress": 0, "done": 2, "cancelled": 0},
}]
def test_limit_keeps_the_most_recently_touched_whatever_their_status():
"""A plan can sit "active" for months; recency is what says it's current."""
rows = [_milestone(1, "active", 1), _milestone(2, "done", 9),
_milestone(3, "active", 5), _milestone(4, "done", 7)]
brief, omitted = brief_milestone_summary(rows, limit=2)
assert omitted == 2
assert [r["id"] for r in brief] == [2, 4] # most recent first
def test_no_limit_keeps_every_row_in_order():
brief, omitted = brief_milestone_summary(_history(8))
assert omitted == 0
assert [r["id"] for r in brief] == list(range(8))
def _task(tid: int, milestone_id: int | None) -> MagicMock:
t = MagicMock()
t.id = tid
t.title = f"T{tid}"
t.status = "todo"
t.milestone_id = milestone_id
return t
def _enter_stubs(project, milestones: list[dict], tasks: list, *, rules=None, systems=None,
design=None):
applicable = rules or {"rules": [], "project_rules": [], "truncated": False,
"subscribed_rulebooks": []}
return [
patch("scribe.mcp.tools.projects.projects_svc.get_project",
AsyncMock(return_value=project)),
patch("scribe.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
AsyncMock(return_value=applicable)),
patch("scribe.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
AsyncMock(return_value=milestones)),
patch("scribe.mcp.tools.projects.notes_svc.list_notes",
AsyncMock(return_value=(tasks, len(tasks)))),
patch("scribe.mcp.tools.projects.systems_svc.list_systems",
AsyncMock(return_value=systems or [])),
patch("scribe.mcp.tools.projects.systems_tools.bootstrap_systems_ask",
AsyncMock(return_value=None)),
patch("scribe.mcp.tools.projects.design_systems_svc.design_context",
AsyncMock(return_value=design)),
patch("scribe.mcp.tools.projects.coverage_svc.cached_coverage",
AsyncMock(return_value=None)),
patch("scribe.mcp.tools.projects.spawn"),
]
async def _enter(*stubs):
with contextlib.ExitStack() as stack:
mocks = [stack.enter_context(cm) for cm in stubs]
out = await enter_project(project_id=5)
return out, mocks
@pytest.mark.asyncio
async def test_enter_project_stays_small_however_large_the_project():
"""The ceiling is the point: a long-lived project's handshake must not
grow with its history. 200 milestones with 5k-character plans, 60 project
rules and 40 Systems would be well over 1M characters in the old shape."""
project = fake_project(id=5, design_system_id=9, goal=GOAL,
description="Background. " * 300)
rules = {
"rules": [{"id": i, "title": f"r{i}", "statement": PLAN} for i in range(50)],
"project_rules": [{"id": 100 + i, "title": f"pr{i}", "statement": PLAN,
"when_to_apply": PLAN} for i in range(60)],
"truncated": True, "subscribed_rulebooks": [{"id": 1, "title": "Family"}],
"suppressed_rules": [], "suppressed_topics": [],
}
systems = []
for i in range(40):
s = MagicMock()
s.id, s.name, s.description = i, f"Area {i}", PLAN
systems.append(s)
design = {"id": 9, "title": "Kit", "description": "", "inherits_from": ["House"],
"guidance": [{"design_system_id": 9, "title": "Kit", "guidance": PLAN * 2}],
"token_count": 111, "token_groups": ["accent", "surface"]}
tasks = [_task(1000 + i, i % 200) for i in range(10)]
out, _ = await _enter(*_enter_stubs(project, _history(200), tasks, rules=rules,
systems=systems, design=design))
assert len(out["milestone_summary"]) == 5
assert out["milestone_summary_omitted"].startswith("195 other milestone(s)")
assert len(out["open_tasks"]) == 10
assert "applicable_rules" not in out and "recent_notes" not in out
assert "guidance" not in out["design_system"]
# The fixed parts are bounded by their caps; what's left to grow is the
# goal, the rule and System titles, and the ask keys when they apply.
size = len(json.dumps(out, indent=2))
assert size < 16_000, size
@pytest.mark.asyncio
async def test_open_tasks_name_their_milestone_even_when_it_is_not_listed():
"""The milestone list is capped at 5; a task's milestone can fall outside
it, and its id must not arrive without a name."""
milestones = _history(20)
tasks = [_task(1, 0), _task(2, None)] # milestone 0 is the least recent
out, mocks = await _enter(*_enter_stubs(fake_project(id=5), milestones, tasks))
assert 0 not in [m["id"] for m in out["milestone_summary"]]
assert out["open_tasks"] == [
{"id": 1, "title": "T1", "status": "todo", "milestone_id": 0, "milestone_title": "M0"},
{"id": 2, "title": "T2", "status": "todo", "milestone_id": None, "milestone_title": None},
]
list_notes = mocks[3]
assert list_notes.await_args.kwargs["sort"] == "touched"
assert list_notes.await_args.kwargs["limit"] == 10
@pytest.mark.asyncio
async def test_omitted_key_is_absent_when_nothing_was_left_out():
"""Attached only when it applies (#2483)."""
out, _ = await _enter(*_enter_stubs(fake_project(id=5), _history(3), []))
assert len(out["milestone_summary"]) == 3
assert "milestone_summary_omitted" not in out
@pytest.mark.asyncio
async def test_get_project_lists_every_milestone_without_plans():
with patch("scribe.mcp.tools.projects.projects_svc.get_project",
AsyncMock(return_value=fake_project(id=5))), \
patch("scribe.mcp.tools.projects.milestones_svc.get_project_milestone_summary",
AsyncMock(return_value=_history(10))), \
patch("scribe.mcp.tools.projects.rulebooks_svc.get_applicable_rules",
AsyncMock(return_value={"rules": [], "truncated": False,
"subscribed_rulebooks": []})):
out = await get_project(project_id=5)
assert len(out["milestone_summary"]) == 10
assert all("body" not in m for m in out["milestone_summary"])
@pytest.mark.asyncio
async def test_list_milestones_lists_every_milestone_without_plans():
"""The call milestone_summary_omitted points to."""
with patch("scribe.mcp.tools.milestones.milestones_svc.get_project_milestone_summary",
AsyncMock(return_value=_history(30))):
out = await list_milestones(project_id=5)
assert len(out["milestones"]) == 30
assert all("body" not in m for m in out["milestones"])
+27
View File
@@ -341,6 +341,33 @@ def test_rules_payload_records_both_the_family_and_project_halves():
assert kw["source"] == "enter_project" assert kw["source"] == "enter_project"
def test_brief_rules_payload_lists_titles_and_records_only_what_it_shows():
"""The handshake's form (#4045): project rules by id and title, the
subscribed rulebooks, nothing else. A subscription-derived rule it doesn't
show must not count as surfaced."""
from scribe.services import rulebooks as svc
rec = MagicMock()
with patch.object(svc, "record_rule_surfaced", rec):
out = svc.rules_payload(
{
"rules": [{"id": 10, "title": "family", "statement": "s"}],
"project_rules": [{"id": 12, "title": "own", "statement": "s"}],
"truncated": False,
"subscribed_rulebooks": [{"id": 1, "title": "Family"}],
},
user_id=1,
source="enter_project",
brief=True,
)
assert out == {
"project_rules": [{"id": 12, "title": "own"}],
"subscribed_rulebooks": [{"id": 1, "title": "Family"}],
}
assert rec.call_args.kwargs["rule_ids"] == [12]
def test_every_rules_payload_caller_names_itself(): def test_every_rules_payload_caller_names_itself():
"""`source` is the CALLER's name, so the readout can still separate the """`source` is the CALLER's name, so the readout can still separate the
session handshake from a mid-session milestone read. A shared constant here session handshake from a mid-session milestone read. A shared constant here
+3
View File
@@ -176,6 +176,9 @@ async def test_build_session_context_pushes_the_projects_design_system():
# returns those rather than the resolved tokens. # returns those rather than the resolved tokens.
assert "resolve_design_system(9)" in ctx assert "resolve_design_system(9)" in ctx
assert "get_design_system_stylesheet(9)" in ctx assert "get_design_system_stylesheet(9)" in ctx
# The prose pointer names the call that has the MERGED guidance, not
# enter_project, which carries only the summary now (#4045).
assert "`get_design_system(9)` → `resolved_guidance`" in ctx
@pytest.mark.asyncio @pytest.mark.asyncio
+14 -3
View File
@@ -14,6 +14,7 @@ So these tests assert the number of sessions opened, not only the values
returned. A version that produced identical output while opening a session per returned. A version that produced identical output while opening a session per
project would pass a correctness test and reproduce the outage. project would pass a correctness test and reproduce the outage.
""" """
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
@@ -99,17 +100,27 @@ async def test_milestone_summaries_for_many_projects_open_ONE_session():
a session per MILESTONE, which is what turned 25 into ~250.""" a session per MILESTONE, which is what turned 25 into ~250."""
from scribe.services import milestones as svc from scribe.services import milestones as svc
m1 = MagicMock(id=10, project_id=1) written = datetime(2026, 9, 1, tzinfo=timezone.utc)
m1 = MagicMock(id=10, project_id=1, updated_at=written)
m1.to_dict = MagicMock(return_value={"id": 10, "title": "A"}) m1.to_dict = MagicMock(return_value={"id": 10, "title": "A"})
m2 = MagicMock(id=11, project_id=2) m2 = MagicMock(id=11, project_id=2, updated_at=written)
m2.to_dict = MagicMock(return_value={"id": 11, "title": "B"}) m2.to_dict = MagicMock(return_value={"id": 11, "title": "B"})
step_closed = datetime(2026, 9, 14, tzinfo=timezone.utc)
opened = [0] opened = [0]
rows = [[m1, m2], [(10, "done", 2), (10, "todo", 1), (11, "cancelled", 1)]] rows = [[m1, m2], [
(10, "done", 2, step_closed),
(10, "todo", 1, datetime(2026, 9, 2, tzinfo=timezone.utc)),
(11, "cancelled", 1, datetime(2026, 8, 1, tzinfo=timezone.utc)),
]]
with patch.object(svc, "async_session", _session_factory(opened, rows)): with patch.object(svc, "async_session", _session_factory(opened, rows)):
out = await svc.get_project_milestone_summaries(1, [1, 2]) out = await svc.get_project_milestone_summaries(1, [1, 2])
assert opened[0] == 1 assert opened[0] == 1
# Touched is the later of the milestone's own edit and its newest step
# update (#4045): steps closing today count, an old step doesn't pull it back.
assert out[1][0]["last_touched_at"] == step_closed.isoformat()
assert out[2][0]["last_touched_at"] == written.isoformat()
assert out[1][0]["completed"] == 2 and out[1][0]["total"] == 3 assert out[1][0]["completed"] == 2 and out[1][0]["total"] == 3
# Cancelled is excluded from the denominator, so a milestone whose only # Cancelled is excluded from the denominator, so a milestone whose only
# task was cancelled reads as complete rather than stalled at 0%. # task was cancelled reads as complete rather than stalled at 0%.