feat(family): the promotion engine - triggers, the three criteria, the decision log and undo (milestone 463 step 3, #4989)

The agent promotes a family idea when all three criteria hold, and no person approves it. The criteria are product text: services/family.py states them, and the promote tool's docstring names every criterion the service enforces.

- Criteria: each one vetoes on its own when its reasoning is blank. Platform terms also needs an applies-when and a platform scope; proven also needs named evidence. A veto keeps the idea a candidate and is logged, so it becomes precedent.
- Precedent: every promotion stores the decisions on the nearest ideas by meaning, plus any the caller names.
- Promotion sets canon, the applicability test and the platform scope, and opens an unassessed ledger row for each member project the promoter can write. Re-promotion moves the version past every version the idea has held.
- Retire and undo: undo reverses only the latest idea-level decision, restores its recorded before-state, and logs itself with the undone decision as its precedent. Settled here: leaving canon closes the unassessed rows but keeps the judged ones, which read as needing a recheck after a re-promotion.
- Triggers open evaluations but never promote:
  - a cross-project lineage citation ("matching #N") on a note or task write;
  - a same-meaning record in another project on a shared platform, on create, at 0.80 (measured: the known pattern's builds scored 0.79-0.82, an unrelated project's best match 0.65);
  - a milestone closing on a platform.
  Each fails open and rides the response as family_hint.
- Doors: seven MCP tools, /api/family REST endpoints, and a Family page (nav, /family) showing the criteria, the ideas, and the decision log with undo and retire.
- utils/recordHref.ts holds the one copy of "where a record opens", now shared with LessonDetailView.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/conftest.py

@@ -215,6 +215,30 @@ def _no_lesson_rule_links(request):
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_family_triggers(request):
+    """Stub family canon's write-time triggers (milestone 463 step 3).
+
+    create_note, create_task, create_snippet, the updates that change a body,
+    and update_milestone(status="done") now look for a pattern carried between
+    projects, and each look is a database read (the repeat check also embeds).
+    All are fail-open, so unstubbed they would cost a slow failed connect per
+    unit test rather than a failure. Patched at the trigger functions, beneath
+    `attach_family_hint`, so the attach itself — its order and its fail-open —
+    still runs. tests/test_family.py binds the real pure parts; the triggers
+    themselves are exercised against Postgres in
+    tests/test_integration_family_promotion.py, which this skips.
+    """
+    if request.node.get_closest_marker("integration"):
+        yield
+        return
+    with patch("scribe.services.family.citation_trigger", AsyncMock(return_value=None)), \
+         patch("scribe.services.family.repeat_trigger", AsyncMock(return_value=None)), \
+         patch("scribe.services.family.milestone_trigger", AsyncMock(return_value=None)), \
+         patch("scribe.services.family.milestone_is_open", AsyncMock(return_value=False)):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def _no_moment_delivery(request):
     """Stub the moment lookup the mapped MCP tools attach (milestone 458).

tests/test_family.py

@@ -0,0 +1,163 @@
+"""The promotion engine without a database (milestone 463 step 3).
+
+The parts that decide — which citations claim lineage, which criteria veto —
+are pure and pinned here, beside the doors' wiring: the triggers ride the
+write tools, fail open, and the criteria the service enforces are the ones
+the agent is told. The state machine against Postgres is in
+tests/test_integration_family_promotion.py.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.mcp.server import _READ_ONLY_TOOLS, _WRITE_TOOLS
+from scribe.mcp.tools import family as family_tools
+from scribe.mcp.tools.milestones import update_milestone
+from scribe.services import family as family_svc
+from scribe.services.family import CRITERIA_KEYS, lineage_citations, vetoes
+from tests.helpers import fake_milestone
+
+pytestmark = pytest.mark.usefixtures("_bind_user")
+
+
+# --- which citations claim lineage ---------------------------------------------
+
+@pytest.mark.parametrize("text", [
+    "Signed release lane, matching roundtable-android (task #1615).",
+    "Ported from #1615, with the keystore step moved first.",
+    "Same shape as #1615: the APK is baked into the image.",
+    "This mirrors #1615 for the desktop client.",
+    "Modelled on #1615.",
+])
+def test_a_citation_with_a_lineage_word_names_its_source(text):
+    assert lineage_citations(text) == [1615]
+
+
+@pytest.mark.parametrize("text", [
+    "See #1615 for context.",
+    "#1615 is related.",
+    "Closes #1615.",
+    # The lineage word is in the PREVIOUS sentence — a different claim.
+    "This is matching the old flow. Unrelated: #1615.",
+    "Matching the old flow\nsee #1615",
+    "",
+])
+def test_a_citation_without_lineage_is_silent(text):
+    assert lineage_citations(text) == []
+
+
+def test_lineage_citations_are_in_order_and_unique():
+    text = "Based on #20; ported from #10. Same pattern as #20."
+    assert lineage_citations(text) == [20, 10]
+
+
+# --- each criterion vetoes on its own -----------------------------------------------
+
+GOOD = dict(
+    applies_when="any app that installs its own updates",
+    platforms=["android-app"],
+    criteria={
+        "platform_terms": "stated as an Android distribution concern",
+        "platform_problem": "Play Protect and signature checks are the platform's",
+        "proven": "shipped in two apps",
+    },
+    evidence=["CI run 8274 green", "#4775"],
+)
+
+
+def test_all_three_criteria_held_means_no_veto():
+    assert vetoes(**GOOD) == []
+
+
+@pytest.mark.parametrize("key", CRITERIA_KEYS)
+def test_each_criterion_left_unsupported_vetoes_on_its_own(key):
+    case = {**GOOD, "criteria": {**GOOD["criteria"], key: "   "}}
+    assert vetoes(**case) == [key]
+
+
+def test_platform_terms_needs_an_applicability_test_and_a_scope():
+    assert vetoes(**{**GOOD, "applies_when": ""}) == ["platform_terms"]
+    assert vetoes(**{**GOOD, "platforms": []}) == ["platform_terms"]
+
+
+def test_proven_needs_named_evidence():
+    assert vetoes(**{**GOOD, "evidence": []}) == ["proven"]
+    assert vetoes(**{**GOOD, "evidence": ["  "]}) == ["proven"]
+
+
+def test_the_criteria_the_agent_is_told_are_the_ones_enforced():
+    """Rule 119: the criteria are product text. The promote tool's docstring
+    must name every criterion the service checks, by its parameter name."""
+    doc = family_tools.promote_family_idea.__doc__
+    for key in CRITERIA_KEYS:
+        assert key in doc, f"promote_family_idea's docstring never names {key}"
+    assert len(family_svc.CRITERIA) == 3
+
+
+# --- the doors ----------------------------------------------------------------------
+
+def test_every_family_tool_is_classified():
+    reads = {"list_family_ideas", "get_family_idea", "list_family_decisions"}
+    writes = {"propose_family_idea", "promote_family_idea", "retire_family_idea",
+              "undo_family_decision"}
+    assert reads <= _READ_ONLY_TOOLS
+    assert writes <= _WRITE_TOOLS
+
+
+def _note(**kw):
+    base = dict(id=7, project_id=2, title="t", body="b", is_task=False)
+    return SimpleNamespace(**{**base, **kw})
+
+
+async def test_a_citation_hint_wins_and_repeat_is_not_asked():
+    data: dict = {}
+    repeat = AsyncMock(return_value="repeat")
+    with patch.object(family_svc, "citation_trigger", AsyncMock(return_value="cited")), \
+         patch.object(family_svc, "repeat_trigger", repeat):
+        await family_svc.attach_family_hint(1, data, _note(), created=True)
+    assert data == {"family_hint": "cited"}
+    repeat.assert_not_awaited()
+
+
+async def test_the_repeat_check_runs_only_on_a_create():
+    repeat = AsyncMock(return_value="repeat")
+    with patch.object(family_svc, "citation_trigger", AsyncMock(return_value=None)), \
+         patch.object(family_svc, "repeat_trigger", repeat):
+        edited: dict = {}
+        await family_svc.attach_family_hint(1, edited, _note(), created=False)
+        created: dict = {}
+        await family_svc.attach_family_hint(1, created, _note(), created=True)
+    assert edited == {}
+    assert created == {"family_hint": "repeat"}
+
+
+async def test_a_failing_trigger_never_breaks_the_write():
+    data: dict = {"id": 7}
+    with patch.object(family_svc, "citation_trigger", AsyncMock(side_effect=RuntimeError("db down"))):
+        await family_svc.attach_family_hint(1, data, _note(), created=True)
+    assert data == {"id": 7}
+
+
+async def test_closing_a_milestone_on_a_platform_carries_the_hint():
+    closed = fake_milestone(id=5, project_id=3, status="done")
+    with patch("scribe.mcp.tools.milestones.milestones_svc.update_milestone",
+               AsyncMock(return_value=closed)), \
+         patch.object(family_svc, "milestone_is_open", AsyncMock(return_value=True)), \
+         patch.object(family_svc, "milestone_trigger", AsyncMock(return_value="evaluate")):
+        out = await update_milestone(project_id=3, milestone_id=5, status="done")
+    assert out["family_hint"] == "evaluate"
+
+
+async def test_re_saving_a_closed_milestone_asks_nothing():
+    closed = fake_milestone(id=5, project_id=3, status="done")
+    trigger = AsyncMock(return_value="evaluate")
+    with patch("scribe.mcp.tools.milestones.milestones_svc.update_milestone",
+               AsyncMock(return_value=closed)), \
+         patch.object(family_svc, "milestone_is_open", AsyncMock(return_value=False)), \
+         patch.object(family_svc, "milestone_trigger", trigger):
+        out = await update_milestone(project_id=3, milestone_id=5, status="done")
+    assert "family_hint" not in out
+    trigger.assert_not_awaited()

tests/test_integration_family_promotion.py

@@ -0,0 +1,293 @@
+"""The promotion engine against real Postgres (milestone 463 step 3).
+
+What the unit lane can't show: a promotion opens the right ledger rows and
+no others, a veto changes nothing but the log, an undo puts back exactly the
+recorded prior state, and each trigger fires on its fixture and stays silent
+on the near-miss beside it.
+
+Precedent search and the repeat trigger both rank by meaning, and the lane
+has no embedding model. The search is stubbed to "nothing similar" for every
+test here (`_no_meaning`); the repeat tests stub it to one hit, to pin what
+the trigger does WITH a hit. The ranking itself is the shared semantic
+search, tested where that lives.
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.family import (
+    FamilyAdoption, FamilyDecision, FamilyIdea, Platform, ProjectPlatform,
+)
+from scribe.models.milestone import Milestone
+from scribe.models.note import Note
+from scribe.models.project import Project
+from scribe.models.user import User
+from scribe.services import family as family_svc
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER = "family_promotion_owner"
+OUTSIDER = "family_promotion_outsider"
+
+CRITERIA = {
+    "platform_terms": "stated for any Android app that distributes its own APK",
+    "platform_problem": "signature continuity is the platform's rule, not one app's",
+    "proven": "shipped and updated in place on a device",
+}
+
+
+async def _purge(username: str) -> None:
+    """SETUP ONLY, as the backup round-trip siblings do: a database call
+    after a `yield` in an autouse fixture orphans a pooled connection."""
+    async with async_session() as s:
+        for user in (await s.execute(select(User).where(User.username == username))).scalars():
+            for note in (await s.execute(select(Note).where(Note.user_id == user.id))).scalars():
+                await s.delete(note)
+            for project in (await s.execute(select(Project).where(Project.user_id == user.id))).scalars():
+                await s.delete(project)
+        await s.commit()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _clean():
+    await _purge(OWNER)
+    await _purge(OUTSIDER)
+
+
+@pytest.fixture(autouse=True)
+def _no_meaning():
+    with patch("scribe.services.embeddings.semantic_search_notes", AsyncMock(return_value=[])):
+        yield
+
+
+async def _platform_id(slug: str) -> int:
+    async with async_session() as s:
+        return await s.scalar(select(Platform.id).where(
+            Platform.slug == slug, Platform.deleted_at.is_(None)))
+
+
+@pytest_asyncio.fixture
+async def family():
+    """Three projects: two Android apps and a Go service, the pattern note
+    in the first, and an outsider's Android app the owner cannot write."""
+    android, go = await _platform_id("android-app"), await _platform_id("go")
+    async with async_session() as s:
+        owner = await ensure_user(s, OWNER)
+        outsider = await ensure_user(s, OUTSIDER)
+        a = Project(user_id=owner.id, title="android one")
+        b = Project(user_id=owner.id, title="android two")
+        c = Project(user_id=owner.id, title="go service")
+        theirs = Project(user_id=outsider.id, title="someone else's android app")
+        s.add_all([a, b, c, theirs])
+        await s.flush()
+        s.add_all([
+            ProjectPlatform(project_id=a.id, platform_id=android, state="declared"),
+            ProjectPlatform(project_id=b.id, platform_id=android, state="detected"),
+            ProjectPlatform(project_id=c.id, platform_id=go, state="declared"),
+            ProjectPlatform(project_id=theirs.id, platform_id=android, state="declared"),
+        ])
+        pattern = Note(user_id=owner.id, project_id=a.id, title="Signed APK lane",
+                       body="One keystore, two channels, in-place update.")
+        s.add(pattern)
+        await s.commit()
+        return {"owner": owner.id, "outsider": outsider.id, "a": a.id, "b": b.id,
+                "c": c.id, "theirs": theirs.id, "note": pattern.id}
+
+
+async def _promote(f, **overrides):
+    kw = dict(
+        applies_when="any Android app that ships its own APK",
+        platforms=["android-app"], criteria=CRITERIA,
+        evidence=["CI green on the release lane", "verified on a device"],
+        reason="built twice, proven, stated for the platform",
+    )
+    kw.update(overrides)
+    return await family_svc.promote(f["owner"], f["note"], **kw)
+
+
+async def _ledger(note_id: int) -> dict[int, str]:
+    async with async_session() as s:
+        rows = (await s.execute(
+            select(FamilyAdoption.project_id, FamilyAdoption.status)
+            .where(FamilyAdoption.idea_id == note_id)
+        )).all()
+    return dict(rows)
+
+
+async def _idea(note_id: int) -> FamilyIdea:
+    async with async_session() as s:
+        return await s.get(FamilyIdea, note_id)
+
+
+# --- promotion ------------------------------------------------------------------
+
+async def test_promotion_opens_a_row_for_every_member_project_the_promoter_can_write(family):
+    out = await _promote(family)
+    assert out["promoted"] is True
+    # Declared and detected both count as membership; the Go service is not
+    # on the platform; the outsider's app is not the promoter's to write.
+    assert await _ledger(family["note"]) == {family["a"]: "unassessed",
+                                            family["b"]: "unassessed"}
+    idea = await _idea(family["note"])
+    assert idea.status == "canon" and idea.canon_version == 1
+    decision = out["decision"]
+    assert decision["action"] == "promote"
+    assert decision["after"]["platforms"] == ["android-app"]
+    assert decision["evidence"]["criteria"]["proven"] == CRITERIA["proven"]
+    assert decision["evidence"]["ledger_rows_opened"] == 2
+
+
+@pytest.mark.parametrize("key", list(family_svc.CRITERIA_KEYS))
+async def test_each_criterion_vetoes_on_its_own_and_the_veto_is_logged(family, key):
+    out = await _promote(family, criteria={**CRITERIA, key: ""})
+    assert out["promoted"] is False and out["vetoed_by"] == [key]
+    assert (await _idea(family["note"])).status == "candidate"
+    assert await _ledger(family["note"]) == {}
+    assert out["decision"]["action"] == "propose"
+    assert out["decision"]["evidence"]["vetoed_by"] == [key]
+
+
+async def test_an_unknown_platform_is_refused_before_anything_is_written(family):
+    with pytest.raises(ValueError, match="unknown platform"):
+        await _promote(family, platforms=["android-app", "no-such-platform"])
+    assert await _idea(family["note"]) is None
+
+
+async def test_someone_who_cannot_write_the_note_cannot_promote_it(family):
+    with pytest.raises(ValueError, match="no write access"):
+        await family_svc.promote(
+            family["outsider"], family["note"], applies_when="x", platforms=["android-app"],
+            criteria=CRITERIA, evidence=["e"], reason="r",
+        )
+
+
+# --- undo and retirement ------------------------------------------------------------
+
+async def test_undoing_a_promotion_restores_the_prior_state_and_keeps_judged_rows(family):
+    out = await _promote(family)
+    async with async_session() as s:
+        row = (await s.execute(select(FamilyAdoption).where(
+            FamilyAdoption.idea_id == family["note"],
+            FamilyAdoption.project_id == family["a"]))).scalars().one()
+        row.status, row.canon_version, row.decided_via = "adopted", 1, "agent"
+        await s.commit()
+
+    await family_svc.undo(family["owner"], out["decision"]["id"], reason="promoted too early")
+    idea = await _idea(family["note"])
+    # Promoted directly, with no proposal before it: the prior state was "not
+    # an idea", which an undo records as retired rather than deleting the
+    # idea and its log with it.
+    assert idea.status == "retired" and idea.applies_when is None
+    # The unjudged row goes; the judged one stays as history.
+    assert await _ledger(family["note"]) == {family["a"]: "adopted"}
+
+    again = await _promote(family)
+    # Re-promotion moves PAST every version this idea has held, so the kept
+    # answer reads as needing a recheck rather than as still agreeing.
+    assert again["idea"]["canon_version"] == 2
+    assert await _ledger(family["note"]) == {family["a"]: "adopted", family["b"]: "unassessed"}
+
+
+async def test_retiring_closes_unjudged_rows_and_undoing_it_reopens_them(family):
+    await _promote(family)
+    out = await family_svc.retire(family["owner"], family["note"], reason="superseded")
+    assert (await _idea(family["note"])).status == "retired"
+    assert await _ledger(family["note"]) == {}
+    await family_svc.undo(family["owner"], out["decision"]["id"], reason="not superseded after all")
+    assert (await _idea(family["note"])).status == "canon"
+    assert set((await _ledger(family["note"])).values()) == {"unassessed"}
+
+
+async def test_only_the_latest_idea_decision_can_be_undone(family):
+    first = await _promote(family)
+    await family_svc.retire(family["owner"], family["note"], reason="superseded")
+    with pytest.raises(ValueError, match="came after it"):
+        await family_svc.undo(family["owner"], first["decision"]["id"], reason="r")
+
+
+async def test_an_undo_names_what_it_reversed_as_its_precedent(family):
+    out = await _promote(family)
+    undo = await family_svc.undo(family["owner"], out["decision"]["id"], reason="r")
+    assert undo["decision"]["action"] == "undo"
+    assert undo["decision"]["precedent_ids"] == [out["decision"]["id"]]
+    async with async_session() as s:
+        actions = (await s.execute(select(FamilyDecision.action).where(
+            FamilyDecision.idea_id == family["note"]).order_by(FamilyDecision.id))).scalars().all()
+    assert actions == ["promote", "undo"]
+
+
+# --- triggers -------------------------------------------------------------------------
+
+async def _note_in(project_id: int, owner_id: int, body: str) -> Note:
+    async with async_session() as s:
+        note = Note(user_id=owner_id, project_id=project_id, title="the second build", body=body)
+        s.add(note)
+        await s.commit()
+        await s.refresh(note)
+        return note
+
+
+async def test_a_cross_project_lineage_citation_opens_an_evaluation(family):
+    citing = await _note_in(family["b"], family["owner"],
+                            f"Release lane, matching the first app (#{family['note']}).")
+    hint = await family_svc.citation_trigger(family["owner"], citing)
+    assert hint and f"#{family['note']}" in hint
+    idea = await _idea(family["note"])
+    assert idea is not None and idea.status == "candidate"
+    async with async_session() as s:
+        d = (await s.execute(select(FamilyDecision).where(
+            FamilyDecision.idea_id == family["note"]))).scalars().one()
+    assert d.decided_via == "system" and d.evidence["trigger"] == "citation"
+
+
+async def test_a_citation_without_lineage_or_within_one_project_is_silent(family):
+    pointer = await _note_in(family["b"], family["owner"], f"See #{family['note']} for context.")
+    same_project = await _note_in(family["a"], family["owner"], f"Matching #{family['note']}.")
+    assert await family_svc.citation_trigger(family["owner"], pointer) is None
+    assert await family_svc.citation_trigger(family["owner"], same_project) is None
+    assert await _idea(family["note"]) is None
+
+
+async def test_a_repeat_on_a_shared_platform_opens_an_evaluation(family):
+    """The meaning match is stubbed — the lane has no model — so this pins
+    what the trigger does with a hit: only a hit in ANOTHER project that
+    shares a platform counts."""
+    repeat = await _note_in(family["b"], family["owner"], "One keystore, two channels.")
+    async with async_session() as s:
+        original = await s.get(Note, family["note"])
+    with patch("scribe.services.embeddings.semantic_search_notes",
+               AsyncMock(return_value=[(0.86, original)])):
+        hint = await family_svc.repeat_trigger(family["owner"], repeat)
+    assert hint and f"#{family['note']}" in hint
+    assert (await _idea(family["note"])).status == "candidate"
+
+
+async def test_a_repeat_with_no_shared_platform_is_silent(family):
+    unrelated = await _note_in(family["c"], family["owner"], "One keystore, two channels.")
+    async with async_session() as s:
+        original = await s.get(Note, family["note"])
+    with patch("scribe.services.embeddings.semantic_search_notes",
+               AsyncMock(return_value=[(0.86, original)])):
+        assert await family_svc.repeat_trigger(family["owner"], unrelated) is None
+    assert await _idea(family["note"]) is None
+
+
+async def test_a_milestone_closing_on_a_platform_asks_and_one_off_platform_does_not(family):
+    async with async_session() as s:
+        on_platform = Milestone(user_id=family["owner"], project_id=family["a"], title="m1")
+        bare = Project(user_id=family["owner"], title="no platforms")
+        s.add_all([on_platform, bare])
+        await s.flush()
+        off_platform = Milestone(user_id=family["owner"], project_id=bare.id, title="m2")
+        s.add(off_platform)
+        await s.commit()
+        await s.refresh(on_platform)
+        await s.refresh(off_platform)
+    hint = await family_svc.milestone_trigger(family["owner"], on_platform)
+    assert hint and "Android app" in hint
+    assert await family_svc.milestone_trigger(family["owner"], off_platform) is None
+    assert await family_svc.milestone_is_open(family["owner"], on_platform.id) is True

tests/test_routes_family.py

@@ -0,0 +1,29 @@
+"""Structural tests for the family blueprint (milestone 463 step 3) — every
+endpoint is routed, and every write hands the service its caller, where the
+note's write gate lives. What the writes do is in
+tests/test_integration_family_promotion.py."""
+import inspect
+
+
+def test_family_blueprint_routes_every_endpoint():
+    from scribe.app import create_app
+    app = create_app()
+    assert "family" in app.blueprints
+    rules = {(r.rule, m) for r in app.url_map.iter_rules() for m in r.methods}
+    for rule, method in (
+        ("/api/family/ideas", "GET"),
+        ("/api/family/ideas/<int:note_id>", "GET"),
+        ("/api/family/ideas/<int:note_id>/propose", "POST"),
+        ("/api/family/ideas/<int:note_id>/promote", "POST"),
+        ("/api/family/ideas/<int:note_id>/retire", "POST"),
+        ("/api/family/decisions", "GET"),
+        ("/api/family/decisions/<int:decision_id>/undo", "POST"),
+    ):
+        assert (rule, method) in rules, f"{method} {rule} is not routed"
+
+
+def test_every_engine_write_takes_the_caller_and_who_decided():
+    from scribe.services import family as svc
+    for name in ("propose", "promote", "retire", "undo"):
+        params = inspect.signature(getattr(svc, name)).parameters
+        assert "user_id" in params and "decided_via" in params, name