feat(usage): count what happened away from a record's own project — the readout #3735 needs

Milestone 385 step 8 (#3735 "a lesson is recalled on a project it was not
written on") is defined by opened-on-another-project. The reader's project has
been recorded on every usage event since 0c8e109, but nothing compared it with
the record's own, so the criterion was still unreadable.

- note_usage.usage_for_notes: a second aggregate in the same session joins
  notes and counts surfaced_away_count / pulled_away_count. Counted only where
  both projects are known and differ; ranked surfacings only (#2477). The
  first aggregate is untouched, so events on deleted notes still count.
- empty_usage carries both keys zero-filled; every door that attaches usage
  (list_lessons, get_lesson, snippets, knowledge) gets them through
  attach_usage.
- UsageBadge tooltip says "On other projects: surfaced N×, opened M×" when it
  happened, and nothing when it did not.
- Tests: the mocked split test feeds both aggregates; a unit test for the
  away counters; a real-Postgres test that home, unreported and ambient
  events are all left out.

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

frontend/src/components/UsageBadge.vue

@@ -42,9 +42,15 @@ const title = () => {
     ? `Last opened ${new Date(u.last_pulled_at).toLocaleDateString()}.`
     : "Never opened.";
   const verdict = isDeadWeight() ? ` ${props.deadWeightAdvice}` : "";
+  // Said only when it happened: "0× elsewhere" on every row would read as a
+  // finding about records that simply have not been near another project.
+  const awaySurfaced = u.surfaced_away_count ?? 0;
+  const away = awaySurfaced
+    ? ` On other projects: surfaced ${awaySurfaced}×, opened ${u.pulled_away_count ?? 0}×.`
+    : "";
   return (
     `Surfaced to an agent ${u.surfaced_count}×, opened in full ` +
-    `${u.pull_count}×. ${last}${verdict}`
+    `${u.pull_count}×.${away} ${last}${verdict}`
   );
 };
 </script>

frontend/src/types/usage.ts

@@ -20,4 +20,9 @@ export interface RecordUsage {
   pull_count: number;
   last_surfaced_at: string | null;
   last_pulled_at: string | null;
+  /** The subsets that happened on a project other than the record's own —
+   *  the evidence it transferred (milestone 385). Notes and snippets carry
+   *  them; rules are global, so their readout does not. */
+  surfaced_away_count?: number;
+  pulled_away_count?: number;
 }

src/scribe/services/note_usage.py

@@ -35,6 +35,7 @@ from collections.abc import Sequence
 from sqlalchemy import case, func, select
 
 from scribe.models import async_session
+from scribe.models.note import Note
 from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
 from scribe.models.base import iso
 from scribe.services.background import report_telemetry_failure
@@ -193,6 +194,13 @@ def empty_usage() -> dict:
     record. `ambient_count` is the rest (see AMBIENT_SOURCES). The split is the
     readout half of #2477: the "high surfaced, zero pulls → dead weight"
     reading is only valid over surfacings that were choices.
+
+    `surfaced_away_count` / `pulled_away_count` are the subsets that happened
+    on a project other than the one the record was written on — the evidence
+    that a record TRANSFERRED, which is the claim the lesson kind rests on
+    (milestone 385, #3735). Counted only where both projects are known: an
+    event with no reader project, or a record with no project of its own,
+    cannot speak to "away" and is left out rather than guessed.
     """
     return {
         "surfaced_count": 0,
@@ -200,6 +208,8 @@ def empty_usage() -> dict:
         "pull_count": 0,
         "last_surfaced_at": None,
         "last_pulled_at": None,
+        "surfaced_away_count": 0,
+        "pulled_away_count": 0,
     }
 
 
@@ -247,6 +257,30 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
                     )
                 )
             ).all()
+            # Away from home: the reader's project against the record's own.
+            # A second aggregate rather than a column on the first, because
+            # it needs the join to `notes` and the first must keep counting
+            # events whose note has since been deleted (the table is FK-free
+            # so that evidence outlives the row). Ranked surfacings only, for
+            # the same reason surfaced_count is (#2477).
+            away_rows = (
+                await session.execute(
+                    select(
+                        NoteUsageEvent.note_id,
+                        NoteUsageEvent.event,
+                        func.count().label("n"),
+                    )
+                    .join(Note, Note.id == NoteUsageEvent.note_id)
+                    .where(
+                        NoteUsageEvent.note_id.in_(ids),
+                        NoteUsageEvent.project_id.is_not(None),
+                        Note.project_id.is_not(None),
+                        NoteUsageEvent.project_id != Note.project_id,
+                        NoteUsageEvent.source.not_in(AMBIENT_SOURCES),
+                    )
+                    .group_by(NoteUsageEvent.note_id, NoteUsageEvent.event)
+                )
+            ).all()
     except Exception:
         # A telemetry readout must not be able to break the list it decorates —
         # but it must say it failed, or a broken readout is indistinguishable
@@ -271,6 +305,14 @@ async def usage_for_notes(note_ids: list[int]) -> dict[int, dict]:
             latest = iso(last_at)
             if latest and (slot["last_pulled_at"] or "") < latest:
                 slot["last_pulled_at"] = latest
+    for note_id, event, n in away_rows:
+        slot = out.get(int(note_id))
+        if slot is None:
+            continue
+        if event == SURFACED:
+            slot["surfaced_away_count"] = int(n)
+        elif event == PULLED:
+            slot["pulled_away_count"] = int(n)
     return out
 
 

tests/test_note_usage.py

@@ -98,9 +98,12 @@ async def test_usage_for_notes_splits_counts_by_event():
         (3, "pulled", 2, ts, False),
     ]
     session = MagicMock()
-    session.execute = AsyncMock(
-        return_value=MagicMock(all=MagicMock(return_value=rows))
-    )
+    # Two aggregates in one session: every event, then the away-from-home
+    # subset (note_id, event, count) — none here.
+    session.execute = AsyncMock(side_effect=[
+        MagicMock(all=MagicMock(return_value=rows)),
+        MagicMock(all=MagicMock(return_value=[])),
+    ])
     ctx = MagicMock()
     ctx.__aenter__ = AsyncMock(return_value=session)
     ctx.__aexit__ = AsyncMock(return_value=False)
@@ -115,6 +118,36 @@ async def test_usage_for_notes_splits_counts_by_event():
     assert out[3]["last_pulled_at"] == ts.isoformat()
 
 
+async def test_usage_for_notes_counts_what_happened_away_from_home():
+    """Surfaced or opened on a project other than the record's own is the
+    evidence a record transferred (#3735). It is a SUBSET of the totals, read
+    from the second aggregate, and it must land on the right counter."""
+    from datetime import datetime, timezone
+
+    ts = datetime(2026, 10, 1, tzinfo=timezone.utc)
+    session = MagicMock()
+    session.execute = AsyncMock(side_effect=[
+        MagicMock(all=MagicMock(return_value=[
+            (3, "surfaced", 9, ts, False),
+            (3, "pulled", 4, ts, False),
+            (4, "surfaced", 2, ts, False),
+        ])),
+        MagicMock(all=MagicMock(return_value=[
+            (3, "surfaced", 5),
+            (3, "pulled", 1),
+        ])),
+    ])
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=session)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    with patch.object(note_usage, "async_session", return_value=ctx):
+        out = await usage_for_notes([3, 4])
+    assert (out[3]["surfaced_away_count"], out[3]["pulled_away_count"]) == (5, 1)
+    assert (out[3]["surfaced_count"], out[3]["pull_count"]) == (9, 4)
+    # Used only at home: the away counters stay zero, not missing.
+    assert (out[4]["surfaced_away_count"], out[4]["pulled_away_count"]) == (0, 0)
+
+
 async def test_usage_readout_failure_degrades_to_zeroes():
     """A telemetry readout must not be able to break the list it decorates."""
     with patch.object(note_usage, "async_session", side_effect=RuntimeError("boom")):
@@ -303,3 +336,53 @@ async def test_record_pulled_lands_end_to_end_from_a_running_loop(_dispose_engin
         assert out[nid]["pull_count"] == 1
     finally:
         await _purge(nid)
+
+
+@pytest.mark.integration
+async def test_away_counts_compare_the_readers_project_with_the_records(_dispose_engine):
+    """The real join: an event counts as away only when the reader's project is
+    known and differs from the record's own. Home, unreported and ambient
+    events are all left out — each would otherwise read as transfer."""
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.note import Note
+    from scribe.models.project import Project
+    from scribe.services.note_usage import _insert_events
+    from tests.helpers import ensure_user
+
+    async with async_session() as s:
+        owner = await ensure_user(s, "usage_away_owner")
+        home = Project(user_id=owner.id, title="usage home")
+        away = Project(user_id=owner.id, title="usage away")
+        s.add_all([home, away])
+        await s.flush()
+        lesson = Note(user_id=owner.id, project_id=home.id, title="a lesson",
+                      note_type="lesson")
+        s.add(lesson)
+        await s.flush()
+        nid, home_id, away_id, uid = lesson.id, home.id, away.id, owner.id
+        await s.commit()
+    try:
+        def ev(event, source, pid):
+            return {"user_id": uid, "note_id": nid, "event": event,
+                    "source": source, "project_id": pid}
+        await _insert_events([
+            ev("surfaced", "lesson_slot", away_id),     # away: counts
+            ev("surfaced", "lesson_slot", home_id),     # home
+            ev("surfaced", "lesson_slot", None),        # unreported
+            ev("surfaced", "enter_project", away_id),   # ambient
+            ev("pulled", "mcp_get_lesson", away_id),    # away: counts
+            ev("pulled", "mcp_get_lesson", None),       # unreported
+        ])
+        out = await usage_for_notes([nid])
+        assert out[nid]["surfaced_away_count"] == 1
+        assert out[nid]["pulled_away_count"] == 1
+        assert out[nid]["surfaced_count"] == 3
+        assert out[nid]["pull_count"] == 2
+    finally:
+        await _purge(nid)
+        async with async_session() as s:
+            await s.execute(delete(Note).where(Note.id == nid))
+            await s.execute(delete(Project).where(Project.id.in_([home_id, away_id])))
+            await s.commit()