diff --git a/alembic/versions/0116_moment_mappings.py b/alembic/versions/0116_moment_mappings.py
new file mode 100644
index 00000000..438b08d1
--- /dev/null
+++ b/alembic/versions/0116_moment_mappings.py
@@ -0,0 +1,47 @@
+"""moment_mappings — an install's own corrections to which actions reach which
+moment (milestone 458 step 2)
+
+Revision ID: 0116
+Revises: 0115
+Create Date: 2026-10-05
+
+The moments and their default actions are code; this table holds what an
+install adds ("make ship" is a deliver here) and the defaults it switches off.
+User-owned configuration, so it cascades with the user. No CHECK on `effect`,
+which the service validates (rule 36).
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0116"
+down_revision = "0115"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.create_table(
+ "moment_mappings",
+ sa.Column("id", sa.BigInteger(), primary_key=True),
+ sa.Column(
+ "created_at", sa.DateTime(timezone=True), nullable=False,
+ server_default=sa.text("now()"),
+ ),
+ sa.Column(
+ "user_id", sa.Integer(),
+ sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False,
+ ),
+ sa.Column("tool", sa.Text(), nullable=False),
+ sa.Column("match", sa.Text(), nullable=False, server_default=""),
+ sa.Column("moment", sa.Text(), nullable=False),
+ sa.Column("effect", sa.Text(), nullable=False, server_default="add"),
+ sa.Column("reason", sa.Text(), nullable=False, server_default=""),
+ sa.Column("actor", sa.Text(), nullable=False, server_default="model"),
+ sa.UniqueConstraint(
+ "user_id", "tool", "match", "moment", name="uq_moment_mapping",
+ ),
+ )
+
+
+def downgrade() -> None:
+ op.drop_table("moment_mappings")
diff --git a/src/scribe/mcp/server.py b/src/scribe/mcp/server.py
index a397e689..2585d8db 100644
--- a/src/scribe/mcp/server.py
+++ b/src/scribe/mcp/server.py
@@ -207,6 +207,9 @@ _WRITE_TOOLS = frozenset({
# reads, and it appends the reason to the audit trail (#4102).
"tune_retrieval",
"migrate_retrieval_floor",
+ # Which actions reach which moment on this install (milestone 458). Each
+ # changes what fires for every later session, so a read key is refused.
+ "map_action", "unmap_action",
# A reviewer's verdicts on logged menu lines (#4772) — rows carrying free
# prose the agent authored, `rule_outcome`'s reason for being a write.
"judge_menu",
diff --git a/src/scribe/mcp/tools/moments.py b/src/scribe/mcp/tools/moments.py
index 212342a5..e92df37e 100644
--- a/src/scribe/mcp/tools/moments.py
+++ b/src/scribe/mcp/tools/moments.py
@@ -1,17 +1,24 @@
-"""The moment catalog as an MCP tool (milestone 458 step 1).
+"""Moments as MCP tools: the catalog, and the in-session corrections to it (milestone 458).
-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.
+A session needs the vocabulary in hand to mount a rule, to read a line that
+names a moment, and to fix a misfire — and the operator's ruling is that the
+fix happens in the session, not on a settings page:
+
+ "making the user leave the session to fix a misfire is not desirable and
+ the llm session should be able to offer corrections"
+
+So the mapping writes are tools, built to be offered mid-work and made on the
+operator's yes.
"""
from __future__ import annotations
+from scribe.mcp._context import current_user_id
+from scribe.services import moment_actions as actions_svc
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.
+ """The moments of work that rules mount on, and which actions reach each one here.
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
@@ -20,17 +27,85 @@ async def list_moments() -> dict:
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 `.`; a named
- procedure (a skill or a stored process) is its own moment,
- `skill.`, listed under `families`.
+ names a moment and you want its meaning, or when an action reached the
+ wrong moment — or none — and you are about to correct it with
+ `map_action` / `unmap_action`.
- 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.
+ Names are `.`; a named procedure (a skill or a stored process)
+ is its own moment, `skill.`, listed under `families`. Each moment
+ says what is happening at it (`means`) and the kinds of action that
+ typically reach it (`reached_by`).
+
+ `actions` maps each moment to the actions that reach it on this install:
+ `via: "default"` ships with the product, `via: "install"` is this
+ install's own mapping. `removed_defaults` are shipped defaults this install
+ switched off.
"""
- return moments_svc.catalog()
+ out = moments_svc.catalog()
+ out.update(await actions_svc.actions_by_moment(current_user_id()))
+ return out
+
+
+async def map_action(tool: str, moment: str, match: str = "", reason: str = "") -> dict:
+ """Make an action reach a moment on this install — the in-session fix for a missed moment.
+
+ Use it when an action plainly happened at a moment and the moment did not
+ fire: the operator ships with `make ship`, and nothing mounted on
+ `work.deliver` arrived. Offer the mapping when you notice, in one line
+ ("that `make ship` was a deliver and nothing fired — map it?"), and make it
+ on their yes. The correction lasts: every later session on this install
+ gets it.
+
+ Args:
+ tool: the tool as the harness names it — `Bash`, `Edit`,
+ `update_task`. An MCP server prefix is ignored.
+ moment: a moment from `list_moments`, or `skill.`.
+ match: which calls of the tool. Empty = every call. For a tool that
+ runs a command, how the command starts (`make ship`, `./deploy.sh`)
+ — it is checked against each part of a compound command line. For
+ any other tool, its arguments as `field=value` pairs, comma-separated
+ (`status=done`).
+ reason: what misfired, or what this action is for here. Optional; it
+ is shown beside the mapping when someone reviews it later.
+
+ Returns the change made and `now_reaches`: every moment that action
+ reaches after the change, so you can confirm it in the same reply.
+ Mapping a shipped default this install had switched off switches it back
+ on.
+ """
+ return await actions_svc.map_action(
+ current_user_id(), tool, match, moment, reason=reason, actor="model",
+ )
+
+
+async def unmap_action(tool: str, moment: str, match: str = "", reason: str = "") -> dict:
+ """Stop an action reaching a moment on this install — the fix for a moment that fires wrongly.
+
+ Use it when a moment fires on an action that is not that moment here — a
+ `curl` to a local test server reaching `env.reach`, say — and the rules it
+ brings are noise every time. Offer it when you notice, and make it on the
+ operator's yes.
+
+ The arguments name the mapping exactly as `list_moments` shows it. This
+ install's own mapping is removed; a shipped default is switched off for
+ this install only, and stays off across upgrades. `map_action` with the
+ same arguments switches a default back on.
+
+ Args:
+ tool: the tool as `list_moments` names it.
+ moment: the moment it should stop reaching.
+ match: the mapping's match, as listed (empty for a whole-tool mapping).
+ reason: why it misfires here — worth giving for a default, since it is
+ the record of why this install differs from the product.
+
+ Returns the change made and `now_reaches` for that action.
+ """
+ return await actions_svc.unmap_action(
+ current_user_id(), tool, match, moment, reason=reason, actor="model",
+ )
def register(mcp) -> None:
mcp.tool(name="list_moments")(list_moments)
+ mcp.tool(name="map_action")(map_action)
+ mcp.tool(name="unmap_action")(unmap_action)
diff --git a/src/scribe/models/__init__.py b/src/scribe/models/__init__.py
index 81d0d024..3b30da1d 100644
--- a/src/scribe/models/__init__.py
+++ b/src/scribe/models/__init__.py
@@ -62,6 +62,7 @@ from scribe.models.note_usage import NoteUsageEvent # noqa: E402, F401
from scribe.models.rule_usage import RuleUsageEvent # noqa: E402, F401
from scribe.models.system_usage import SystemUsageEvent # noqa: E402, F401
from scribe.models.retrieval_judgment import RetrievalJudgment # noqa: E402, F401
+from scribe.models.moment_mapping import MomentMapping # noqa: E402, F401
from scribe.models.project import Project # noqa: E402, F401
from scribe.models.milestone import Milestone # noqa: E402, F401
from scribe.models.task_log import TaskLog # noqa: E402, F401
diff --git a/src/scribe/models/moment_mapping.py b/src/scribe/models/moment_mapping.py
new file mode 100644
index 00000000..2e1e5abf
--- /dev/null
+++ b/src/scribe/models/moment_mapping.py
@@ -0,0 +1,75 @@
+from sqlalchemy import BigInteger, ForeignKey, Integer, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import CreatedAtMixin, iso
+
+
+class MomentMapping(Base, CreatedAtMixin):
+ """One install's correction to which actions reach which moment (milestone 458).
+
+ The moments are the product's vocabulary and the shipped defaults say how
+ the common actions reach them (`services/moment_actions.py`). What no
+ default can know is how THIS operator works: one delivers with a push,
+ another with `make ship`, a third by publishing a document. A row here is
+ that local knowledge, written in-session the moment a misfire is noticed.
+
+ `effect` is "add" (this action reaches this moment) or "remove" (a shipped
+ default that misfires here is switched off). Removal is a row rather than
+ an edit to the defaults because the defaults are code: an install can only
+ say "not here", and saying so must survive an upgrade that ships the same
+ default again.
+
+ Text rather than a CHECK on `effect`, for retrieval_tuning_events' reason
+ (rule 36): the service validates it, and a constraint would buy nothing but
+ a migration the day a third effect is wanted.
+
+ ONE ROW PER (user, tool, match, moment), so mapping the same thing twice
+ is an update of the reason rather than a duplicate, and an add and a remove
+ of the same mapping cannot both stand.
+
+ CASCADES with the user, unlike the telemetry tables: this is the user's own
+ configuration, not a history that should outlive them.
+ """
+
+ __tablename__ = "moment_mappings"
+
+ id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+ user_id: Mapped[int] = mapped_column(
+ Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False,
+ )
+ # The tool as the harness names it, with any MCP server prefix stripped
+ # (`mcp__plugin_x__update_task` → `update_task`), so a mapping does not
+ # depend on what an install called its server.
+ tool: Mapped[str] = mapped_column(Text, nullable=False)
+ # "" = every call of the tool. For a tool that runs a command, how the
+ # command starts ("make ship"); for any other tool, `field=value` pairs.
+ match: Mapped[str] = mapped_column(Text, nullable=False, default="")
+ moment: Mapped[str] = mapped_column(Text, nullable=False)
+ effect: Mapped[str] = mapped_column(Text, nullable=False, default="add")
+ # Why — what misfired, or what this action is for here. Optional at the
+ # boundary: the correction is usually self-explanatory, and a required
+ # reason would be friction on the exact in-session fix this table exists
+ # to make cheap.
+ reason: Mapped[str] = mapped_column(Text, nullable=False, default="")
+ # "model" | "human", for retrieval_tuning_events' reason: both act as the
+ # same user, and "did I do this or did the session?" is the first question.
+ actor: Mapped[str] = mapped_column(Text, nullable=False, default="model")
+
+ __table_args__ = (
+ UniqueConstraint(
+ "user_id", "tool", "match", "moment", name="uq_moment_mapping",
+ ),
+ )
+
+ def to_dict(self) -> dict:
+ return {
+ "id": self.id,
+ "tool": self.tool,
+ "match": self.match,
+ "moment": self.moment,
+ "effect": self.effect,
+ "reason": self.reason,
+ "actor": self.actor,
+ "created_at": iso(self.created_at),
+ }
diff --git a/src/scribe/routes/retrieval.py b/src/scribe/routes/retrieval.py
index 1f76417d..88ffe0c7 100644
--- a/src/scribe/routes/retrieval.py
+++ b/src/scribe/routes/retrieval.py
@@ -17,6 +17,7 @@ import logging
from quart import Blueprint, jsonify, request
from scribe.auth import get_current_user_id, login_required
+from scribe.services import moment_actions as moment_actions_svc
from scribe.services import moments as moments_svc
from scribe.services.retrieval_tuning import set_dial, current_settings, tuning_history
@@ -110,10 +111,43 @@ async def tuning_history_route():
@retrieval_bp.route("/moments", methods=["GET"])
@login_required
async def moments_route():
- """The moments of work that rules mount on (milestone 458).
+ """The moments of work that rules mount on (milestone 458), with the
+ actions that reach each one on this install.
- 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.
+ The same payload `list_moments` returns, from the same services, so the
+ session and the Settings view cannot name different moments.
"""
- return jsonify(moments_svc.catalog())
+ out = moments_svc.catalog()
+ out.update(await moment_actions_svc.actions_by_moment(get_current_user_id()))
+ return jsonify(out)
+
+
+async def _mapping_change(change):
+ """map/unmap from the browser: the MCP tools' service, recorded as human."""
+ data = await request.get_json()
+ if not isinstance(data, dict):
+ return jsonify({"error": "Expected a JSON object"}), 400
+ if not data.get("tool") or not data.get("moment"):
+ return jsonify({"error": "tool and moment are required"}), 400
+ try:
+ result = await change(
+ get_current_user_id(), str(data["tool"]), str(data.get("match") or ""),
+ str(data["moment"]), reason=str(data.get("reason") or ""), actor="human",
+ )
+ except ValueError as e:
+ # Unknown moment, a match the tool cannot take, nothing to remove —
+ # the service says what would work, so pass it through.
+ return jsonify({"error": str(e)}), 400
+ return jsonify(result)
+
+
+@retrieval_bp.route("/moments/mappings", methods=["POST"])
+@login_required
+async def map_action_route():
+ return await _mapping_change(moment_actions_svc.map_action)
+
+
+@retrieval_bp.route("/moments/mappings", methods=["DELETE"])
+@login_required
+async def unmap_action_route():
+ return await _mapping_change(moment_actions_svc.unmap_action)
diff --git a/src/scribe/services/backup.py b/src/scribe/services/backup.py
index 2fd3282e..7bdd7e08 100644
--- a/src/scribe/services/backup.py
+++ b/src/scribe/services/backup.py
@@ -14,6 +14,7 @@ from scribe.models.design_system import DesignSystem, DesignToken
from scribe.models.note_usage import NoteUsageEvent
from scribe.models.rule_usage import RuleUsageEvent
from scribe.models.system_usage import SystemUsageEvent
+from scribe.models.moment_mapping import MomentMapping
from scribe.models.retrieval_tuning import RetrievalTuningEvent
from scribe.models.canonical_system import CanonicalSystem
from scribe.models.rulebook import RuleRelation, rule_systems as rule_systems_t
@@ -93,8 +94,12 @@ logger = logging.getLogger(__name__)
# 444): the files that are each area, and whether an area's rulings were read
# once shown. The usage rows restore through the SYSTEM map, for the reason
# the rule twin restores through the rule map.
+# v21 (2026-10) added moment_mappings (milestone 458): which of this install's
+# actions reach which moment, and the shipped defaults it switched off. Each
+# row is a correction an operator made in-session; a restore that dropped them
+# would silently put every misfire back.
# Bump when the serialized schema changes.
-BACKUP_VERSION = 20
+BACKUP_VERSION = 21
# Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
# below, these two lists must together account for the entire schema — which is
@@ -139,6 +144,8 @@ _BACKED_UP = [
# v20 (2026-10): System usage telemetry (milestone 444), for the reason
# its note and rule twins travel.
"system_usage_events",
+ # v21 (2026-10): an install's action → moment corrections (milestone 458).
+ "moment_mappings",
]
# Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -247,6 +254,8 @@ _COLUMN_EXCLUSIONS: dict[str, set[str]] = {
# Same again — and everything else travels, because each remaining column
# is part of the argument: what moved, from what, to what, by whom, why.
"retrieval_tuning_events": {"id"},
+ # Every other column is the correction itself, or who made it and why.
+ "moment_mappings": {"id"},
"design_systems": {"deleted_at", "deleted_batch_id", "created_at", "updated_at"},
"design_tokens": {"deleted_at", "deleted_batch_id", "created_at", "updated_at"},
"repo_bindings": {"id", "created_at", "updated_at"},
@@ -334,6 +343,7 @@ _IMPORT_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
"rule_usage_events": {"id"},
"system_usage_events": {"id"},
"retrieval_tuning_events": {"id"},
+ "moment_mappings": {"id"},
"design_systems": {
"id", "deleted_at", "deleted_batch_id", "created_at", "updated_at",
},
@@ -486,6 +496,21 @@ def _rule_usage_event_rows(rows) -> list[dict]:
]
+def _moment_mapping_rows(rows) -> list[dict]:
+ """An install's action → moment corrections (milestone 458). Not
+ `to_dict()`, for _retrieval_tuning_event_rows' reason: the restore needs
+ `user_id` to remap, and the MCP reader omits it."""
+ return [
+ {
+ "user_id": r.user_id, "tool": r.tool, "match": r.match,
+ "moment": r.moment, "effect": r.effect, "reason": r.reason,
+ "actor": r.actor,
+ "created_at": r.created_at.isoformat() if r.created_at else None,
+ }
+ for r in rows
+ ]
+
+
def _retrieval_tuning_event_rows(rows) -> list[dict]:
"""The record of why a retrieval dial is where it is (#4102).
@@ -836,6 +861,9 @@ async def export_full_backup() -> dict:
retrieval_tuning_events = (await session.execute(
select(RetrievalTuningEvent).order_by(RetrievalTuningEvent.id)
)).scalars().all()
+ moment_mappings = (await session.execute(
+ select(MomentMapping).order_by(MomentMapping.id)
+ )).scalars().all()
repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
code_shapes = (await session.execute(select(CodeShape))).scalars().all()
code_shape_events = (await session.execute(
@@ -886,6 +914,7 @@ async def export_full_backup() -> dict:
"retrieval_tuning_events": _retrieval_tuning_event_rows(
retrieval_tuning_events
),
+ "moment_mappings": _moment_mapping_rows(moment_mappings),
"repo_bindings": _repo_binding_rows(repo_bindings),
"note_supersessions": _note_supersession_rows(supersessions),
"code_shapes": _code_shape_rows(code_shapes),
@@ -1035,6 +1064,13 @@ async def export_user_backup(user_id: int) -> dict:
.where(RetrievalTuningEvent.user_id == user_id)
.order_by(RetrievalTuningEvent.id)
)).scalars().all()
+ # The user's own configuration: user_id is the owner, nothing to
+ # route around.
+ moment_mappings = (await session.execute(
+ select(MomentMapping)
+ .where(MomentMapping.user_id == user_id)
+ .order_by(MomentMapping.id)
+ )).scalars().all()
rule_relations = (await session.execute(
select(RuleRelation).where(
RuleRelation.from_rule_id.in_(_rule_ids),
@@ -1093,6 +1129,7 @@ async def export_user_backup(user_id: int) -> dict:
"retrieval_tuning_events": _retrieval_tuning_event_rows(
retrieval_tuning_events
),
+ "moment_mappings": _moment_mapping_rows(moment_mappings),
"repo_bindings": _repo_binding_rows(repo_bindings),
"note_supersessions": _note_supersession_rows(supersessions),
"code_shapes": _code_shape_rows(code_shapes),
@@ -1301,6 +1338,23 @@ def _build_setting(row: dict, maps: _Maps) -> Setting | None:
return Setting(user_id=uid, key=row["key"], value=row.get("value", ""))
+def _build_moment_mapping(row: dict, maps: _Maps) -> MomentMapping | None:
+ """No remapping beyond the user: `tool` and `moment` are names, not keys."""
+ uid = maps.users.get(row.get("user_id") or 0)
+ if uid is None or not row.get("tool") or not row.get("moment"):
+ return None
+ return MomentMapping(
+ user_id=uid,
+ tool=row["tool"],
+ match=row.get("match") or "",
+ moment=row["moment"],
+ effect=row.get("effect") or "add",
+ reason=row.get("reason") or "",
+ actor=row.get("actor") or "model",
+ created_at=_dt(row.get("created_at")),
+ )
+
+
def _build_retrieval_tuning_event(row: dict, maps: _Maps) -> RetrievalTuningEvent | None:
"""No id remapping beyond the user: `surface` is a registry NAME, not a
foreign key, which is what lets this history survive a restore into an
@@ -1874,7 +1928,7 @@ async def _restore_v2(data: dict) -> dict:
"code_shape_uses": 0, "canonical_systems": 0,
"rule_systems": 0, "rule_relations": 0, "rule_versions": 0,
"retrieval_tuning_events": 0, "lesson_rule_links": 0,
- "lesson_no_rule": 0,
+ "lesson_no_rule": 0, "moment_mappings": 0,
}
async with async_session() as session:
@@ -1984,6 +2038,15 @@ async def _restore_v2(data: dict) -> dict:
session.add(event)
stats["retrieval_tuning_events"] += 1
+ # 8c. Moment mappings (v21) — the install's corrections to which
+ # actions reach which moment. Names only, so only the user remaps.
+ for mm_data in data.get("moment_mappings", []):
+ mapping = _build_moment_mapping(mm_data, maps)
+ if mapping is None:
+ continue
+ session.add(mapping)
+ stats["moment_mappings"] += 1
+
# 9. Rulebooks (v3)
for rb_data in data.get("rulebooks", []):
rb = _build_rulebook(rb_data, maps)
diff --git a/src/scribe/services/moment_actions.py b/src/scribe/services/moment_actions.py
new file mode 100644
index 00000000..6130866f
--- /dev/null
+++ b/src/scribe/services/moment_actions.py
@@ -0,0 +1,393 @@
+"""Which actions reach which moment: shipped defaults plus each install's own (milestone 458 step 2).
+
+The catalog (`services/moments.py`) says what the moments ARE. This module says
+how the work gets there. One call can reach several moments: `kubectl apply`
+is a run, a deliver and a reach outside the workspace all at once, and a rule
+mounted on any of them should arrive.
+
+AN ACTION is a tool plus an optional `match`:
+
+- `match` empty — every call of that tool (`Edit` → work.change).
+- A tool that runs a command (its input carries `command`) — how the command
+ starts, tested against each segment of a compound line, so
+ `cd app && make ship` reaches what `make ship` reaches. Leading `VAR=value`
+ assignments are skipped; a word boundary is required, so `git push` does
+ not match `git pushd`.
+- Any other tool — `field=value` pairs, comma-separated, all of which must
+ hold (`update_task` with `status=done`).
+
+Tool names are compared without any MCP server prefix and without case:
+`mcp__plugin_x__update_task` and `update_task` are one tool, because what an
+install called its server is not something a mapping should depend on.
+
+WHY THE DEFAULTS ARE CODE AND THE CORRECTIONS ARE ROWS
+
+The defaults cover the actions every install shares — the harness's own tools,
+Scribe's own tools, the commonest command shapes. They ship, and improve, with
+the product. What no default can know is how one operator works, so an install
+ADDS mappings and REMOVES defaults that misfire for it, in-session, through
+`map_action` / `unmap_action`. A removal is stored rather than applied to the
+defaults so it survives an upgrade that ships the same default again.
+
+The concrete commands below are examples of reaching a moment, which is why
+they may name particular tools when the moments themselves may not.
+"""
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass
+from typing import Iterable
+
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.moment_mapping import MomentMapping
+from scribe.services import moments as catalog
+
+logger = logging.getLogger(__name__)
+
+ADD, REMOVE = "add", "remove"
+EFFECTS = (ADD, REMOVE)
+DEFAULT, INSTALL = "default", "install"
+
+# The input field that makes a tool a command-runner, and the tools known to
+# be one. The field drives MATCHING (any tool whose call carries a command is
+# matched by prefix); the names drive VALIDATION, so a prefix-style match on a
+# tool that takes no command is refused rather than stored to never fire.
+COMMAND_FIELD = "command"
+COMMAND_TOOLS = frozenset({"bash"})
+
+# The harness's skill loader: loading a procedure reaches `skill.`,
+# derived from the call rather than listed, since the names are the
+# procedures' own.
+SKILL_TOOL, SKILL_FIELD = "skill", "skill"
+
+_SEGMENT_SPLIT = re.compile(r"&&|\|\||[;|\n]")
+_ENV_ASSIGN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=\S*\s+")
+
+
+@dataclass(frozen=True)
+class Action:
+ tool: str
+ match: str
+ moment: str
+
+
+def _defaults() -> tuple[Action, ...]:
+ out: list[Action] = []
+
+ def on(moment: str, tool: str, *matches: str) -> None:
+ for m in matches or ("",):
+ out.append(Action(tool, m, moment))
+
+ # The harness's own tools.
+ on("work.change", "Edit")
+ on("work.change", "Write")
+ on("work.change", "MultiEdit")
+ on("work.change", "NotebookEdit")
+ on("work.run", "Bash")
+ on("work.delegate", "Task")
+ on("work.delegate", "Agent")
+ on("work.plan", "EnterPlanMode")
+ on("work.plan", "ExitPlanMode")
+ on("reply.ask", "AskUserQuestion")
+
+ # Scribe's own tools — every install has these, so their moments ship.
+ on("work.start", "update_task", "status=in_progress")
+ on("work.finish", "update_task", "status=done", "status=cancelled")
+ on("work.finish", "update_milestone", "status=done")
+ on("work.plan", "start_planning")
+ on("work.plan", "create_milestone")
+ for tool in ("create_note", "create_lesson", "create_rule",
+ "create_project_rule", "create_preference", "create_snippet",
+ "create_process"):
+ on("work.record", tool)
+
+ # The commonest command shapes. An install's own (`make ship`,
+ # `./deploy.sh`) are what map_action is for.
+ on("work.deliver", "Bash",
+ "git push", "git merge", "gh pr merge", "gh release create",
+ "docker push", "kubectl apply", "helm install", "helm upgrade",
+ "terraform apply", "npm publish", "twine upload", "cargo publish")
+ on("work.verify", "Bash",
+ "pytest", "python -m pytest", "npm test", "npm run test", "yarn test",
+ "pnpm test", "go test", "go vet", "cargo test", "make test",
+ "make check", "ruff check", "mypy", "terraform plan",
+ "terraform validate")
+ on("env.reach", "Bash",
+ "ssh", "scp", "curl", "wget", "kubectl", "helm")
+ return tuple(out)
+
+
+DEFAULT_ACTIONS: tuple[Action, ...] = _defaults()
+
+
+# ── matching ──────────────────────────────────────────────────────────────
+
+def tool_key(name: str) -> str:
+ """The tool's bare, lowercased name: no MCP server prefix."""
+ clean = (name or "").strip()
+ if clean.startswith("mcp__"):
+ clean = clean.rsplit("__", 1)[-1]
+ return clean.lower()
+
+
+def _squash(text: str) -> str:
+ return " ".join((text or "").split())
+
+
+def _segments(command: str) -> list[str]:
+ out = []
+ for part in _SEGMENT_SPLIT.split(command or ""):
+ seg = part.strip().lstrip("(").strip()
+ while _ENV_ASSIGN.match(seg):
+ seg = _ENV_ASSIGN.sub("", seg, count=1)
+ seg = _squash(seg)
+ if seg:
+ out.append(seg)
+ return out
+
+
+def _pairs(match: str) -> dict[str, str] | None:
+ """`a=b, c=d` → {"a": "b", "c": "d"}; None when it is not that shape."""
+ out: dict[str, str] = {}
+ for part in (match or "").split(","):
+ key, sep, value = part.partition("=")
+ if not sep or not key.strip():
+ return None
+ out[key.strip().lower()] = value.strip().lower()
+ return out or None
+
+
+def action_matches(action: Action, tool: str, tool_input: dict | None) -> bool:
+ if tool_key(action.tool) != tool_key(tool):
+ return False
+ if not action.match:
+ return True
+ tool_input = tool_input or {}
+ command = tool_input.get(COMMAND_FIELD)
+ if isinstance(command, str):
+ want = _squash(action.match)
+ return any(seg == want or seg.startswith(want + " ")
+ for seg in _segments(command))
+ pairs = _pairs(action.match)
+ if pairs is None:
+ return False
+ given = {str(k).lower(): v for k, v in tool_input.items()}
+ return all(
+ str(given.get(k, "")).strip().lower() == v for k, v in pairs.items()
+ )
+
+
+def effective_actions(mappings: Iterable) -> list[tuple[Action, str]]:
+ """The defaults this install has not removed, then its own additions.
+
+ `mappings` are MomentMapping rows, or anything with the same four fields.
+ """
+ removed: set[tuple[str, str, str]] = set()
+ added: list[Action] = []
+ for m in mappings:
+ key = (tool_key(m.tool), _squash(m.match), m.moment)
+ if m.effect == REMOVE:
+ removed.add(key)
+ elif m.effect == ADD:
+ added.append(Action(m.tool, _squash(m.match), m.moment))
+ out = [
+ (a, DEFAULT) for a in DEFAULT_ACTIONS
+ if (tool_key(a.tool), a.match, a.moment) not in removed
+ ]
+ out.extend((a, INSTALL) for a in added)
+ return out
+
+
+def resolve(tool: str, tool_input: dict | None, mappings: Iterable = ()) -> list[dict]:
+ """Every moment this call reaches, each with the action that reached it.
+
+ One entry per moment, in catalog order. When two actions reach the same
+ moment the more specific one — the longer match — is the one named, since
+ "reached by `git push`" says more than "reached by Bash".
+ """
+ best: dict[str, dict] = {}
+ for action, via in effective_actions(mappings):
+ if not action_matches(action, tool, tool_input):
+ continue
+ held = best.get(action.moment)
+ if held is None or len(action.match) > len(held["match"]):
+ best[action.moment] = {
+ "moment": action.moment, "tool": action.tool,
+ "match": action.match, "via": via,
+ }
+
+ if tool_key(tool) == SKILL_TOOL:
+ name = str((tool_input or {}).get(SKILL_FIELD) or "").strip().lower()
+ moment = catalog.SKILL_PREFIX + name
+ if name and catalog.is_moment(moment):
+ best[moment] = {"moment": moment, "tool": tool, "match": f"skill={name}",
+ "via": DEFAULT}
+
+ order = {name: i for i, name in enumerate(catalog.MOMENTS)}
+ return sorted(best.values(), key=lambda h: (order.get(h["moment"], len(order)), h["moment"]))
+
+
+def _sample_input(tool: str, match: str) -> dict:
+ """A call that this action would describe — for reporting what it reaches."""
+ if not match:
+ return {}
+ if tool_key(tool) in COMMAND_TOOLS:
+ return {COMMAND_FIELD: match}
+ return dict(_pairs(match) or {})
+
+
+def _clean(tool: str, match: str, moment: str) -> tuple[str, str, str]:
+ """Validate a mapping's three parts, or refuse with what would work."""
+ bare = (tool or "").strip()
+ if bare.startswith("mcp__"):
+ bare = bare.rsplit("__", 1)[-1]
+ if not bare:
+ raise ValueError("tool is required — the tool as the harness names it, e.g. Bash or update_task")
+ clean_match = _squash(match)
+ if clean_match and tool_key(bare) not in COMMAND_TOOLS and _pairs(clean_match) is None:
+ raise ValueError(
+ f"{bare} does not run a command, so `match` names its arguments as "
+ f"field=value pairs (e.g. status=done), not {clean_match!r}. Leave "
+ f"it empty to map every {bare} call."
+ )
+ return bare, clean_match, catalog.require_moment(moment)
+
+
+def _is_default(tool: str, match: str, moment: str) -> bool:
+ return any(
+ tool_key(a.tool) == tool_key(tool) and a.match == match and a.moment == moment
+ for a in DEFAULT_ACTIONS
+ )
+
+
+# ── the install's own ─────────────────────────────────────────────────────
+
+async def list_mappings(user_id: int) -> list[MomentMapping]:
+ async with async_session() as session:
+ rows = await session.execute(
+ select(MomentMapping)
+ .where(MomentMapping.user_id == user_id)
+ .order_by(MomentMapping.id)
+ )
+ return list(rows.scalars().all())
+
+
+async def moments_for(user_id: int, tool: str, tool_input: dict | None) -> list[dict]:
+ """The moments a call reaches for this user.
+
+ Fails OPEN to the defaults: an unreadable mapping table must cost the
+ install its corrections, not every moment.
+ """
+ try:
+ mappings = await list_mappings(user_id)
+ except Exception:
+ logger.warning("moment mappings unreadable; using the defaults", exc_info=True)
+ mappings = []
+ return resolve(tool, tool_input, mappings)
+
+
+async def _find(session, user_id: int, tool: str, match: str, moment: str):
+ rows = await session.execute(
+ select(MomentMapping).where(
+ MomentMapping.user_id == user_id,
+ MomentMapping.moment == moment,
+ MomentMapping.match == match,
+ )
+ )
+ return next((r for r in rows.scalars().all() if tool_key(r.tool) == tool_key(tool)), None)
+
+
+async def _reaches(user_id: int, tool: str, match: str) -> list[dict]:
+ return resolve(tool, _sample_input(tool, match), await list_mappings(user_id))
+
+
+async def map_action(
+ user_id: int, tool: str, match: str, moment: str,
+ *, reason: str = "", actor: str = "model",
+) -> dict:
+ """Make an action reach a moment for this install.
+
+ Mapping a shipped default this install had removed restores it; mapping
+ one already in force changes nothing and says so.
+ """
+ tool, match, moment = _clean(tool, match, moment)
+ reason = (reason or "").strip()
+ async with async_session() as session:
+ row = await _find(session, user_id, tool, match, moment)
+ if _is_default(tool, match, moment):
+ if row is not None and row.effect == REMOVE:
+ await session.delete(row)
+ await session.commit()
+ change = "restored the shipped default"
+ else:
+ change = "already a shipped default — nothing to add"
+ elif row is not None:
+ row.reason = reason or row.reason
+ row.actor = actor
+ await session.commit()
+ change = "already mapped — reason updated" if reason else "already mapped"
+ else:
+ session.add(MomentMapping(
+ user_id=user_id, tool=tool, match=match, moment=moment,
+ effect=ADD, reason=reason, actor=actor,
+ ))
+ await session.commit()
+ change = "mapped"
+ return {
+ "change": change, "tool": tool, "match": match, "moment": moment,
+ "now_reaches": await _reaches(user_id, tool, match),
+ }
+
+
+async def unmap_action(
+ user_id: int, tool: str, match: str, moment: str,
+ *, reason: str = "", actor: str = "model",
+) -> dict:
+ """Stop an action reaching a moment for this install.
+
+ An install's own mapping is deleted. A shipped default is switched off by
+ a stored removal, so the next release does not switch it back on.
+ """
+ tool, match, moment = _clean(tool, match, moment)
+ reason = (reason or "").strip()
+ async with async_session() as session:
+ row = await _find(session, user_id, tool, match, moment)
+ if row is not None and row.effect == ADD:
+ await session.delete(row)
+ await session.commit()
+ change = "removed this install's mapping"
+ elif _is_default(tool, match, moment):
+ if row is None:
+ session.add(MomentMapping(
+ user_id=user_id, tool=tool, match=match, moment=moment,
+ effect=REMOVE, reason=reason, actor=actor,
+ ))
+ await session.commit()
+ change = "switched off the shipped default"
+ else:
+ change = "the shipped default was already off"
+ else:
+ raise ValueError(
+ f"nothing maps {tool}{' ' + repr(match) if match else ''} onto "
+ f"{moment}, so there is nothing to remove. list_moments shows "
+ f"which actions reach each moment."
+ )
+ return {
+ "change": change, "tool": tool, "match": match, "moment": moment,
+ "now_reaches": await _reaches(user_id, tool, match),
+ }
+
+
+async def actions_by_moment(user_id: int) -> dict:
+ """Each moment's actions in force, and the defaults this install removed."""
+ mappings = await list_mappings(user_id)
+ by_moment: dict[str, list[dict]] = {}
+ for action, via in effective_actions(mappings):
+ by_moment.setdefault(action.moment, []).append(
+ {"tool": action.tool, "match": action.match, "via": via}
+ )
+ removed = [m.to_dict() for m in mappings if m.effect == REMOVE]
+ return {"actions": by_moment, "removed_defaults": removed}
diff --git a/tests/test_integration_moment_mappings.py b/tests/test_integration_moment_mappings.py
new file mode 100644
index 00000000..2446a52a
--- /dev/null
+++ b/tests/test_integration_moment_mappings.py
@@ -0,0 +1,119 @@
+"""Real-Postgres tests for an install's moment mappings (milestone 458 step 2).
+
+What these pin is what the in-session correction promises the operator: a
+mapping made is in force on the next call, a default switched off stays off,
+switching it back on leaves no residue, one user's corrections are not
+another's, and asking to remove what nothing maps is refused rather than
+recorded. Every one of those is a claim about rows, so none of it is mocked.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import delete, select
+
+from scribe.models import async_session
+from scribe.models.moment_mapping import MomentMapping
+from scribe.services import moment_actions as ma
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER_USERNAME = "moment_mapping_owner"
+STRANGER_USERNAME = "moment_mapping_stranger"
+
+SHIP = {"command": "make ship"}
+CURL = {"command": "curl localhost:8000"}
+
+
+@pytest_asyncio.fixture
+async def users():
+ async with async_session() as s:
+ owner = await ensure_user(s, OWNER_USERNAME)
+ stranger = await ensure_user(s, STRANGER_USERNAME)
+ await s.commit()
+ ids = (owner.id, stranger.id)
+ # At SETUP: the lane shares one database, so a previous run's rows
+ # are cleared before this one reads anything.
+ await s.execute(delete(MomentMapping).where(MomentMapping.user_id.in_(ids)))
+ await s.commit()
+ return ids
+
+
+async def _reached(uid, tool, tool_input):
+ return [h["moment"] for h in await ma.moments_for(uid, tool, tool_input)]
+
+
+async def _rows(uid):
+ async with async_session() as s:
+ return (await s.execute(
+ select(MomentMapping).where(MomentMapping.user_id == uid)
+ )).scalars().all()
+
+
+async def test_a_mapping_is_in_force_on_the_next_call(users):
+ uid, _ = users
+ assert "work.deliver" not in await _reached(uid, "Bash", SHIP)
+
+ out = await ma.map_action(uid, "Bash", "make ship", "work.deliver",
+ reason="this install ships with make")
+ assert out["change"] == "mapped"
+ assert "work.deliver" in [h["moment"] for h in out["now_reaches"]]
+ assert "work.deliver" in await _reached(uid, "Bash", SHIP)
+
+
+async def test_mapping_twice_is_one_row(users):
+ uid, _ = users
+ await ma.map_action(uid, "Bash", "make ship", "work.deliver")
+ again = await ma.map_action(uid, "Bash", "make ship", "work.deliver", reason="why")
+ assert again["change"].startswith("already mapped")
+ rows = await _rows(uid)
+ assert len(rows) == 1 and rows[0].reason == "why"
+
+
+async def test_one_users_corrections_are_not_anothers(users):
+ uid, other = users
+ await ma.map_action(uid, "Bash", "make ship", "work.deliver")
+ assert "work.deliver" not in await _reached(other, "Bash", SHIP)
+
+
+async def test_unmapping_an_installs_mapping_deletes_it(users):
+ uid, _ = users
+ await ma.map_action(uid, "Bash", "make ship", "work.deliver")
+ out = await ma.unmap_action(uid, "Bash", "make ship", "work.deliver")
+ assert out["change"] == "removed this install's mapping"
+ assert await _rows(uid) == []
+ assert "work.deliver" not in await _reached(uid, "Bash", SHIP)
+
+
+async def test_a_default_switched_off_stays_off_and_switches_back_cleanly(users):
+ uid, _ = users
+ assert "env.reach" in await _reached(uid, "Bash", CURL)
+
+ off = await ma.unmap_action(uid, "Bash", "curl", "env.reach",
+ reason="curl here only hits the local dev server")
+ assert off["change"] == "switched off the shipped default"
+ assert await _reached(uid, "Bash", CURL) == ["work.run"]
+ [row] = await _rows(uid)
+ assert row.effect == ma.REMOVE
+
+ by_moment = await ma.actions_by_moment(uid)
+ assert [r["match"] for r in by_moment["removed_defaults"]] == ["curl"]
+ assert {"tool": "Bash", "match": "curl", "via": ma.DEFAULT} not in by_moment["actions"]["env.reach"]
+
+ on = await ma.map_action(uid, "Bash", "curl", "env.reach")
+ assert on["change"] == "restored the shipped default"
+ assert await _rows(uid) == []
+ assert "env.reach" in await _reached(uid, "Bash", CURL)
+
+
+async def test_mapping_a_default_already_in_force_writes_nothing(users):
+ uid, _ = users
+ out = await ma.map_action(uid, "update_task", "status=done", "work.finish")
+ assert out["change"].startswith("already a shipped default")
+ assert await _rows(uid) == []
+
+
+async def test_removing_what_nothing_maps_is_refused(users):
+ uid, _ = users
+ with pytest.raises(ValueError, match="nothing to remove"):
+ await ma.unmap_action(uid, "Bash", "make ship", "work.deliver")
+ assert await _rows(uid) == []
diff --git a/tests/test_moment_actions.py b/tests/test_moment_actions.py
new file mode 100644
index 00000000..985e25ec
--- /dev/null
+++ b/tests/test_moment_actions.py
@@ -0,0 +1,181 @@
+"""Which actions reach which moment (milestone 458 step 2) — the pure half.
+
+Matching and resolution take no database: the defaults are code and an
+install's mappings arrive as rows, so every case below hands `resolve` the
+rows it needs. The writes that store those rows are exercised against real
+Postgres in `test_integration_moment_mappings.py`.
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services import moment_actions as ma
+from scribe.services import moments
+
+
+def _row(tool, match, moment, effect=ma.ADD):
+ return SimpleNamespace(tool=tool, match=match, moment=moment, effect=effect)
+
+
+def _reached(tool, tool_input=None, mappings=()):
+ return [h["moment"] for h in ma.resolve(tool, tool_input, mappings)]
+
+
+# ── the defaults ─────────────────────────────────────────────────────────
+
+def test_every_default_reaches_a_real_moment():
+ """A default naming a moment that is not in the catalog would never be
+ mounted on — the typo require_moment refuses at the door."""
+ bad = [a for a in ma.DEFAULT_ACTIONS if not moments.is_moment(a.moment)]
+ assert not bad, bad
+
+
+def test_no_default_is_listed_twice():
+ keys = [(ma.tool_key(a.tool), a.match, a.moment) for a in ma.DEFAULT_ACTIONS]
+ assert len(keys) == len(set(keys))
+
+
+def test_every_default_is_a_mapping_the_door_would_accept():
+ """The defaults pass the same validation an install's mapping does, so a
+ default and an install's copy of it can never disagree about its shape."""
+ for a in ma.DEFAULT_ACTIONS:
+ assert ma._clean(a.tool, a.match, a.moment) == (a.tool, a.match, a.moment)
+
+
+# ── matching ─────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("command", [
+ "git push origin dev",
+ "git push",
+ "cd app && git push",
+ "make lint; git push --tags",
+ "GIT_TRACE=1 git push",
+ " git push ",
+])
+def test_a_command_prefix_matches_any_segment_of_the_line(command):
+ assert "work.deliver" in _reached("Bash", {"command": command})
+
+
+@pytest.mark.parametrize("command", ["git pushd", "echo git push", "git status"])
+def test_a_command_prefix_needs_a_word_boundary_and_the_head_of_a_segment(command):
+ assert "work.deliver" not in _reached("Bash", {"command": command})
+
+
+def test_one_action_reaches_every_moment_it_is():
+ """`kubectl apply` is a run, a deliver and a reach outside the workspace."""
+ assert _reached("Bash", {"command": "kubectl apply -f deploy.yaml"}) == [
+ "work.run", "work.deliver", "env.reach",
+ ]
+
+
+def test_the_most_specific_action_is_the_one_named():
+ hits = ma.resolve("Bash", {"command": "git push"})
+ deliver = next(h for h in hits if h["moment"] == "work.deliver")
+ assert deliver["match"] == "git push"
+ run = next(h for h in hits if h["moment"] == "work.run")
+ assert run["match"] == ""
+
+
+def test_an_ordinary_command_is_only_a_run():
+ assert _reached("Bash", {"command": "ls -la"}) == ["work.run"]
+
+
+def test_an_unknown_tool_reaches_nothing():
+ assert _reached("SomeOtherTool", {"x": 1}) == []
+
+
+@pytest.mark.parametrize("name", [
+ "mcp__plugin_scribe_scribe__update_task", "mcp__scribe__update_task",
+ "update_task", "Update_Task",
+])
+def test_the_server_prefix_and_case_do_not_matter(name):
+ assert _reached(name, {"task_id": 1, "status": "done"}) == ["work.finish"]
+
+
+@pytest.mark.parametrize("status,expected", [
+ ("done", ["work.finish"]),
+ ("cancelled", ["work.finish"]),
+ ("in_progress", ["work.start"]),
+ ("todo", []),
+ ("", []),
+])
+def test_arguments_select_the_moment(status, expected):
+ assert _reached("update_task", {"task_id": 1, "status": status}) == expected
+
+
+def test_a_whole_tool_mapping_ignores_its_arguments():
+ assert _reached("Edit", {"file_path": "x", "old_string": "a"}) == ["work.change"]
+
+
+def test_loading_a_procedure_reaches_its_own_moment():
+ assert _reached("Skill", {"skill": "scribe:writing-plans"}) == [
+ "skill.scribe:writing-plans",
+ ]
+
+
+def test_a_procedure_name_that_is_not_a_moment_reaches_nothing():
+ assert _reached("Skill", {"skill": "two words"}) == []
+ assert _reached("Skill", {}) == []
+
+
+# ── an install's own ─────────────────────────────────────────────────────
+
+def test_an_install_mapping_adds_a_moment():
+ rows = [_row("Bash", "make ship", "work.deliver")]
+ assert "work.deliver" in _reached("Bash", {"command": "make ship"}, rows)
+ assert "work.deliver" not in _reached("Bash", {"command": "make ship"})
+
+
+def test_removing_a_default_removes_only_that_default():
+ """`curl` stops reaching env.reach here; it is still a run, and `ssh`
+ still reaches out — a removal overrides nothing it did not name."""
+ rows = [_row("Bash", "curl", "env.reach", ma.REMOVE)]
+ assert _reached("Bash", {"command": "curl localhost:8000"}, rows) == ["work.run"]
+ assert "env.reach" in _reached("Bash", {"command": "ssh box"}, rows)
+
+
+def test_a_removal_matches_the_default_whatever_the_tool_is_called():
+ rows = [_row("mcp__x__update_task", "status=done", "work.finish", ma.REMOVE)]
+ assert _reached("update_task", {"status": "done"}, rows) == []
+
+
+def test_an_install_mapping_is_named_as_the_installs():
+ rows = [_row("Bash", "make ship", "work.deliver")]
+ hit = next(h for h in ma.resolve("Bash", {"command": "make ship"}, rows)
+ if h["moment"] == "work.deliver")
+ assert (hit["via"], hit["match"]) == (ma.INSTALL, "make ship")
+
+
+# ── validation ───────────────────────────────────────────────────────────
+
+def test_a_prefix_on_a_tool_that_runs_no_command_is_refused():
+ """It would be stored and never match — a mapping that looks made and is
+ not, which is the misfire this table exists to fix."""
+ with pytest.raises(ValueError, match="field=value"):
+ ma._clean("update_task", "done", "work.finish")
+
+
+def test_an_unknown_moment_is_refused():
+ with pytest.raises(ValueError, match="unknown moment"):
+ ma._clean("Bash", "make ship", "work.ship")
+
+
+def test_a_mapping_is_normalised_before_it_is_stored():
+ assert ma._clean("mcp__x__update_task", " status=done ", " Work.Finish") == (
+ "update_task", "status=done", "work.finish",
+ )
+ assert ma._clean("Bash", "make ship", "work.deliver")[1] == "make ship"
+
+
+def test_a_tool_is_required():
+ with pytest.raises(ValueError, match="tool is required"):
+ ma._clean(" ", "", "work.run")
+
+
+# ── failing open ─────────────────────────────────────────────────────────
+
+async def test_an_unreadable_mapping_table_costs_the_corrections_not_the_moments():
+ with patch.object(ma, "list_mappings", AsyncMock(side_effect=RuntimeError("db down"))):
+ hits = await ma.moments_for(7, "Bash", {"command": "git push"})
+ assert [h["moment"] for h in hits] == ["work.run", "work.deliver"]
diff --git a/tests/test_moments.py b/tests/test_moments.py
index 1655e890..f4e12c8a 100644
--- a/tests/test_moments.py
+++ b/tests/test_moments.py
@@ -93,27 +93,42 @@ def test_the_catalog_payload_carries_every_moment_and_the_family():
assert data["families"][0]["prefix"] == moments.SKILL_PREFIX
-async def test_the_tool_returns_the_service_catalog():
+async def test_the_tool_returns_the_catalog_with_this_installs_actions():
+ from unittest.mock import AsyncMock, patch
+
from scribe.mcp.tools import moments as tool
- assert await tool.list_moments() == moments.catalog()
+ by_moment = {"actions": {"work.change": []}, "removed_defaults": []}
+ with patch.object(tool, "current_user_id", lambda: 7), \
+ patch.object(tool.actions_svc, "actions_by_moment",
+ AsyncMock(return_value=by_moment)) as read:
+ out = await tool.list_moments()
+ read.assert_awaited_once_with(7)
+ assert out["moments"] == moments.catalog()["moments"]
+ assert out["actions"] == by_moment["actions"]
+ assert out["removed_defaults"] == []
-def test_the_tool_is_registered_and_read_only():
- from scribe.mcp.server import _READ_ONLY_TOOLS
+def test_the_tools_are_registered_and_classified():
+ from scribe.mcp.server import _READ_ONLY_TOOLS, _WRITE_TOOLS
from scribe.mcp.tools import moments as tool
mcp = FakeMCP()
tool.register(mcp)
- assert mcp.names == ["list_moments"]
+ assert mcp.names == ["list_moments", "map_action", "unmap_action"]
assert "list_moments" in _READ_ONLY_TOOLS
+ assert {"map_action", "unmap_action"} <= _WRITE_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."""
+ """Rule 33 parity: the tool and the routes call the same services, so the
+ session and the Settings view cannot name different moments or disagree
+ about what a mapping does."""
from scribe.mcp.tools import moments as tool
from scribe.routes import retrieval as routes
+ from scribe.services import moment_actions
assert tool.moments_svc is moments
assert routes.moments_svc is moments
+ assert tool.actions_svc is moment_actions
+ assert routes.moment_actions_svc is moment_actions
diff --git a/tests/test_routes_retrieval_tuning.py b/tests/test_routes_retrieval_tuning.py
index 883fdb52..22ab4b14 100644
--- a/tests/test_routes_retrieval_tuning.py
+++ b/tests/test_routes_retrieval_tuning.py
@@ -43,6 +43,7 @@ def test_every_endpoint_is_reachable_on_the_app():
"/api/retrieval/surfaces/",
"/api/retrieval/tuning-history",
"/api/retrieval/moments",
+ "/api/retrieval/moments/mappings",
}
diff --git a/tests/test_services_backup.py b/tests/test_services_backup.py
index b1c78df3..ac3e7689 100644
--- a/tests/test_services_backup.py
+++ b/tests/test_services_backup.py
@@ -27,7 +27,7 @@ def test_backup_version_is_current():
(Named for the number it asserted until v10, which is exactly the drift a
name-carrying-a-value invites; it now says what it checks.)"""
- assert backup.BACKUP_VERSION == 20
+ assert backup.BACKUP_VERSION == 21
def _exportable_note(**over):
@@ -141,6 +141,7 @@ def _column_guard_targets():
from scribe.models.rule_usage import RuleUsageEvent
from scribe.models.system_usage import SystemUsageEvent
from scribe.models.retrieval_tuning import RetrievalTuningEvent
+ from scribe.models.moment_mapping import MomentMapping
from scribe.models.note_version import NoteVersion
from scribe.models.rule_version import RuleVersion
from scribe.models.project import Project
@@ -177,6 +178,7 @@ def _column_guard_targets():
"retrieval_tuning_events": (
RetrievalTuningEvent, backup._retrieval_tuning_event_rows,
),
+ "moment_mappings": (MomentMapping, backup._moment_mapping_rows),
"design_systems": (DesignSystem, backup._design_system_rows),
"design_tokens": (DesignToken, backup._design_token_rows),
"repo_bindings": (RepoBinding, backup._repo_binding_rows),
@@ -294,6 +296,7 @@ def _import_guard_targets():
"rule_usage_events": backup._build_rule_usage_event,
"system_usage_events": backup._build_system_usage_event,
"retrieval_tuning_events": backup._build_retrieval_tuning_event,
+ "moment_mappings": backup._build_moment_mapping,
"design_systems": backup._build_design_system,
"design_tokens": backup._build_design_token,
"repo_bindings": backup._build_repo_binding,