feat(tasks): SessionStart reads the claim — a compaction gets its work back (milestone 381 step 3)

The claim now pays rent to the session that set it. The SessionStart hook
sends the host's `source` and the session id; the server renders a claim
section by source (task_claims.render_claims):

- compact / clear: the work this session had claimed, each with its two
  latest log entries, so a compacted session resumes from the record
  rather than from a count of open tasks.
- startup / clear: other sessions' live claims (may still be running) and
  in-progress tasks whose claim went quiet (abandoned mid-task).
- fork: the same, framed as "two sessions may now hold this".
- resume, or no source sent: nothing.

The task sidebar shows a live claim as "Being worked" and a dead one on
open work as "Went quiet", so the operator sees a session die mid-task.

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

src/scribe/routes/plugin.py

@@ -70,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)
 

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,

src/scribe/services/task_claims.py

@@ -123,3 +123,161 @@ async def bind_session(user_id: int, task_id: int, session_id: str) -> dict | No
         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)