diff --git a/alembic/versions/0109_embeddings_record_their_model.py b/alembic/versions/0109_embeddings_record_their_model.py
new file mode 100644
index 0000000..334b298
--- /dev/null
+++ b/alembic/versions/0109_embeddings_record_their_model.py
@@ -0,0 +1,42 @@
+"""embeddings_record_their_model — a vector says whose space it lives in (#4132)
+
+Revision ID: 0109
+Revises: 0108
+Create Date: 2026-09-23
+
+Every embedding table stamps `chunker_version`, so a change to the document
+shape is caught per row and re-embedded. None stamped the MODEL. The column is
+`vector(384)` — a width, not an identity — so swapping bge-small for any other
+384-dim model would write a second geometry beside the first with no error,
+and search would go on ranking by cosines between the two, which mean nothing.
+
+`embedding_model` is the other half of `calibration_stamp()`, stored per row on
+all four tables. The startup backfill now re-embeds on either half moving.
+
+THE BACKFILL LITERAL IS SAFE BECAUSE NO INSTALL HAS EVER CHANGED MODELS. The
+name is hardcoded in `services/embeddings.py`, and every vector ever written was
+written by it — so stamping every existing row with that name states a fact,
+not a guess. It is frozen here rather than imported: a migration records what
+was true when it ran, and a later model change must not rewrite history.
+"""
+from alembic import op
+
+revision = "0109"
+down_revision = "0108"
+branch_labels = None
+depends_on = None
+
+_TABLES = ("note_embeddings", "rule_embeddings", "milestone_embeddings", "system_embeddings")
+_MODEL = "BAAI/bge-small-en-v1.5"
+
+
+def upgrade() -> None:
+    for table in _TABLES:
+        op.execute(f"ALTER TABLE {table} ADD COLUMN embedding_model text")
+        op.execute(f"UPDATE {table} SET embedding_model = '{_MODEL}'")
+        op.execute(f"ALTER TABLE {table} ALTER COLUMN embedding_model SET NOT NULL")
+
+
+def downgrade() -> None:
+    for table in _TABLES:
+        op.execute(f"ALTER TABLE {table} DROP COLUMN embedding_model")
diff --git a/alembic/versions/0110_task_claims.py b/alembic/versions/0110_task_claims.py
new file mode 100644
index 0000000..3b95078
--- /dev/null
+++ b/alembic/versions/0110_task_claims.py
@@ -0,0 +1,38 @@
+"""task_claims — a task can say a session is working it (milestone 381 step 2)
+
+Revision ID: 0110
+Revises: 0109
+Create Date: 2026-09-23
+
+`status` is durable and nothing clears it, so `in_progress` cannot also mean
+"someone is on this now". The claim is that second meaning, stored as WHO and
+WHEN rather than as a flag, so it can be read as dead without anything having
+cleared it (`services/task_claims.py` has the semantics).
+
+Four nullable columns, no backfill: no existing row has ever been claimed, and
+inventing a claim for one would assert attention nobody observed. No CHECK
+enum is touched.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0110"
+down_revision = "0109"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("notes", sa.Column(
+        "claimed_by", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True,
+    ))
+    op.add_column("notes", sa.Column("claimed_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column("notes", sa.Column("claim_touched_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column("notes", sa.Column("claim_session", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("notes", "claim_session")
+    op.drop_column("notes", "claim_touched_at")
+    op.drop_column("notes", "claimed_at")
+    op.drop_column("notes", "claimed_by")
diff --git a/frontend/src/assets/sweep-shared.css b/frontend/src/assets/sweep-shared.css
new file mode 100644
index 0000000..a2f81df
--- /dev/null
+++ b/frontend/src/assets/sweep-shared.css
@@ -0,0 +1,137 @@
+/* The sweep-row visual language (#3207).
+ *
+ * A "sweep" is a pane that lists records in the order they most need a human
+ * — the note verification sweep, the rule sweep, preference drift — as one
+ * raised row per record: an italic title that opens it, an age or date pushed
+ * right, a small grid of facts, and a row of plain buttons. It was written
+ * out four times, scoped, before this sheet existed.
+ *
+ * Imported UNSCOPED, once, from main.ts, so every
+ * class is prefixed `sweep-`: `.row-title`, `.lede`, `.age` and `.actions`
+ * also exist, scoped, in views that have nothing to do with sweeps, and an
+ * unprefixed global would leak into all of them. What a pane does
+ * differently stays in its own scoped block — spacing remainders, a `code`
+ * style — rather than growing a modifier here per pane.
+ */
+
+/* The pane itself: header, filters, rows and footnote stacked. */
+.sweep { display: flex; flex-direction: column; gap: var(--fs-space-3); }
+
+.sweep-lede {
+  margin: 0;
+  max-width: 62ch;
+  font-size: var(--fs-size-body-sm);
+  color: var(--fs-text-secondary);
+  line-height: var(--fs-leading-body);
+}
+
+.sweep-filters {
+  display: flex;
+  gap: var(--fs-space-5);
+  align-items: center;
+  flex-wrap: wrap;
+}
+.sweep-filter {
+  display: flex;
+  align-items: center;
+  gap: var(--fs-space-2);
+  font-size: 0.82rem;
+  color: var(--fs-text-secondary);
+}
+.sweep-filter input[type="checkbox"] { accent-color: var(--fs-accent); }
+
+.sweep-state {
+  margin: 0;
+  font-size: var(--fs-size-body-sm);
+  color: var(--fs-text-secondary);
+}
+.sweep-state.empty { color: var(--fs-text-tertiary); }
+
+.sweep-rows {
+  list-style: none;
+  margin: 0;
+  padding: 0;
+  display: flex;
+  flex-direction: column;
+  gap: var(--fs-space-3);
+}
+.sweep-row {
+  background: var(--fs-surface-raised);
+  border-radius: var(--fs-radius-md);
+  padding: var(--fs-space-3);
+}
+.sweep-row-head {
+  display: flex;
+  align-items: baseline;
+  gap: var(--fs-space-2);
+  flex-wrap: wrap;
+}
+.sweep-row-title {
+  background: none;
+  border: none;
+  padding: 0;
+  cursor: pointer;
+  font-family: Fraunces, serif;
+  font-style: italic;
+  font-size: 1.02rem;
+  color: var(--fs-text-primary);
+  text-align: left;
+}
+.sweep-row-title:hover { text-decoration: underline; }
+
+/* The ORDER carries urgency — the top of a sweep is the thing that most needs
+   a look. No red/amber ramp: it would restate the ordering and force an
+   invented "stale after N days" threshold. "Never" is marked because it is
+   categorically DIFFERENT from a date, not a worse one. */
+.sweep-age {
+  margin-left: auto;
+  font-size: 0.78rem;
+  color: var(--fs-text-secondary);
+  font-variant-numeric: tabular-nums;
+}
+.sweep-age.unchecked { font-style: italic; color: var(--fs-text-tertiary); }
+
+.sweep-check {
+  display: grid;
+  grid-template-columns: auto 1fr;
+  gap: 0.15rem var(--fs-space-3);
+  margin: var(--fs-space-3) 0 0;
+}
+.sweep-check dt {
+  font-size: 0.7rem;
+  text-transform: uppercase;
+  letter-spacing: 0.05em;
+  color: var(--fs-text-tertiary);
+}
+.sweep-check dd {
+  margin: 0;
+  font-size: 0.82rem;
+  color: var(--fs-text-primary);
+  min-width: 0;
+}
+
+.sweep-actions {
+  display: flex;
+  gap: var(--fs-space-2);
+  margin-top: var(--fs-space-3);
+}
+.sweep-actions button {
+  cursor: pointer;
+  font: inherit;
+  font-size: 0.78rem;
+  background: var(--fs-surface-page);
+  color: var(--fs-text-primary);
+  border: 1px solid var(--fs-border-color);
+  border-radius: var(--fs-radius-md);
+  padding: 0.25rem 0.6rem;
+}
+.sweep-actions button:hover:not(:disabled) { background: var(--fs-surface-hover); }
+.sweep-actions button:disabled { opacity: var(--fs-disabled-opacity); cursor: default; }
+
+.sweep-footnote {
+  margin: 0;
+  max-width: 62ch;
+  font-size: 0.78rem;
+  color: var(--fs-text-tertiary);
+  line-height: 1.45;
+}
diff --git a/frontend/src/components/NoteSweepPane.vue b/frontend/src/components/NoteSweepPane.vue
index cd18eb9..f7d4d2a 100644
--- a/frontend/src/components/NoteSweepPane.vue
+++ b/frontend/src/components/NoteSweepPane.vue
@@ -83,42 +83,42 @@ defineExpose({ reload });
   <section class="sweep">
     <header>
       <h2>Due for verification</h2>
-      <p class="lede">
+      <p class="sweep-lede">
         Notes that assert a fact about something outside your control — what a
         service does, how a tool behaves. Most notes are decisions and never
         appear here; they have no truth value to go stale.
       </p>
     </header>
 
-    <div class="filters">
-      <label class="filter">
+    <div class="sweep-filters">
+      <label class="sweep-filter">
         <input v-model="neverOnly" type="checkbox" @change="reload" />
         <span>Never checked only</span>
       </label>
     </div>
 
-    <p v-if="loading" class="state">Loading…</p>
+    <p v-if="loading" class="sweep-state">Loading…</p>
 
     <!-- An empty sweep is GOOD NEWS and must not read like a broken page. -->
-    <p v-else-if="!rows.length" class="state empty">
+    <p v-else-if="!rows.length" class="sweep-state empty">
       Nothing to check.
       {{ neverOnly
         ? "Every note that carries a check has been confirmed at least once."
         : "No note carries a check yet — add one to a note that asserts a fact." }}
     </p>
 
-    <ol v-else class="rows">
-      <li v-for="n in rows" :key="n.id" class="row">
-        <div class="row-head">
-          <button class="row-title" @click="emit('open-note', n.id)">{{ n.title }}</button>
-          <span class="age" :class="{ unchecked: n.days_since_verified === null }">
+    <ol v-else class="sweep-rows">
+      <li v-for="n in rows" :key="n.id" class="sweep-row">
+        <div class="sweep-row-head">
+          <button class="sweep-row-title" @click="emit('open-note', n.id)">{{ n.title }}</button>
+          <span class="sweep-age" :class="{ unchecked: n.days_since_verified === null }">
             {{ n.days_since_verified === null
               ? "never checked"
               : `${n.days_since_verified}d ago` }}
           </span>
         </div>
 
-        <dl class="check">
+        <dl class="sweep-check">
           <dt>Check</dt>
           <dd>{{ n.verify_with }}</dd>
           <template v-if="n.expires_when">
@@ -127,14 +127,14 @@ defineExpose({ reload });
           </template>
         </dl>
 
-        <div class="actions">
+        <div class="sweep-actions">
           <button :disabled="busyId === n.id" @click="verify(n.id, true)">Still true</button>
           <button :disabled="busyId === n.id" @click="verify(n.id, false)">No longer true</button>
         </div>
       </li>
     </ol>
 
-    <p v-if="rows.length" class="footnote">
+    <p v-if="rows.length" class="sweep-footnote">
       Record a result only after actually running the check. “No longer true” stores nothing
       on purpose — the note is wrong rather than in a state worth recording, so it keeps its
       place here until you correct it, supersede it, or remove its check.
@@ -143,56 +143,8 @@ defineExpose({ reload });
 </template>
 
 <style scoped>
-.sweep { display: flex; flex-direction: column; gap: var(--fs-space-3); }
 h2 { margin: 0; font-size: 1.05rem; }
-.lede {
-  margin: 0.35rem 0 0;
-  max-width: 62ch;
-  font-size: 0.85rem;
-  color: var(--fs-text-secondary);
-  line-height: 1.5;
-}
-
-.filters { display: flex; gap: var(--fs-space-5); align-items: center; flex-wrap: wrap; }
-.filter { display: flex; align-items: center; gap: var(--fs-space-2); font-size: 0.82rem; color: var(--fs-text-secondary); }
-.filter input[type="checkbox"] { accent-color: var(--fs-accent); }
-
-.state { margin: 0; font-size: 0.9rem; color: var(--fs-text-secondary); }
-.state.empty { color: var(--fs-text-tertiary); }
-
-.rows { list-style: none; margin: 0; padding: 0; display: flex; flex-direction: column; gap: var(--fs-space-3); }
-.row {
-  background: var(--fs-surface-raised);
-  border-radius: var(--fs-radius-md);
-  padding: var(--fs-space-3);
-}
-.row-head { display: flex; align-items: baseline; gap: var(--fs-space-2); flex-wrap: wrap; }
-.row-title {
-  background: none; border: none; padding: 0; cursor: pointer;
-  font-family: Fraunces, serif; font-style: italic; font-size: 1.02rem;
-  color: var(--fs-text-primary); text-align: left;
-}
-.row-title:hover { text-decoration: underline; }
-/* The ORDER carries urgency — the top of this list is the least-confirmed
-   thing in the corpus. No red/amber ramp: it would restate the ordering and
-   force an invented "stale after N days" threshold. "Never" is marked because
-   it is categorically DIFFERENT from a date, not a worse one. */
-.age { margin-left: auto; font-size: 0.78rem; color: var(--fs-text-secondary); font-variant-numeric: tabular-nums; }
-.age.unchecked { font-style: italic; color: var(--fs-text-tertiary); }
-
-.check { display: grid; grid-template-columns: auto 1fr; gap: 0.15rem var(--fs-space-3); margin: var(--fs-space-3) 0 0; }
-.check dt { font-size: 0.7rem; text-transform: uppercase; letter-spacing: 0.05em; color: var(--fs-text-tertiary); }
-.check dd { margin: 0; font-size: 0.82rem; color: var(--fs-text-primary); min-width: 0; overflow-wrap: anywhere; }
-
-.actions { display: flex; gap: var(--fs-space-2); margin-top: var(--fs-space-3); }
-.actions button {
-  cursor: pointer; font: inherit; font-size: 0.78rem;
-  background: var(--fs-surface-page); color: var(--fs-text-primary);
-  border: 1px solid var(--fs-border-color); border-radius: var(--fs-radius-md);
-  padding: 0.25rem 0.6rem;
-}
-.actions button:hover:not(:disabled) { background: var(--fs-surface-hover); }
-.actions button:disabled { opacity: var(--fs-disabled-opacity); cursor: default; }
-
-.footnote { margin: 0; max-width: 62ch; font-size: 0.78rem; color: var(--fs-text-tertiary); line-height: 1.45; }
+.sweep-lede { margin-top: 0.35rem; }
+/* A check can be a long URL or command with no break point. */
+.sweep-check dd { overflow-wrap: anywhere; }
 </style>
diff --git a/frontend/src/components/rules/PreferenceDriftPane.vue b/frontend/src/components/rules/PreferenceDriftPane.vue
index 187e0cf..897c234 100644
--- a/frontend/src/components/rules/PreferenceDriftPane.vue
+++ b/frontend/src/components/rules/PreferenceDriftPane.vue
@@ -77,7 +77,7 @@ onMounted(() => store.fetchDrift());
   <section class="pane drift">
     <header>
       <h2>Recent changes</h2>
-      <p class="lede">
+      <p class="sweep-lede">
         Preferences Scribe rewrote while working, most recently changed first. A
         preference is how you want work done, so sessions keep it current
         without asking — this is where you see what they decided. Rules are not
@@ -85,18 +85,18 @@ onMounted(() => store.fetchDrift());
       </p>
     </header>
 
-    <p v-if="store.loading" class="state">Loading…</p>
+    <p v-if="store.loading" class="sweep-state">Loading…</p>
 
     <!-- Nothing changed is the ordinary state and must not read as a fault. -->
-    <p v-else-if="!store.drift.length" class="state empty">
+    <p v-else-if="!store.drift.length" class="sweep-state empty">
       Nothing has been rewritten. A preference appears here the first time a
       session changes one — until then there is nothing to review.
     </p>
 
-    <ol v-else class="rows">
-      <li v-for="row in store.drift" :key="row.rule.id" class="row">
-        <div class="row-head">
-          <button class="row-title" @click="emit('open-rule', row.rule.id)">
+    <ol v-else class="sweep-rows">
+      <li v-for="row in store.drift" :key="row.rule.id" class="sweep-row">
+        <div class="sweep-row-head">
+          <button class="sweep-row-title" @click="emit('open-rule', row.rule.id)">
             {{ row.rule.title }}
           </button>
           <span class="when">{{ stamp(row.previous.created_at) }}</span>
@@ -122,10 +122,10 @@ onMounted(() => store.fetchDrift());
             <span class="was">Was:</span> {{ row.previous.when_to_apply || "nothing" }}
           </p>
           <DiffView v-if="diffFor(row).length" :diff="diffFor(row)" />
-          <p v-else class="state">
+          <p v-else class="sweep-state">
             The statement is unchanged — this edit moved another field.
           </p>
-          <div class="actions">
+          <div class="sweep-actions">
             <button
               :disabled="busyId === row.rule.id"
               @click="restore(row)"
@@ -150,27 +150,6 @@ onMounted(() => store.fetchDrift());
 <style src="@/assets/rules-shared.css" />
 <style scoped>
 .drift { display: flex; flex-direction: column; gap: var(--fs-space-3); }
-.lede {
-  margin: 0; max-width: 62ch; font-size: var(--fs-size-body-sm);
-  color: var(--fs-text-secondary); line-height: var(--fs-leading-body);
-}
-
-.state { margin: 0; font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); }
-.state.empty { color: var(--fs-text-tertiary); }
-
-.rows { list-style: none; margin: 0; padding: 0; display: flex; flex-direction: column; gap: var(--fs-space-3); }
-.row {
-  background: var(--fs-surface-raised);
-  border-radius: var(--fs-radius-md);
-  padding: var(--fs-space-3);
-}
-.row-head { display: flex; align-items: baseline; gap: var(--fs-space-2); flex-wrap: wrap; }
-.row-title {
-  background: none; border: none; padding: 0; cursor: pointer;
-  font-family: Fraunces, serif; font-style: italic; font-size: 1.02rem;
-  color: var(--fs-text-primary); text-align: left;
-}
-.row-title:hover { text-decoration: underline; }
 .when {
   margin-left: auto; font-size: var(--fs-size-tiny);
   color: var(--fs-text-secondary); font-variant-numeric: tabular-nums;
@@ -205,19 +184,13 @@ onMounted(() => store.fetchDrift());
 }
 .was { color: var(--fs-text-tertiary); }
 
-.actions { display: flex; align-items: baseline; gap: var(--fs-space-3); flex-wrap: wrap; }
-.actions button {
-  cursor: pointer; font: inherit; font-size: 0.78rem;
-  background: var(--fs-surface-page); color: var(--fs-text-primary);
-  border: 1px solid var(--fs-border-color); border-radius: var(--fs-radius-md);
-  padding: 0.25rem 0.6rem;
-}
-.actions button:hover:not(:disabled) { background: var(--fs-surface-hover); }
-.actions button:disabled { opacity: var(--fs-disabled-opacity); cursor: default; }
 .actions-note {
   flex: 1; min-width: 18ch; font-size: var(--fs-size-tiny);
   color: var(--fs-text-tertiary); line-height: var(--fs-leading-body);
 }
 
 .footnote { margin: 0; max-width: 62ch; font-size: var(--fs-size-tiny); color: var(--fs-text-tertiary); line-height: var(--fs-leading-body); }
+/* Drift's buttons sit beside a note inside the expanded detail, not under a
+   row of facts. */
+.sweep-actions { align-items: baseline; gap: var(--fs-space-3); flex-wrap: wrap; margin-top: 0; }
 </style>
diff --git a/frontend/src/components/rules/RuleHistoryPanel.vue b/frontend/src/components/rules/RuleHistoryPanel.vue
index a7592f8..156379b 100644
--- a/frontend/src/components/rules/RuleHistoryPanel.vue
+++ b/frontend/src/components/rules/RuleHistoryPanel.vue
@@ -144,10 +144,10 @@ watch(() => props.ruleId, () => { selected.value = null; load(); });
     </button>
 
     <div v-if="expanded" class="body">
-      <p v-if="loading" class="state">Loading…</p>
+      <p v-if="loading" class="sweep-state">Loading…</p>
 
       <!-- Never reworded is the ordinary case, and must not read as a fault. -->
-      <p v-else-if="!versions.length" class="state empty">
+      <p v-else-if="!versions.length" class="sweep-state empty">
         This rule has never been reworded. Nothing was recorded before the history
         existed, so an older rule starts empty too.
       </p>
@@ -172,7 +172,7 @@ watch(() => props.ruleId, () => { selected.value = null; load(); });
             </button>
 
             <div v-if="selected?.id === v.id" class="detail">
-              <p v-if="loadingDetail" class="state">Loading…</p>
+              <p v-if="loadingDetail" class="sweep-state">Loading…</p>
               <template v-else>
                 <p v-if="checkChanged(i)" class="warn">
                   This edit changed the rule's check, which cleared its verification
@@ -188,7 +188,7 @@ watch(() => props.ruleId, () => { selected.value = null; load(); });
                 </dl>
                 <h4>Statement</h4>
                 <DiffView v-if="diff.length" :diff="diff" />
-                <p v-else class="state">The statement did not change in this edit.</p>
+                <p v-else class="sweep-state">The statement did not change in this edit.</p>
               </template>
             </div>
           </li>
@@ -213,8 +213,6 @@ watch(() => props.ruleId, () => { selected.value = null; load(); });
 }
 
 .body { margin-top: var(--fs-space-3); display: flex; flex-direction: column; gap: var(--fs-space-3); }
-.state { margin: 0; font-size: var(--fs-size-body-sm); color: var(--fs-text-secondary); }
-.state.empty { color: var(--fs-text-tertiary); }
 .lede {
   margin: 0; max-width: 62ch; font-size: var(--fs-size-tiny);
   color: var(--fs-text-tertiary); line-height: var(--fs-leading-body);
diff --git a/frontend/src/components/rules/RuleSweepPane.vue b/frontend/src/components/rules/RuleSweepPane.vue
index 1610db7..beae189 100644
--- a/frontend/src/components/rules/RuleSweepPane.vue
+++ b/frontend/src/components/rules/RuleSweepPane.vue
@@ -39,37 +39,37 @@ onMounted(reload);
   <section class="pane sweep">
     <header>
       <h2>Due for verification</h2>
-      <p class="lede">
+      <p class="sweep-lede">
         Rules that assert a fact about something outside your control. Most rules are
         decisions and never appear here — they have no truth value to go stale.
       </p>
     </header>
 
-    <div class="filters">
-      <label class="filter">
+    <div class="sweep-filters">
+      <label class="sweep-filter">
         <input v-model="neverOnly" type="checkbox" @change="reload" />
         <span>Never checked only</span>
       </label>
     </div>
 
-    <p v-if="store.loading" class="state">Loading…</p>
+    <p v-if="store.loading" class="sweep-state">Loading…</p>
 
     <!-- An empty sweep is GOOD NEWS, and must not read like a broken page. -->
-    <p v-else-if="!store.rulesDue.length" class="state empty">
+    <p v-else-if="!store.rulesDue.length" class="sweep-state empty">
       Nothing to check.
       {{ neverOnly ? "No rule matches these filters." : "No rule carries a check yet — add one to a rule that asserts a fact." }}
     </p>
 
-    <ol v-else class="rows">
-      <li v-for="r in store.rulesDue" :key="r.id" class="row">
-        <div class="row-head">
-          <button class="row-title" @click="emit('open-rule', r.id)">{{ r.title }}</button>
+    <ol v-else class="sweep-rows">
+      <li v-for="r in store.rulesDue" :key="r.id" class="sweep-row">
+        <div class="sweep-row-head">
+          <button class="sweep-row-title" @click="emit('open-rule', r.id)">{{ r.title }}</button>
           <span
             v-if="!r.when_to_apply"
             class="rule-chip rule-chip-inert"
             title="No trigger, so nothing can retrieve it — this rule will never reach a session"
           >never surfaces</span>
-          <span class="age" :class="{ unchecked: r.days_since_verified === null }">
+          <span class="sweep-age" :class="{ unchecked: r.days_since_verified === null }">
             {{ r.days_since_verified === null
               ? "never checked"
               : `${r.days_since_verified}d ago` }}
@@ -78,7 +78,7 @@ onMounted(reload);
 
         <p class="statement">{{ r.statement }}</p>
 
-        <dl class="check">
+        <dl class="sweep-check">
           <dt>Check</dt>
           <dd><code>{{ r.verify_with }}</code></dd>
           <template v-if="r.expires_when">
@@ -87,14 +87,14 @@ onMounted(reload);
           </template>
         </dl>
 
-        <div class="actions">
+        <div class="sweep-actions">
           <button :disabled="busyId === r.id" @click="verify(r.id, true)">Still true</button>
           <button :disabled="busyId === r.id" @click="verify(r.id, false)">No longer true</button>
         </div>
       </li>
     </ol>
 
-    <p v-if="store.rulesDue.length" class="footnote">
+    <p v-if="store.rulesDue.length" class="sweep-footnote">
       Record a result only after actually running the check. “No longer true” stores nothing
       on purpose — the rule is wrong rather than in a state worth recording, so it keeps its
       place here until you correct or retire it.
@@ -104,70 +104,22 @@ onMounted(reload);
 
 <style src="@/assets/rules-shared.css" />
 <style scoped>
-.sweep { display: flex; flex-direction: column; gap: var(--fs-space-3); }
-.lede {
-  margin: 0;
-  max-width: 62ch;
-  font-size: 0.85rem;
-  color: var(--fs-text-secondary);
-  line-height: 1.5;
-}
-
-.filters { display: flex; gap: var(--fs-space-5); align-items: center; flex-wrap: wrap; }
-.filter { display: flex; align-items: center; gap: var(--fs-space-2); font-size: 0.82rem; color: var(--fs-text-secondary); }
-.filter input[type="checkbox"] { accent-color: var(--fs-accent); }
-.filter select {
-  font: inherit; font-size: 0.82rem;
-  background: var(--fs-surface-page); color: var(--fs-text-primary);
-  border: 1px solid var(--fs-border-color); border-radius: var(--fs-radius-md);
-  padding: 0.2rem 0.4rem;
-}
-
-.state { margin: 0; font-size: 0.9rem; color: var(--fs-text-secondary); }
-.state.empty { color: var(--fs-text-tertiary); }
-
-.rows { list-style: none; margin: 0; padding: 0; display: flex; flex-direction: column; gap: var(--fs-space-3); }
-.row {
-  background: var(--fs-surface-raised);
-  border-radius: var(--fs-radius-md);
-  padding: var(--fs-space-3);
-}
-.row-head { display: flex; align-items: baseline; gap: var(--fs-space-2); flex-wrap: wrap; }
-.row-title {
-  background: none; border: none; padding: 0; cursor: pointer;
-  font-family: Fraunces, serif; font-style: italic; font-size: 1.02rem;
-  color: var(--fs-text-primary); text-align: left;
-}
-.row-title:hover { text-decoration: underline; }
-/* The ORDER carries urgency — the top of this list is the most overdue thing
-   in the rulebook. No red/amber ramp: it would restate the ordering and force
-   an invented "stale after N days" threshold. "Never" is marked because it is
-   categorically different from a date, not a worse one. */
-.age { margin-left: auto; font-size: 0.78rem; color: var(--fs-text-secondary); font-variant-numeric: tabular-nums; }
-.age.unchecked { font-style: italic; color: var(--fs-text-tertiary); }
-
 .statement { margin: 0.35rem 0 0; font-size: 0.88rem; color: var(--fs-text-secondary); }
 
-.check { display: grid; grid-template-columns: auto 1fr; gap: 0.15rem var(--fs-space-3); margin: var(--fs-space-3) 0 0; }
-.check dt { font-size: 0.7rem; text-transform: uppercase; letter-spacing: 0.05em; color: var(--fs-text-tertiary); }
-.check dd { margin: 0; font-size: 0.82rem; color: var(--fs-text-primary); min-width: 0; }
-.check code {
+.sweep-filter select {
+  font: inherit;
+  font-size: 0.82rem;
+  background: var(--fs-surface-page);
+  color: var(--fs-text-primary);
+  border: 1px solid var(--fs-border-color);
+  border-radius: var(--fs-radius-md);
+  padding: 0.2rem 0.4rem;
+}
+.sweep-check code {
   font-family: var(--fs-font-mono);
   background: var(--fs-surface-code-inline);
   border-radius: var(--fs-radius-sm);
   padding: 0.05rem 0.3rem;
   overflow-wrap: anywhere;
 }
-
-.actions { display: flex; gap: var(--fs-space-2); margin-top: var(--fs-space-3); }
-.actions button {
-  cursor: pointer; font: inherit; font-size: 0.78rem;
-  background: var(--fs-surface-page); color: var(--fs-text-primary);
-  border: 1px solid var(--fs-border-color); border-radius: var(--fs-radius-md);
-  padding: 0.25rem 0.6rem;
-}
-.actions button:hover:not(:disabled) { background: var(--fs-surface-hover); }
-.actions button:disabled { opacity: var(--fs-disabled-opacity); cursor: default; }
-
-.footnote { margin: 0; max-width: 62ch; font-size: 0.78rem; color: var(--fs-text-tertiary); line-height: 1.45; }
 </style>
diff --git a/frontend/src/main.ts b/frontend/src/main.ts
index 865c447..a0831bb 100644
--- a/frontend/src/main.ts
+++ b/frontend/src/main.ts
@@ -6,6 +6,9 @@ import "./assets/theme.css";
 // After theme.css — it consumes the tokens declared there.
 import "./assets/components.css";
 import "./assets/prose.css";
+// Global rather than `<style src>` per pane: plugin-vue cannot compile one
+// src stylesheet shared by several SFCs (#3207 broke the image build).
+import "./assets/sweep-shared.css";
 
 const app = createApp(App);
 app.use(createPinia());
diff --git a/frontend/src/types/note.ts b/frontend/src/types/note.ts
index 2850ea3..5bcc33b 100644
--- a/frontend/src/types/note.ts
+++ b/frontend/src/types/note.ts
@@ -29,6 +29,10 @@ export interface Note {
   due_date: string | null;
   started_at: string | null;
   completed_at: string | null;
+  // Which session is working this task now (milestone 381). Null when no
+  // session ever claimed it; `live` false when the claim's lease ran out —
+  // a session that went quiet mid-task, which is worth seeing, not hiding.
+  claim?: TaskClaim | null;
   recurrence_rule: Record<string, unknown> | null;
   recurrence_next_spawn_at: string | null;
   is_task: boolean;
@@ -53,3 +57,11 @@ export interface NoteListResponse {
   notes: Note[];
   total: number;
 }
+
+export interface TaskClaim {
+  held_by: number | null;
+  session: string | null;
+  since: string | null;
+  touched: string | null;
+  live: boolean;
+}
diff --git a/frontend/src/views/TaskEditorView.vue b/frontend/src/views/TaskEditorView.vue
index 0f0ca64..c984a73 100644
--- a/frontend/src/views/TaskEditorView.vue
+++ b/frontend/src/views/TaskEditorView.vue
@@ -15,7 +15,8 @@ import type { TaskStatus, TaskPriority } from "@/types/task";
 import type { TaskKind } from "@/types/note";
 import { useSystemsStore } from "@/stores/systems";
 import type { System } from "@/api/systems";
-import type { Note } from "@/types/note";
+import type { Note, TaskClaim } from "@/types/note";
+import { relativeTime } from "@/composables/useRelativeTime";
 import type { Editor } from "@tiptap/vue-3";
 import MarkdownToolbar from "@/components/MarkdownToolbar.vue";
 import TiptapEditor from "@/components/TiptapEditor.vue";
@@ -54,6 +55,7 @@ const parentId = ref<number | null>(null);
 const parentTitle = ref("");
 const startedAt = ref<string | null>(null);
 const completedAt = ref<string | null>(null);
+const claim = ref<TaskClaim | null>(null);
 const recurrenceRule = ref<Record<string, unknown> | null>(null);
 const parentSearchQuery = ref("");
 const parentSearchResults = ref<{ id: number; title: string }[]>([]);
@@ -318,6 +320,7 @@ onMounted(async () => {
       const noteTask = store.currentTask as unknown as Note;
       startedAt.value = noteTask.started_at ?? null;
       completedAt.value = noteTask.completed_at ?? null;
+      claim.value = noteTask.claim ?? null;
       recurrenceRule.value = noteTask.recurrence_rule ?? null;
       savedTitle = title.value;
       savedBody = body.value;
@@ -592,7 +595,21 @@ useEditorGuards(dirty, save);
               <option v-if="kind === 'plan'" value="plan">Plan (legacy)</option>
             </select>
           </div>
-          <div v-if="startedAt || completedAt" class="sb-timestamps">
+          <div v-if="startedAt || completedAt || claim" class="sb-timestamps">
+            <div v-if="claim && claim.live" class="sb-timestamp">
+              <span class="sb-ts-label">Being worked</span>
+              <span class="sb-ts-value">
+                by a session{{ claim.session ? ` (${claim.session.slice(0, 8)})` : "" }},
+                last active {{ claim.touched ? relativeTime(claim.touched) : "recently" }}
+              </span>
+            </div>
+            <div v-else-if="claim" class="sb-timestamp">
+              <span class="sb-ts-label">Went quiet</span>
+              <span class="sb-ts-value">
+                the session working this stopped
+                {{ claim.touched ? relativeTime(claim.touched) : "" }} without finishing it
+              </span>
+            </div>
             <div v-if="startedAt" class="sb-timestamp">
               <span class="sb-ts-label">Started</span>
               <span class="sb-ts-value">{{ new Date(startedAt).toLocaleString() }}</span>
diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json
index 4beef23..86ab450 100644
--- a/plugin/.claude-plugin/plugin.json
+++ b/plugin/.claude-plugin/plugin.json
@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
-  "version": "2026.09.23.2127",
+  "version": "2026.09.24.1042",
   "author": {
     "name": "Bryan Van Deusen"
   },
diff --git a/plugin/hooks/hooks.json b/plugin/hooks/hooks.json
index 8d1ffcd..5c6b7ef 100644
--- a/plugin/hooks/hooks.json
+++ b/plugin/hooks/hooks.json
@@ -71,6 +71,15 @@
             "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_record_outcome.sh\""
           }
         ]
+      },
+      {
+        "matcher": "mcp__.*__(update_task|add_task_log)",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_claim_session.sh\""
+          }
+        ]
       }
     ],
     "PreCompact": [
@@ -92,6 +101,16 @@
           }
         ]
       }
+    ],
+    "SessionEnd": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_session_end.sh\""
+          }
+        ]
+      }
     ]
   }
 }
diff --git a/plugin/hooks/scribe_claim_session.sh b/plugin/hooks/scribe_claim_session.sh
new file mode 100755
index 0000000..c765a56
--- /dev/null
+++ b/plugin/hooks/scribe_claim_session.sh
@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+# Scribe — say WHICH session holds a task's claim (milestone 381 step 2).
+#
+# The server stamps a claim on the write that is the work — a task reaching
+# `in_progress`, a work log on an open task — so every MCP client gets a claim
+# that dies on its own once its lease runs out. What the server cannot know is
+# which SESSION made the write: an MCP caller is a user and nothing more.
+#
+# The harness knows. This PostToolUse hook watches `update_task` and
+# `add_task_log` and reports the event's `session_id` against the task the
+# call named, so the claim can say "this session" rather than "someone,
+# recently". Same evidence class as scribe_record_opened.sh: a tool call
+# happened and the harness reported it; nothing here asks the model anything.
+#
+# It cannot CREATE a claim. The server binds the session only to a live claim
+# the caller already holds, so a call that closed the task, or one on a task
+# someone else is working, binds nothing.
+#
+# EXIT 0 AND SILENT, ALWAYS. This decorates a record; a PostToolUse hook that
+# spoke would put a line after every task write, and a failure here must never
+# turn a successful tool call into a hook error.
+set -uo pipefail
+
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
+command -v curl >/dev/null 2>&1 || exit 0
+scribe_config || exit 0
+
+event=$(cat 2>/dev/null || true)
+[ -n "$event" ] || exit 0
+
+event_flat=$(printf '%s' "$event" | scribe_json_flat)
+session_id=$(scribe_json_pick "$event_flat" '.session_id')
+[ -n "$session_id" ] || exit 0
+
+task_id=$(scribe_json_pick "$event_flat" '.tool_input.task_id')
+task_id=$(printf '%s' "$task_id" | tr -cd '0-9')
+[ -n "$task_id" ] || exit 0
+
+sid_enc=$(printf '%s' "$session_id" | scribe_urlenc) || exit 0
+curl -fsS --max-time 4 \
+  -H "Authorization: Bearer ${token}" \
+  "${url%/}/api/plugin/claim-session?task_id=${task_id}&session_id=${sid_enc}" \
+  >/dev/null 2>&1 || true
+exit 0
diff --git a/plugin/hooks/scribe_session_context.sh b/plugin/hooks/scribe_session_context.sh
index d269c4d..d672e31 100755
--- a/plugin/hooks/scribe_session_context.sh
+++ b/plugin/hooks/scribe_session_context.sh
@@ -187,8 +187,14 @@ if [ -n "$url" ] && [ -n "$token" ] && command -v curl >/dev/null 2>&1; then
   marker_why=${marker_read#*$'\t'}
   repo=$(git -C "$repo_dir" remote get-url origin 2>/dev/null || true)
   scope=$(scribe_scope_query "$repo_dir")
-  q=""
-  [ -n "$scope" ] && q="?${scope}"
+  # The source and the session id decide what the claim section says
+  # (milestone 381): a compaction gets back the work this session had claimed,
+  # with its latest logs; a new session hears about other sessions' claims.
+  sid_now=$(scribe_json_pick "$event_flat" '.session_id')
+  q="source=$(printf '%s' "$source" | scribe_urlenc)"
+  [ -n "$sid_now" ] && q="${q}&session_id=$(printf '%s' "$sid_now" | scribe_urlenc)"
+  [ -n "$scope" ] && q="${q}&${scope}"
+  q="?${q}"
   # ONE FETCH, NAMED WHEN IT FAILS (#4366). This used to be `curl -f … ||
   # body=""`, which folded a timeout, an HTTP error and a refused key into one
   # sentence — so a session that started blind could not say why, and neither
diff --git a/plugin/hooks/scribe_session_end.sh b/plugin/hooks/scribe_session_end.sh
new file mode 100755
index 0000000..db21695
--- /dev/null
+++ b/plugin/hooks/scribe_session_end.sh
@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+# Scribe — release this session's task claims as it ends (milestone 381 step 4).
+#
+# A claim records a session's attention, and a session that ends has none left
+# to give. This is the mechanical half of the hand-off: it needs nothing from
+# the model, only the session id the harness reports. The other half — writing
+# down where the work stands — only the model can do, and it is stated as a
+# practice in the skill and the static context, not here.
+#
+# A TIDY-UP, NOT THE GUARANTEE. SessionEnd does not fire on a crash, a killed
+# terminal or a dropped connection, and those are exactly the cases the claim
+# was designed around. The lease is what makes a dead session's claim read as
+# dead; this only keeps the ordinary exit from leaving a claim to run out.
+#
+# NOT ON /clear. A clear ends one conversation and starts the next in the same
+# terminal, and SessionStart(source=clear) pushes back the work this session
+# had claimed. Releasing here would leave that push with nothing to say. The
+# next write moves the claim wherever the work actually continues.
+#
+# EXIT 0 AND SILENT, ALWAYS. Nobody reads a SessionEnd hook's output, and the
+# session is ending whether this succeeds or not.
+set -uo pipefail
+
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
+command -v curl >/dev/null 2>&1 || exit 0
+scribe_config || exit 0
+
+event=$(cat 2>/dev/null || true)
+[ -n "$event" ] || exit 0
+
+event_flat=$(printf '%s' "$event" | scribe_json_flat)
+reason=$(scribe_json_pick "$event_flat" '.reason')
+[ "$reason" = "clear" ] && exit 0
+
+session_id=$(scribe_json_pick "$event_flat" '.session_id')
+[ -n "$session_id" ] || exit 0
+
+sid_enc=$(printf '%s' "$session_id" | scribe_urlenc) || exit 0
+curl -fsS --max-time 4 \
+  -H "Authorization: Bearer ${token}" \
+  "${url%/}/api/plugin/release-session?session_id=${sid_enc}" \
+  >/dev/null 2>&1 || true
+exit 0
diff --git a/plugin/hooks/scribe_static_context.md b/plugin/hooks/scribe_static_context.md
index d20ddde..062af16 100644
--- a/plugin/hooks/scribe_static_context.md
+++ b/plugin/hooks/scribe_static_context.md
@@ -23,6 +23,9 @@ What only Claude Code needs said:
   long session, log it to Scribe, then tell the operator it's a good moment to
   `/compact` and name what you logged. You can't run it yourself; suggest it at
   seams, not every turn.
+- **When the operator wraps up, hand off first.** Log where each task you
+  worked stands before the session ends; the plugin releases your claims at
+  session end, but only you can say what happened (using-scribe, "Hand off").
 - **Stored Processes arrive as skills** (`scribe-proc-*`), refreshed at session
   start. After a Process is added or edited, `/scribe:sync` makes it available
   straight away.
diff --git a/plugin/skills/using-scribe/SKILL.md b/plugin/skills/using-scribe/SKILL.md
index 984a30e..da75f16 100644
--- a/plugin/skills/using-scribe/SKILL.md
+++ b/plugin/skills/using-scribe/SKILL.md
@@ -175,6 +175,16 @@ Two constraints on *how* that's achieved:
    **complete** a task and when you **hit or discover a problem**, so a change
    of direction is on the record and not only the successes.
 
+   **Hand off before this session's context stops existing.** A compaction, a
+   `/clear`, the operator wrapping up for the day — each is the last moment the
+   reasoning behind the work lives anywhere but here. Log on the task you were
+   holding where it stands, what you tried and ruled out, and the next move:
+   write down what the next session needs, because it arrives with Scribe's
+   record and nothing else. Moving a task to `in_progress` or logging on it also
+   claims it for this session — that claim is what hands the work back to you
+   after a compaction, and it ends on its own when you stop, so there is nothing
+   to release by hand.
+
 6. **Fixes are issues, not work-logs.** When you fix a problem — even one solved
    in passing — record it as its own issue (`create_task(kind="issue")`) with
    symptom → root cause → fix, optionally linked to the task it arose from
diff --git a/scripts/check_plugin.py b/scripts/check_plugin.py
index b65f48a..33bac05 100755
--- a/scripts/check_plugin.py
+++ b/scripts/check_plugin.py
@@ -386,6 +386,21 @@ SMOKE_EVENTS: dict[str, str] = {
          "tool_input": {"rule_id": 1, "outcome": "applied"},
          "tool_response": {}}
     ),
+    # The claim-session binder (milestone 381). Silent like the ledgers above —
+    # a PostToolUse hook that spoke would add a line after every task write —
+    # and with no instance configured it must exit before reaching for one.
+    "scribe_claim_session.sh": json.dumps(
+        {"session_id": "smoke", "cwd": ".",
+         "tool_name": "mcp__scribe__add_task_log",
+         "tool_input": {"task_id": 1, "content": "smoke"},
+         "tool_response": {}}
+    ),
+    # The SessionEnd claim release (milestone 381 step 4). Silent: nobody reads
+    # a SessionEnd hook's output, and with no instance it must exit first.
+    "scribe_session_end.sh": json.dumps(
+        {"session_id": "smoke", "cwd": ".", "hook_event_name": "SessionEnd",
+         "reason": "prompt_input_exit"}
+    ),
     # The shared library is sourced, never run; executed bare it defines
     # functions and exits — silent by construction.
     "scribe_defs.sh": "",
diff --git a/src/scribe/mcp/tools/notes.py b/src/scribe/mcp/tools/notes.py
index 158cbb0..c7942d9 100644
--- a/src/scribe/mcp/tools/notes.py
+++ b/src/scribe/mcp/tools/notes.py
@@ -183,11 +183,14 @@ async def create_note(
             "When Forgejo issues run numbers per workflow rather than per
             repository", not "in six months". Constraints expire when the
             ground moves, not on a schedule.
-        force: Bypass the near-duplicate gate. By default, if a title- or
-            meaning-similar note already exists in the same project, creation is
-            BLOCKED and the existing note's id is returned so you update it
-            instead (no duplicate bloat / no stale RAG copies). Set true only
-            when you're sure this is a genuinely distinct note.
+        force: Bypass the near-duplicate gate. By default, a note with the
+            same title, or one that reads as a copy, in the same project BLOCKS
+            the create and its id is returned so you update it instead. A note
+            that reads CLOSE but not identical does not block: the note is
+            created and the reply carries `overlaps` — open the top one and
+            judge it. Same thing: fold into it and delete the new one. A
+            sibling (the next dev-log, another part of one design): keep both.
+            Set force true only when a blocked note is genuinely distinct.
 
     AN ID EXISTS ONLY ONCE A CREATE RETURNS IT. A body citing a `#N` that has
     not been assigned yet is refused — every session and user draws from one
@@ -203,10 +206,11 @@ async def create_note(
     """
     uid = current_user_id()
     await refuse_guessed_ids(title, body)
+    overlaps: list = []
     if not force:
         dup = await dedup_svc.find_duplicate_note(
             uid, title, body, project_id=project_id or None,
-            is_task=False, note_type="note",
+            is_task=False, note_type="note", overlaps=overlaps,
         )
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "note")
@@ -231,6 +235,7 @@ async def create_note(
     data = note.to_dict()
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
     await supersession_svc.attach_relations(uid, note.id, data, hint=True)
+    data.update(dedup_svc.note_overlap_response(overlaps, "note"))
     return data
 
 
diff --git a/src/scribe/mcp/tools/tasks.py b/src/scribe/mcp/tools/tasks.py
index e33e56e..98dd9ee 100644
--- a/src/scribe/mcp/tools/tasks.py
+++ b/src/scribe/mcp/tools/tasks.py
@@ -304,10 +304,14 @@ async def create_task(
         arose_from_id: For an issue, the id of the task/feature it arose from;
             for a spike, the record that raised the question — including a
             standing rule whose check just failed. 0 = none.
-        force: Bypass the near-duplicate gate. By default, if a title- or
-            meaning-similar task already exists in the same project, creation is
-            BLOCKED and the existing task's id is returned so you update it
-            instead. Set true only for a genuinely distinct task.
+        force: Bypass the near-duplicate gate. By default, a task with the
+            same title, or one that reads as a copy, in the same project BLOCKS
+            the create and its id is returned so you update it instead. A task
+            that reads CLOSE but not identical does not block: the task is
+            created and the reply carries `overlaps` — open the top one and
+            judge whether it is the same work (fold in, delete the new one) or
+            separate work (keep both). Set force true only when a blocked task
+            is genuinely distinct.
 
     AN ID EXISTS ONLY ONCE A CREATE RETURNS IT. Never write the id you expect
     a record to get: every session and user draws from one sequence, so the
@@ -332,10 +336,11 @@ async def create_task(
             "task with create_task(milestone_id=<that milestone>)."
         )
     await refuse_guessed_ids(title, body)
+    overlaps: list = []
     if not force:
         dup = await dedup_svc.find_duplicate_note(
             uid, title, body, project_id=project_id or None,
-            is_task=True, note_type="note",
+            is_task=True, note_type="note", overlaps=overlaps,
         )
         if dup is not None:
             return dedup_svc.duplicate_response(dup, "task")
@@ -356,6 +361,7 @@ async def create_task(
         await systems_svc.set_record_systems(uid, note.id, system_ids)
     data = note.to_dict()
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
+    data.update(dedup_svc.note_overlap_response(overlaps, "task"))
     return await placement_svc.attach_placement(uid, data, note)
 
 
@@ -477,6 +483,11 @@ async def add_task_log(task_id: int, content: str) -> dict:
     cannot get from the body or the diff is what you tried, what you ruled
     out, and where it actually stands.
 
+    A log on an open task also marks it as being worked by this session — its
+    `claim` (milestone 381). That is what brings the task and its newest
+    entries back to you after a compaction; it lapses by itself once the
+    session stops writing, and a status of done, cancelled or todo clears it.
+
     The response shows the task's `systems` — or, if the task is an untagged
     project record, the `systems_hint` question: logging work IS working in
     some area, so answer it (update_task with system_ids, or create_system
@@ -556,13 +567,23 @@ def _batch_items(records: list[dict], *, what: str = "record") -> list[batch_svc
     return items
 
 
-async def _first_duplicate(uid: int, items: list, project_id: int | None) -> dict | None:
-    """The duplicate gate over a whole batch — the first hit blocks all of it."""
+async def _first_duplicate(
+    uid: int, items: list, project_id: int | None,
+    overlaps: dict | None = None,
+) -> dict | None:
+    """The duplicate gate over a whole batch — the first hit blocks all of it.
+
+    `overlaps`, when given, collects each record's near matches below the copy
+    band by its 1-based position (#4306), for the reply once the batch is
+    created."""
     for i, item in enumerate(items, start=1):
+        found: list = []
         dup = await dedup_svc.find_duplicate_note(
             uid, item.title, item.body, project_id=project_id,
-            is_task=item.is_task, note_type="note",
+            is_task=item.is_task, note_type="note", overlaps=found,
         )
+        if found and overlaps is not None:
+            overlaps[i] = found
         if dup is not None:
             payload = dedup_svc.duplicate_response(dup, "task" if item.is_task else "note")
             payload["record"] = i
@@ -616,14 +637,17 @@ async def create_records(
     uid = current_user_id()
     items = _batch_items(records)
     await refuse_guessed_ids(*[t for item in items for t in (item.title, item.body)])
+    overlaps: dict = {}
     if not force:
-        dup = await _first_duplicate(uid, items, project_id or None)
+        dup = await _first_duplicate(uid, items, project_id or None, overlaps)
         if dup is not None:
             return dup
     _ms, notes = await batch_svc.create_batch(
         uid, items, project_id=project_id or None, milestone_id=milestone_id or None,
     )
-    return {"ids": [n.id for n in notes], "records": [n.to_dict() for n in notes]}
+    out = {"ids": [n.id for n in notes], "records": [n.to_dict() for n in notes]}
+    out.update(dedup_svc.batch_overlap_response(overlaps))
+    return out
 
 
 async def start_planning(
@@ -699,14 +723,18 @@ async def start_planning(
         )
         if match is not None:
             return match
+    overlaps: dict = {}
     if items and not force:
-        dup = await _first_duplicate(uid, items, project_id or None)
+        dup = await _first_duplicate(uid, items, project_id or None, overlaps)
         if dup is not None:
             return dup
-    return await planning_svc.start_planning(
+    result = await planning_svc.start_planning(
         user_id=uid, project_id=project_id, title=title,
         body=body or None, steps=items or None,
     )
+    if isinstance(result, dict):
+        result.update(dedup_svc.batch_overlap_response(overlaps))
+    return result
 
 
 async def delete_task(task_id: int) -> dict:
diff --git a/src/scribe/models/embedding.py b/src/scribe/models/embedding.py
index de5797c..4691a30 100644
--- a/src/scribe/models/embedding.py
+++ b/src/scribe/models/embedding.py
@@ -41,6 +41,14 @@ class NoteEmbedding(Base):
     # any note whose rows carry a stale version — shape changes become a
     # version bump instead of a table wipe.
     chunker_version: Mapped[int] = mapped_column(Integer, nullable=False)
+    # embeddings.EMBEDDING_MODEL at write time — the SPACE the vector lives in
+    # (#4132). The column is `vector(384)`, a width and not an identity, so a
+    # swap to another 384-dim model writes a second geometry beside the first
+    # with no error, and cosine across the two is a number that means nothing.
+    # The version above says what text was embedded; this says in whose
+    # geometry. Either one moving makes the row stale, and the backfill
+    # re-embeds on both.
+    embedding_model: Mapped[str] = mapped_column(Text, nullable=False)
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),
@@ -91,6 +99,7 @@ class RuleEmbedding(Base):
     # centroid (measured in note 2485).
     chunk_text: Mapped[str] = mapped_column(Text, nullable=False)
     chunker_version: Mapped[int] = mapped_column(Integer, nullable=False)
+    embedding_model: Mapped[str] = mapped_column(Text, nullable=False)  # see NoteEmbedding
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),
@@ -124,6 +133,7 @@ class MilestoneEmbedding(Base):
     embedding: Mapped[list] = mapped_column(Vector(EMBEDDING_DIM), nullable=False)
     chunk_text: Mapped[str] = mapped_column(Text, nullable=False)
     chunker_version: Mapped[int] = mapped_column(Integer, nullable=False)
+    embedding_model: Mapped[str] = mapped_column(Text, nullable=False)  # see NoteEmbedding
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),
@@ -163,6 +173,7 @@ class SystemEmbedding(Base):
     embedding: Mapped[list] = mapped_column(Vector(EMBEDDING_DIM), nullable=False)
     chunk_text: Mapped[str] = mapped_column(Text, nullable=False)
     chunker_version: Mapped[int] = mapped_column(Integer, nullable=False)
+    embedding_model: Mapped[str] = mapped_column(Text, nullable=False)  # see NoteEmbedding
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),
diff --git a/src/scribe/models/note.py b/src/scribe/models/note.py
index 0133276..58cda56 100644
--- a/src/scribe/models/note.py
+++ b/src/scribe/models/note.py
@@ -76,6 +76,16 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
     recurrence_next_spawn_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )
+    # The claim — which session is working this task NOW (milestone 381).
+    # Orthogonal to `status`: status is the work's state and durable, the claim
+    # is a session's attention and dies on read once `claim_touched_at` is past
+    # the lease. Semantics in services/task_claims.py.
+    claimed_by: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True
+    )
+    claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    claim_touched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    claim_session: Mapped[str | None] = mapped_column(Text, nullable=True)
     # WHAT KIND of record this is, on the note/entity axis. Task-ness is tracked
     # by `status`, not here (person/place/list entity types removed 2026-07):
     #   note    (default) — authored prose, findable by what it is ABOUT
@@ -184,6 +194,7 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
             "completed_at": iso(self.completed_at),
             "recurrence_rule": self.recurrence_rule,
             "recurrence_next_spawn_at": iso(self.recurrence_next_spawn_at),
+            "claim": _claim_state(self),
             "is_task": self.is_task,
             "note_type": self.note_type or "note",
             "task_kind": self.task_kind,
@@ -198,3 +209,9 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
             "created_at": iso(self.created_at),
             "updated_at": iso(self.updated_at),
         }
+
+
+def _claim_state(note: "Note") -> dict | None:
+    from scribe.services.task_claims import claim_state
+
+    return claim_state(note)
diff --git a/src/scribe/routes/plugin.py b/src/scribe/routes/plugin.py
index 5768d32..a9cd7de 100644
--- a/src/scribe/routes/plugin.py
+++ b/src/scribe/routes/plugin.py
@@ -15,6 +15,7 @@ from scribe.config import Config
 from scribe.services import plugin_context as plugin_ctx_svc
 from scribe.services import repo_bindings as repo_bindings_svc
 from scribe.services import report_check as report_check_svc
+from scribe.services import task_claims as task_claims_svc
 from scribe.services.settings import get_admin_setting, set_setting
 
 plugin_bp = Blueprint("plugin", __name__, url_prefix="/api/plugin")
@@ -69,10 +70,16 @@ async def session_context():
             send it when a `.scribe` marker file names a project. Takes
             precedence over `repo`. Access-checked like any other read — an id
             this account cannot read loads no project rather than failing.
+        source (optional str) — the host's SessionStart source (startup,
+            resume, compact, clear, fork); decides what the claim section says.
+        session_id (optional str) — the session's id, so a claim bound to it
+            reads as this session's own (milestone 381).
     """
     project_id, _repo, unbound_repo = await _project_scope()
     result = await plugin_ctx_svc.build_session_context(
-        g.user.id, project_id, unbound_repo=unbound_repo
+        g.user.id, project_id, unbound_repo=unbound_repo,
+        source=(request.args.get("source") or "").strip()[:20],
+        session_id=(request.args.get("session_id") or "").strip()[:200],
     )
     return jsonify(result)
 
@@ -352,6 +359,54 @@ async def report_check():
     return jsonify(body)
 
 
+@plugin_bp.get("/claim-session")
+@login_required
+async def claim_session():
+    """Bind the harness's session id to the caller's live claim on a task (milestone 381).
+
+    Called by `scribe_claim_session.sh` after `update_task` / `add_task_log`.
+    The server has already stamped the claim on that write; this only says
+    WHICH session made it — an id the harness reported, not one the model
+    asserted. A GET for the reason every plugin endpoint is one: a read-scoped
+    key must be enough to run the plugin.
+
+    Query:
+        task_id (int)    — the task the tool call named.
+        session_id (str) — the Claude Code session id from the hook event.
+
+    Returns the claim when one was bound, `{"claim": null}` when there was
+    nothing to bind to (no live claim of the caller's, or not writable).
+    """
+    try:
+        task_id = int(request.args.get("task_id") or "")
+    except ValueError:
+        return jsonify({"error": "task_id must be an integer"}), 400
+    session_id = (request.args.get("session_id") or "").strip()
+    if not session_id:
+        return jsonify({"error": "session_id is required"}), 400
+    claim = await task_claims_svc.bind_session(g.user.id, task_id, session_id)
+    return jsonify({"claim": claim})
+
+
+@plugin_bp.get("/release-session")
+@login_required
+async def release_session():
+    """Release the claims a session held, as it ends (milestone 381 step 4).
+
+    Called by `scribe_session_end.sh`. Best-effort by design: SessionEnd does
+    not fire on a crash, so the claim's lease — not this call — is what makes a
+    dead session's claim read as dead. This only makes the common case tidy.
+
+    Query:
+        session_id (str) — the ending session's id, from the hook event.
+    """
+    session_id = (request.args.get("session_id") or "").strip()
+    if not session_id:
+        return jsonify({"error": "session_id is required"}), 400
+    released = await task_claims_svc.release_session(g.user.id, session_id)
+    return jsonify({"released": released})
+
+
 @plugin_bp.get("/processes")
 @login_required
 async def process_manifest():
diff --git a/src/scribe/services/backup.py b/src/scribe/services/backup.py
index 0f515fc..81e470c 100644
--- a/src/scribe/services/backup.py
+++ b/src/scribe/services/backup.py
@@ -180,7 +180,13 @@ _COLUMN_EXCLUSIONS: dict[str, set[str]] = {
         "forge_connection_id",
     },
     "milestones": {"deleted_at", "deleted_batch_id"},
-    "notes": {"deleted_at", "deleted_batch_id"},
+    "notes": {
+        "deleted_at", "deleted_batch_id",
+        # A claim is a session's attention, not the work's state (milestone
+        # 381). Restoring one would assert that a session on another install,
+        # possibly long gone, is working the task right now.
+        "claimed_by", "claimed_at", "claim_touched_at", "claim_session",
+    },
     "task_logs": set(),
     "note_drafts": set(),
     "note_versions": set(),
@@ -268,7 +274,10 @@ _IMPORT_COLUMN_EXCLUSIONS: dict[str, set[str]] = {
         "design_system_id", "inception",
     },
     "milestones": {"id", "deleted_at", "deleted_batch_id"},
-    "notes": {"id", "deleted_at", "deleted_batch_id"},
+    "notes": {
+        "id", "deleted_at", "deleted_batch_id",
+        "claimed_by", "claimed_at", "claim_touched_at", "claim_session",
+    },
     "task_logs": {"id"},
     "note_drafts": {"id"},
     "note_versions": {"id"},
diff --git a/src/scribe/services/dedup.py b/src/scribe/services/dedup.py
index f03ef70..5f6f24d 100644
--- a/src/scribe/services/dedup.py
+++ b/src/scribe/services/dedup.py
@@ -94,6 +94,25 @@ _SNIPPET_SEMANTIC_THRESHOLD = 0.96
 # records that can still be merged by hand.
 _LESSON_SEMANTIC_THRESHOLD = 0.96
 
+# NOTES AND TASKS BLOCK ONLY A COPY, and surface the rest (#4306). Measured
+# 2026-09-22 with find_duplicate_records(note, 0.85): of the 74 note pairs at or
+# above the old 0.90 bar, almost all were DISTINCT siblings — consecutive
+# dev-logs (0.90–0.94), sub-notes of one design (0.90–0.94), research parts
+# (0.90–0.97), lore entries (0.95–0.98). The one clear copy sat at 0.997. A
+# block in that band refused the next dev-log and taught force=true, the same
+# finding #4134 made for rules. So only the copy band blocks; below it, a
+# near match is shown on the create reply for the session to judge.
+_NOTE_COPY_THRESHOLD = 0.98
+# Where a near match starts being worth reading. The measured pair counts
+# climb steeply under 0.87 (36 pairs at 0.87 against 200 capped at 0.85), and
+# the reply lists at most _NOTE_OVERLAP_LIMIT, so this is a cost floor — what
+# decides whether a match matters is the session reading it.
+_NOTE_OVERLAP_FLOOR = 0.87
+_NOTE_OVERLAP_LIMIT = 3
+# The note_types the copy band applies to. A process is prose too, but its
+# gate was not part of the measurement, so it keeps the general bar.
+_COPY_BAND_TYPES = {"note"}
+
 # The gate queries per CHUNK of the candidate (#280) — this caps how many
 # searches one save may cost. Eight chunks ≈ five thousand words of candidate;
 # a duplicate hiding past that is the duplicate report's job to find, not a
@@ -236,9 +255,20 @@ def _semantic_threshold(note_type: str) -> float:
         return _SNIPPET_SEMANTIC_THRESHOLD
     if note_type == LESSON_NOTE_TYPE:
         return _LESSON_SEMANTIC_THRESHOLD
+    if note_type in _COPY_BAND_TYPES:
+        return _NOTE_COPY_THRESHOLD
     return _SEMANTIC_THRESHOLD
 
 
+@dataclass
+class NoteOverlap:
+    """An existing note or task close enough to read before keeping a new one,
+    and not close enough to be called a copy."""
+    id: int
+    title: str
+    similarity: float
+
+
 async def find_duplicate_note(
     user_id: int,
     title: str,
@@ -249,6 +279,7 @@ async def find_duplicate_note(
     code: str = "",
     locations: list[dict] | None = None,
     data: dict | None = None,
+    overlaps: list[NoteOverlap] | None = None,
 ) -> DuplicateMatch | None:
     """Best near-duplicate of (title, body) within the same owner + project +
     kind, or None. Title match first (cheap, exact), then — for snippets — the
@@ -264,6 +295,11 @@ async def find_duplicate_note(
     carries the trigger, which the TITLE no longer does (milestone 427): the
     title check compares names, and the semantic check rebuilds the embedded
     document from `data`.
+
+    `overlaps`, when given, is filled with the near matches below the copy band
+    (#4306) — from the SAME searches, so asking costs nothing extra. Only the
+    kinds in _COPY_BAND_TYPES collect them. The caller creates the record and
+    returns them with `note_overlap_response`.
     """
     norm = " ".join((title or "").split()).lower()
 
@@ -319,6 +355,9 @@ async def find_duplicate_note(
         # under its name and embedded under `name — trigger`, so the query
         # document is built the way the corpus was, from `data`.
         doc_title = embeddings_svc.document_title(title, note_type, data, body)
+        block_at = _semantic_threshold(note_type)
+        collect = overlaps is not None and note_type in _COPY_BAND_TYPES
+        near: dict[int, NoteOverlap] = {}
         for query in embeddings_svc.chunk_document(doc_title, body)[:_GATE_MAX_CHUNKS]:
             # Scope the semantic check the same way as the title check: a record
             # in project P compares only to P; a project-less (orphan) record
@@ -330,7 +369,7 @@ async def find_duplicate_note(
                 user_id, query, project_id=project_id, is_task=is_task,
                 orphan_only=(project_id is None),
                 limit=3,
-                threshold=_semantic_threshold(note_type),
+                threshold=_NOTE_OVERLAP_FLOOR if collect else block_at,
                 # Owner-only, deliberately: this gate BLOCKS a create and tells
                 # the caller to update the match instead. Matching someone
                 # else's record would refuse their write and point them at
@@ -346,12 +385,69 @@ async def find_duplicate_note(
             for score, note in hits:
                 # semantic_search_notes doesn't filter note_type — enforce it
                 # here so a note doesn't shadow a task of the same wording, etc.
-                if note.note_type == note_type:
+                if note.note_type != note_type:
+                    continue
+                if score >= block_at:
                     return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")
+                # Best chunk wins per record: one long note matching in two
+                # sections is one overlap, not two.
+                prior = near.get(note.id)
+                if prior is None or score > prior.similarity:
+                    near[note.id] = NoteOverlap(note.id, note.title, round(score, 3))
+        if collect:
+            overlaps.extend(sorted(
+                near.values(), key=lambda o: o.similarity, reverse=True,
+            )[:_NOTE_OVERLAP_LIMIT])
 
     return None
 
 
+def note_overlap_response(overlaps: list[NoteOverlap], kind: str) -> dict:
+    """The keys a note or task create adds to its reply when an existing record
+    reads closely like the one just written (#4306). Empty when none.
+
+    The judgement is the session's: the embedding cannot tell a restatement
+    from the next dev-log in a series, and a reader can in one look."""
+    if not overlaps:
+        return {}
+    top = overlaps[0]
+    named = "; ".join(f'#{o.id} "{o.title}" ({o.similarity})' for o in overlaps)
+    return {
+        "overlaps": [
+            {"id": o.id, "title": o.title, "similarity": o.similarity}
+            for o in overlaps
+        ],
+        "overlap_note": (
+            f"Created — and it reads closely like: {named}. Open #{top.id} and "
+            f"judge it. If it records the same thing, fold what is new into it "
+            f"(update_{kind}) and delete this {kind}: two copies are found "
+            f"apart and drift apart. If it is a sibling — the next entry in a "
+            f"series, another part of one design — keep both."
+        ),
+    }
+
+
+def batch_overlap_response(per_record: dict[int, list[NoteOverlap]]) -> dict:
+    """`note_overlap_response` for a batch create: each overlap names the
+    1-based record it belongs to, so the reader knows which new id to judge."""
+    rows = [
+        {"record": i, "id": o.id, "title": o.title, "similarity": o.similarity}
+        for i, found in sorted(per_record.items()) for o in found
+    ]
+    if not rows:
+        return {}
+    return {
+        "overlaps": rows,
+        "overlap_note": (
+            "Created — and some records read closely like existing ones (see "
+            "`overlaps`, by record). Open each and judge it: the same thing "
+            "means fold what is new into the existing record and delete the "
+            "new one; a sibling (the next entry in a series, another part of "
+            "one design) means keep both."
+        ),
+    }
+
+
 # --- corpus-wide near-duplicate report (#2088) -------------------------------
 # The gate above PREVENTS a new duplicate; merge_snippets CURES one you point it
 # at. Neither FINDS the duplicates already sitting in the record — someone had to
diff --git a/src/scribe/services/embeddings.py b/src/scribe/services/embeddings.py
index fe1675e..ea14d71 100644
--- a/src/scribe/services/embeddings.py
+++ b/src/scribe/services/embeddings.py
@@ -18,7 +18,7 @@ from collections.abc import Sequence
 
 from typing import TYPE_CHECKING
 
-from sqlalchemy import delete, func, or_, select
+from sqlalchemy import and_, delete, func, or_, select
 
 from scribe.models import async_session
 from scribe.models.embedding import NoteEmbedding, RuleEmbedding
@@ -367,6 +367,34 @@ def calibration_stamp() -> dict:
     """
     return {"embedding_model": EMBEDDING_MODEL, "shape_version": CHUNKER_VERSION}
 
+
+def is_current_stamp(table):
+    """The rows written by THIS chunker in THIS model's space (#4132).
+
+    Every embedding table stores both halves of `calibration_stamp()` per row,
+    and a row is current only when both match. One predicate for all four
+    tables, because a backfill that checked the version alone is exactly how a
+    same-width model swap would have gone unnoticed.
+    """
+    return and_(
+        table.chunker_version == CHUNKER_VERSION,
+        table.embedding_model == EMBEDDING_MODEL,
+    )
+
+
+async def rows_off_the_live_model(table) -> int:
+    """How many rows of an embedding table were NOT written in the live space.
+
+    Nonzero means the corpus is part-way through a model change: a search over
+    it compares vectors from two geometries, and a statistic computed from it
+    (`retrieval_migration.migrate_floor`) is a blend of both.
+    """
+    async with async_session() as session:
+        return int((await session.execute(
+            select(func.count()).select_from(table)
+            .where(table.embedding_model != EMBEDDING_MODEL)
+        )).scalar_one())
+
 # Character budget approximating the model window. Tokens-per-char varies by
 # content — ~4 chars/token for prose, closer to 3 for code and tables — so 1400
 # chars sits at roughly 350-470 tokens, leaving headroom for the title prefixed
@@ -633,6 +661,7 @@ async def upsert_note_embedding(
                         embedding=vector,
                         chunk_text=chunk,
                         chunker_version=CHUNKER_VERSION,
+                        embedding_model=EMBEDDING_MODEL,
                     )
                 )
             await session.commit()
@@ -1092,7 +1121,7 @@ async def backfill_note_embeddings() -> None:
                 for row in (
                     await session.execute(
                         select(NoteEmbedding.note_id).where(
-                            NoteEmbedding.chunker_version == CHUNKER_VERSION
+                            is_current_stamp(NoteEmbedding)
                         )
                     )
                 ).fetchall()
@@ -1294,6 +1323,7 @@ async def upsert_rule_embedding(
                         embedding=vector,
                         chunk_text=chunk,
                         chunker_version=CHUNKER_VERSION,
+                        embedding_model=EMBEDDING_MODEL,
                     )
                 )
             await session.commit()
@@ -1464,7 +1494,7 @@ async def backfill_rule_embeddings() -> None:
     try:
         async with async_session() as session:
             current = select(RuleEmbedding.rule_id).where(
-                RuleEmbedding.chunker_version == CHUNKER_VERSION
+                is_current_stamp(RuleEmbedding)
             )
             # IDS ONLY — the text is re-read per rule below (#4262).
             by_version = {
@@ -1563,6 +1593,7 @@ async def upsert_milestone_embedding(
                 session.add(MilestoneEmbedding(
                     milestone_id=milestone_id, chunk_index=index, embedding=vector,
                     chunk_text=chunk, chunker_version=CHUNKER_VERSION,
+                    embedding_model=EMBEDDING_MODEL,
                 ))
             await session.commit()
     except Exception:
@@ -1725,6 +1756,7 @@ async def upsert_system_embedding(
                 session.add(SystemEmbedding(
                     system_id=system_id, chunk_index=index, embedding=vector,
                     chunk_text=chunk, chunker_version=CHUNKER_VERSION,
+                    embedding_model=EMBEDDING_MODEL,
                 ))
             await session.commit()
     except Exception:
@@ -1841,7 +1873,7 @@ async def backfill_system_embeddings() -> None:
     try:
         async with async_session() as session:
             current = select(SystemEmbedding.system_id).where(
-                SystemEmbedding.chunker_version == CHUNKER_VERSION
+                is_current_stamp(SystemEmbedding)
             )
             # IDS ONLY — the charter is re-read per System below (#4262).
             by_version = {
@@ -1884,7 +1916,7 @@ async def backfill_milestone_embeddings() -> None:
     try:
         async with async_session() as session:
             current = select(MilestoneEmbedding.milestone_id).where(
-                MilestoneEmbedding.chunker_version == CHUNKER_VERSION
+                is_current_stamp(MilestoneEmbedding)
             )
             # IDS ONLY — the plan is re-read per milestone below (#4262).
             by_version = {
diff --git a/src/scribe/services/notes.py b/src/scribe/services/notes.py
index fe3c1b6..4c460a5 100644
--- a/src/scribe/services/notes.py
+++ b/src/scribe/services/notes.py
@@ -284,7 +284,7 @@ def build_note(
         except ValueError:
             raise ValueError(f"Invalid priority: {priority!r}. Must be one of: {[p.value for p in TaskPriority]}")
 
-    return Note(
+    note = Note(
         user_id=user_id,
         title=title,
         body=body,
@@ -304,6 +304,49 @@ def build_note(
         verify_with=verify_with,
         expires_when=expires_when,
     )
+    # A create that names a status is the same transition an update to it is
+    # (#3683) — otherwise `create_task(status="in_progress")` writes a row no
+    # update could produce: started, with no `started_at`.
+    if status is not None:
+        apply_status_transition(note, user_id)
+    return note
+
+
+def apply_status_transition(note: Note, user_id: int | None = None) -> None:
+    """Stamp what reaching `note.status` implies — the ONE statement of it.
+
+    Called by the update path whenever `status` is written and by `build_note`
+    whenever a create names one, so a task created at a status is
+    indistinguishable from one that reached it by update. Two copies of "what
+    a status implies" are where the two drift, and #3683 was that drift.
+
+    `user_id` is whoever is making the change: reaching `in_progress` is the
+    moment a session takes the work on, so it stamps that user's claim, and a
+    status that ends or un-starts the work releases it (milestone 381).
+    """
+    from scribe.services.task_claims import release_claim, stamp_claim
+
+    _now = datetime.now(timezone.utc)
+    if note.status == TaskStatus.in_progress.value:
+        if note.started_at is None:
+            note.started_at = _now
+        if user_id is not None:
+            stamp_claim(note, user_id, _now)
+    elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
+        release_claim(note)
+        note.completed_at = _now
+        if note.recurrence_rule:
+            from scribe.services.recurrence import calculate_next_due
+            base = note.due_date or _now.date()
+            next_due = calculate_next_due(note.recurrence_rule, base)
+            note.recurrence_next_spawn_at = datetime(
+                next_due.year, next_due.month, next_due.day, tzinfo=timezone.utc
+            )
+    elif note.status == TaskStatus.todo.value:
+        release_claim(note)
+        note.started_at = None
+        note.completed_at = None
+        note.recurrence_next_spawn_at = None
 
 
 async def get_note(user_id: int, note_id: int) -> Note | None:
@@ -653,25 +696,8 @@ async def update_note(
             recompose = _mirror_recomposers().get(note.note_type or "")
             if recompose is not None:
                 note.data = recompose(note)
-        # Auto-set lifecycle timestamps on status transitions
         if "status" in fields:
-            _now = datetime.now(timezone.utc)
-            if note.status == TaskStatus.in_progress.value:
-                if note.started_at is None:
-                    note.started_at = _now
-            elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
-                note.completed_at = _now
-                if note.recurrence_rule:
-                    from scribe.services.recurrence import calculate_next_due
-                    base = note.due_date or _now.date()
-                    next_due = calculate_next_due(note.recurrence_rule, base)
-                    note.recurrence_next_spawn_at = datetime(
-                        next_due.year, next_due.month, next_due.day, tzinfo=timezone.utc
-                    )
-            elif note.status == TaskStatus.todo.value:
-                note.started_at = None
-                note.completed_at = None
-                note.recurrence_next_spawn_at = None
+            apply_status_transition(note, user_id)
         note.updated_at = datetime.now(timezone.utc)
         await session.commit()
         await session.refresh(note)
diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py
index cda7e84..dccf733 100644
--- a/src/scribe/services/plugin_context.py
+++ b/src/scribe/services/plugin_context.py
@@ -26,6 +26,7 @@ from scribe.services import notes as notes_svc
 from scribe.services import projects as projects_svc
 from scribe.services import shape_ledger as shape_ledger_svc
 from scribe.services import snippets as snippets_svc
+from scribe.services import task_claims as task_claims_svc
 from scribe.services.access import label_shared_items, owner_names_for
 from scribe.services.embeddings import (
     document_title,
@@ -3055,7 +3056,8 @@ def _goal_line(goal: str, project_id: int) -> str:
 
 
 async def build_session_context(
-    user_id: int, project_id: int = 0, unbound_repo: str = ""
+    user_id: int, project_id: int = 0, unbound_repo: str = "",
+    source: str = "", session_id: str = "",
 ) -> dict:
     """Render the SessionStart context for a user, optionally project-scoped.
 
@@ -3069,6 +3071,12 @@ async def build_session_context(
         unbound_repo: when the hook sent a repo remote that maps to no project,
             its normalized key — triggers a one-line "bind this repo" hint so
             the binding is self-healing.
+        source / session_id: the host's SessionStart `source` and the
+            session's id, when the adapter sends them. They decide what the
+            claim section says (milestone 381 step 3, `task_claims.
+            render_claims`): a compaction gets back the work it had claimed
+            with its latest logs; a new session hears about other sessions'
+            live and abandoned claims; a resume hears nothing.
 
     Returns {"context": str, "project": dict | None}.
 
@@ -3135,6 +3143,17 @@ async def build_session_context(
                         f"`get_design_system({design['id']})` → "
                         f"`resolved_guidance`.",
                     ]
+    # The claim section goes after the project block and before any "nothing
+    # loaded" note: claimed work is the most specific thing this session can be
+    # told, and it is true whether or not a project resolved. Best-effort — a
+    # session start never fails on it.
+    try:
+        lines += await task_claims_svc.claims_for_session_start(
+            user_id, project_dict["id"] if project_dict else 0, source, session_id,
+        )
+    except Exception:  # noqa: BLE001 - context is best-effort
+        logger.warning("claim section skipped", exc_info=True)
+
     # Nothing loaded — say which nothing (#4085). This used to hang off the
     # `if project_id:` above as an `elif`, which meant an id that was SENT and
     # did not resolve produced no message at all: the outer branch was taken,
diff --git a/src/scribe/services/retrieval_migration.py b/src/scribe/services/retrieval_migration.py
index ba20473..00881b6 100644
--- a/src/scribe/services/retrieval_migration.py
+++ b/src/scribe/services/retrieval_migration.py
@@ -58,9 +58,11 @@ import logging
 from sqlalchemy import select
 
 from scribe.models import async_session
+from scribe.models.embedding import NoteEmbedding, RuleEmbedding
 from scribe.models.retrieval_log import RetrievalLog
 from scribe.services.embeddings import (
     calibration_stamp,
+    rows_off_the_live_model,
     semantic_search_notes,
     semantic_search_rules,
 )
@@ -117,6 +119,19 @@ _RESCORERS = {
     "report_preference": lambda u, q, p: _rescore_rules(u, q, p, "preference"),
 }
 
+# The embedding table each surface's re-scorer reads. A migration from a corpus
+# that is still part-way through a model change re-scores against a blend of
+# two geometries and writes a floor computed from it, with a confident reason
+# attached (#4132) — so the corpus has to be wholly in the live space first.
+_CORPUS = {
+    "auto_inject": NoteEmbedding,
+    "write_path": NoteEmbedding,
+    "write_path_rule": RuleEmbedding,
+    "pre_tool_rule": RuleEmbedding,
+    "prompt_rule": RuleEmbedding,
+    "report_preference": RuleEmbedding,
+}
+
 
 def _floor_admitting(scores: list[float], fraction: float) -> float:
     """The floor that admits `fraction` of `scores`, on this scale.
@@ -161,6 +176,21 @@ async def migrate_floor(
             "arm's own corpus filters."
         )
 
+    off_model = await rows_off_the_live_model(_CORPUS[surface])
+    if off_model:
+        # Refused before sampling anything. The order of operations — re-embed,
+        # THEN migrate — used to be held only in the operator's memory.
+        stamp = calibration_stamp()
+        return {
+            "surface": surface, "migrated": False,
+            "why": f"{off_model} embedding row(s) this surface searches are not "
+                   f"yet in {stamp['embedding_model']}'s space. Re-scoring now "
+                   "would measure a blend of two models; let the startup "
+                   "backfill finish re-embedding, then migrate",
+            "rows_off_model": off_model,
+            "calibration": stamp,
+        }
+
     old_floor = await floor_for(user_id, surface)
     async with async_session() as session:
         rows = (await session.execute(
diff --git a/src/scribe/services/task_claims.py b/src/scribe/services/task_claims.py
new file mode 100644
index 0000000..5124295
--- /dev/null
+++ b/src/scribe/services/task_claims.py
@@ -0,0 +1,313 @@
+"""A task's claim — which session is working it right now (milestone 381 step 2).
+
+`status` says where THE WORK stands and is durable: `in_progress` means
+committed to, not finished. It cannot also say "someone is on this now",
+because nothing ever clears it — a session that crashes, is killed or simply
+moves on leaves `in_progress` behind, and the row goes on asserting attention
+nobody is paying. The claim is the other half, a property of a session's
+attention rather than of the work, and it is built so that nothing has to
+clear it either.
+
+WHO STAMPS IT. The server, on the write that IS the work: a transition to
+`in_progress` and a work log on an open task. No tool asks the model to claim
+anything, because a claim the model has to remember is the flag this replaces.
+Any MCP client gets the lease; the Claude Code plugin's PostToolUse hook then
+binds the harness's session id to it (`bind_session`), so the claim can say
+WHICH session and not only "someone, recently" — the harness reports the id,
+the model asserts nothing.
+
+HOW IT DIES. On read. A claim is live while its last touch is inside
+`CLAIM_LEASE`; past that it reads as dead, whatever the row still holds. There
+is no sweep: a job that tidies claims would reintroduce exactly the dependency
+on something running that this is designed out of. A status that ends or
+un-starts the work (done, cancelled, todo) releases it outright.
+
+`in_progress` with no live claim is the state this exists to make sayable:
+committed to, and nobody on it.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from scribe.models.note import Note
+
+# How long a claim stays live after its last touch. Long enough that a
+# compaction, a resume or a long read does not kill it; short enough that a
+# session gone overnight reads as gone. The cost either way is stated rather
+# than hidden: readers show the age beside `live`, never the boolean alone.
+CLAIM_LEASE = timedelta(hours=2)
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def claim_is_live(note: Note, now: datetime | None = None) -> bool:
+    touched = note.claim_touched_at
+    return touched is not None and (now or _now()) - touched < CLAIM_LEASE
+
+
+def stamp_claim(note: Note, user_id: int, now: datetime | None = None) -> None:
+    """Record that `user_id` is working this task now.
+
+    The most recent worker holds it: a live claim by someone else is taken
+    over rather than refused, because the write that stamps a claim has already
+    happened — it is the evidence of who is on the task, and refusing the claim
+    would only make the record less true. `claimed_at` restarts whenever the
+    holder changes or the previous claim had died, so "since" means since THIS
+    stretch of attention, not since the task was first touched.
+    """
+    now = now or _now()
+    if note.claimed_by != user_id or not claim_is_live(note, now):
+        note.claimed_by = user_id
+        note.claimed_at = now
+        note.claim_session = None
+    note.claim_touched_at = now
+
+
+def release_claim(note: Note) -> None:
+    """Clear the claim. Idempotent — releasing nothing is not an error."""
+    note.claimed_by = None
+    note.claimed_at = None
+    note.claim_touched_at = None
+    note.claim_session = None
+
+
+def claim_state(note: Note, now: datetime | None = None) -> dict | None:
+    """The claim as a reader sees it, or None when there has never been one.
+
+    A dead claim is returned, not hidden: "last worked by that session three
+    days ago" is what a resuming session needs to know, and `live: false` says
+    no-one should read it as current.
+    """
+    if note.claimed_at is None:
+        return None
+    from scribe.models.base import iso
+
+    return {
+        "held_by": note.claimed_by,
+        "session": note.claim_session,
+        "since": iso(note.claimed_at),
+        "touched": iso(note.claim_touched_at),
+        "live": claim_is_live(note, now),
+    }
+
+
+async def bind_session(user_id: int, task_id: int, session_id: str) -> dict | None:
+    """Attach a harness-reported session id to the caller's live claim.
+
+    Called by the plugin's PostToolUse hook after `update_task` or
+    `add_task_log`. A no-op — returning None — when the task is not writable
+    by the caller, has no live claim, or the live claim is someone else's: the
+    hook reports what happened, it cannot create a claim the server did not
+    stamp. A different session taking over a live claim restarts `since`,
+    for the reason `stamp_claim` gives.
+    """
+    from scribe.models import async_session
+    from scribe.models.note import Note
+    from scribe.services.access import can_write_note
+
+    session_id = (session_id or "").strip()[:200]
+    if not session_id or not await can_write_note(user_id, task_id):
+        return None
+    async with async_session() as session:
+        note = await session.get(Note, task_id)
+        if note is None or note.claimed_by != user_id or not claim_is_live(note):
+            return None
+        now = _now()
+        if note.claim_session not in (None, session_id):
+            note.claimed_at = now
+        note.claim_session = session_id
+        note.claim_touched_at = now
+        await session.commit()
+        return claim_state(note, now)
+
+
+# --- The readers (milestone 381 step 3) ---------------------------------------
+#
+# A claim nobody reads is the state before this milestone. SessionStart is the
+# reader that pays rent to the session that set it, and it branches on the
+# `source` the host sends, because the same claim means different things
+# depending on what just happened to the context:
+#
+#   compact  the context was summarised away and the claim is certainly ours.
+#            Push the claimed work AND its latest log entries — the state a
+#            compaction destroys, which the record already holds. A count of
+#            open tasks cannot answer "where were we".
+#   clear    the context was wiped, so the same push; and other sessions' claims
+#            are worth knowing about, as on a startup.
+#   startup  a new session. Claims held by OTHER sessions are the news: live
+#            ones may be running right now, dead ones were abandoned mid-task.
+#   fork     the session carries a conversation that held claims under another
+#            id. Two sessions now believe they hold the same work, so the
+#            live claims are named as possibly-the-parent's, with what a write
+#            does about it.
+#   resume   the context was restored intact. Say nothing.
+
+# What a session is told, per source. Pure data so the branch is one lookup.
+_PUSH_OWN = {"compact", "clear"}
+_NAME_OTHERS = {"startup", "clear", "fork"}
+
+# Caps: a session-start block is read by every session, so it is sized for the
+# few claims that matter rather than for the worst case.
+_OWN_CAP = 5
+_OTHERS_CAP = 5
+_LOGS_PER_TASK = 2
+_LOG_CHARS = 600
+
+
+def _age(when: datetime | None, now: datetime) -> str:
+    if when is None:
+        return "at an unknown time"
+    secs = max(0, int((now - when).total_seconds()))
+    if secs < 90:
+        return "just now"
+    if secs < 5400:
+        return f"{secs // 60}m ago"
+    if secs < 2 * 86400:
+        return f"{secs // 3600}h ago"
+    return f"{secs // 86400}d ago"
+
+
+def render_claims(
+    source: str,
+    session_id: str,
+    claims: list,
+    logs: dict[int, list],
+    now: datetime | None = None,
+) -> list[str]:
+    """The claim section of the SessionStart context, as markdown lines.
+
+    `claims` are the caller's claimed tasks (objects with id, title, status and
+    the claim columns); `logs` maps a task id to its newest log entries
+    (objects with `created_at` and `content`), newest first. Empty when there is
+    nothing this source should say — silence is the right answer on a resume,
+    and on any start with no claims.
+    """
+    from scribe.services.text import elide
+
+    now = now or _now()
+    source = (source or "").strip()
+    session_id = (session_id or "").strip()
+    ours = [c for c in claims if claim_is_live(c, now)
+            and (c.claim_session == session_id or c.claim_session is None)]
+    others_live = [c for c in claims if claim_is_live(c, now)
+                   and c.claim_session not in (None, session_id)]
+    abandoned = [c for c in claims if not claim_is_live(c, now)
+                 and c.status == "in_progress"]
+
+    lines: list[str] = []
+    if source in _PUSH_OWN and session_id and ours:
+        lines += [
+            "",
+            "## In flight — the work this session had claimed",
+            "Scribe's record of what you were doing before the context was "
+            "lost. Carry on from here; the full log is `get_task(id)`.",
+        ]
+        for c in ours[:_OWN_CAP]:
+            lines.append(
+                f"- #{c.id} \"{c.title}\" ({c.status}) — claimed "
+                f"{_age(c.claimed_at, now)}, last touched {_age(c.claim_touched_at, now)}"
+            )
+            for entry in logs.get(c.id, [])[:_LOGS_PER_TASK]:
+                text, _ = elide(" ".join((entry.content or "").split()), _LOG_CHARS)
+                lines.append(f"  - log {_age(entry.created_at, now)}: {text}")
+    if source in _NAME_OTHERS and (others_live or abandoned):
+        lines += ["", "## Work other sessions were doing"]
+        if source == "fork":
+            lines.append(
+                "This session was forked, so a live claim below may be the "
+                "session you were forked from — two sessions now think they "
+                "hold it. Your next log or status change on a task moves its "
+                "claim here; leave it alone if the other session is still on it."
+            )
+        for c in others_live[:_OTHERS_CAP]:
+            lines.append(
+                f"- #{c.id} \"{c.title}\" — claimed by another session, last "
+                f"touched {_age(c.claim_touched_at, now)}. It may still be "
+                "running; check before working the same task."
+            )
+        for c in abandoned[:_OTHERS_CAP]:
+            lines.append(
+                f"- #{c.id} \"{c.title}\" — in progress, but the session "
+                f"working it went quiet {_age(c.claim_touched_at, now)} without "
+                "finishing. Read its log and continue it, or set it back to todo."
+            )
+    return lines
+
+
+async def claims_for_session_start(
+    user_id: int, project_id: int, source: str, session_id: str,
+) -> list[str]:
+    """Load the caller's claims (in the active project, when one resolved) and
+    their newest logs, and render them for this `source`.
+
+    "The caller's claims" is `claimed_by == user_id` — a statement about whose
+    attention a claim records, not an access filter: a claim is only ever
+    stamped by a write the caller was already allowed to make.
+    """
+    from sqlalchemy import select
+
+    from scribe.models import async_session
+    from scribe.models.note import Note
+    from scribe.models.task_log import TaskLog
+
+    # No source means the caller did not ask — another client, or an adapter
+    # older than this section — and a resume restored everything already.
+    if (source or "") in ("", "resume"):
+        return []
+    async with async_session() as session:
+        q = select(Note).where(
+            Note.claimed_by == user_id,
+            Note.claimed_at.is_not(None),
+            Note.deleted_at.is_(None),
+        )
+        if project_id:
+            q = q.where(Note.project_id == project_id)
+        claims = list((await session.execute(
+            q.order_by(Note.claim_touched_at.desc()).limit(_OWN_CAP + 2 * _OTHERS_CAP)
+        )).scalars().all())
+        logs: dict[int, list] = {}
+        if claims:
+            rows = (await session.execute(
+                select(TaskLog)
+                .where(TaskLog.task_id.in_([c.id for c in claims]))
+                .order_by(TaskLog.created_at.desc())
+            )).scalars().all()
+            for row in rows:
+                bucket = logs.setdefault(row.task_id, [])
+                if len(bucket) < _LOGS_PER_TASK:
+                    bucket.append(row)
+    return render_claims(source, session_id, claims, logs)
+
+
+async def release_session(user_id: int, session_id: str) -> int:
+    """Release every claim the caller holds under `session_id` (milestone 381 step 4).
+
+    Called by the plugin's SessionEnd hook: the session's context is about to
+    stop existing, so the attention its claims record is ending too. A TIDY-UP,
+    not the guarantee — SessionEnd does not fire on a crash, a killed terminal
+    or a dropped connection, and those are what the lease is for. Returns how
+    many were released; 0 is the ordinary answer for a session that claimed
+    nothing.
+    """
+    from sqlalchemy import select
+
+    from scribe.models import async_session
+    from scribe.models.note import Note
+
+    session_id = (session_id or "").strip()[:200]
+    if not session_id:
+        return 0
+    async with async_session() as session:
+        held = (await session.execute(
+            select(Note).where(
+                Note.claimed_by == user_id, Note.claim_session == session_id,
+            )
+        )).scalars().all()
+        for note in held:
+            release_claim(note)
+        await session.commit()
+        return len(held)
diff --git a/src/scribe/services/task_logs.py b/src/scribe/services/task_logs.py
index 61853aa..fc96d3e 100644
--- a/src/scribe/services/task_logs.py
+++ b/src/scribe/services/task_logs.py
@@ -6,8 +6,9 @@ from sqlalchemy import func, select
 
 from scribe.models import async_session
 from scribe.models.task_log import TaskLog
-from scribe.models.note import Note
-from scribe.services.access import can_read_note, readable_notes_clause
+from scribe.models.note import Note, TaskStatus
+from scribe.services.access import can_read_note, can_write_note, readable_notes_clause
+from scribe.services.task_claims import stamp_claim
 
 logger = logging.getLogger(__name__)
 
@@ -50,13 +51,21 @@ async def create_log(
     content: str,
     duration_minutes: int | None = None,
 ) -> TaskLog:
+    # Whoever may WRITE the task may log on it (rule #78) — a collaborator on a
+    # shared project included. This used to be a bare owner filter, which
+    # refused exactly the person a shared task exists for.
+    if not await can_write_note(user_id, task_id):
+        raise ValueError(f"Task {task_id} not found")
     async with async_session() as session:
-        # Verify task exists and belongs to user
-        result = await session.execute(
-            select(Note).where(Note.id == task_id, Note.user_id == user_id)
-        )
-        if result.scalars().first() is None:
+        result = await session.execute(select(Note).where(Note.id == task_id))
+        task = result.scalars().first()
+        if task is None:
             raise ValueError(f"Task {task_id} not found")
+        # Logging IS working the task, so it stamps the claim (milestone 381) —
+        # unless the work is over: a retrospective note on a closed task is not
+        # a session picking it up.
+        if task.status not in (TaskStatus.done.value, TaskStatus.cancelled.value):
+            stamp_claim(task, user_id)
         log = TaskLog(
             task_id=task_id,
             user_id=user_id,
diff --git a/tests/helpers.py b/tests/helpers.py
index 54d7271..88a964d 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -183,6 +183,9 @@ def fake_note(**attrs) -> MagicMock:
         # Milestone 317: a truthy mock here reads as "this note carries a
         # check", which trips the guard on records that may not have one.
         "verify_with": None, "expires_when": None, "verified_at": None,
+        # Milestone 381: a truthy mock would read as a live claim.
+        "claimed_by": None, "claimed_at": None, "claim_touched_at": None,
+        "claim_session": None,
     }, attrs)
 
 
@@ -193,6 +196,9 @@ def fake_task(**attrs) -> MagicMock:
         "tags": [], "parent_id": None, "project_id": None, "is_task": True,
         "task_kind": "work", "user_id": 7, "deleted_at": None,
         "verify_with": None, "expires_when": None, "verified_at": None,
+        # Milestone 381: a truthy mock would read as a live claim.
+        "claimed_by": None, "claimed_at": None, "claim_touched_at": None,
+        "claim_session": None,
     }, attrs)
 
 
diff --git a/tests/test_chunking.py b/tests/test_chunking.py
index 10d881a..5836164 100644
--- a/tests/test_chunking.py
+++ b/tests/test_chunking.py
@@ -228,6 +228,7 @@ async def test_upsert_stores_one_versioned_row_per_chunk():
     assert [r.chunk_index for r in rows] == list(range(len(chunks)))
     assert [r.chunk_text for r in rows] == chunks
     assert {r.chunker_version for r in rows} == {emb.CHUNKER_VERSION}
+    assert {r.embedding_model for r in rows} == {emb.EMBEDDING_MODEL}
     assert {r.user_id for r in rows} == {42}
     session.execute.assert_awaited()  # the delete that makes replacement atomic
 
@@ -314,3 +315,17 @@ async def test_backfill_reembeds_notes_with_a_stale_chunker_version():
 
     embedded = [call.args[0] for call in upsert.call_args_list]
     assert embedded == [2], "only the stale note is re-embedded"
+
+
+def test_a_row_is_current_only_in_the_live_models_space():
+    """#4132: the column is a width, not an identity, so a same-width model
+    swap writes a second geometry with no error. The backfill's "current"
+    predicate has to name BOTH halves of the stamp, on every embedding table."""
+    from scribe.models.embedding import (
+        MilestoneEmbedding, NoteEmbedding, RuleEmbedding, SystemEmbedding,
+    )
+    from scribe.services import embeddings as emb
+
+    for table in (NoteEmbedding, RuleEmbedding, MilestoneEmbedding, SystemEmbedding):
+        clause = str(emb.is_current_stamp(table))
+        assert "chunker_version" in clause and "embedding_model" in clause, table
diff --git a/tests/test_guidance_ownership.py b/tests/test_guidance_ownership.py
index 99241be..db86c97 100644
--- a/tests/test_guidance_ownership.py
+++ b/tests/test_guidance_ownership.py
@@ -155,6 +155,11 @@ TOPICS: tuple[Topic, ...] = (
     Topic("a retrieved rule outranks a default habit", U, ("outranks a default habit",),
           "a retrieved rule outranks a default habit"),
     Topic("log on completion and on a problem", U, ("hit or discover a problem",), "hit or discover a problem"),
+    # Milestone 381 step 4: the half of a hand-off only the model can do. The
+    # release is mechanical (SessionEnd hook); saying what happened is not.
+    Topic("hand off before the context stops existing", U,
+          ("hand off", "claims it for this session"),
+          "write down what the next session needs"),
     Topic("the project's design system binds ui", U, ("resolve_design_system",),
           "building ui: the project's design system binds", index=("resolve_design_system",)),
     Topic("name the record, never just its number", U, ("name the record",),
diff --git a/tests/test_integration_lesson_reach.py b/tests/test_integration_lesson_reach.py
index 24c7c40..7f996d5 100644
--- a/tests/test_integration_lesson_reach.py
+++ b/tests/test_integration_lesson_reach.py
@@ -23,7 +23,7 @@ from scribe.models.embedding import EMBEDDING_DIM, NoteEmbedding
 from scribe.models.note import Note
 from scribe.models.project import Project
 from scribe.services import lessons as lessons_svc
-from scribe.services.embeddings import CHUNKER_VERSION, semantic_search_notes
+from scribe.services.embeddings import CHUNKER_VERSION, EMBEDDING_MODEL, semantic_search_notes
 from tests.helpers import ensure_user
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
@@ -71,6 +71,7 @@ async def corpus():
                 note_id=note.id, chunk_index=0, user_id=owner.id,
                 embedding=QUERY_VEC, chunk_text=note.title,
                 chunker_version=CHUNKER_VERSION,
+                embedding_model=EMBEDDING_MODEL,
             ))
         ids = {k: n.id for k, n in rows.items()}
         ids["owner"], ids["a"], ids["b"] = owner.id, a.id, b.id
diff --git a/tests/test_integration_milestone_search.py b/tests/test_integration_milestone_search.py
index 94fe9b4..b786a92 100644
--- a/tests/test_integration_milestone_search.py
+++ b/tests/test_integration_milestone_search.py
@@ -17,7 +17,7 @@ from scribe.models.embedding import EMBEDDING_DIM, MilestoneEmbedding
 from scribe.models.milestone import Milestone
 from scribe.models.project import Project
 from scribe.services import dedup as dedup_svc
-from scribe.services.embeddings import CHUNKER_VERSION, semantic_search_milestones
+from scribe.services.embeddings import CHUNKER_VERSION, EMBEDDING_MODEL, semantic_search_milestones
 from tests.helpers import ensure_user
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")]
@@ -48,7 +48,8 @@ async def roadmap():
         await s.flush()
         for ms, vec in ((m3, NEAR), (done, NEAR), (unrelated, FAR), (foreign, NEAR)):
             s.add(MilestoneEmbedding(milestone_id=ms.id, chunk_index=0, embedding=vec,
-                                     chunk_text=ms.title, chunker_version=CHUNKER_VERSION))
+                                     chunk_text=ms.title, chunker_version=CHUNKER_VERSION,
+                                     embedding_model=EMBEDDING_MODEL))
         ids = {"owner": owner.id, "stranger": stranger.id, "mine": mine.id,
                "m3": m3.id, "done": done.id, "unrelated": unrelated.id, "foreign": foreign.id}
         await s.commit()
diff --git a/tests/test_integration_pgvector_search.py b/tests/test_integration_pgvector_search.py
index 4748c1e..d36ee63 100644
--- a/tests/test_integration_pgvector_search.py
+++ b/tests/test_integration_pgvector_search.py
@@ -33,7 +33,7 @@ def _vec(*nonzero_first):
 
 def _emb(note_id, user_id, chunk_index, vec):
     """A chunk row at the current chunker version (#280, migration 0077)."""
-    from scribe.services.embeddings import CHUNKER_VERSION
+    from scribe.services.embeddings import CHUNKER_VERSION, EMBEDDING_MODEL
 
     return NoteEmbedding(
         note_id=note_id,
@@ -42,6 +42,7 @@ def _emb(note_id, user_id, chunk_index, vec):
         embedding=vec,
         chunk_text=f"chunk {chunk_index} of note {note_id}",
         chunker_version=CHUNKER_VERSION,
+        embedding_model=EMBEDDING_MODEL,
     )
 
 
diff --git a/tests/test_integration_rule_scope.py b/tests/test_integration_rule_scope.py
index 27ca02d..41b928b 100644
--- a/tests/test_integration_rule_scope.py
+++ b/tests/test_integration_rule_scope.py
@@ -18,7 +18,7 @@ from scribe.models.embedding import EMBEDDING_DIM, RuleEmbedding
 from scribe.models.project import Project
 from scribe.models.share import ProjectShare
 from scribe.services import rulebooks as rulebooks_svc
-from scribe.services.embeddings import CHUNKER_VERSION, semantic_search_rules
+from scribe.services.embeddings import CHUNKER_VERSION, EMBEDDING_MODEL, semantic_search_rules
 from tests.helpers import ensure_user
 
 pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
@@ -66,6 +66,7 @@ async def homes():
             s.add(RuleEmbedding(
                 rule_id=rule.id, chunk_index=0, embedding=QUERY_VEC,
                 chunk_text=rule.title, chunker_version=CHUNKER_VERSION,
+                embedding_model=EMBEDDING_MODEL,
             ))
         await s.commit()
     ids.update(glob=glob.id, on_a=on_a.id, on_b=on_b.id)
diff --git a/tests/test_lesson_write_path.py b/tests/test_lesson_write_path.py
index 03a7159..1032218 100644
--- a/tests/test_lesson_write_path.py
+++ b/tests/test_lesson_write_path.py
@@ -152,9 +152,12 @@ def test_a_lesson_is_judged_at_the_trigger_dominated_bar():
         "about one area would refuse each other"
     )
     assert _LESSON_SEMANTIC_THRESHOLD > _SEMANTIC_THRESHOLD
-    # An ordinary note is untouched — the carve-out is per kind, not a
-    # loosening of the gate.
-    assert _semantic_threshold("note") == _SEMANTIC_THRESHOLD
+    # The carve-out is per kind, not a loosening of the gate: a lesson's bar is
+    # its own, apart from the note's copy band (#4306).
+    from scribe.services.dedup import _NOTE_COPY_THRESHOLD
+
+    assert _semantic_threshold("note") == _NOTE_COPY_THRESHOLD
+    assert _NOTE_COPY_THRESHOLD != _LESSON_SEMANTIC_THRESHOLD
 
 
 @pytest.mark.asyncio
diff --git a/tests/test_retrieval_migration.py b/tests/test_retrieval_migration.py
index e46820e..cc41aa4 100644
--- a/tests/test_retrieval_migration.py
+++ b/tests/test_retrieval_migration.py
@@ -15,6 +15,9 @@ WHAT THIS PINS
      mid-backfill would end up with every bar at zero.
   4. **Every registry surface can be migrated.** A seventh arm that nobody adds
      a re-scorer for is one whose floor silently cannot survive a model change.
+  5. **A half-migrated corpus is a refusal (#4132).** While any row the surface
+     searches is stamped with another model, a re-score measures a blend of two
+     geometries — so nothing is sampled until the backfill has finished.
 """
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -43,11 +46,42 @@ def _session_with(rows):
     return session
 
 
+@pytest.fixture(autouse=True)
+def _corpus_on_the_live_model():
+    """Every test below starts from a wholly re-embedded corpus; the one that
+    is about a half-migrated corpus patches this again."""
+    with patch.object(rm, "rows_off_the_live_model", AsyncMock(return_value=0)):
+        yield
+
+
 def test_every_surface_has_a_rescorer():
     """Otherwise a surface's floor cannot cross a model change at all."""
     assert set(rm._RESCORERS) == set(SURFACES)
 
 
+def test_every_surface_names_the_corpus_it_searches():
+    """Otherwise the half-migrated check has no table to count for it."""
+    assert set(rm._CORPUS) == set(SURFACES)
+
+
+@pytest.mark.asyncio
+async def test_a_corpus_part_way_through_a_model_change_refuses_before_sampling():
+    rows = _logs([("q1", None, 0.9), ("q2", None, 0.1)])
+    rescore = AsyncMock(return_value=0.4)
+    with patch.object(rm, "rows_off_the_live_model", AsyncMock(return_value=17)) as off, \
+         patch.object(rm, "async_session", MagicMock(return_value=_session_with(rows))), \
+         patch.object(rm, "floor_for", AsyncMock(return_value=0.5)), \
+         patch.object(rm, "set_dial", AsyncMock()) as set_dial, \
+         patch.dict(rm._RESCORERS, {"prompt_rule": rescore}):
+        out = await rm.migrate_floor(1, "prompt_rule", apply=True)
+
+    assert out["migrated"] is False
+    assert out["rows_off_model"] == 17
+    assert off.await_args.args[0] is rm.RuleEmbedding
+    rescore.assert_not_called()
+    set_dial.assert_not_called()
+
+
 def test_the_floor_that_admits_a_fraction_is_an_observed_score():
     scores = [0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1, 0.05]
     # 30% of ten is three; the third-best score is the bar that admits exactly
diff --git a/tests/test_services_dedup.py b/tests/test_services_dedup.py
index f7c2b33..3a13cff 100644
--- a/tests/test_services_dedup.py
+++ b/tests/test_services_dedup.py
@@ -43,8 +43,9 @@ async def test_short_body_skips_semantic_check():
 
 @pytest.mark.asyncio
 async def test_semantic_match_when_body_substantial():
+    # A note blocks only in the copy band (#4306).
     hit = fake_note(id=20, title="Existing", note_type="note")
-    sem = AsyncMock(return_value=[(0.93, hit)])
+    sem = AsyncMock(return_value=[(0.99, hit)])
     with patch("scribe.services.dedup.async_session",
                return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
@@ -54,7 +55,80 @@ async def test_semantic_match_when_body_substantial():
     assert dup is not None
     assert dup.id == 20
     assert dup.reason == "semantic"
-    assert dup.similarity == 0.93
+    assert dup.similarity == 0.99
+
+
+@pytest.mark.asyncio
+async def test_a_near_note_is_surfaced_not_blocked():
+    """#4306: sibling notes — the next dev-log, another part of one design —
+    measured 0.90–0.98, so a match there is shown for the session to judge
+    instead of refusing the write."""
+    from scribe.services.dedup import _NOTE_OVERLAP_FLOOR
+
+    hit = fake_note(id=20, title="Dev-log day 3", note_type="note")
+    sem = AsyncMock(return_value=[(0.93, hit)])
+    overlaps: list = []
+    with patch("scribe.services.dedup.async_session",
+               return_value=session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(
+            7, "Dev-log day 4", body="x" * 250, project_id=2, is_task=False,
+            note_type="note", overlaps=overlaps,
+        )
+    assert dup is None
+    assert [(o.id, o.similarity) for o in overlaps] == [(20, 0.93)]
+    # Asked at the overlap floor, so the one search serves both answers.
+    assert sem.await_args.kwargs["threshold"] == _NOTE_OVERLAP_FLOOR
+
+
+@pytest.mark.asyncio
+async def test_a_record_matching_in_two_chunks_is_one_overlap():
+    para = ("A paragraph long enough for the chunker to keep as its own "
+            "section of real content in this test body. ") * 4
+    body = "\n\n".join(f"## Part {i}\n\n{para} (p{i})" for i in range(8))
+    hit = fake_note(id=31, title="Design part one", note_type="note")
+    sem = AsyncMock(return_value=[(0.9, hit)])
+    overlaps: list = []
+    with patch("scribe.services.dedup.async_session",
+               return_value=session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        await find_duplicate_note(
+            7, "Design part two", body=body, project_id=2, note_type="note",
+            overlaps=overlaps,
+        )
+    assert [o.id for o in overlaps] == [31]
+
+
+@pytest.mark.asyncio
+async def test_without_an_overlaps_list_the_gate_asks_at_the_copy_band():
+    """A caller that only wants the block (create_process) does not pay for
+    a wider search it will not read."""
+    from scribe.services.dedup import _NOTE_COPY_THRESHOLD
+
+    sem = AsyncMock(return_value=[])
+    with patch("scribe.services.dedup.async_session",
+               return_value=session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        await find_duplicate_note(7, "T", body="x" * 250, note_type="note")
+    assert sem.await_args.kwargs["threshold"] == _NOTE_COPY_THRESHOLD
+
+
+def test_the_overlap_reply_leaves_the_judgement_to_the_reader():
+    from scribe.services.dedup import NoteOverlap, note_overlap_response
+
+    assert note_overlap_response([], "note") == {}
+    out = note_overlap_response([NoteOverlap(9, "Dev-log day 3", 0.93)], "task")
+    assert out["overlaps"] == [{"id": 9, "title": "Dev-log day 3", "similarity": 0.93}]
+    assert "update_task" in out["overlap_note"]
+    assert "keep both" in out["overlap_note"]
+
+
+def test_a_batch_overlap_names_its_record():
+    from scribe.services.dedup import NoteOverlap, batch_overlap_response
+
+    assert batch_overlap_response({}) == {}
+    out = batch_overlap_response({2: [NoteOverlap(9, "X", 0.9)]})
+    assert out["overlaps"] == [{"record": 2, "id": 9, "title": "X", "similarity": 0.9}]
 
 
 @pytest.mark.asyncio
@@ -73,7 +147,7 @@ async def test_gate_catches_a_duplicate_hiding_in_a_later_chunk():
 
     hit = fake_note(id=30, title="The existing decision", note_type="note")
     # Every chunk misses except the LAST one the gate will ask about.
-    sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.94, hit)]])
+    sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.99, hit)]])
     with patch("scribe.services.dedup.async_session",
                return_value=session_returning(None)), \
          patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
diff --git a/tests/test_services_notes.py b/tests/test_services_notes.py
index e4b9de9..f467fc0 100644
--- a/tests/test_services_notes.py
+++ b/tests/test_services_notes.py
@@ -8,6 +8,8 @@ MCP, recurrence, snippets — gets it by construction rather than by remembering
 These test the helper directly. The point of the change is that there is now ONE
 place to test.
 """
+import pytest
+
 from scribe.services import notes as notes_svc
 
 # --- inline embedding (#2056) -----------------------------------------------
@@ -69,3 +71,36 @@ def test_embed_note_swallows_an_indexing_failure():
     note = MagicMock(id=5, user_id=42, title="T", body="B")
     with patch("asyncio.create_task", side_effect=ValueError("model gone")):
         notes_svc.embed_note(note)          # must not raise
+
+
+# --- #3683: a create that names a status is the transition an update is ------
+
+_LIFECYCLE = ("started_at", "completed_at", "recurrence_next_spawn_at")
+
+
+@pytest.mark.parametrize("status", ["todo", "in_progress", "done", "cancelled"])
+def test_a_task_created_at_a_status_matches_one_updated_to_it(status):
+    """The invariant, not the instance that surfaced it: which lifecycle
+    stamps a row carries must not depend on whether it was created at a status
+    or reached it by update. Recurrence is included because done/cancelled
+    schedule the next spawn."""
+    rule = {"type": "interval", "unit": "week", "every": 1}
+    created = notes_svc.build_note(1, title="t", status=status, recurrence_rule=rule)
+
+    updated = notes_svc.build_note(1, title="t", recurrence_rule=rule)
+    updated.status = status
+    notes_svc.apply_status_transition(updated)
+
+    for field in _LIFECYCLE:
+        assert (getattr(created, field) is None) == (getattr(updated, field) is None), (status, field)
+
+
+def test_a_task_created_in_progress_knows_when_it_started():
+    note = notes_svc.build_note(1, title="t", status="in_progress")
+    assert note.started_at is not None
+    assert note.completed_at is None
+
+
+def test_a_note_with_no_status_gets_no_lifecycle_stamps():
+    note = notes_svc.build_note(1, title="t")
+    assert all(getattr(note, f) is None for f in _LIFECYCLE)
diff --git a/tests/test_task_claims.py b/tests/test_task_claims.py
new file mode 100644
index 0000000..ffe0580
--- /dev/null
+++ b/tests/test_task_claims.py
@@ -0,0 +1,301 @@
+"""A task's claim — which session is working it, and how it dies (milestone 381 step 2).
+
+WHAT THIS PINS
+
+  1. **The write that is the work stamps it.** Reaching `in_progress` — by
+     update OR by a create that names it (#3683) — claims the task for whoever
+     made the change. Nothing asks the model to claim anything.
+  2. **Ending or un-starting the work releases it**, and releasing is
+     idempotent.
+  3. **A claim dies on read.** Past the lease it reads `live: false` with
+     nothing having run — no sweep, which is the whole design.
+  4. **The latest worker holds it**, and `since` restarts when the holder
+     changes; the same holder touching a live claim keeps `since`.
+  5. **The session is bound, never created.** `bind_session` attaches a
+     harness-reported id only to a live claim the caller already holds.
+"""
+import json
+import re
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+
+from scribe.services import notes as notes_svc
+from scribe.services import task_claims as tc
+from tests.helpers import ensure_user
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _task(status="todo"):
+    return notes_svc.build_note(1, title="t", status=status)
+
+
+# --- 1. the write that is the work stamps it ---------------------------------
+
+def test_a_task_created_in_progress_is_claimed_by_its_creator():
+    note = notes_svc.build_note(42, title="t", status="in_progress")
+    state = tc.claim_state(note)
+    assert state["held_by"] == 42 and state["live"] is True
+    assert state["session"] is None, "the server never invents a session"
+
+
+def test_reaching_in_progress_by_update_claims_for_the_acting_user():
+    note = _task()
+    note.status = "in_progress"
+    notes_svc.apply_status_transition(note, user_id=9)
+    assert tc.claim_state(note)["held_by"] == 9
+
+
+def test_a_task_never_worked_has_no_claim():
+    assert tc.claim_state(_task()) is None
+    assert _task().to_dict()["claim"] is None
+
+
+# --- 2. ending the work releases it ------------------------------------------
+
+@pytest.mark.parametrize("status", ["done", "cancelled", "todo"])
+def test_ending_or_unstarting_the_work_releases_the_claim(status):
+    note = notes_svc.build_note(42, title="t", status="in_progress")
+    note.status = status
+    notes_svc.apply_status_transition(note, user_id=42)
+    assert tc.claim_state(note) is None
+
+
+def test_releasing_nothing_is_not_an_error():
+    note = _task()
+    tc.release_claim(note)
+    tc.release_claim(note)
+    assert tc.claim_state(note) is None
+
+
+# --- 3. a claim dies on read -------------------------------------------------
+
+def test_a_claim_past_its_lease_reads_dead_with_nothing_having_run():
+    then = datetime.now(timezone.utc) - tc.CLAIM_LEASE - timedelta(minutes=1)
+    note = _task("in_progress")
+    tc.stamp_claim(note, 42, now=then)
+    state = tc.claim_state(note)
+    assert state["live"] is False
+    assert state["held_by"] == 42, "a dead claim is still reported, not hidden"
+
+
+def test_in_progress_and_unclaimed_is_now_representable():
+    """The state the milestone exists to make sayable: committed, nobody on it."""
+    then = datetime.now(timezone.utc) - timedelta(days=3)
+    note = _task("in_progress")
+    tc.stamp_claim(note, 42, now=then)
+    assert note.status == "in_progress" and not tc.claim_is_live(note)
+
+
+# --- 4. the latest worker holds it -------------------------------------------
+
+def test_the_same_holder_touching_a_live_claim_keeps_since():
+    t0 = datetime.now(timezone.utc) - timedelta(minutes=30)
+    note = _task("in_progress")
+    tc.stamp_claim(note, 42, now=t0)
+    tc.stamp_claim(note, 42)
+    assert note.claimed_at == t0
+    assert note.claim_touched_at > t0
+
+
+def test_another_worker_takes_the_claim_over_and_since_restarts():
+    t0 = datetime.now(timezone.utc) - timedelta(minutes=30)
+    note = _task("in_progress")
+    tc.stamp_claim(note, 42, now=t0)
+    note.claim_session = "a-session"
+    tc.stamp_claim(note, 7)
+    assert note.claimed_by == 7
+    assert note.claimed_at > t0
+    assert note.claim_session is None, "the old holder's session is not the new one's"
+
+
+def test_a_dead_claim_restarts_rather_than_resumes():
+    t0 = datetime.now(timezone.utc) - tc.CLAIM_LEASE - timedelta(hours=1)
+    note = _task("in_progress")
+    tc.stamp_claim(note, 42, now=t0)
+    tc.stamp_claim(note, 42)
+    assert note.claimed_at > t0
+
+
+# --- the plugin half ---------------------------------------------------------
+
+def test_the_binder_hook_watches_the_two_writes_that_stamp_a_claim():
+    hooks = json.loads((ROOT / "plugin/hooks/hooks.json").read_text())["hooks"]
+    blocks = [b for b in hooks["PostToolUse"]
+              if any("scribe_claim_session.sh" in h["command"] for h in b["hooks"])]
+    assert len(blocks) == 1
+    matcher = re.compile(blocks[0]["matcher"])
+    assert matcher.fullmatch("mcp__plugin_scribe_scribe__update_task")
+    assert matcher.fullmatch("mcp__scribe__add_task_log")
+    assert not matcher.fullmatch("mcp__scribe__get_task")
+
+
+# --- 5. binding a session (integration) --------------------------------------
+
+@pytest_asyncio.fixture
+async def users(_dispose_engine):
+    from scribe.models import async_session
+
+    async with async_session() as session:
+        a = (await ensure_user(session, "claims_itest_a")).id
+        b = (await ensure_user(session, "claims_itest_b")).id
+        await session.commit()
+    return a, b
+
+
+@pytest.mark.integration
+async def test_a_session_binds_to_its_own_live_claim_only(users):
+    owner, stranger = users
+    task = await notes_svc.create_note(owner, title="claim bind", status="in_progress")
+
+    assert await tc.bind_session(stranger, task.id, "their-session") is None
+
+    bound = await tc.bind_session(owner, task.id, "sess-1")
+    assert bound["session"] == "sess-1" and bound["live"] is True
+
+    # A different session taking over a live claim restarts `since`.
+    again = await tc.bind_session(owner, task.id, "sess-2")
+    assert again["session"] == "sess-2"
+    assert again["since"] >= bound["since"]
+
+
+@pytest.mark.integration
+async def test_a_closed_task_binds_nothing(users):
+    owner, _ = users
+    task = await notes_svc.create_note(owner, title="claim closed", status="in_progress")
+    await notes_svc.update_note(owner, task.id, status="done")
+    assert await tc.bind_session(owner, task.id, "sess") is None
+
+
+# --- step 3: the readers ----------------------------------------------------
+
+from types import SimpleNamespace  # noqa: E402
+
+_NOW = datetime(2026, 9, 24, 12, 0, tzinfo=timezone.utc)
+
+
+def _claimed(id, session, touched_ago, status="in_progress", title=None):
+    t = _NOW - touched_ago
+    return SimpleNamespace(
+        id=id, title=title or f"task {id}", status=status,
+        claimed_by=1, claimed_at=t, claim_touched_at=t, claim_session=session,
+    )
+
+
+def _log(content, ago=timedelta(minutes=5)):
+    return SimpleNamespace(content=content, created_at=_NOW - ago)
+
+
+def _render(source, claims, logs=None, sid="me"):
+    return "\n".join(tc.render_claims(source, sid, claims, logs or {}, now=_NOW))
+
+
+def test_a_compaction_gets_back_its_own_claimed_work_and_latest_logs():
+    """The measurement the milestone names: a compacted session comes back
+    holding its own state, without being told to go looking."""
+    mine = _claimed(10, "me", timedelta(minutes=3), title="wire the reader")
+    out = _render("compact", [mine], {10: [_log("ruled out the cache theory")]})
+    assert "#10" in out and "wire the reader" in out
+    assert "ruled out the cache theory" in out
+
+
+def test_a_resume_says_nothing():
+    mine = _claimed(10, "me", timedelta(minutes=3))
+    assert _render("resume", [mine]) == ""
+
+
+def test_a_startup_names_other_sessions_live_and_abandoned_claims():
+    live = _claimed(11, "other", timedelta(minutes=10))
+    gone = _claimed(12, "older", timedelta(days=3))
+    out = _render("startup", [live, gone])
+    assert "#11" in out and "may still be running" in out
+    assert "#12" in out and "went quiet" in out
+
+
+def test_a_startup_does_not_push_this_sessions_own_work():
+    """A new session id owns nothing yet; the own-work push is for a context
+    that was lost, not one that never existed."""
+    mine = _claimed(10, "me", timedelta(minutes=3))
+    assert "In flight" not in _render("startup", [mine])
+
+
+def test_a_fork_is_told_two_sessions_may_hold_the_same_claim():
+    parent = _claimed(13, "parent", timedelta(minutes=2))
+    out = _render("fork", [parent], sid="child")
+    assert "forked" in out and "#13" in out
+
+
+def test_a_dead_claim_on_finished_work_is_not_news():
+    done = _claimed(14, "older", timedelta(days=3), status="done")
+    assert _render("startup", [done]) == ""
+
+
+def test_the_session_start_hook_sends_the_source_and_the_session():
+    """The reader branches on what the hook sends; a hook that stopped sending
+    either would leave every session on the no-claims path, silently."""
+    text = (ROOT / "plugin/hooks/scribe_session_context.sh").read_text()
+    assert re.search(r'q="source=\$\(printf', text)
+    assert "session_id=$(printf" in text
+
+
+@pytest.mark.integration
+async def test_session_start_after_a_compaction_carries_the_claimed_task(users):
+    from scribe.services import task_logs
+    from scribe.services.plugin_context import build_session_context
+
+    owner, _ = users
+    task = await notes_svc.create_note(owner, title="claimed then compacted",
+                                       status="in_progress")
+    await task_logs.create_log(owner, task.id, "halfway: the migration is written")
+    await tc.bind_session(owner, task.id, "sess-compact")
+
+    ctx = (await build_session_context(
+        owner, source="compact", session_id="sess-compact"))["context"]
+    assert "claimed then compacted" in ctx
+    assert "the migration is written" in ctx
+
+
+# --- step 4: the hand-off ----------------------------------------------------
+
+def test_session_end_releases_claims_but_not_on_clear():
+    """The mechanical half of the hand-off. A /clear keeps the claim, because
+    SessionStart(source=clear) pushes the claimed work straight back."""
+    hooks = json.loads((ROOT / "plugin/hooks/hooks.json").read_text())["hooks"]
+    commands = [h["command"] for b in hooks.get("SessionEnd", []) for h in b["hooks"]]
+    assert any("scribe_session_end.sh" in c for c in commands)
+    text = (ROOT / "plugin/hooks/scribe_session_end.sh").read_text()
+    assert '[ "$reason" = "clear" ] && exit 0' in text
+    assert "/api/plugin/release-session" in text
+
+
+@pytest.mark.integration
+async def test_ending_a_session_releases_only_that_sessions_claims(users):
+    owner, _ = users
+    kept = await notes_svc.create_note(owner, title="other session's", status="in_progress")
+    gone = await notes_svc.create_note(owner, title="ending session's", status="in_progress")
+    await tc.bind_session(owner, kept.id, "sess-stays")
+    await tc.bind_session(owner, gone.id, "sess-ends")
+
+    assert await tc.release_session(owner, "sess-ends") >= 1
+    assert (await notes_svc.get_note(owner, gone.id)).claimed_at is None
+    assert (await notes_svc.get_note(owner, kept.id)).claim_session == "sess-stays"
+    assert await tc.release_session(owner, "") == 0
+
+
+@pytest.mark.integration
+async def test_a_collaborator_with_write_access_can_log_and_so_claims(users):
+    """create_log used to filter on the task's OWNER (a rule #78 violation), so
+    a collaborator on a shared task could not log on it — nor, since logging
+    stamps the claim, ever be seen working it."""
+    from scribe.services import sharing, task_logs
+
+    owner, collaborator = users
+    task = await notes_svc.create_note(owner, title="shared work", status="todo")
+    await sharing.share_note(owner, task.id, target_user_id=collaborator,
+                             permission="editor")
+    await task_logs.create_log(collaborator, task.id, "picked this up")
+    got = await notes_svc.get_note(owner, task.id)
+    assert got.claimed_by == collaborator
diff --git a/tests/test_task_document_shape.py b/tests/test_task_document_shape.py
index 92dc9f0..affa0ee 100644
--- a/tests/test_task_document_shape.py
+++ b/tests/test_task_document_shape.py
@@ -125,7 +125,7 @@ def test_a_short_task_with_a_short_log_is_still_one_sharp_chunk():
 # is #4241's half-surface one layer down: the entry is readable, and the search
 # still answers as though it were never written.
 
-from unittest.mock import MagicMock, patch          # noqa: E402
+from unittest.mock import AsyncMock, MagicMock, patch  # noqa: E402
 
 import pytest                                       # noqa: E402
 
@@ -141,6 +141,7 @@ async def test_writing_a_work_log_refreshes_the_tasks_embedding():
     session = session_returning(note)
     with (
         patch.object(svc, "async_session", return_value=session),
+        patch.object(svc, "can_write_note", AsyncMock(return_value=True)),
         patch("scribe.services.notes.embed_note") as embed,
     ):
         await svc.create_log(42, 7, "what I tried")
@@ -193,6 +194,7 @@ async def test_a_failed_refresh_does_not_fail_the_log_that_saved():
     session = session_returning(fake_note(id=7, title="T", body="b", is_task=True))
     with (
         patch.object(svc, "async_session", return_value=session),
+        patch.object(svc, "can_write_note", AsyncMock(return_value=True)),
         patch("scribe.services.notes.embed_note", side_effect=RuntimeError("boom")),
     ):
         log = await svc.create_log(42, 7, "what I tried")