feat(ledger): write-path stamping — a pulled canon the session then instantiates lands as a hook instance row (#2791, milestone 294 step 5)

The prior-art hook now names the shapes being written (shapes=kind:name —
every definition in the payload, or the one enclosing an Edit found by
walking the file upward) and the server stamps them as instance rows when
the session PULLED a snippet inside PULL_WINDOW that the payload references
by symbol or that the semantic arm scored for this very payload.
classified_by=hook, evidence in reason; never overrides a judgment or a
canonical row, overridable by classify_shapes. Offered-but-unopened stamps
nothing. Pulled-and-already-seen snippets stay in the semantic query as
evidence without re-entering the deduped menu. A brand-new shape gets a
provisional row the next sync confirms or vanishes. Read-scoped keys get
the hint, never the stamp. Plugin 0.1.34.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_integration_shape_classify.py

@@ -219,3 +219,133 @@ async def test_sync_refiles_rows_whose_snippet_was_purged(seeded):
         ))).scalar_one()
     assert row.status == "unclassified"
     assert row.snippet_id is None
+
+
+# --- #2791: the write-path feed lands hook evidence as rows -------------------
+
+
+@pytest.mark.integration
+async def test_write_path_stamp_is_evidence_that_yields_to_judgment(seeded):
+    """Pulled + referenced → every named shape of the snippet's kind becomes
+    an instance row, classified_by=hook, carrying the evidence as reason. A
+    later agent judgment on one of them stands against a re-stamp; the hook
+    may only overwrite nobody's judgment or its own. The outsider stamps
+    nothing (write-gated like every other ledger write)."""
+    from datetime import datetime, timezone
+
+    from scribe.services.shape_ledger import stamp_write_path_instances
+
+    owner, other, pid, sid = (
+        seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
+    )
+    pulled = {sid: datetime.now(timezone.utc)}
+    code = "app = factory()\nreturn app\n"          # references the snippet's symbol
+
+    assert await stamp_write_path_instances(
+        other, pid, path="src/app.py", shapes=[("sym", "make_app")],
+        code=code, pulled=pulled,
+    ) == []
+
+    stamped = await stamp_write_path_instances(
+        owner, pid, path="src/app.py",
+        shapes=[("sym", "make_app"), ("sym", "Config"), ("css", "nope")],
+        code=code, pulled=pulled,
+    )
+    assert {s["symbol"] for s in stamped} == {"make_app", "Config"}   # css skipped: no css canon
+    rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
+    by_symbol = {r.symbol: r for r in rows}
+    assert by_symbol["make_app"].status == "instance"
+    assert by_symbol["make_app"].classified_by == "hook"
+    assert by_symbol["make_app"].reason == f"hook: pulled #{sid}; payload references `factory`"
+
+    # A judgment lands; the next stamp must leave it alone but may re-stamp
+    # its own earlier row.
+    await classify_shapes(owner, pid, [
+        {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
+         "reason": "the app factory is its own thing"},
+    ])
+    again = await stamp_write_path_instances(
+        owner, pid, path="src/app.py",
+        shapes=[("sym", "make_app"), ("sym", "Config")], code=code, pulled=pulled,
+    )
+    assert {s["symbol"] for s in again} == {"Config"}
+    rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
+    by_symbol = {r.symbol: r for r in rows}
+    assert by_symbol["make_app"].status == "exempt"
+    assert by_symbol["Config"].status == "instance"
+
+    # Neither pulled nor in play → nothing, even with shapes named.
+    assert await stamp_write_path_instances(
+        owner, pid, path="src/util.py", shapes=[("sym", "helper")],
+        code="print('unrelated')", pulled=pulled,
+    ) == []
+
+
+@pytest.mark.integration
+async def test_a_brand_new_shape_gets_a_provisional_row_the_sync_settles(seeded):
+    """The shape being written right now has no ledger row yet. With the
+    hook's repo key it gets a provisional one — seen markers unset — so the
+    stamp survives until the next sync, which confirms it (sets the marker)
+    or stamps it vanished. Without a repo key only existing rows are touched."""
+    from datetime import datetime, timezone
+
+    from scribe.services.shape_ledger import stamp_write_path_instances
+
+    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
+    pulled = {sid: datetime.now(timezone.utc)}
+    code = "def build():\n    return factory()\n"
+
+    assert await stamp_write_path_instances(
+        owner, pid, path="src/new.py", shapes=[("sym", "build")],
+        code=code, pulled=pulled,                              # no repo_key
+    ) == []
+    stamped = await stamp_write_path_instances(
+        owner, pid, path="src/new.py", shapes=[("sym", "build")],
+        code=code, pulled=pulled, repo_key=REPO,
+    )
+    assert [s["symbol"] for s in stamped] == ["build"]
+    async with async_session() as s:
+        row = (await s.execute(select(CodeShape).where(
+            CodeShape.project_id == pid, CodeShape.path == "src/new.py",
+        ))).scalar_one()
+        assert row.status == "instance" and row.classified_by == "hook"
+        assert row.first_seen_commit is None and row.last_seen_commit is None
+
+    # The sync sees the shape in the tree → confirmed, stamp intact.
+    await sync_repo_shapes(
+        pid, REPO, SHAPES + [("src/new.py", "sym", "build")], seen_marker="abc123",
+    )
+    rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
+    assert rows[0].status == "instance" and rows[0].last_seen_commit == "abc123"
+
+    # The sync no longer sees it → vanished, out of the live accounting.
+    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="def456")
+    rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
+    assert rows == []
+    rows, _ = await list_project_shapes(owner, pid, path="src/new.py", include_vanished=True)
+    assert rows[0].vanished_at is not None
+
+
+@pytest.mark.integration
+async def test_recent_pulls_reads_the_usage_stream(seeded):
+    """The "actually pulled it" half is the PULLED usage event, inside the
+    window; a surfacing alone is not a pull."""
+    from datetime import datetime, timedelta, timezone
+
+    from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
+    from scribe.services.shape_ledger import recent_pulls
+
+    owner, sid = seeded["owner"], seeded["snippet"]
+    now = datetime.now(timezone.utc)
+    async with async_session() as s:
+        s.add_all([
+            NoteUsageEvent(user_id=owner, note_id=sid, event=PULLED, source="mcp_get_snippet"),
+            NoteUsageEvent(user_id=owner, note_id=sid + 1000, event=SURFACED, source="auto_inject"),
+            NoteUsageEvent(user_id=owner, note_id=sid + 2000, event=PULLED,
+                           source="mcp_get_snippet", created_at=now - timedelta(days=2)),
+        ])
+        await s.commit()
+    pulls = await recent_pulls(owner)
+    assert sid in pulls
+    assert sid + 1000 not in pulls
+    assert sid + 2000 not in pulls