feat(family): the shape ledger judges against family ideas, across languages (milestone 463 step 5, #4991)

- classify_shapes takes idea_id: the shape is judged against the idea's
  reference in its language, or its first.
- A shape classified against a canon idea's reference moves the project's
  adoption row (instance -> adopted, variant -> variant). Withdrawing the
  shapes that gave an answer returns it to unassessed. This runs on
  classify_shapes, the sweep, confirm_proposals and the coverage refresh.
- assess and undo refuse an answer the shapes contradict. A conflict
  resolution re-judges the variant shapes on both sides.
- The proposer's family arm offers a canon idea's reference on a shared
  platform, in any language. It proposes only when the reference is the top
  hit over every readable snippet and scores at least 0.70. The pairs this
  was measured on are recorded beside _FAMILY_FLOOR. The proposer is now v6.
- The matrix cell shows the code that answers it.

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

tests/test_divergence_meaning_gate.py

@@ -64,14 +64,14 @@ def _patch(mock: AsyncMock):
 
 async def test_a_hit_returns_the_canon() -> None:
     with _patch(_hits((0.88, CANON))):
-        assert await _semantic_canon(1, BODY, {CANON}) == (CANON, 0.88)
+        assert await _semantic_canon(1, BODY, {CANON}) == (CANON, 0.88, "semantic")
 
 
 async def test_an_allowed_canon_below_the_top_hit_still_wins() -> None:
     """The scan is over the whole result set, so a disallowed snippet ranking
     first does not hide an allowed one behind it."""
     with _patch(_hits((0.95, OTHER), (0.83, CANON))):
-        assert await _semantic_canon(1, BODY, {CANON}) == (CANON, 0.83)
+        assert await _semantic_canon(1, BODY, {CANON}) == (CANON, 0.83, "semantic")
 
 
 async def test_a_miss_is_none_and_nothing_more() -> None:

tests/test_family_shapes.py

@@ -0,0 +1,142 @@
+"""The shape ledger judging against family ideas, without a database
+(milestone 463 step 5).
+
+Pinned here: what a project's shapes say about an idea, when an answer
+contradicts them, which reference a shape in a given language is judged
+against, and the family arm's rule — measured on 2026-10-06 (#4991), and
+replayed below as the table it was chosen from. The two ledgers moving
+together against Postgres are in tests/test_integration_family_adoption.py.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from scribe.services.family_adoption import (
+    _contradiction, path_speaks, shape_outcome, shapes_say,
+)
+from scribe.services.shape_ledger import _FAMILY_FLOOR, _semantic_canon, pick_semantic
+from tests.helpers import tool_doc
+
+
+def _shape(status: str, symbol: str = "f", reason: str = "", snippet_id: int = 7):
+    return SimpleNamespace(id=1, path=f"src/{symbol}.py", symbol=symbol, status=status,
+                           reason=reason, snippet_id=snippet_id)
+
+
+# --- what the shapes say ---------------------------------------------------------
+
+@pytest.mark.parametrize("statuses,expected", [
+    (["instance"], "adopted"),
+    (["canonical"], "adopted"),
+    (["variant", "instance"], "adopted"),
+    (["variant"], "variant"),
+    ([], None),
+])
+def test_shape_outcome(statuses, expected):
+    assert shape_outcome(statuses) == expected
+
+
+def test_no_shape_is_silence_not_owed():
+    assert shapes_say([]) is None
+
+
+def test_an_adopted_reason_is_fixed_text_so_another_instance_logs_nothing():
+    one = shapes_say([_shape("instance", "a")])
+    two = shapes_say([_shape("instance", "a"), _shape("instance", "b")])
+    assert one["reason"] == two["reason"]
+    assert two["evidence"] == ["src/a.py::a", "src/b.py::b"]
+
+
+def test_a_variant_carries_the_shapes_why_and_only_variants_are_its_evidence():
+    said = shapes_say([_shape("variant", "k", reason="a kiosk installs it")])
+    assert said["outcome"] == "variant" and "a kiosk installs it" in said["reason"]
+    mixed = shapes_say([_shape("variant", "k", reason="x"), _shape("instance", "i")])
+    assert mixed["outcome"] == "adopted" and mixed["evidence"] == ["src/i.py::i"]
+
+
+# --- the two ledgers never disagree -----------------------------------------------
+
+def test_an_answer_the_shapes_contradict_names_them_and_the_way_out():
+    said = shapes_say([_shape("instance", "a")])
+    refused = _contradiction("owed", said, idea_id=5)
+    assert "src/a.py::a" in refused and "classify_shapes" in refused and "#5" in refused
+
+
+@pytest.mark.parametrize("outcome,said", [
+    ("adopted", shapes_say([_shape("instance")])),
+    ("owed", None),
+    ("exempt", None),
+])
+def test_an_agreeing_answer_or_silent_shapes_refuse_nothing(outcome, said):
+    assert _contradiction(outcome, said, idea_id=5) is None
+
+
+# --- which reference a shape is judged against --------------------------------------
+
+@pytest.mark.parametrize("path,language,expected", [
+    ("tools/release.py", "python", True),
+    ("app/Updater.kt", "kotlin", True),
+    ("cmd/main.go", "go", True),
+    ("cmd/main.go", "kotlin", False),
+    ("web/App.vue", "vue", True),
+    ("README", "python", False),
+    ("x.py", "", False),
+])
+def test_path_speaks(path, language, expected):
+    assert path_speaks(path, language) is expected
+
+
+# --- the family arm, and the measurement it was chosen from -----------------------
+
+OWN, REF, OTHER = 1, 2, 3
+
+
+@pytest.mark.parametrize("name,hits,expected", [
+    # #4991's pairs, the reference as REF and the best other hit as OTHER.
+    ("A py->go", [(0.728, REF), (0.703, OTHER)], "family"),
+    ("B py->go", [(0.715, REF), (0.675, OTHER)], "family"),
+    ("D kt->kt", [(0.816, REF), (0.734, OTHER)], "family"),
+    ("E go->go", [(0.610, REF), (0.604, OTHER)], None),
+    # No reference exists: the best hit is unrelated, the reference trails.
+    ("N4", [(0.715, OTHER), (0.706, REF)], None),
+    ("N1", [(0.666, OTHER)], None),
+])
+def test_the_family_arm_replays_its_measurement(name, hits, expected):
+    found = pick_semantic(hits, allowed=set(), family={REF}, floor=0.68)
+    assert (found[2] if found else None) == expected, name
+
+
+def test_a_family_reference_must_be_the_top_hit_not_merely_the_first_allowed():
+    assert pick_semantic([(0.80, OTHER), (0.79, REF)], set(), {REF}, floor=0.68) is None
+
+
+def test_the_own_project_arm_still_looks_past_disallowed_hits():
+    assert pick_semantic([(0.80, OTHER), (0.70, OWN)], {OWN}, {REF}, floor=0.68) == (
+        OWN, 0.7, "semantic")
+
+
+def test_the_family_floor_sits_between_the_bands_it_was_measured_on():
+    assert 0.610 < _FAMILY_FLOOR <= 0.715
+
+
+async def test_a_project_on_a_shared_platform_is_searched_with_no_own_canon():
+    """The own-project allowed set is often empty (the language gate); the
+    family references alone are reason to search."""
+    body = "def throttle(key: str) -> bool:\n    return attempts[key] < LIMIT and not locked(key)\n"
+    hit = SimpleNamespace(id=REF)
+    with patch("scribe.services.embeddings.semantic_search_notes",
+               AsyncMock(return_value=[(0.72, hit)])):
+        assert await _semantic_canon(1, body, set(), {REF}) == (REF, 0.72, "family")
+
+
+# --- the agent-facing contract -----------------------------------------------------
+
+def test_the_tools_teach_idea_id_and_the_family_basis():
+    classify = tool_doc("scribe.mcp.tools.shapes", "classify_shapes")
+    assert "idea_id" in classify and "shapes contradict" in classify
+    assert "family" in tool_doc("scribe.mcp.tools.shapes", "list_shapes")
+    assert '"family"' in tool_doc("scribe.mcp.tools.shapes", "confirm_shape_proposals")
+    assert "classify_shapes" in tool_doc("scribe.mcp.tools.family", "assess_family_adoption")

tests/test_integration_family_adoption.py

@@ -387,3 +387,191 @@ async def test_the_matrix_shows_every_member_and_who_has_not_been_asked(family):
 async def test_the_matrix_hides_what_the_caller_cannot_read(family):
     matrix = await adoption_svc.adoption_matrix(family["outsider"])
     assert all(i["note_id"] != family["note"] for i in matrix["ideas"])
+
+
+# --- the shape ledger as evidence (step 5) -------------------------------------------
+
+REPO = "git.example/android-two"
+KOTLIN_BODY = (
+    "fun installUpdate(context: Context, apk: File) {\n"
+    "    val installer = context.packageManager.packageInstaller\n"
+    "    val session = installer.openSession(installer.createSession(params()))\n"
+    "    apk.inputStream().use { src -> session.openWrite(\"apk\", 0, apk.length()).use(src::copyTo) }\n"
+    "}\n"
+)
+B_SHAPES = [
+    ("app/src/main/kotlin/Updater.kt", "sym", "selfUpdate"),
+    ("tools/release.py", "sym", "push_update"),
+]
+
+
+async def _reference(f, *, language: str = "kotlin", name: str = "installUpdate") -> int:
+    from scribe.services import snippets as snippets_svc
+
+    snippet = await snippets_svc.create_snippet(
+        f["owner"], name=f"{name} ({language})", code=KOTLIN_BODY, language=language,
+        project_id=f["a"],
+    )
+    return int(snippet.id)
+
+
+@pytest_asyncio.fixture
+async def shapes(family):
+    """Android two's ledger has two live shapes, and the idea has a Kotlin
+    reference in android one."""
+    from scribe.services.shape_ledger import sync_repo_shapes
+
+    await sync_repo_shapes(family["b"], REPO, B_SHAPES, seen_marker="main")
+    family["kotlin"] = await _reference(family)
+    await adoption_svc.set_references(family["owner"], family["note"], [family["kotlin"]])
+    return family
+
+
+async def _classify(f, project: str, symbol: str, status: str, **item):
+    from scribe.services import shape_ledger
+
+    path = next(p for p, _, s in B_SHAPES if s == symbol)
+    return await shape_ledger.classify_shapes(
+        f["owner"], f[project], [{"path": path, "symbol": symbol, "status": status, **item}])
+
+
+async def test_a_shape_classified_against_the_idea_answers_it_adopted(shapes):
+    """Across languages: a Python shape is an instance of an idea whose only
+    reference is Kotlin, and the project's answer moves with it."""
+    out = await _classify(shapes, "b", "push_update", "instance", idea_id=shapes["note"])
+    assert out["classified"] == 1
+    assert [m["status"] for m in out["family"]] == ["adopted"]
+    row = await _row(shapes, "b")
+    assert (row.status, row.decided_via, row.canon_version) == ("adopted", "system", 1)
+    [decision] = await _decisions(shapes, "b")
+    assert decision.evidence["source"] == adoption_svc.SHAPE_SOURCE
+    assert decision.evidence["evidence"] == ["tools/release.py::push_update"]
+    assert decision.evidence["shapes"][0]["snippet_id"] == shapes["kotlin"]
+    # The matrix shows the code that answers the cell.
+    matrix = await adoption_svc.adoption_matrix(shapes["owner"], project_id=shapes["b"])
+    [cell] = [c for c in matrix["cells"] if c["idea_id"] == shapes["note"]]
+    assert cell["shapes"] == [
+        {"path": "tools/release.py", "symbol": "push_update", "status": "instance"}]
+    # A second instance agrees with the standing answer: nothing more is logged.
+    await _classify(shapes, "b", "selfUpdate", "instance", idea_id=shapes["note"])
+    assert len(await _decisions(shapes, "b")) == 1
+
+
+async def test_idea_id_prefers_the_reference_in_the_shapes_language(shapes):
+    from scribe.services.shape_ledger import live_rows
+
+    python_ref = await _reference(shapes, language="python", name="install_update")
+    await adoption_svc.set_references(shapes["owner"], shapes["note"],
+                                      [shapes["kotlin"], python_ref])
+    await _classify(shapes, "b", "push_update", "instance", idea_id=shapes["note"])
+    await _classify(shapes, "b", "selfUpdate", "instance", idea_id=shapes["note"])
+    by_symbol = {r.symbol: r.snippet_id for r in await live_rows(shapes["b"])}
+    assert by_symbol == {"push_update": python_ref, "selfUpdate": shapes["kotlin"]}
+
+
+async def test_an_idea_with_no_reference_is_refused_and_nothing_applies(family):
+    from scribe.services.shape_ledger import live_rows, sync_repo_shapes
+
+    await sync_repo_shapes(family["b"], REPO, B_SHAPES, seen_marker="main")
+    with pytest.raises(ValueError, match="no reference implementation"):
+        await _classify(family, "b", "push_update", "instance", idea_id=family["note"])
+    assert {r.status for r in await live_rows(family["b"])} == {"unclassified"}
+
+
+async def test_a_variant_shape_answers_variant_and_withdrawing_it_unassesses(shapes):
+    await _classify(shapes, "b", "selfUpdate", "variant", idea_id=shapes["note"],
+                    reason="installs through the device-owner API: it is a kiosk")
+    row = await _row(shapes, "b")
+    assert row.status == "variant" and "device-owner API" in row.reason
+    out = await _classify(shapes, "b", "selfUpdate", "unclassified")
+    assert [(m["idea_id"], m["status"]) for m in out["family"]] == [
+        (shapes["note"], "unassessed")]
+    row = await _row(shapes, "b")
+    assert (row.status, row.canon_version, row.decided_via) == ("unassessed", None, None)
+    assert (await _decisions(shapes, "b"))[-1].evidence["source"] == adoption_svc.SHAPE_SOURCE
+
+
+async def test_the_shapes_close_an_owed_task_as_done(shapes):
+    owed = await _assess(shapes, "b", "owed", reason="no in-place update yet")
+    await _classify(shapes, "b", "push_update", "instance", idea_id=shapes["note"])
+    assert (await _task(owed["owed_task"]["id"])).status == "done"
+
+
+async def test_an_answer_the_shapes_contradict_is_refused(shapes):
+    await _classify(shapes, "b", "push_update", "instance", idea_id=shapes["note"])
+    with pytest.raises(ValueError, match="never disagree"):
+        await _assess(shapes, "b", "owed", reason="the agent thinks otherwise")
+    # The agreeing answer is not refused.
+    await _assess(shapes, "b", "adopted", reason="release.py does it",
+                  evidence=["tools/release.py"])
+
+
+async def test_an_undo_that_would_contradict_the_shapes_is_refused(shapes):
+    await _assess(shapes, "b", "owed", reason="not yet")
+    await _classify(shapes, "b", "push_update", "instance", idea_id=shapes["note"])
+    latest = (await _decisions(shapes, "b"))[-1]
+    with pytest.raises(ValueError, match="never disagree"):
+        await family_svc.undo(shapes["owner"], latest.id, reason="go back")
+
+
+async def test_an_answer_given_on_other_evidence_survives_the_shapes_leaving(shapes):
+    """No shape is not the same as owed: withdrawing a shape only withdraws
+    an answer the shape ledger gave."""
+    await _assess(shapes, "b", "adopted", reason="CI proves it",
+                  evidence=["ci run 12"])
+    await _classify(shapes, "b", "push_update", "instance", idea_id=shapes["note"])
+    out = await _classify(shapes, "b", "push_update", "unclassified")
+    assert "family" not in out
+    assert (await _row(shapes, "b")).status == "adopted"
+
+
+async def test_nothing_is_answered_or_proposed_across_projects_sharing_no_platform(shapes):
+    """The Go service shares no platform with the idea: its shape judged
+    against the reference is plain canon use, and the proposer never offers
+    the reference to it."""
+    from scribe.services import shape_ledger
+
+    await shape_ledger.sync_repo_shapes(shapes["c"], "git.example/go", [
+        ("cmd/update.go", "sym", "PushUpdate")], seen_marker="main")
+    out = await shape_ledger.classify_shapes(shapes["owner"], shapes["c"], [
+        {"path": "cmd/update.go", "symbol": "PushUpdate", "status": "instance",
+         "snippet_id": shapes["kotlin"]}])
+    assert out["classified"] == 1 and "family" not in out
+    async with async_session() as s:
+        assert (await s.execute(select(FamilyAdoption).where(
+            FamilyAdoption.project_id == shapes["c"]))).first() is None
+    assert await adoption_svc.family_reference_ids(shapes["c"]) == set()
+    assert await adoption_svc.family_reference_ids(shapes["b"]) == {shapes["kotlin"]}
+
+
+async def test_the_proposer_offers_the_family_reference_as_the_top_hit(shapes):
+    from scribe.services import shape_ledger
+
+    body = ("def push_update(apk_path: str) -> None:\n"
+            "    session = installer.open_session(installer.create_session())\n"
+            "    session.write_stream('apk', open(apk_path, 'rb'))\n"
+            "    session.commit()\n")
+    defs = [("tools/release.py", "sym", "push_update", "def push_update(apk_path: str) -> None:",
+             "sha-push", body)]
+    ref = type("Hit", (), {"id": shapes["kotlin"]})()
+    with patch("scribe.services.embeddings.semantic_search_notes",
+               AsyncMock(return_value=[(0.73, ref)])):
+        await shape_ledger.propose_for_repo(shapes["owner"], shapes["b"], REPO, defs)
+    row = next(r for r in await shape_ledger.live_rows(shapes["b"]) if r.symbol == "push_update")
+    assert (row.proposed_snippet_id, row.proposal_basis) == (shapes["kotlin"], "family")
+    # Confirming it answers the idea.
+    out = await shape_ledger.confirm_proposals(shapes["owner"], shapes["b"], basis="family")
+    assert out["confirmed"] == 1 and out["family"][0]["status"] == "adopted"
+
+
+async def test_a_conflict_brings_the_losing_sides_variant_shapes_back_to_the_todo(shapes):
+    from scribe.services.shape_ledger import live_rows
+
+    await _classify(shapes, "b", "selfUpdate", "variant", idea_id=shapes["note"],
+                    reason="signs in CI with a throwaway key")
+    out = await _resolve(shapes, "covers_failure")
+    assert out["shapes_rejudged"] == {str(shapes["b"]): 1}
+    b = await _row(shapes, "b")
+    assert b.status == "owed"
+    row = next(r for r in await live_rows(shapes["b"]) if r.symbol == "selfUpdate")
+    assert row.status == "unclassified"