feat(moments): the moment catalog rules will mount on, readable in-session (milestone 458 step 1, #4919)
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 15s
CI & Build / TypeScript typecheck (push) Successful in 55s
CI & Build / integration (push) Successful in 58s
CI & Build / Python tests (push) Successful in 1m53s
CI & Build / Build & push image (push) Successful in 32s

Fourteen generic moments of work (session.start, work.start … reply.ask)
plus the skill.<name> family, each with what it means and the kinds of
action that reach it, written for any kind of work rather than software
alone. The catalog is code because every install needs the same mount
points; which actions reach a moment is per-install data (step 2).

require_moment refuses an unknown name with the catalog listed, so a
typo cannot become a mount that never fires. list_moments (read-only)
and GET /api/retrieval/moments hand out the same catalog.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-05 10:50:45 -04:00
co-authored by Claude Opus 5.5
parent 0720ab6dcf
commit 2ff7f2f34f
7 changed files with 333 additions and 1 deletions
+5
View File
@@ -154,6 +154,11 @@ _READ_ONLY_TOOLS = frozenset({
# most: it is the surface the "ask before acting" reflex calls, and a key # most: it is the surface the "ask before acting" reflex calls, and a key
# that could not reach it would be denied exactly the check it should run. # that could not reach it would be denied exactly the check it should run.
"what_might_apply", "what_might_apply",
# The moment catalog (milestone 458). A pure read of a constant; the
# mount and mapping writes that use it are separate tools. Spelled out for
# retrieval_telemetry's reason, and needed by a read key so that a line
# naming a moment can be understood by whoever was shown it.
"list_moments",
}) })
# Every tool that WRITES, by name. Nothing reads this set at runtime — a tool # Every tool that WRITES, by name. Nothing reads this set at runtime — a tool
+2 -1
View File
@@ -6,7 +6,7 @@ from `mcp.server.build_mcp_server`.
""" """
from scribe.mcp.tools import ( from scribe.mcp.tools import (
design_systems, lessons, milestones, notes, processes, projects, recent, repos, design_systems, lessons, milestones, notes, processes, projects, recent, repos,
retrieval_review, retrieval_tuning, moments, retrieval_review, retrieval_tuning,
wide_net, wide_net,
rulebooks, search, shapes, snippets, systems, tags, tasks, trash, rulebooks, search, shapes, snippets, systems, tags, tasks, trash,
) )
@@ -18,6 +18,7 @@ def register_all(mcp) -> None:
retrieval_tuning.register(mcp) retrieval_tuning.register(mcp)
retrieval_review.register(mcp) retrieval_review.register(mcp)
wide_net.register(mcp) wide_net.register(mcp)
moments.register(mcp)
notes.register(mcp) notes.register(mcp)
tasks.register(mcp) tasks.register(mcp)
projects.register(mcp) projects.register(mcp)
+36
View File
@@ -0,0 +1,36 @@
"""The moment catalog as an MCP tool (milestone 458 step 1).
A session needs the vocabulary in hand to mount a rule, to correct which
action reaches which moment, and to read a moment line it was shown — and it
needs it without leaving the session for a settings page. So the catalog is a
tool from the first step, ahead of the writes that will use it.
"""
from __future__ import annotations
from scribe.services import moments as moments_svc
async def list_moments() -> dict:
"""The moments of work that rules mount on — their names and what each means.
A rule mounted on a moment arrives whenever that moment happens, whatever
the words of the work look like. That is how a rule reaches you when it is
about WHEN something is done rather than WHAT it is about: a rule on
finishing work belongs at `work.finish` and `reply.report`, where nothing
said needs to resemble it.
Read this when you are about to mount a rule, when a line you were shown
names a moment and you want its meaning, or when you are correcting which
of your actions reaches which moment. Names are `<area>.<verb>`; a named
procedure (a skill or a stored process) is its own moment,
`skill.<name>`, listed under `families`.
Each moment says what is happening at it (`means`) and the kinds of action
that typically reach it (`reached_by`). The actions are examples: which of
YOUR actions reach a moment varies by install and is corrected in-session.
"""
return moments_svc.catalog()
def register(mcp) -> None:
mcp.tool(name="list_moments")(list_moments)
+13
View File
@@ -17,6 +17,7 @@ import logging
from quart import Blueprint, jsonify, request from quart import Blueprint, jsonify, request
from scribe.auth import get_current_user_id, login_required from scribe.auth import get_current_user_id, login_required
from scribe.services import moments as moments_svc
from scribe.services.retrieval_tuning import set_dial, current_settings, tuning_history from scribe.services.retrieval_tuning import set_dial, current_settings, tuning_history
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -104,3 +105,15 @@ async def tuning_history_route():
except ValueError as e: except ValueError as e:
return jsonify({"error": str(e)}), 400 return jsonify({"error": str(e)}), 400
return jsonify({"events": events, "total": len(events)}) return jsonify({"events": events, "total": len(events)})
@retrieval_bp.route("/moments", methods=["GET"])
@login_required
async def moments_route():
"""The moments of work that rules mount on (milestone 458).
The same catalog `list_moments` returns, from the same service: the
Settings view of mounts and mappings reads its vocabulary here, so the two
doors cannot name different moments.
"""
return jsonify(moments_svc.catalog())
+157
View File
@@ -0,0 +1,157 @@
"""The moments of work that rules mount on (milestone 458).
WHY THIS EXISTS
Semantic retrieval answers "what does this text resemble?", and most rules are
not ABOUT a topic — they belong to a MOMENT in the work. A rule about when work
counts as finished governs the moment work is sent out, checked, closed and
reported, and nothing said at those moments needs to resemble the rule. Matched
on topic, it scores below every bar; mounted on its moments, it cannot miss.
So a rule names the moments it belongs to, and when one happens every rule
mounted on it arrives — by lookup, with no embedding and no score. Semantic
retrieval stays the second net, for whatever nobody mounted.
WHY THE CATALOG IS CODE, NOT DATA
The moments are the product's vocabulary, not an install's. Every install needs
the same mount points, or a rule written on one could not mean the same thing on
another, and the plugin's hooks, the skills that declare moments and the
instruction surfaces all name them. What varies per install is which ACTIONS
reach a moment — one operator delivers with a push, another with a deploy
script, a third by sending a document — and that mapping is data (step 2). The
moment is generic; the action that reaches it is local.
HOW TO WRITE ONE
Domain-neutral (rule 115): Scribe's work may be software, configuration,
infrastructure or anything else a person drives through an agent, so a
definition names the moment in words every one of those recognises. The test
beside this module refuses software-only vocabulary in both fields. Concrete
actions belong in the default mappings, where they are examples of reaching a
moment rather than the meaning of it.
Names are expensive to change once rules point at them. Adding one is cheap;
renaming one is a migration of every mount.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
@dataclass(frozen=True)
class Moment:
"""One point in the work that rules can mount on."""
name: str
"""`<area>.<verb>` — the string a mount stores and a hook reports."""
means: str
"""What is happening at this moment, in one line, for any kind of work."""
reached_by: str
"""The kinds of action that typically reach it, said generically."""
def _m(name: str, means: str, reached_by: str) -> tuple[str, Moment]:
return name, Moment(name=name, means=means, reached_by=reached_by)
MOMENTS: dict[str, Moment] = dict([
_m("session.start",
"a working session begins, or resumes after its context was summarized",
"opening a session; resuming after the earlier conversation was summarized"),
_m("work.start",
"taking up a piece of work",
"marking a task as in progress; beginning a step of a plan"),
_m("work.plan",
"designing an approach before carrying it out",
"opening a plan; entering a planning mode; loading a planning procedure"),
_m("work.change",
"altering the thing being worked on",
"editing or creating a file; changing a setting or a document"),
_m("work.run",
"carrying out an action in the environment",
"running a command or a script"),
_m("work.verify",
"checking that the work does what it should",
"running a check; reading the result of an automated check; "
"loading a verification procedure"),
_m("work.deliver",
"sending work beyond the place it was made",
"publishing, merging, releasing, deploying or sending it to someone"),
_m("work.finish",
"declaring a piece of work done, or abandoning it",
"closing a task or an issue; marking it done or cancelled"),
_m("work.record",
"writing down what was learned or decided",
"creating a note, a lesson, a rule or a reusable example"),
_m("work.debug",
"working out why something does not behave as expected",
"reading a failure; loading a diagnosis procedure"),
_m("work.delegate",
"handing part of the work to another agent",
"dispatching a helper agent"),
_m("env.reach",
"touching a machine or a service outside the workspace",
"connecting to a host; calling a remote service; inspecting what runs there"),
_m("reply.report",
"reporting results to the person the work is for",
"the reply that ends a turn; loading a reporting procedure"),
_m("reply.ask",
"asking the person the work is for to decide or to act",
"a question in the reply; a structured question to the person"),
])
# A FAMILY rather than entries: one moment per named procedure, reached when it
# is loaded. The names are the procedures' own (bundled skills, stored
# processes), so they cannot be listed here — only their shape can.
SKILL_PREFIX = "skill."
SKILL_FAMILY = Moment(
name=f"{SKILL_PREFIX}<name>",
means="a named procedure is loaded",
reached_by="loading a skill or a stored process by its name",
)
_SKILL_NAME = re.compile(r"^[a-z0-9][a-z0-9_:-]*$")
def is_moment(name: str) -> bool:
"""Is this a name a rule can mount on — a catalog moment, or skill.<name>?"""
if name in MOMENTS:
return True
if name.startswith(SKILL_PREFIX):
return bool(_SKILL_NAME.match(name[len(SKILL_PREFIX):]))
return False
def require_moment(name: str) -> str:
"""The name, normalised, or a refusal that lists what is available.
A typo'd moment must not be mountable: the mount would be stored, read
back, and never fire — a rule that looks attached and is not, which is the
silent failure moments exist to end.
"""
clean = (name or "").strip().lower()
if is_moment(clean):
return clean
raise ValueError(
f"unknown moment {name!r}. Moments are: "
+ ", ".join(MOMENTS)
+ f", or {SKILL_FAMILY.name} for a named procedure"
)
def catalog() -> dict:
"""The catalog as plain data, for the MCP tool and the REST door alike."""
return {
"moments": [
{"name": m.name, "means": m.means, "reached_by": m.reached_by}
for m in MOMENTS.values()
],
"families": [
{"name": SKILL_FAMILY.name, "prefix": SKILL_PREFIX,
"means": SKILL_FAMILY.means, "reached_by": SKILL_FAMILY.reached_by},
],
"total": len(MOMENTS),
}
+119
View File
@@ -0,0 +1,119 @@
"""The moment catalog — the vocabulary rules mount on (milestone 458 step 1).
What this pins is what a mount depends on: every name is well-formed and
resolvable, an unknown name is refused rather than stored, the catalog speaks
for any kind of work rather than only software, and both doors — the MCP tool
and the REST read — hand out the same catalog.
"""
import re
import pytest
from scribe.services import moments
from tests.helpers import FakeMCP
_NAME = re.compile(r"^[a-z]+\.[a-z]+$")
def test_the_catalog_is_not_empty():
"""The sweeps below are vacuous over an empty catalog (rule 167)."""
assert len(moments.MOMENTS) >= 10
def test_every_moment_is_keyed_by_its_own_well_formed_name():
for key, m in moments.MOMENTS.items():
assert key == m.name, f"{key!r} is filed under a name it does not carry"
assert _NAME.match(key), f"{key!r} is not <area>.<verb>"
def test_every_moment_says_what_it_means_and_how_it_is_reached():
for m in moments.MOMENTS.values():
assert m.means.strip() and m.reached_by.strip(), m.name
assert "\n" not in m.means, f"{m.name}: `means` is one line"
def test_the_skill_family_is_not_a_catalog_entry():
"""A family is a shape, not a moment — no rule mounts on `skill.<name>`
literally, and a catalog key starting with the prefix would shadow it."""
assert not any(k.startswith(moments.SKILL_PREFIX) for k in moments.MOMENTS)
# Software-only vocabulary, the same guard the completion query carries. The
# moments are named for any work a person drives through an agent; the actions
# particular to one kind of work belong in the mappings, not in the meaning.
_DEV_ONLY = (r"\bCI\b", r"\bcommit", r"\bpull request", r"\bcode\b", r"\btest",
r"\bgit\b", r"\brepo(s|sitor\w*)?\b", r"\bbranch", r"\bcompil", r"\bbuild\b")
@pytest.mark.parametrize("field", ["means", "reached_by"])
def test_the_catalog_assumes_no_particular_domain(field):
found = {
m.name: hits
for m in [*moments.MOMENTS.values(), moments.SKILL_FAMILY]
if (hits := [w for w in _DEV_ONLY
if re.search(w, getattr(m, field), re.IGNORECASE)])
}
assert not found, f"software-only vocabulary in `{field}`: {found}"
def test_the_guard_can_fail():
assert re.search(_DEV_ONLY[1], "after the commit lands", re.IGNORECASE)
@pytest.mark.parametrize("name", list(moments.MOMENTS))
def test_every_catalog_moment_is_mountable(name):
assert moments.require_moment(name) == name
@pytest.mark.parametrize("raw,clean", [
(" Work.Finish ", "work.finish"),
("skill.verification", "skill.verification"),
("skill.scribe:writing-plans", "skill.scribe:writing-plans"),
("skill.scribe-proc-release_notes", "skill.scribe-proc-release_notes"),
])
def test_a_name_is_normalised_before_it_is_judged(raw, clean):
assert moments.require_moment(raw) == clean
@pytest.mark.parametrize("bad", [
"work.finished", "finish", "", "skill.", "skill.two words", "work.",
])
def test_an_unknown_moment_is_refused_with_the_catalog(bad):
"""A typo stored as a mount never fires — the silent failure moments end."""
with pytest.raises(ValueError) as exc:
moments.require_moment(bad)
assert "work.finish" in str(exc.value)
assert moments.SKILL_FAMILY.name in str(exc.value)
def test_the_catalog_payload_carries_every_moment_and_the_family():
data = moments.catalog()
assert [m["name"] for m in data["moments"]] == list(moments.MOMENTS)
assert data["total"] == len(moments.MOMENTS)
assert data["families"][0]["prefix"] == moments.SKILL_PREFIX
async def test_the_tool_returns_the_service_catalog():
from scribe.mcp.tools import moments as tool
assert await tool.list_moments() == moments.catalog()
def test_the_tool_is_registered_and_read_only():
from scribe.mcp.server import _READ_ONLY_TOOLS
from scribe.mcp.tools import moments as tool
mcp = FakeMCP()
tool.register(mcp)
assert mcp.names == ["list_moments"]
assert "list_moments" in _READ_ONLY_TOOLS
def test_both_doors_read_one_catalog():
"""Rule 33 parity: the tool and the route call the same service, so the
session and the Settings view cannot name different moments."""
from scribe.mcp.tools import moments as tool
from scribe.routes import retrieval as routes
assert tool.moments_svc is moments
assert routes.moments_svc is moments
+1
View File
@@ -42,6 +42,7 @@ def test_every_endpoint_is_reachable_on_the_app():
"/api/retrieval/surfaces", "/api/retrieval/surfaces",
"/api/retrieval/surfaces/<surface>", "/api/retrieval/surfaces/<surface>",
"/api/retrieval/tuning-history", "/api/retrieval/tuning-history",
"/api/retrieval/moments",
} }