goals: failure cascade — fail_goal cancels non-terminal tasks, narrated

A failing goal now cascades exactly like a cancelled one: drafts are
deleted (no orphan stays dispatchable), queued/working/blocked tasks are
cancelled, the run task fails first, and the narrative records a cascade
event alongside the single goal-failed event.

src/Services/GoalService.vala

@@ -476,11 +476,15 @@ namespace ClankerSaddle.Services {
         /**
          * The goal agent failed the goal (`saddle failure` in goal
          * context): status 'failed', the report stored, the goal-run
-         * task failed (its session closes and saddle frees) and every
-         * user notified. Idempotent on terminal goals (plan 13 B2):
-         * a replayed failure push — redelivered OR a fresh entry id
+         * task failed (its session closes and saddle frees), every
+         * non-terminal task cascades exactly like a cancellation
+         * (plan 13 B3: drafts are deleted — no orphan stays
+         * dispatchable — queued/working/blocked tasks are cancelled)
+         * and the narrative records the cascade. Every user is
+         * notified. Idempotent on terminal goals (plan 13 B2): a
+         * replayed failure push — redelivered OR a fresh entry id
          * from the honest burst — re-runs no bookkeeping, so exactly
-         * one goal-failed event, one notification, ever.
+         * one goal-failed event, one cascade, one notification, ever.
          */
         public async void fail_goal(int64 goal_id, string? report) throws Error {
             var goal = yield require_goal(goal_id);
@@ -499,8 +503,30 @@ namespace ClankerSaddle.Services {
                 yield tasks.fail_task(((!)run).id,
                     report != null && report.length > 0 ? report : "The goal agent failed the goal");
             }
+            // Plan 13 B3: the failure cascade — the cancel-cascade
+            // semantics (the run task just failed above, so the loop's
+            // terminal skip covers it).
+            int64 cascaded = 0;
+            foreach (var task in yield db.query<TaskEntity>()
+                    .where(expr("goal_id == $0", elem<int64?>(goal_id)))
+                    .order_by(expr("id"))
+                    .materialise_async()) {
+                if (task.status == STATUS_DONE || task.status == STATUS_FAILED) {
+                    continue;
+                }
+                if (task.status == STATUS_DRAFT) {
+                    yield tasks.delete_task_quietly(task.id);
+                } else {
+                    yield tasks.cancel_task(task.id);
+                }
+                cascaded++;
+            }
             yield record_event(goal_id, "goal-failed",
                 report != null && report.length > 0 ? report : "The goal agent failed the goal");
+            if (cascaded > 0) {
+                yield record_event(goal_id, "cascade",
+                    @"Goal failure cancelled $(cascaded) outstanding task(s)");
+            }
             push.notify_global(PushService.goal_terminal_notification(
                 ((!)goal).title, goal_id, false));
             yield pulse.bump(0, "goal failed");

test/web_stack_test.py

@@ -5700,6 +5700,79 @@ def bundle_goals(ctx):
                          "brief": "b", "branch_name": "goal/doomed"})
     check(reuse > goal2, "a cancelled goal's branch name is reusable")
 
+    print("== plan 13 B3: goal failure cascades to non-terminal tasks ==")
+    # A goal holding a draft, a queued task and a working task: the
+    # agent's failure cascades like a cancellation — drafts deleted (no
+    # orphan stays dispatchable), queued/working cancelled — and the
+    # narrative records the cascade.
+    collapse = create_goal({"project_id": "1", "title": "Collapse goal",
+                            "brief": "Fails with tasks in flight.",
+                            "branch_name": "goal/collapse"})
+    fake.goal_script = [("suspend",)]
+    act(auth, goal_state(collapse)["start"], {})
+    conn = db()
+    run3 = conn.execute("select id from tasks where goal_id = ? and kind = 'goal_run'",
+                        (collapse,)).fetchone()[0]
+    conn.close()
+    check(wait_for(lambda: task_state(auth, run3)["status"] == "suspended" or False,
+                   timeout=60, what="collapse goal agent parks") is not None,
+          "the collapse goal's agent parks first")
+    conn = db()
+    run3_session = conn.execute(
+        "select id from sessions where task_id = ? and ended_at is null",
+        (run3,)).fetchone()[0]
+    conn.close()
+
+    def agent_task(title):
+        status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
+            "session_id": run3_session, "title": title, "mandate": "m"})
+        task_id = int(body["task_id"])
+        saddle_api("POST", "/api/saddle/task/enqueue", fake_token,
+                   {"session_id": run3_session, "task_id": task_id})
+        return task_id
+
+    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
+        "session_id": run3_session, "title": "Collapse draft", "mandate": "m"})
+    collapse_draft = int(body["task_id"])  # stays draft (never enqueued)
+    collapse_queued = agent_task("Collapse queued")
+    collapse_worker = agent_task("Collapse worker")
+    # Both queued tasks race for the one saddle; freeze the daemon the
+    # moment the first dispatches so one stays genuinely WORKING (and
+    # the other queued) when the goal fails.
+    fake.hold_state = True
+    def collapse_working():
+        for task_id in (collapse_queued, collapse_worker):
+            if task_state(auth, task_id)["status"] == "working":
+                return task_id
+        return False
+    running_one = wait_for(collapse_working, timeout=60, what="a collapse task works")
+    check(running_one is not None,
+          "a collapse task reaches working before the failure")
+    queued_one = collapse_worker if running_one == collapse_queued else collapse_queued
+
+    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
+                           {"session_id": run3_session, "state": "failed",
+                            "summary": "the foundation crumbled"})
+    check(status == 200 and goal_state(collapse)["status"] == "failed",
+          "the agent's failure fails the goal")
+    check(task_state(auth, run3)["status"] == "failed",
+          "the failure fails the goal agent's run task")
+    check(wait_for(lambda: task_state(auth, queued_one)["status"] == "failed"
+                   and task_state(auth, running_one)["status"] == "failed" or False,
+                   timeout=30, what="failure cascade settles") is not None,
+          "the failure cascade cancels the queued and working tasks")
+    check(not any(t["id"] == collapse_draft
+                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
+          "the failure cascade deletes the goal's draft tasks")
+    state = goal_state(collapse)
+    check(any(e["kind"] == "cascade" for e in state["events"]),
+          "the narrative records the failure cascade")
+    check(state["draft_tasks"] == 0 and state["queued_tasks"] == 0,
+          "nothing under the failed goal stays dispatchable (no orphan drafts)")
+    check(len([e for e in state["events"] if e["kind"] == "goal-failed"]) == 1,
+          "the failing goal records exactly one goal-failed event")
+    fake.hold_state = False
+
     print("== plan 12: auto-enqueue-after for human tasks ==")
     act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
         "project_id": "1", "title": "Predecessor task", "priority": "4"})