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