"""The allocation 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: allocation — the ask/wake/allocate machinery: concurrent ask # race, token revocation, workspace affinity, the lost-wake guard, the # slow-ask regression, assignment capability context + the tag # catalogue, and the pinned-blocks/activity dashboard. # ============================================================================= def run(ctx): auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token app = ctx.app r4_token = fake_token # gpu rides the primary fake from the start (the routing bundle's # early attach, replayed): its attach row precedes lostwake's, so # the assignment's saddle_tags order matches the assertion in the # capability-context section below. saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "gpu"}) act(auth, saddle_card(auth, "fake-saddle")["attach_capability"], {"capability_id": str(int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"] if c["name"] == "gpu"][0]))}) print("== phase 1: ask/allocation — token auth + concurrent race ==") # Plan 10 phase 1. Registration mints per-saddle tokens; the # ask endpoint authenticates by them; concurrent asks race to # exactly one winner through the allocation transaction. status, _ = saddle_api("POST", "/api/saddle/next-task", None) check(status == 401, "next-task rejects a missing token") status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-bogus-token") check(status == 401, "next-task rejects a wrong token") status, body = saddle_api("POST", "/api/saddle/next-task", fake_token) check(status == 204, "next-task answers 204 (nothing offerable — daemon sleeps)") # Two more racers: real daemons on their own ports; fake2 reports # its ssh key at /register so the key lands on the row directly. fake2 = FakeSaddle() fake2.ssh_public_key = "ssh-ed25519 AAAAFake2RacerKey saddle-2@fedora" fake2.serve(port=free_port()) fake3 = FakeSaddle() fake3.serve(port=free_port()) saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["register"], {"name": "racer-two", "host": "127.0.0.1:%d" % fake2.port}) act(auth, saddles_state["register"], {"name": "racer-three", "host": "127.0.0.1:%d" % fake3.port}) check(wait_for(lambda: len(fake2.register_log) > 0 and len(fake3.register_log) > 0 or False, timeout=15, what="racers registered") is not None, "both racers received /register") check(wait_for(lambda: "AAAAFake2RacerKey" in [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-two"][0].get("ssh_public_key", "") or False, timeout=15, what="register-reported key") is not None, "the ssh key reported at /register lands on the saddle row") racer_cards = {} for name, daemon in (("fake-saddle", fake), ("racer-two", fake2), ("racer-three", fake3)): card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == name][0] racer_cards[daemon] = {"id": int(card["id"]), "token": card["token"]} # Revocation: a registered saddle's token works until the saddle # is removed, then nothing authorises its calls. fake4 = FakeSaddle() fake4.ask_on_register = False # a pre-flip daemon: never asks fake4.serve(port=free_port()) saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["register"], {"name": "old-daemon", "host": "127.0.0.1:%d" % fake4.port}) old_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "old-daemon"][0] status, _ = saddle_api("POST", "/api/saddle/next-task", old_card["token"]) check(status in (200, 204), "the old daemon's fresh token authenticates next-task") status, _ = saddle_api("POST", "/api/saddle/note", old_card["token"], {"project": "demo", "body": "from the old daemon"}) check(status == 200, "note accepts any registered saddle's token") act(auth, old_card["unregister"], {}) status, _ = saddle_api("POST", "/api/saddle/next-task", old_card["token"]) check(status == 401, "removing the saddle revokes its token") status, _ = saddle_api("POST", "/api/saddle/note", old_card["token"], {"project": "demo", "body": "should fail"}) check(status == 401, "revocation covers the note endpoint too") # Deterministic race: settle the queue first (this shard runs the # race before any parked work exists, so only in-flight asks need # draining), freeze every daemon's self-ask triggers, put all three # to sleep (204), then fire one task worth of concurrent asks. for daemon in (fake, fake2, fake3): daemon.ask_on_register = False daemon.ask_on_wake = False daemon.ask_on_terminal = False daemon.ask_on_idle_status = False daemon.settle_asks() # converted sleep: in-flight asks land (204) def queued_count(): conn = db() try: return conn.execute( "select count(*) from tasks where status = 'queued'").fetchone()[0] finally: conn.close() check(wait_for(lambda: queued_count() == 0 or False, timeout=60, what="queue settles before the race") is not None, "the queue settles before the race") sleep_statuses = [daemon.ask_now()[0] for daemon in (fake, fake2, fake3)] check(sleep_statuses == [204, 204, 204], "all three racers sleep (204) before the task exists") task_id_tmp = create_queued(auth, { "project_id": "1", "title": "Race task", "priority": "2", }) race_task = task_id_tmp race_results = [] def race_ask(daemon): # A real racing daemon accepts the assignment it wins — the # manual ask path does exactly that on a 200. race_results.append(daemon.ask_now()) threads = [threading.Thread(target=race_ask, args=(daemon,)) for daemon in (fake, fake2, fake3)] for thread in threads: thread.start() for thread in threads: thread.join() race_statuses = sorted(r[0] for r in race_results) check(race_statuses == [200, 204, 204], "concurrent asks race to exactly one winner (%s)" % race_statuses) winner = [r for r in race_results if r[0] == 200][0][1] check(int(winner["task"]["id"]) == race_task, "the winner's payload carries the task") check(int(winner["session_id"]) > 0, "the assignment carries the session id") conn = db() race_sessions = conn.execute( "select count(*) from sessions where task_id = ?", (race_task,)).fetchone()[0] conn.close() check(race_sessions == 1, "exactly one session row — the transaction serialised the claims (%d)" % race_sessions) check(wait_task_done(auth, race_task, timeout=60) is not None, "the raced task completes on the winning daemon") for daemon in (fake, fake2, fake3): daemon.ask_on_register = True daemon.ask_on_wake = True daemon.ask_on_terminal = True daemon.ask_on_idle_status = True # Re-establish sleeping explicitly: the winner's post-race # re-ask timing is not deterministic, and the affinity section # below needs every racer provably asleep. Any stray offerable # drains into these asks first. def all_asleep(): return all(daemon.ask_now()[0] == 204 for daemon in (fake, fake2, fake3)) check(wait_for(lambda: all_asleep() or False, timeout=30, what="racers re-sleep after the race") is not None, "all racers re-sleep after the race") print("== phase 1: workspace affinity pins parked tasks ==") # A suspended parent parked on racer-two may be resumed ONLY by # racer-two: a tag-capable non-affine saddle asking gets 204, and # the wake targets the affine saddle. Only racer-two carries the # tag at dispatch time (the race's winner is nondeterministic, so # the park must not depend on which racers are free); racer-three # gains it afterwards to prove the pinning is by workspace, not # tags. saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "affinity-tag"}) affinity_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"] if c["name"] == "affinity-tag"][0]) racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-two"][0] racer3_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-three"][0] act(auth, racer2_card["attach_capability"], {"capability_id": str(affinity_cap)}) # Plan 12: the parked workspace is the goal agent's run task. The # other racers sleep (frozen ask triggers), so racer-two claims # the goal run and parks it; the goal-task event then resumes the # agent AFFINE to racer-two — racer-three (tag-capable or not) # gets 204, the dispatcher's wake targets racer-two. fake.ask_on_wake = False fake.ask_on_idle_status = False fake.settle_asks() fake3.ask_on_wake = False fake3.ask_on_idle_status = False fake3.settle_asks() act(auth, slot_public(auth, "/goals", "goals")["create"], { "project_id": "1", "title": "Affinity goal", "brief": "Parks on racer-two.", "branch_name": "goal/affinity"}) conn = db() affinity_goal = conn.execute( "select id from goals where title = 'Affinity goal'").fetchone()[0] conn.close() fake2.goal_script = [("plan", "# Affinity plan\n\n1. Park")] act(auth, slot_public(auth, "/goals/%d" % affinity_goal, "goal")["start"], {}) check(wait_for(lambda: slot_public(auth, "/goals/%d" % affinity_goal, "goal")["status"] == "plan_review" or False, timeout=60, what="affinity goal plans") is not None, "the affinity goal reaches plan review (racer-two hosting)") fake2.goal_script = [("suspend",)] # Freeze racer-two's ask triggers across the approval too (the # later wake measurement already does): the mirror-sync fix makes # the run resume-offerable the moment its goal row carries the # verdict, so a wake-triggered background ask would race this # manual ask_now for the claim. fake2.ask_on_wake = False fake2.ask_on_idle_status = False fake2.settle_asks() act(auth, slot_public(auth, "/goals/%d" % affinity_goal, "goal") ["plan_pending_block"]["answer_action"], {"answer": "Plan approved. Proceed with implementation."}) conn = db() affinity_parent = conn.execute( "select id from tasks where goal_id = ? and kind = 'goal_run'", (affinity_goal,)).fetchone()[0] agent_session = conn.execute( "select max(id) from sessions where task_id = ?", (affinity_parent,)).fetchone()[0] conn.close() status, resume_payload = fake2.ask_now() check(status == 200 and int(resume_payload["task"]["id"]) == affinity_parent, "the approved plan resumes the agent on racer-two (its workspace)") check(wait_for(lambda: task_state(auth, affinity_parent).get("status") == "suspended" or False, timeout=60, what="affinity agent parks") is not None, "the affinity goal agent parks on racer-two") conn = db() affinity_session = conn.execute( "select saddle_id from sessions where task_id = ?" " order by id desc limit 1", (affinity_parent,)).fetchone()[0] conn.close() check(affinity_session == int(racer2_card["id"]), "the parked workspace sits on racer-two") # Now racer-three carries the tag too — the pinning is by # workspace, not tags. With nothing due, a non-affine ask gets 204. act(auth, racer3_card["attach_capability"], {"capability_id": str(affinity_cap)}) status, _ = fake3.ask_now() check(status == 204, "a tag-capable non-affine saddle is refused the parked agent (204)") # A goal task completing makes the parked agent resume-offerable — # affine to racer-two only. The other racers wake back up (they # host the worker; their claims of the parked agent get refused as # non-affine), while racer-two's ask triggers stay frozen so the # wake measurement stays clean. A wake may have hit the frozen # racers while the goal run queued (the lost-wake guard marks them # offline): the refresh probe brings them back before the worker. fake.ask_on_wake = True fake.ask_on_idle_status = True fake3.ask_on_wake = True fake3.ask_on_idle_status = True for racer_name in ("fake-saddle", "racer-three"): act(auth, saddle_card(auth, racer_name)["refresh"], {}) check(wait_for(lambda name=racer_name: [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == name][0]["status"] == "idle" or False, timeout=15, what="racer re-onlined") is not None, "%s re-onlined after its frozen-window wake" % racer_name) fake2.ask_on_wake = False fake2.ask_on_idle_status = False wakes_before_2 = len(fake2.wake_log) fake2.settle_asks() # converted sleep: park-time ask lands (204) status, body = saddle_api("POST", "/api/saddle/task/create", fake2.token, { "session_id": agent_session, "title": "Affinity child", "mandate": "finish the affinity fixture"}) check(status == 200, "the parked agent fans its worker out via the CLI") affinity_child = int(body["task_id"]) saddle_api("POST", "/api/saddle/task/enqueue", fake2.token, {"session_id": agent_session, "task_id": affinity_child}) check(wait_task_done(auth, affinity_child, timeout=90) is not None, "the affinity worker completes (on a non-affine racer)") check(wait_for(lambda: len(fake2.wake_log) > wakes_before_2 or False, timeout=90, what="affine wake") is not None, "the dispatcher's wake targets the affine saddle") status, resume_payload = fake2.ask_now() fake2.ask_on_wake = True # restored whatever the claim returned fake2.ask_on_idle_status = True check(status == 200 and int(resume_payload["task"]["id"]) == affinity_parent, "the affine saddle's ask claims the parked agent") check("Affinity child" in (resume_payload.get("resume_notice") or "") and "fake daemon finished" in (resume_payload.get("resume_notice") or ""), "the resume assignment carries the goal's batched report notice") check(wait_for(lambda: task_state(auth, affinity_parent).get("status") == "suspended" or False, timeout=60, what="affinity agent re-parks") is not None, "the resumed agent re-parks (script)") act(auth, slot_public(auth, "/goals/%d" % affinity_goal, "goal")["cancel"], {}) check(wait_for(lambda: slot_public(auth, "/goals/%d" % affinity_goal, "goal")["status"] == "cancelled" or False, timeout=30, what="affinity goal cancelled") is not None, "the affinity goal cancels (drain)") print("== phase 1: lost wake -> offline + re-wake of the next capable ==") # racer-three sleeps through wakes (a daemon that died after its # last 204): the guard window (SADDLE_LOST_WAKE_S=2) passes with # no ask, the saddle goes offline, and the work moves to the # next capable sleeping saddle. fake3.ask_on_wake = False fake3.ask_on_idle_status = False # sleeps through everything saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "lostwake"}) lostwake_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"] if c["name"] == "lostwake"][0]) racer3_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-three"][0] act(auth, racer3_card["attach_capability"], {"capability_id": str(lostwake_cap)}) # Racer-three must be asleep to be woken: park it with a manual # 204 ask (nothing offerable for it — the task below is not # created yet). status, _ = fake3.ask_now() check(status == 204, "racer-three sleeps before the lost-wake run") task_id_tmp = create_queued(auth, { "project_id": "1", "title": "Lost wake task", "priority": "2", "capability_id": [str(lostwake_cap)], }) lost_task = task_id_tmp check(wait_for(lambda: len(fake3.wake_log) > 0 or False, timeout=30, what="wake sent to racer-three") is not None, "the only capable sleeping saddle is woken") check(wait_for(lambda: [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-three"][0]["status"] == "offline" or False, timeout=30, what="lost-wake -> offline") is not None, "no ask after the wake: the saddle goes offline (lost-wake guard)") check(task_state(auth, lost_task)["status"] == "queued", "the task is untouched while its wake was pending") # Re-wake the next capable: the healthy saddle gains the tag and # picks the work up via its own wake -> ask -> allocate roundtrip. fake_card = saddle_card(auth, "fake-saddle") act(auth, fake_card["attach_capability"], {"capability_id": str(lostwake_cap)}) wakes_before = len(fake.wake_log) check(wait_for(lambda: len(fake.wake_log) > wakes_before or False, timeout=30, what="re-wake of the next capable") is not None, "the next capable sleeping saddle is woken for the task") check(wait_task_done(auth, lost_task, timeout=60) is not None, "wake -> ask -> allocate: the task completes on the healthy saddle") check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == lost_task for p in fake.start_payloads) or False, timeout=10, what="lost task assignment captured") is not None, "the re-woken saddle's ask produced the assignment") fake3.ask_on_wake = True fake3.ask_on_idle_status = True print("== phase 3: capability context, catalogue, subtask tags ==") # Assignment capability context (plan 10 phase 3): every # assignment carries {saddle_tags: the allocated saddle's tags, # fleet_tags: every tag name known across the fleet} — the # daemon renders it into its system prompt. fake-saddle carries # gpu + lostwake; the catalogue is affinity-tag, gpu, lostwake. fake2.ask_on_register = False fake2.ask_on_wake = False fake2.ask_on_terminal = False fake2.ask_on_idle_status = False fake2.settle_asks() # converted sleep: in-flight trigger asks land task_id_tmp = create_queued(auth, { "project_id": "1", "title": "Capabilities context", "priority": "2", }) cap_task = task_id_tmp check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == cap_task for p in fake.start_payloads) or False, timeout=60, what="capabilities-context task dispatched") is not None, "the capabilities-context task dispatches") cap_payload = [p for p in fake.start_payloads if int(p.get("task", {}).get("id") or 0) == cap_task][-1] context = cap_payload.get("capabilities") or {} check(context.get("saddle_tags") == ["gpu", "lostwake"], "assignment saddle_tags = the allocated saddle's tags (%s)" % context.get("saddle_tags")) check(context.get("fleet_tags") == ["affinity-tag", "gpu", "lostwake"], "assignment fleet_tags = every tag name known across the fleet (%s)" % context.get("fleet_tags")) check(wait_task_done(auth, cap_task, timeout=60) is not None, "capabilities-context task completes") # GET /api/saddle/capabilities feeds the daemon's `saddle # capabilities` CLI: per-saddle-token auth, tag order by name, # each tag with the saddles carrying it (name order). status, _ = saddle_api("GET", "/api/saddle/capabilities", None) check(status == 401, "capabilities endpoint rejects a missing token") status, body = saddle_api("GET", "/api/saddle/capabilities", "saddle-bogus") check(status == 401, "capabilities endpoint rejects a wrong token") status, body = saddle_api("GET", "/api/saddle/capabilities", fake_token) catalogue = {t["name"]: t["saddles"] for t in body.get("tags", [])} check(status == 200 and body.get("ok") is True and [t["name"] for t in body.get("tags", [])] == ["affinity-tag", "gpu", "lostwake"], "catalogue lists every tag by name") check(catalogue.get("gpu") == ["fake-saddle"] and catalogue.get("affinity-tag") == ["racer-three", "racer-two"] and catalogue.get("lostwake") == ["fake-saddle", "racer-three"], "each tag names its carrying saddles in name order (%s)" % catalogue) # Task CLI capabilities (`task create --capability`): validated # against the operator tag set; unknown tags reject with the valid # list in the error (the agent reads it verbatim in the CLI # output). A goal agent's parked session hosts the calls. fake.suspend_on_title = "Tag host" tag_host = create_queued(auth, { "project_id": "1", "title": "Tag host", "priority": "3"}) check(wait_for(lambda: task_state(auth, tag_host)["status"] == "suspended" or False, timeout=60, what="tag host parks") is not None, "a parked plain task hosts nothing (no goal scope)") # The goal callbacks need a real goal agent: reuse the sequencing # goal pattern — a fresh parked agent under this shard. act(auth, slot_public(auth, "/goals", "goals")["create"], { "project_id": "1", "title": "Tag goal", "brief": "b", "branch_name": "goal/tags"}) conn = db() tag_goal = conn.execute("select id from goals where title = 'Tag goal'").fetchone()[0] conn.close() fake.goal_script = [("suspend",)] act(auth, slot_public(auth, "/goals/%d" % tag_goal, "goal")["start"], {}) conn = db() tag_run = conn.execute( "select id from tasks where goal_id = ? and kind = 'goal_run'", (tag_goal,)).fetchone()[0] conn.close() check(wait_for(lambda: task_state(auth, tag_run)["status"] == "suspended" or False, timeout=60, what="tag goal agent parks") is not None, "the tag goal agent parks") conn = db() tag_session = conn.execute( "select max(id) from sessions where task_id = ?", (tag_run,)).fetchone()[0] conn.close() conn = db() tag_host_saddle = conn.execute( "select saddle_id from sessions where id = ?", (tag_session,)).fetchone()[0] conn.close() racer2_card_id = int([c for c in slot_public(auth, "/saddles", "saddles")["saddles"] if c["name"] == "racer-two"][0]["id"]) racer3_card_id = int([c for c in slot_public(auth, "/saddles", "saddles")["saddles"] if c["name"] == "racer-three"][0]["id"]) host_daemon = {int(fake_id): fake, racer2_card_id: fake2, racer3_card_id: fake3}[int(tag_host_saddle)] status, body = saddle_api("POST", "/api/saddle/task/create", host_daemon.token, { "session_id": tag_session, "title": "Bad tag worker", "mandate": "m", "capabilities": ["gpu", "warp-drive"]}) check(status == 400, "an unknown capability tag rejects the task call") error_text = body.get("error", "") check("warp-drive" in error_text and "valid tags" in error_text and "gpu" in error_text and "lostwake" in error_text, "the rejection names the unknown tag and the valid list (%s)" % error_text) status, _ = saddle_api("POST", "/api/saddle/task/create", host_daemon.token, { "session_id": tag_session, "title": "Bad shape worker", "mandate": "m", "capabilities": "gpu"}) check(status == 400, "a non-array capabilities member rejects") # A valid tag creates the goal worker in draft with exactly the # explicit tag attached; the goal agent enqueues it. status, body = saddle_api("POST", "/api/saddle/task/create", host_daemon.token, { "session_id": tag_session, "title": "Explicit tag worker", "mandate": "do the tagged work", "capabilities": ["gpu"]}) check(status == 200 and body.get("ok") is True, "a valid tag creates the task synchronously") tagged_worker = int(body["task_id"]) detail = task_state(auth, tagged_worker) check([c["name"] for c in detail.get("capabilities", [])] == ["gpu"], "the worker carries exactly the explicit tag") conn = db() tag_rows = conn.execute( "select c.name from task_capabilities tc" " join capabilities c on c.id = tc.capability_id" " where tc.task_id = ?", (tagged_worker,)).fetchall() conn.close() check([r[0] for r in tag_rows] == ["gpu"], "the task_capabilities rows record the explicit tag") check(detail["status"] == "draft", "the callback's task lands in draft until its agent enqueues") saddle_api("POST", "/api/saddle/task/enqueue", host_daemon.token, {"session_id": tag_session, "task_id": tagged_worker}) check(wait_task_done(auth, tagged_worker, timeout=90) is not None, "the tagged worker dispatches and completes") act(auth, slot_public(auth, "/goals/%d" % tag_goal, "goal")["cancel"], {}) act(auth, task_state(auth, tag_host)["cancel"], {}) print("== phase 3: dashboard — pinned blocks + activity timeline ==") # Pinned OPEN blocking items as link-through cards, newest # first; no inline answer forms (the card links to the task # page). Fixture: one open question + one open review raised on a # local task (stand-ins for the auto-approve grandchild's # leftovers) join the probe RFC raised below. conn = db() conn.execute("insert into tasks (project_id, title, status, priority, created_at)" " values (1, 'Dash fixture task', 'working', 5, strftime('%s','now'))") dash_task = conn.execute("select max(id) from tasks").fetchone()[0] conn.execute("insert into sessions (task_id, saddle_id, started_at)" " values (?, ?, strftime('%s','now'))", (dash_task, fake_id)) dash_session = conn.execute("select max(id) from sessions").fetchone()[0] conn.commit() conn.close() def raise_on_dash(question, kind="question", body=None, summary=None): """Push-era raises on the dashboard fixture task (the poll-era /update channel is gone): a blocked results push carries the block payload; a review verb rides a review entry.""" if kind == "review" and summary is not None: return saddle_api("POST", "/api/saddle/results", fake_token, { "session_id": dash_session, "state": "working", "entries": [{"id": 9317, "command": "saddle review", "output": summary, "type": "review", "timestamp": time.strftime( "%Y-%m-%dT%H:%M:%SZ", time.gmtime())}]})[0] block = {"question": question, "type": kind} if body is not None: block["body"] = body return saddle_api("POST", "/api/saddle/results", fake_token, { "session_id": dash_session, "state": "blocked", "block": block})[0] raise_on_dash("Dash fixture question?", kind="question") raise_on_dash("dash fixture review", kind="review", summary="dash fixture review") # The activity timeline: last <= 50 events, newest first, fed # from the existing tables. Fixture rows a few seconds old # guarantee the progress kind and a human-authored comment ride # along; the probe block below proves same-source ordering. conn = db() conn.execute("insert into task_comments (task_id, author, body, created_at)" " values (?, 'user:admin', 'timeline fixture comment by a human'," " strftime('%s','now','-5 seconds'))", (cap_task,)) conn.execute("insert into task_progress (task_id, summary, created_at)" " values (?, 'timeline fixture progress'," " strftime('%s','now','-5 seconds'))", (cap_task,)) conn.commit() conn.close() # The probe RFC raised last: the newest pinned card and the newest # timeline event (its raise+answer below prove same-source ordering # both ways). The timeline sorts same-second events by source # materialisation order, so the probe must land in a LATER clock # second than the fixtures — wait for the boundary (bounded). fixtures_epoch = int(time.time()) wait_for(lambda: int(time.time()) > fixtures_epoch or False, timeout=3, what="next clock second before the probe raise") raise_on_dash("Timeline probe: approve the dashboard flow", kind="rfc", body="# Probe") dash = slot_public(auth, "/", "dashboard") pinned = dash.get("pinned", []) pinned_ids = [int(b["id"]) for b in pinned] check(len(pinned_ids) >= 3 and pinned_ids == sorted(pinned_ids, reverse=True), "pinned open blocks order newest-first (%s)" % pinned_ids) check({b["block_type"] for b in pinned} >= {"question", "rfc", "review"}, "pinned cards cover questions, RFCs and reviews") check(all(b["question"] and b["task_title"] and int(b["task_id"]) > 0 for b in pinned) and all("answer_action" not in b for b in pinned), "pinned cards are link-through only (task title, kind badge, question)") activity = dash.get("activity", []) epochs = [int(e["at_epoch"]) for e in activity] check(0 < len(activity) <= 50 and epochs == sorted(epochs, reverse=True), "the timeline shows the last <=50 events newest-first (%d events)" % len(activity)) kinds = {e["kind"] for e in activity} check({"created", "started", "dispatch", "comment", "progress", "completed"} <= kinds, "the timeline covers status changes, dispatches, comments, progress and" " completions (%s)" % sorted(kinds)) check(any(e["kind"] == "comment" and e["text"].startswith("user:admin:") and "fixture comment" in e["text"] for e in activity), "comment events carry author + snippet") # The freshly raised probe pinned as the newest card and landed # atop the timeline as block-raised; answering it records # block-answered. check(dash["pinned"][0]["question"] == "Timeline probe: approve the dashboard flow" and int(dash["pinned"][0]["task_id"]) == dash_task, "the freshest open block pins as the first card") check(dash["activity"][0]["kind"] == "block-raised" and "Timeline probe" in dash["activity"][0]["text"], "the raised block lands atop the timeline") probe_block = [b for b in task_state(auth, dash_task)["blocks"] if b["question"] == "Timeline probe: approve the dashboard flow"][0] act(auth, probe_block["answer_action"], {"answer": "Ship the dashboard"}) dash = slot_public(auth, "/", "dashboard") activity = dash.get("activity", []) raised_at = next((i for i, e in enumerate(activity) if e["kind"] == "block-raised" and "Timeline probe" in e["text"]), -1) answered_at = next((i for i, e in enumerate(activity) if e["kind"] == "block-answered" and "Ship the dashboard" in e["text"]), -1) check(answered_at >= 0 and raised_at >= 0 and answered_at < raised_at, "the answered event sorts above its raised event (newest-first, same source)") check(all(b["question"] != "Timeline probe: approve the dashboard flow" for b in dash["pinned"]), "the answered block leaves the pinned cards") print("== phase 3 regression: a slow ask after a wake is never dropped ==") # The daemon worker's lost-wake-guard regression: a saddle whose # ask arrives SLOWLY after the wake (busy winding down when the # wake lands) must not lose the wake — the delayed ask (inside # the guard window) claims the task, no re-wake storm fires # while it is pending, and the saddle stays healthy. The # never-asks path (guard -> offline + re-wake of the next # capable) is proven by the racer-three section above. check(wait_for(lambda: queued_count() == 0 or False, timeout=60, what="queue settles before the slow-wake run") is not None, "the queue settles before the slow-wake run") status, _ = fake2.ask_now() check(status == 204, "the slow-ask racer sleeps before its wake") saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "slowwake"}) slowwake_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles") ["capabilities"] if c["name"] == "slowwake"][0]) racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-two"][0] act(auth, racer2_card["attach_capability"], {"capability_id": str(slowwake_cap)}) fake2.ask_delay = 1.2 # slow: asks well after the wake, inside the 2s guard fake2.ask_on_wake = True wakes_before_slow = len(fake2.wake_log) task_id_tmp = create_queued(auth, { "project_id": "1", "title": "Slow wake task", "priority": "2", "capability_id": [str(slowwake_cap)], }) slow_task = task_id_tmp check(wait_for(lambda: len(fake2.wake_log) > wakes_before_slow or False, timeout=30, what="the slow-ask racer woken") is not None, "the only capable sleeping saddle is woken") check(wait_task_done(auth, slow_task, timeout=120) is not None, "the delayed ask claims the task — the wake was not dropped") check(any(a.get("status") == 200 and int((a.get("payload") or {}).get("task", {}).get("id") or 0) == slow_task for a in fake2.asks), "the slow ask's 200 carries the task") check(len(fake2.wake_log) == wakes_before_slow + 1, "the pending-wake skip re-wakes nobody while the slow ask is in flight") racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-two"][0] check(racer2_card["status"] in ("idle", "busy"), "the slow asker never went offline (guard did not fire)") fake2.ask_delay = 0.0 fake2.ask_on_wake = False # stays out of the pool below print("== goal 4 t1: plain-task force wake — offerable on its affine saddle ==") # A suspended PLAIN task stays parked until the human force-wakes # it; the resuming claim carries the explicit wake notice and # consumes the flag (one wake per press). fake2.ask_on_wake = False fake2.ask_on_idle_status = False fake2.settle_asks() conn = db() conn.execute("insert into tasks (project_id, kind, title, status, priority, mode," " creator, created_at, suspended_at) values (1, 'task'," " 'Park me', 'suspended', 5, 'implement', 'human'," " strftime('%s','now'), strftime('%s','now'))") plain_parked = conn.execute("select max(id) from tasks").fetchone()[0] # parked_offerable keys the affine saddle off the newest SESSION # row (plan 16 D4): a parked round's row is closed 'parked' — give # the fixture one so the wake offers on the fake saddle. conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at," " outcome) values (?, ?, strftime('%s','now')," " strftime('%s','now'), 'parked')", (plain_parked, fake_id)) conn.commit() conn.close() task_page = http_get_page(auth, "/tasks/%d" % plain_parked) check('stm-action="task.force_wake"' in task_page and 'stm-if="task.force_wake != null"' in task_page, "the task page authors the Force wake button for a suspended plain task") # No press, no wake: frozen-ask racer-two gets 204. status, _ = fake2.ask_now() check(status == 204, "without a press the parked plain task stays parked (204)") detail = task_state(auth, plain_parked) check(detail.get("force_wake") is not None, "the task state exposes the offered force-wake action") act(auth, detail["force_wake"], {}) status, resume_payload = fake2.ask_now() check(status == 200 and int(resume_payload["task"]["id"]) == plain_parked, "the force-woke plain task is claimable again") woke_notice = resume_payload.get("resume_notice") or "" check("force-woke this parked task" in woke_notice and "resume where you left off" in woke_notice, "the plain-task resume carries the explicit wake notice") conn = db() flag_row = conn.execute("select force_wake from tasks where id = ?", (plain_parked,)).fetchone() conn.close() check(flag_row is not None and flag_row[0] in (0, False), "the claim consumed the task force-wake flag") # The woken run re-parks (fake daemon finishes then suspends? no — # fake.start_payloads script: the plain task runs to done via the # daemon's own finish path in later suites; here just let it be # requeued by completing it directly to keep the fixture short). check(wait_task_done(auth, plain_parked, timeout=60) is not None, "the woken plain task completes (daemon finish path)") # Guard: a non-suspended task offers no wake action (the template # ships the gated markup statically; absence is a STATE property). check(task_state(auth, plain_parked).get("force_wake") is None, "a done task offers no force-wake action")