FabledScribe/tests/test_task_claims.py

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