Force Wake tests: dedicated forcewake bundle + order registration

The goals/allocation/lifecycle shards die environmentally mid-run
in this container (verified identical on the clean 9341240 base
tree) before reaching their new force-wake sections, so the
behaviour gets a self-contained short-lived bundle (green here:
28 passed, 0 failed, ~15 s):

- goal stall fixture: parked run + a never-enqueued draft (the
  typo'd-enqueue class) + the actionable enqueue-error tail check
  (pinned substring + "nothing was enqueued" + "saddle tasks");
- the wake press: immediate resume, the explicit force-woke lead
  line, the draft named NOT-enqueued with the enqueue fix, the
  standard batched report kept;
- flag consumption: a second quiet window stays quiet;
- guards: a paused goal offers no action and refuses the replayed
  sealed wake; the goal force-wake flag stays clear;
- plain-task wake: 204 without a press, claimable with the
  explicit notice after, the task flag consumed, no action once
  done.

The allocation fixture now carries a parked session row
(parked_offerable keys the affine saddle off the newest session);
goals/allocation/goalui keep their in-place sections for when the
environment is fixed. forcewake registered in BUNDLES_ORDER.

test/bundles/allocation.py

@@ -657,6 +657,13 @@ def run(ctx):
                  " '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)

test/bundles/forcewake.py

@@ -0,0 +1,195 @@
+"""The forcewake bundle (goal 4 task 1): the human-triggered resume.
+
+Self-contained on purpose: the goals/allocation shards die
+environmentally mid-run in this container (the known framework stack
+corruption) long before their new sections, so the force-wake
+behaviour gets its own short-lived app here. The extended sections
+stay in their home bundles for when task 3 fixes the environment.
+"""
+from harness import *  # noqa: F401,F403 — the shared wire helpers
+
+
+def run(ctx):
+    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
+
+    def create_goal(fields):
+        act(auth, slot_public(auth, "/goals", "goals")["create"], fields)
+        rows = [g for g in slot_public(auth, "/goals", "goals")["goals"]
+                if g["title"] == fields["title"]]
+        return int(rows[0]["id"])
+
+    def run_task_of(goal_id):
+        conn = db()
+        try:
+            return int(conn.execute(
+                "select id from tasks where goal_id = ? and kind = 'goal_run'",
+                (goal_id,)).fetchone()[0])
+        finally:
+            conn.close()
+
+    def run_status(goal_id):
+        return task_state(auth, run_task_of(goal_id))["status"]
+
+    print("== goal 4 t1: force wake — the goal-2 stall class ==")
+    # Create → plan → approve → parked: the goalui-proven short flow.
+    fake.goal_script = [("plan", "# Wake plan\n\n1. Park safely")]
+    goal = create_goal({"project_id": "1", "title": "Force-wake goal",
+                        "brief": "Parks with a never-enqueued draft.",
+                        "branch_name": "goal/forcewake"})
+    act(auth, goal_state(goal)["start"], {})
+    check(wait_for(lambda: goal_state(goal)["status"] == "plan_review" or False,
+                   timeout=90, what="goal plans") is not None,
+          "the goal reaches plan review")
+    fake.goal_script = [("suspend",)]
+    pending = goal_state(goal)["plan_pending_block"]
+    act(auth, pending["answer_action"], {"verdict": "accept"})
+    check(wait_for(lambda: goal_state(goal)["status"] == "active" or False,
+                   timeout=60, what="plan approved") is not None,
+          "approving the plan activates the goal")
+    check(wait_for(lambda: run_status(goal) == "suspended" or False,
+                   timeout=60, what="agent parks") is not None,
+          "the approved agent parks (saddle suspend)")
+
+    # The stall fixture: a draft that was never enqueued (the typo'd
+    # enqueue class) — no terminal transition will ever fire.
+    conn = db()
+    agent_session = conn.execute(
+        "select max(id) from sessions where task_id = ?",
+        (run_task_of(goal),)).fetchone()[0]
+    conn.close()
+    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
+        "session_id": agent_session, "title": "QA sweep",
+        "mandate": "Run the QA sweep.", "mode": "test"})
+    check(status == 200 and body.get("ok") is True,
+          "the stall fixture draft task creates (%s)" % body.get("error"))
+    stall_draft = int(body["task_id"])
+    # The actionable error tail (goal 4 t1): the typo'd enqueue names
+    # the recovery path while keeping the pinned substring. The wrong
+    # id is a REAL task owned by nobody's goal (a plain human task):
+    # the ownership check fires before the existence-only path.
+    conn = db()
+    conn.execute("insert into tasks (project_id, kind, title, status, priority,"
+                 " mode, creator, created_at) values (1, 'task', 'Wrong id',"
+                 " 'draft', 5, 'implement', 'human', strftime('%s','now'))")
+    wrong_id = conn.execute("select max(id) from tasks").fetchone()[0]
+    conn.commit()
+    conn.close()
+    status, body = saddle_api("POST", "/api/saddle/task/enqueue", fake_token, {
+        "session_id": agent_session, "task_id": wrong_id})
+    check(status == 400 and "does not belong to your goal" in (body.get("error") or "")
+          and "nothing was enqueued" in (body.get("error") or "")
+          and "saddle tasks" in (body.get("error") or ""),
+          "the enqueue error carries the pinned substring + the actionable tail (%s)"
+          % body.get("error"))
+
+    state = goal_state(goal)
+    check(state.get("force_wake") is not None,
+          "an active goal with a parked run offers the force-wake action")
+    goal_page = http_get_page(auth, "/goals/%d" % goal)
+    check('stm-action="goal.force_wake"' in goal_page
+          and 'stm-if="goal.force_wake != null"' in goal_page,
+          "the goal page authors the gated Force wake button")
+    run_page = http_get_page(auth, "/tasks/%d" % run_task_of(goal))
+    check('stm-action="task.force_wake"' in run_page,
+          "the suspended goal-run row offers the task-page wake (routing to the goal)")
+
+    # No press, no wake: a quiet window passes without any resume.
+    parked_runs = len(fake.goal_assignments)
+    check(quiet_window(quiet_seconds(),
+                       lambda: len(fake.goal_assignments) == parked_runs,
+                       "no press, no wake"),
+          "without a press the parked agent never wakes")
+
+    # Press: the wake fires immediately (no coalesce wait) and leads
+    # with the explicit line; the statuses name the draft as NOT
+    # enqueued with the enqueue fix.
+    act(auth, goal_state(goal)["force_wake"], {})
+    check(wait_for(lambda: len(fake.goal_assignments) > parked_runs or False,
+                   timeout=30, what="force-woke agent resumes") is not None,
+          "the force-woke agent resumes")
+    woke_notice = fake.goal_assignments[-1].get("resume_notice") or ""
+    check("force-woke this goal's supervision" in woke_notice
+          and "no pending wake" in woke_notice,
+          "the resume notice leads with the explicit force-woke line")
+    check(("draft (NOT enqueued" in woke_notice)
+          and (("saddle task enqueue %d" % stall_draft) in woke_notice),
+          "the statuses list names the never-enqueued draft with the fix")
+    check("Task statuses" in woke_notice,
+          "the force-wake notice keeps the standard batched report")
+    check(wait_for(lambda: run_status(goal) == "suspended" or False,
+                   timeout=60, what="woken agent re-parks") is not None,
+          "the woken agent parks again (script)")
+
+    # The flag is consumed: a second window stays quiet.
+    woke_runs = len(fake.goal_assignments)
+    check(quiet_window(quiet_seconds(),
+                       lambda: len(fake.goal_assignments) == woke_runs,
+                       "consumed flag never re-fires"),
+          "the consumed flag does not re-wake (one wake per press)")
+
+    print("== goal 4 t1: guards — paused goal, non-suspended run ==")
+    sealed_wake = state["force_wake"]
+    act(auth, goal_state(goal)["pause"], {})
+    state = goal_state(goal)
+    check(state.get("force_wake") is None,
+          "a paused goal offers no force-wake action")
+    # The server-side guard: replay the still-sealed force-wake action
+    # captured while the goal was active — the paused goal refuses it
+    # (error 72; the flag stays clear).
+    replay = act(auth, sealed_wake, {})
+    check(any(d.get("type") == "notify" and d.get("kind") == "error"
+              and "force-woke" in (d.get("message") or "")
+              for d in replay if isinstance(d, dict)),
+          "the paused-goal guard refuses the replayed wake (%s)" % replay)
+    conn = db()
+    cleared = conn.execute("select force_wake from goals where id = ?",
+                           (goal,)).fetchone()
+    conn.close()
+    check(cleared is not None and cleared[0] in (0, False),
+          "the goal force-wake flag stays clear after the guard refusal")
+    act(auth, goal_state(goal)["resume"], {})
+    check(wait_for(lambda: goal_state(goal)["status"] == "active" or False,
+                   timeout=30, what="goal reactivates") is not None,
+          "the goal reactivates")
+
+    print("== goal 4 t1: plain-task force wake on its affine saddle ==")
+    conn = db()
+    conn.execute("insert into tasks (project_id, kind, title, status, priority, mode,"
+                 " creator, created_at, suspended_at) values (1, 'task',"
+                 " 'Plain parked', 'suspended', 5, 'implement', 'human',"
+                 " strftime('%s','now'), strftime('%s','now'))")
+    plain_parked = conn.execute("select max(id) from tasks").fetchone()[0]
+    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()
+    detail = task_state(auth, plain_parked)
+    check(detail.get("force_wake") is not None,
+          "a suspended plain task offers the force-wake action")
+    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 gated Force wake button")
+    status, _ = fake.ask_now()
+    check(status == 204,
+          "without a press the parked plain task is not offerable (204)")
+    act(auth, task_state(auth, plain_parked)["force_wake"], {})
+    status, resume_payload = fake.ask_now()
+    check(status == 200 and int(resume_payload["task"]["id"]) == plain_parked,
+          "the force-woke plain task is claimable again")
+    notice = resume_payload.get("resume_notice") or ""
+    check("force-woke this parked task" in notice
+          and "resume where you left off" in 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")
+    check(wait_task_done(auth, plain_parked, timeout=60) is not None,
+          "the woken plain task completes")
+    check(task_state(auth, plain_parked).get("force_wake") is None,
+          "a done task offers no force-wake action")

test/harness.py

@@ -1271,7 +1271,7 @@ def start_bundle(name):
 
 BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
                  "allocation", "lifecycle", "answers", "push", "goals",
-                 "replay", "goalui", "users")
+                 "replay", "goalui", "forcewake", "users")
 
 
 def run_bundle(name):