"""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("== goal 8 task 5: per-saddle prompt note ==") # The admin edits the note on the /saddles card; it persists on the # row and rides EVERY later assignment as `prompt_note`. fake_card = saddle_card(auth, "fake-saddle") check(fake_card.get("prompt_note", "") == "", "no prompt note before the admin sets one") check(all(p.get("prompt_note") == "" for p in fake.start_payloads), "unset note carries an empty string (never a null)") act(auth, fake_card["update_prompt_note"], {"prompt_note": "You have root and may install development tooling via DNF."}) fake_card = saddle_card(auth, "fake-saddle") check(fake_card.get("prompt_note") == "You have root and may install development tooling via DNF.", "inline edit saves the prompt note") note_task = create_queued(auth, { "project_id": "1", "title": "Prompt note task", "priority": "2", }) check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == note_task for p in fake.start_payloads) or False, timeout=60, what="prompt-note task dispatched") is not None, "prompt-note task dispatches") note_payload = [p for p in fake.start_payloads if int(p.get("task", {}).get("id") or 0) == note_task][-1] check(note_payload.get("prompt_note") == "You have root and may install development tooling via DNF.", "the assignment carries the saddle's prompt_note verbatim") check(wait_task_done(auth, note_task, timeout=60) is not None, "prompt-note task completes") # Clearing the note empties the member on later assignments. fake_card = saddle_card(auth, "fake-saddle") act(auth, fake_card["update_prompt_note"], {"prompt_note": ""}) check(saddle_card(auth, "fake-saddle").get("prompt_note", "") == "", "clearing the prompt note empties the row") # A fat note is clamped: the row keeps the full text but the wire # serves at most PROMPT_NOTE_MAX_CHARS. fat = "N" * 5000 fake_card = saddle_card(auth, "fake-saddle") act(auth, fake_card["update_prompt_note"], {"prompt_note": fat}) clamp_task = create_queued(auth, { "project_id": "1", "title": "Clamp note task", "priority": "2", }) check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == clamp_task for p in fake.start_payloads) or False, timeout=60, what="clamp-note task dispatched") is not None, "clamp-note task dispatches") clamp_payload = [p for p in fake.start_payloads if int(p.get("task", {}).get("id") or 0) == clamp_task][-1] check(len(clamp_payload.get("prompt_note") or "") == 2000, "a fat prompt_note is clamped to 2000 chars on the wire") check(wait_task_done(auth, clamp_task, timeout=60) is not None, "clamp-note task completes") fake_card = saddle_card(auth, "fake-saddle") act(auth, fake_card["update_prompt_note"], {"prompt_note": ""}) 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")