clanker-saddle/test/bundles/lifecycle.py

"""The lifecycle bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: lifecycle — offline requeues, the hung-request tick survival,
# parked-refresh bookkeeping, cancellation, the health panel, raw-log
# losslessness, git identity and the live blocked-push roundtrip.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
    app = ctx.app

    print("== plan 11: draft state machine, edit matrix, enqueue/delete guards ==")
    # New tasks default to draft; Enqueue is the only path to the
    # queue (there is no agent path — the ingest API refuses to queue).
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Draft lifecycle task",
        "priority": "4",
    })
    draft_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                      if t["title"] == "Draft lifecycle task"][0])
    detail = task_state(auth, draft_task)
    check(detail["status"] == "draft", "a created task lands in draft")
    check(detail.get("enqueue") is not None and detail.get("delete") is not None,
          "draft exposes the enqueue + delete actions")
    check(detail.get("cancel") is not None, "the cancel action stays authored")
    status, page = http("GET", "/tasks/%d" % draft_task, timeout=15)
    check(status == 200 and "Enqueue task" in page and ">Delete<" in page,
          "the task page shows Enqueue + Delete in draft")
    check("Mark done" not in page,
          "the user Mark-complete action is gone (agent-only completion)")
    # Plan 12: the auto-enqueue-after predecessor REPLACES draft
    # dependencies — setting it is allowed while the predecessor is
    # still pending (the enqueue itself succeeds; the DISPATCHER holds
    # dispatch back until the predecessor is done).
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Draft dependency",
        "priority": "4",
    })
    draft_dep = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Draft dependency"][0])
    act(auth, detail["update"], {"enqueue_after_task_id": str(draft_dep)})
    check(int(task_state(auth, draft_task).get("enqueue_after_task_id") or 0)
          == draft_dep,
          "the draft edit sets the auto-enqueue predecessor")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Eager dependent", "priority": "4",
        "enqueue_after_task_id": str(draft_dep)})
    eager = task_id_of("Eager dependent")
    enqueue_task(auth, eager)
    check(task_state(auth, eager)["status"] == "queued",
          "enqueue succeeds while the predecessor is pending (no draft guard)")
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, eager)["status"] == "queued",
                       "pending predecessor holds dispatch"),
          "the pending predecessor holds dispatch back (never assigned)")
    # The edit matrix starts in DRAFT: every field except status.
    act(auth, detail["update"], {
        "title": "Draft lifecycle task (edited)",
        "description": "Now with a description",
        "priority": "2",
        "task_branch": "feature/draft",
        "branch_from": "main",
        "mode": "investigate",
    })
    detail = task_state(auth, draft_task)
    check(detail["title"] == "Draft lifecycle task (edited)"
          and detail["description"] == "Now with a description"
          and detail["priority"] == 2
          and detail["task_branch"] == "feature/draft"
          and detail["branch_from"] == "main"
          and detail["mode"] == "investigate",
          "draft edits apply to every field (incl. the mode)")
    # Clear the predecessor (the empty sentinel): dispatch gates on
    # the hold tag alone now.
    act(auth, detail["update"], {"enqueue_after_task_id": "0"})
    check(not int(task_state(auth, draft_task).get("enqueue_after_task_id") or 0),
          "the draft edit clears the auto-enqueue predecessor")
    # A hold tag nothing carries keeps the dependent QUEUED (never
    # dispatched) through the remaining edit-matrix checks.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "holdlock"})
    holdlock_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                        if c["name"] == "holdlock"][0])
    act(auth, detail["attach_capability"], {"capability_id": str(holdlock_cap)})
    # Enqueue the now-cleared dependent: the hold tag keeps it queued
    # (never dispatched), so the non-draft edit matrix applies.
    enqueue_task(auth, draft_task)
    check(quiet_window(3, lambda: task_state(auth, draft_task)["status"] == "queued",
                       "held dependent stays queued"),
          "the held dependent stays queued (never dispatched)")
    # The predecessor settles for the shard drain.
    enqueue_task(auth, draft_dep)
    check(wait_task_done(auth, draft_dep, timeout=90) is not None,
          "the predecessor settles")

    # Drafts are never dispatched even with an idle saddle available.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Never-dispatched draft",
        "priority": "1",
    })
    silent_draft = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Never-dispatched draft"][0])
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, silent_draft)["status"] == "draft",
                       "draft never dispatched"),
          "an idle saddle never picks a draft up (no agent enqueue path)")

    # Once work can begin (queued here), only priority is editable.
    act(auth, detail["update"], {
        "title": "Should not apply",
        "priority": "3",
    })
    detail = task_state(auth, draft_task)
    check(detail["title"] == "Draft lifecycle task (edited)" and detail["priority"] == 3,
          "non-draft edits apply priority only (title locked)")
    # A terminal task edits nothing.
    conn = db()
    conn.execute("update tasks set status = 'done', completed_at = strftime('%s','now')"
                 " where id = ?", (draft_task,))
    conn.commit()
    conn.close()
    act(auth, task_state(auth, draft_task)["update"], {"priority": "8"})
    check(task_state(auth, draft_task)["priority"] == 3,
          "terminal tasks edit nothing (priority locked)")
    # Capabilities lock once a worker has the job: editable while
    # queued, locked once working.
    act(auth, saddles_state["create_capability"], {"name": "editlock"})
    editlock_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")
                        ["capabilities"] if c["name"] == "editlock"][0])
    act(auth, saddle_card(auth, "fake-saddle")["attach_capability"],
        {"capability_id": str(editlock_cap)})
    locked_task = create_queued(auth, {"project_id": "1", "title": "Caplock task",
                                       "priority": "2"})
    detail = task_state(auth, locked_task)
    act(auth, detail["attach_capability"], {"capability_id": str(editlock_cap)})
    check(wait_for(lambda: any(c["name"] == "editlock"
                               for c in task_state(auth,
                                                   locked_task)["capabilities"])
                   or False, timeout=60, what="editlock attached") is not None,
          "capabilities attach while queued (no worker yet)")
    check(wait_for(lambda: task_state(auth, locked_task).get("status") == "working"
                   or False, timeout=60, what="caplock working") is not None,
          "the caplock task reaches working")
    detach_directives = act(auth, task_state(auth, locked_task)["capabilities"][0]["detach"], {})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              for d in detach_directives),
          "capabilities are locked once a worker has the job")
    # The ingest API has no queue transition either (only working and
    # blocked are accepted states).
    conn = db()
    caplock_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (locked_task,)).fetchone()
    conn.close()
    if caplock_session is not None:
        status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                               {"session_id": caplock_session[0],
                                "state": "queued"})
        check(status == 200
              and task_state(auth, locked_task)["status"] in ("working", "done"),
              "a queued state push is a no-op (only the wire's states drive bookkeeping)")

    # Delete guards: a clean draft deletes; non-draft refuses; a draft
    # with dependents is refused naming them.
    act(auth, task_state(auth, silent_draft)["delete"], {})
    check(not any(t["id"] == silent_draft
                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
          "a clean draft task deletes")
    delete_directives = act(auth, task_state(auth, draft_task)["delete"], {})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              for d in delete_directives),
          "a non-draft task refuses the delete")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Guarded draft", "priority": "4"})
    guarded = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                   if t["title"] == "Guarded draft"][0])
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Dependent namer", "priority": "4"})
    namer = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                 if t["title"] == "Dependent namer"][0])
    act(auth, task_state(auth, namer)["update"],
        {"enqueue_after_task_id": str(guarded)})
    check(int(task_state(auth, namer).get("enqueue_after_task_id") or 0) == guarded,
          "the namer points at the guarded predecessor")
    act(auth, task_state(auth, guarded)["delete"], {})
    check(not any(t["id"] == guarded
                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
          "deleting a predecessor succeeds")
    check(not int(task_state(auth, namer).get("enqueue_after_task_id") or 0),
          "the delete clears the dependent's dangling predecessor reference")

    print("== offline requeues ==")
    # Tasks caught working on a saddle that went offline requeue with a
    # agent comment each tick.
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Offline saddle task",
        "priority": "2",
    })
    offline_task = task_id_tmp
    check(wait_for(lambda: task_state(auth, offline_task).get("status") == "working"
                   or False, timeout=60, what="offline task starts working") is not None,
          "task reaches working before its saddle disappears")
    # Hold the daemon mid-round across the requeue window (the
    # original relied on its 8s work period outlasting the 5s tick):
    # the task completing on its own would race the offline pass and
    # skip the requeue the check exists to observe.
    fake.hold_state = True
    conn = db()
    conn.execute("update saddles set status = 'offline' where name = 'fake-saddle'")
    conn.commit()
    conn.close()
    # The requeue frees the saddle in the same tick, so 'queued' is
    # transient; the durable evidence is the agent comment (and the task
    # completing again on the healthy daemon).
    check(wait_for(lambda: any(c["body"] == "saddle went offline; task requeued"
                               for c in task_state(auth, offline_task).get("comments", []))
                   or False, timeout=30, what="offline requeue comment") is not None,
          "offline saddle requeues the task")
    fake.hold_state = False  # release: the round ends, the daemon re-asks
    check(wait_task_done(auth, offline_task, timeout=60) is not None,
          "requeued task completes again on the healthy daemon")
    # R2 (arch 13): the release owner's outcomes — the requeue closed
    # the held attempt's session 'stopped'; the held daemon's terminal
    # push then completes the task against that closed session (the
    # outcome stands — a closed row never reopens). A task that
    # completes a LIVE round closes its session 'done' (draft_dep
    # above ran one open session to its terminal push).
    conn = db()
    outcome_rows = conn.execute(
        "select outcome from sessions where task_id = ? order by id",
        (offline_task,)).fetchall()
    dep_outcome = conn.execute(
        "select outcome from sessions where task_id = ? order by id desc",
        (draft_dep,)).fetchone()
    conn.close()
    check(outcome_rows and outcome_rows[0][0] == "stopped",
          "the one release path closes the requeued attempt 'stopped'")
    check(dep_outcome is not None and dep_outcome[0] == "done",
          "the terminal push closes a live round's session 'done'")

    print("== R6 F1: a hanging request cannot freeze the tick ==")
    # One wedged /task-update (never responds) blocks the comment/
    # subtask-event delivery pass; the Soup timeout must fail it and
    # later ticks keep dispatching.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Hang survivor', 'queued', 2, strftime('%s','now'))")
    hang_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.close()
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Hang runner",
        "priority": "3",
    })
    hang_runner = task_id_tmp
    # A human comment on the running task gives the tick a
    # /task-update delivery to make; the next one hangs.
    check(wait_for(lambda: task_state(auth, hang_runner).get("status") == "working"
                   or False, timeout=60, what="hang runner working") is not None,
          "hang runner reaches working")
    act(auth, task_state(auth, hang_runner)["comment"],
        {"body": "operator ping while wedged"})
    fake.hang_next["/task-update"] = 1
    check(wait_task_done(auth, hang_runner, timeout=90) is not None,
          "dispatching task survives a hung daemon request")
    check(wait_for(lambda: any(c.get("text") == "operator ping while wedged"
                               or c.get("body") == "operator ping while wedged"
                               for c in fake.comment_log) or False,
                   timeout=30, what="wedged comment eventually delivered") is not None,
          "the comment delivery retries after the hang")
    # The hung delivery itself fails after the timeout; the system
    # kept moving either way.
    with open(os.path.join(WORK, "app.log")) as handle:
        app_log = handle.read()
    check("tick exceeded" not in app_log or "Dispatch tick failed" in app_log,
          "a hung request never freezes the tick latch")

    print("== R6 F2: refresh must not corrupt parked bookkeeping ==")
    fake.suspend_on_title = "Parked refresh parent"
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Parked refresh parent",
        "priority": "2",
    })
    parked = task_id_tmp
    check(wait_for(lambda: task_state(auth, parked).get("status") == "suspended"
                   or False, timeout=60, what="parked parent suspends") is not None,
          "parent parks before the refresh")
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["refresh"], {})
    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card["status"] == "idle" and int(fake_card["current_task_id"] or 0) == 0,
          "refresh on a parked (suspended) daemon leaves the saddle idle")
    # The parked task keeps its open session and stays put: plain
    # tasks have no subtask events to resume them (the goal machinery
    # owns resumes); the human resumes it explicitly.
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, parked)["status"] == "suspended",
                       "parked task stays parked"),
          "a parked plain task stays parked after the refresh (no resume channel)")
    act(auth, task_state(auth, parked)["cancel"], {})
    check(wait_for(lambda: task_state(auth, parked)["status"] == "failed" or False,
                   timeout=30, what="parked task cancelled") is not None,
          "the parked task cancels cleanly after the refresh")

    print("== R7 B1: cancel reaches the daemon ==")
    # Plan 10 phase 2: cancellation is POST /task-update {kind: cancel}
    # to the RUNNING daemon (the abort machinery, workspace kept).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Cancel live",
        "priority": "2",
    })
    cancel_live = task_id_tmp
    check(wait_for(lambda: task_state(auth, cancel_live).get("status") == "working"
                   or False, timeout=60, what="cancel-live working") is not None,
          "task reaches working before the cancel")
    act(auth, task_state(auth, cancel_live)["cancel"], {})
    check(wait_for(lambda: cancel_live in fake.abort_log or False,
                   timeout=30, what="daemon abort recorded") is not None,
          "cancelling POSTs task-update cancel to the daemon")
    check(any(c["body"] == "cancelled; daemon aborted"
              for c in task_state(auth, cancel_live).get("comments", [])),
          "the abort outcome is narrated")
    check(task_state(auth, cancel_live)["status"] == "failed",
          "cancelled task lands failed as before")

    # A 500 on the cancel marks the saddle offline (no fence layer
    # anymore); nothing dispatches to it while offline, and the
    # daemon's own ask flips the row back to idle (proof of life).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Fence runner",
        "priority": "2",
    })
    fence_runner = task_id_tmp
    check(wait_for(lambda: task_state(auth, fence_runner).get("status") == "working"
                   or False, timeout=60, what="fence runner working") is not None,
          "second task reaches working")
    fake.abort_failures = 1
    act(auth, task_state(auth, fence_runner)["cancel"], {})
    check(any(c["body"] == "cancelled; daemon unreachable — saddle marked offline"
              for c in task_state(auth, fence_runner).get("comments", [])),
          "failed cancel narrates the offline marking")
    check(saddle_card(auth, "fake-saddle")["status"] == "offline",
          "saddle marked offline after the failed cancel")
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "During fence",
        "priority": "3",
    })
    during_fence = task_id_tmp
    # Converted sleep(7): a quiet window spanning several dispatch
    # ticks in which the task MUST stay queued — if the offline
    # saddle ever dispatches it, this fails loudly. The daemon is held
    # mid-round for the window (the original relied on its 8s work
    # period outlasting the sleep): a finishing round would ask, prove
    # liveness and flip the offline row — only after the window may it.
    fake.hold_state = True
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, during_fence)["status"] == "queued",
                       "offline saddle does not dispatch during_fence"),
          "offline saddle is not dispatchable")
    fake.hold_state = False
    # The (unaborted) daemon finishes its round, goes idle-awake and
    # asks — the ask is the proof of life that flips the row back.
    check(wait_for(lambda: saddle_card(auth, "fake-saddle")["status"] == "idle" or False,
                   timeout=30, what="ask flips offline to idle") is not None,
          "the daemon's ask flips the offline row back to idle")
    check(wait_task_done(auth, during_fence, timeout=60) is not None,
          "queued work dispatches once the saddle proves it is alive")

    print("== R7 B3: health panel + pushed events from the first round ==")
    # A normally dispatched task pushes its event batch immediately:
    # ledger rows exist from the first round (the session-4 gap stays
    # fixed without any probe machinery).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Raw from first push",
        "priority": "2",
    })
    raw_first = task_id_tmp

    def event_count(task_id):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from events where session_id in"
                " (select id from sessions where task_id = ?)", (task_id,)).fetchone()[0]
        finally:
            conn.close()

    check(wait_for(lambda: event_count(raw_first) > 0 or False,
                   timeout=60, what="event rows from first push") is not None,
          "pushed events land from the first round (session-4 gap)")
    check(wait_task_done(auth, raw_first, timeout=60) is not None,
          "raw-first task completes")

    # Health panel: a manual refresh populates the snapshot.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["refresh"], {})
    card = saddle_card(auth, "fake-saddle")
    check(card.get("health") in ("healthy", "stale")
          and card.get("daemon_state") in ("working", "idle", "done")
          and card.get("uptime_display", "") != ""
          and card.get("last_contact_display", "") not in ("", "never"),
          "saddles page shows daemon state, uptime and last contact")

    print("== R8: pushed chunk ingestion is lossless ==")
    # The daemon's opening flush: a coloured prompt run, plain text,
    # a bare CRLF between prompts and the tail — every non-empty
    # chunk must land, exactly once, in seq order. A mid-flight
    # redelivery of the whole batch (the outbox's at-least-once
    # retry) consumes the duplicates.
    corpus = [
        {"seq": 1, "kind": "command_output",
         "payload": {"text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "}},
        {"seq": 2, "kind": "command_output", "payload": {"text": "echo hello\r\n"}},
        {"seq": 3, "kind": "command_output", "payload": {"text": "\r\n"}},
        {"seq": 4, "kind": "command_output", "payload": {"text": "hello\r\n"}},
        {"seq": 5, "kind": "command_output",
         "payload": {"text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "}},
        {"seq": 6, "kind": "command_output", "payload": {"text": "done\r\n"}},
    ]
    fake.scripted_corpus = corpus
    fake.hold_state = True
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Lossless log task",
        "priority": "2",
    })
    lossless = task_id_tmp

    def stored_seqs(task_id):
        conn = db()
        try:
            return [r[0] for r in conn.execute(
                "select seq from events where session_id in"
                " (select id from sessions where task_id = ?)"
                " and seq is not null order by seq", (task_id,))]
        finally:
            conn.close()

    check(wait_for(lambda: task_state(auth, lossless).get("status") == "working"
                   or False, timeout=60, what="lossless task working") is not None,
          "lossless task dispatches and pushes its opening flush")
    check(wait_for(lambda: stored_seqs(lossless) == [1, 2, 3, 4, 5, 6] or False,
                   timeout=30, what="all six chunks stored") is not None,
          "every pushed event lands in the ledger, CRLF-only included (%s)"
          % stored_seqs(lossless))
    status, _ = saddle_api("POST", "/api/saddle/chunks", fake.token,
                           {"session_id": fake.session_id,
                            "events": [dict(chunk) for chunk in corpus]})
    check(status == 200 and stored_seqs(lossless) == [1, 2, 3, 4, 5, 6],
          "a full-batch redelivery stores nothing new (exactly-once)")
    request = urllib.request.Request(
        WEB + "/tasks/%d/events/stream" % lossless)
    with urllib.request.urlopen(request, timeout=15) as stream:
        texts = []
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and len(texts) < 6:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("kind") == "command_output":
                texts.append(event.get("payload", {}).get("text") or "")
        check("\r\n" in texts and len(texts) == 6,
              "SSE serves the whitespace payload unfiltered")
    fake.hold_state = False
    fake.scripted_corpus = None
    check(wait_task_done(auth, lossless, timeout=60) is not None,
          "lossless task completes")

    print("== R9 B1/B2: git identity + ssh key ==")
    # The key travels on /status into the saddle row and the page view
    # (the manual refresh's exceptional probe).
    fake.ssh_public_key = "ssh-ed25519 AAAAC3NzaC1lZDI1teStKey saddle-1@fedora"
    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card.get("ssh_public_key", "") == "",
          "no key shown before the daemon reports one")
    act(auth, fake_card["refresh"], {})
    check(wait_for(lambda: "AAAAC3NzaC1lZDI1teStKey" in
                   saddle_card(auth, "fake-saddle").get("ssh_public_key", "") or False,
                   timeout=30, what="key lands on the row") is not None,
          "reported ssh_public_key persists onto the saddle row")
    conn = db()
    row_key = conn.execute("select ssh_public_key from saddles where name = 'fake-saddle'"
                           ).fetchone()[0]
    conn.close()
    check(row_key == fake.ssh_public_key,
          "the key is stored verbatim in the database")

    # Identity edit flows through the UI into the assignment payload.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["update_identity"],
        {"git_name": "clanker-bot", "git_email": "bot@unitatem.net"})
    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card.get("git_name") == "clanker-bot"
          and fake_card.get("git_email") == "bot@unitatem.net",
          "inline edit saves the git identity")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Identity task",
        "priority": "2",
    })
    identity_task = task_id_tmp
    check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == identity_task
                               for p in fake.start_payloads) or False,
                   timeout=60, what="identity task dispatched") is not None,
          "identity task dispatches")
    payload = [p for p in fake.start_payloads
               if int(p.get("task", {}).get("id") or 0) == identity_task][-1]
    identity = payload.get("git_identity") or {}
    check(identity.get("name") == "clanker-bot"
          and identity.get("email") == "bot@unitatem.net",
          "the assignment carries the saddle's git_identity")

    # Earlier assignments (identity unset) carry no git_identity member.
    check(all("git_identity" not in p for p in fake.start_payloads[:-1]),
          "unset identity omits the member (daemon fallback applies)")

    # Clearing the identity omits it from later assignments.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["update_identity"], {"git_name": "", "git_email": ""})
    check(saddle_card(auth, "fake-saddle").get("git_name", "") == "",
          "clearing the identity empties the row")
    check(wait_task_done(auth, identity_task, timeout=60) is not None,
          "identity task completes")

    print("== phase 2: blocked push -> block row + pulse ==")
    # The browser flow on each pulse: statum.act(task.refresh) — the
    # sealed GET action must rehydrate the task slot with the new
    # block the daemon just PUSHED (state blocked + block payload).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Live block task",
        "priority": "2",
    })
    live_task = task_id_tmp
    check(wait_for(lambda: task_state(auth, live_task).get("status") == "working"
                   or False, timeout=60, what="live task working") is not None,
          "live-block task reaches working")
    fake.hold_state = True
    fake.state = "blocked"  # the daemon parks on its question
    status = fake.push_results("blocked", block={
        "question": "Pick one", "type": "question", "options": ["A", "B"]})
    check(status == 200, "blocked state push accepted")
    # The parked daemon is idle-awake: it asks (and sleeps on 204).
    fake.maybe_trigger_ask("parked")
    check(wait_for(lambda: any(b.get("question") == "Pick one"
                               for b in task_state(auth, live_task).get("blocks", []))
                   or False, timeout=30, what="block lands in state") is not None,
          "the pushed block reaches the sealed refresh without a reload")
    detail = task_state(auth, live_task)
    check(detail.get("status") == "blocked",
          "the blocked push flips the task to blocked")
    pushed_block = [b for b in detail["blocks"] if b.get("question") == "Pick one"][0]
    conn = db()
    options_json = conn.execute(
        "select options from task_blocks where id = ?", (int(pushed_block["id"]),)
    ).fetchone()[0]
    conn.close()
    check(json.loads(options_json) == ["A", "B"],
          "the block row carries its options payload")
    check(any(s["status"] == "idle"
              for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
          "the blocked push parks the saddle")
    detail = task_state(auth, live_task)
    check(detail.get("refresh") is not None,
          "the sealed GET refresh action is still authored")
    status, _ = http("GET", detail["refresh"]["uri"], timeout=15)
    check(status in (200, 401),
          "the refresh URI resolves as a GET (client roundtrip shape)")
    act(auth, pushed_block["answer_action"], {"answer": "A"})
    # The answer makes the parked task resume-offerable: the affine
    # saddle is woken, asks, and its assignment carries the answer.
    pushes_before = len([p for p in fake.start_payloads
                         if int(p.get("task", {}).get("id") or 0) == live_task])
    check(wait_for(lambda: len([p for p in fake.start_payloads
                                if int(p.get("task", {}).get("id") or 0) == live_task])
                   > pushes_before or False, timeout=60,
                   what="blocked task resumes") is not None,
          "answering the pushed block resumes the parked workspace")
    fake.hold_state = False  # the resumed round runs to completion
    resume_payload = [p for p in fake.start_payloads
                      if int(p.get("task", {}).get("id") or 0) == live_task][-1]
    answers = resume_payload.get("task", {}).get("answers", [])
    check(any(a.get("answer") == "A" for a in answers),
          "the resume assignment carries the human's answer")
    check(wait_task_done(auth, live_task, timeout=60) is not None,
          "live-block task completes after answering")

    print("== R2: the one release path behind unregister ==")
    # Arch 13 + plan 16 item 23: removing a saddle releases the task it
    # is DERIVED to run (its open session) through the same owner as
    # every requeue — never an orphan open session on a deleted row, and
    # no saddle-row cache to key the release on. Route 1: a DEAD host —
    # the abort delivery fails fast, the offline bookkeeping releases
    # (queued + comment + session 'stopped'), the unregister then
    # deletes the row.
    conn = db()
    no_cache_columns = conn.execute(
        "select count(*) from pragma_table_info('saddles')"
        " where name in ('current_task_id', 'allocated_at')").fetchone()[0]
    conn.close()
    check(no_cache_columns == 0,
          "the saddle row carries no work cache (item 23: sessions are the truth)")
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["register"], {"name": "dead-host",
                                          "host": "127.0.0.1:1"})
    dead_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                 if s["name"] == "dead-host"][0]
    dead_id = int(dead_card["id"])
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Dead-host unregister task",
        "priority": "2",
    })
    dead_task = task_id_tmp
    conn = db()
    conn.execute("update tasks set status = 'working' where id = ?",
                 (dead_task,))
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (dead_task, dead_id))
    conn.commit()
    dead_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (dead_task,)).fetchone()[0]
    conn.close()
    act(auth, dead_card["unregister"], {})
    conn = db()
    dead_row = conn.execute("select status from tasks where id = ?",
                            (dead_task,)).fetchone()
    dead_session_row = conn.execute(
        "select outcome, ended_at is not null from sessions where id = ?",
        (dead_session,)).fetchone()
    dead_saddle_count = conn.execute(
        "select count(*) from saddles where id = ?", (dead_id,)).fetchone()[0]
    conn.close()
    check(dead_row is not None and dead_row[0] == "queued",
          "unregister of a dead-host saddle releases the held task to the queue")
    check(dead_session_row is not None and dead_session_row[0] == "stopped"
          and dead_session_row[1],
          "the offline release closes the held task's session 'stopped'")
    check(dead_saddle_count == 0, "the dead-host saddle row is gone")

    # Route 2: a REACHABLE host holding a running session — the abort
    # delivers (the daemon answers 200; its idle loop aborts nothing),
    # and the unregister itself carries the derived release through the
    # one owner (requeue flip + close + row deleted).
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["register"], {"name": "quiet-host",
                                          "host": "127.0.0.1:%d" % fake.port})
    quiet_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                  if s["name"] == "quiet-host"][0]
    quiet_id = int(quiet_card["id"])
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Quiet-host unregister task",
        "priority": "2",
    })
    quiet_task = task_id_tmp
    conn = db()
    conn.execute("update tasks set status = 'working' where id = ?",
                 (quiet_task,))
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (quiet_task, quiet_id))
    conn.commit()
    quiet_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (quiet_task,)).fetchone()[0]
    conn.close()
    act(auth, quiet_card["unregister"], {})
    conn = db()
    quiet_row = conn.execute("select status from tasks where id = ?",
                             (quiet_task,)).fetchone()
    quiet_session_row = conn.execute(
        "select outcome, ended_at is not null from sessions where id = ?",
        (quiet_session,)).fetchone()
    quiet_saddle_count = conn.execute(
        "select count(*) from saddles where id = ?", (quiet_id,)).fetchone()[0]
    conn.close()
    check(quiet_row is not None and quiet_row[0] == "queued",
          "unregister's own release returns the derived held task to the queue")
    check(quiet_session_row is not None and quiet_session_row[0] == "stopped"
          and quiet_session_row[1],
          "the derived release closes the held task's session 'stopped'")
    check(quiet_saddle_count == 0, "the quiet-host saddle row is gone")

    # Route 3 (the derived-view pin): a 'working'-status task with NO
    # session derives NO hold — with the cache gone there is nothing to
    # key a release on, and the unregister must leave the drift row
    # untouched (a shape only fixtures can fabricate now: every real
    # path that flips a task running opens its session in the same
    # breath).
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["register"], {"name": "empty-host",
                                          "host": "127.0.0.1:1"})
    empty_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                  if s["name"] == "empty-host"][0]
    empty_id = int(empty_card["id"])
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Empty-host drift task",
        "priority": "2",
    })
    empty_task = task_id_tmp
    conn = db()
    conn.execute("update tasks set status = 'working' where id = ?",
                 (empty_task,))
    conn.commit()
    conn.close()
    act(auth, empty_card["unregister"], {})
    conn = db()
    empty_row = conn.execute("select status from tasks where id = ?",
                             (empty_task,)).fetchone()
    empty_saddle_count = conn.execute(
        "select count(*) from saddles where id = ?", (empty_id,)).fetchone()[0]
    conn.close()
    check(empty_row is not None and empty_row[0] == "working",
          "no session means no derived hold — unregister releases nothing")
    check(empty_saddle_count == 0, "the empty-host saddle row is gone")