feat(tasks): a task can say a session is working it — the claim (milestone 381 step 2)

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 so it dies on read rather than needing to be cleared.

- notes.claimed_by / claimed_at / claim_touched_at / claim_session (0110).
- The server stamps it on the write that is the work: reaching in_progress
  (update or create, via apply_status_transition) and a work log on an open
  task. done/cancelled/todo release it. Live while touched within
  CLAIM_LEASE (2h); dead on read past it, with no sweep.
- The plugin's PostToolUse hook on update_task/add_task_log binds the
  harness's session_id (GET /api/plugin/claim-session). It binds only to a
  live claim the caller holds; it cannot create one.
- to_dict carries `claim`. Plugin version minted.

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

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")

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.23.2309",
   "author": {
     "name": "Bryan Van Deusen"
   },

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": [

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

scripts/check_plugin.py

@@ -386,6 +386,15 @@ 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 shared library is sourced, never run; executed bare it defines
     # functions and exits — silent by construction.
     "scribe_defs.sh": "",

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)

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")
@@ -352,6 +353,35 @@ 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("/processes")
 @login_required
 async def process_manifest():

src/scribe/services/notes.py

@@ -308,23 +308,32 @@ def build_note(
     # (#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)
+        apply_status_transition(note, user_id)
     return note
 
 
-def apply_status_transition(note: Note) -> None:
+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
@@ -334,6 +343,7 @@ def apply_status_transition(note: Note) -> None:
                 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
@@ -687,7 +697,7 @@ async def update_note(
             if recompose is not None:
                 note.data = recompose(note)
         if "status" in fields:
-            apply_status_transition(note)
+            apply_status_transition(note, user_id)
         note.updated_at = datetime.now(timezone.utc)
         await session.commit()
         await session.refresh(note)

src/scribe/services/task_claims.py

@@ -0,0 +1,125 @@
+"""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)

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.models.note import Note, TaskStatus
 from scribe.services.access import can_read_note, readable_notes_clause
+from scribe.services.task_claims import stamp_claim
 
 logger = logging.getLogger(__name__)
 
@@ -55,8 +56,14 @@ async def create_log(
         result = await session.execute(
             select(Note).where(Note.id == task_id, Note.user_id == user_id)
         )
-        if result.scalars().first() is None:
+        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,

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)
 
 

tests/test_task_claims.py

@@ -0,0 +1,170 @@
+"""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