clanker-saddle/test/bundles/subtasks.py

"""The subtasks 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: subtasks — the plan-12 task graph: plain-task parking (no
# subtask events exist to resume it), the DELETED subtask machinery's
# verbatim refusals, and the goal agent's task CLI in depth
# (capabilities, ownership, the daemon's task-reading API, the
# goal-run block answered from the task page).
# =============================================================================

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

    # A hold tag nothing carries: goal tasks created with it stay
    # queued (a stable parked agent) until released at the settle.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "gpu"})
    gpu_cap_id = [c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                  if c["name"] == "gpu"][0]

    def create_goal_row(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"])

    print("== B7: plain-task parking stays put (no subtask events) ==")
    fake.suspend_on_title = "Suspend plain"
    pushes_before = len(fake.start_payloads)
    plain = create_queued(auth, {
        "project_id": "1",
        "title": "Suspend plain",
        "priority": "2",
    })
    check(wait_for(lambda: task_state(auth, plain).get("status") == "suspended"
                   or False, timeout=60, what="plain task parks") is not None,
          "daemon 'suspended' state parks the task")
    detail = task_state(auth, plain)
    check(detail.get("display") == "parked",
          "a parked plain task displays as parked (%s)" % detail.get("display"))
    check(any(s["status"] == "idle"
              for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
          "suspension frees the saddle")
    first_push = fake.start_payloads[pushes_before]
    check(first_push.get("kind") == "task" and first_push.get("creator") == "human"
          and first_push.get("mode") == "implement",
          "the /start payload carries the plan-12 identity triplet (kind/creator/mode)")
    check("subtasks" not in first_push and "parent" not in first_push,
          "the plan-12 payload drops the subtasks/parent members")

    # Other tasks finishing must NOT resume a parked plain task (the
    # subtask-event channel is gone; only goal events resume goal runs).
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, summary,"
                 " completed_at, created_at) values (1, 'Unrelated finisher', 'done',"
                 " 5, 'finished loudly', strftime('%s','now','+1 second'),"
                 " strftime('%s','now'))")
    conn.commit()
    conn.close()
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, plain)["status"] == "suspended",
                       "plain park survives unrelated finishes"),
          "a parked plain task never resumes on other tasks' events")

    print("== the deleted machinery refuses, verbatim ==")
    status, body = saddle_api("POST", "/api/saddle/subtask", fake_token,
                              {"parent_task_id": plain, "title": "ghost"})
    check(status == 404,
          "the /api/saddle/subtask callback route is gone (404)")
    # The poll-era ingest channel went with it (arch 19): the whole
    # /update surface 404s — subtask and plan verb arms included — and
    # the verdict channel (/poll) with it.
    conn = db()
    plain_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (plain,)).fetchone()[0]
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                              {"type": "subtask", "session_id": plain_session,
                               "data": {"title": "Ghost child"}})
    check(status == 404,
          "the /update ingest surface is gone (404, the subtask arm with it)")
    status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                              {"type": "plan", "session_id": plain_session,
                               "data": {"plan": "# nope"}})
    check(status == 404,
          "the task-level 'plan' update arm is gone with the channel (404)")
    # The daemon's task CLI: no parent/plan members; goal context instead.
    status, body = saddle_api("GET", "/api/saddle/task/%d" % plain, fake_token)
    check(status == 200 and "plan" not in body.get("task", {})
          and "parent_task_id" not in body.get("task", {})
          and "goal_id" in body.get("task", {}),
          "the task CLI detail drops parent/plan and carries goal_id")
    status, body = saddle_api("GET", "/api/saddle/tasks?project_id=1", fake_token)
    check(status == 200 and all("parent_task_id" not in t and "goal_id" in t
                                for t in body.get("tasks", [])),
          "the task CLI list drops parent_task_id and carries goal_id")
    # Settle the parked plain task: nothing will resume it — cancel it.
    act(auth, task_state(auth, plain)["cancel"], {})
    check(wait_for(lambda: task_state(auth, plain)["status"] == "failed" or False,
                   timeout=30, what="parked plain task cancelled") is not None,
          "the parked plain task cancels (the human cleanup path)")

    print("== the goal agent's task CLI in depth ==")
    goal = create_goal_row({"project_id": "1", "title": "Sequencing goal",
                            "brief": "Fan tasks out with the CLI.",
                            "branch_name": "goal/sequence"})
    fake.goal_script = [("plan", "# Plan\n\n1. Fan out")]
    act(auth, goal_state(goal)["start"], {})
    check(wait_for(lambda: goal_state(goal)["status"] == "plan_review" or False,
                   timeout=60, what="sequencing goal plans") is not None,
          "the sequencing goal reaches plan review")
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(goal)["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    conn = db()
    run_task = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (goal,)).fetchone()[0]
    agent_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (run_task,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="sequencing agent parks") is not None,
          "the sequencing goal agent parks")

    # Create with capabilities: attached until assignment (the hold
    # keeps the task queued), mandate + mode stored, ephemeral branch.
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session, "title": "Held worker",
        "mandate": "Do the held work.", "mode": "review",
        "capabilities": ["gpu"]})
    check(status == 200 and body.get("ok") is True,
          "task create with a capability tag succeeds")
    held = int(body["task_id"])
    detail = task_state(auth, held)
    check(detail["status"] == "draft" and detail["mode"] == "review",
          "the held worker drafts with its mode")
    check([c["name"] for c in detail.get("capabilities", [])] == ["gpu"],
          "the capability attaches to the created task")
    check(detail["task_branch"] == "task/%d-%d" % (goal, held)
          and detail["branch_from"] == "goal/sequence",
          "the CLI task's ephemeral branch pair is stored")
    status, body = saddle_api("POST", "/api/saddle/task/enqueue", fake_token, {
        "session_id": agent_session, "task_id": held})
    check(status == 200, "the held worker enqueues")
    check(quiet_window(3, lambda: task_state(auth, held)["status"] == "queued",
                       "unsatisfiable capability holds dispatch"),
          "the unsatisfiable capability holds the queued worker (never assigned)")
    queue_row = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                 if int(t["id"]) == held][0]
    check(queue_row.get("waiting_for_saddle") is True,
          "the queue annotates the held worker as waiting for a saddle")

    # The human edit path: goal tasks never carry enqueue-after (the
    # field is human-tasks-only — the update silently ignores it).
    act(auth, task_state(auth, held)["update"], {
        "priority": "4", "enqueue_after_task_id": str(held - 1)})
    check(task_state(auth, held).get("enqueue_after_task_id") in (None, 0, "0"),
          "a goal task ignores the auto-enqueue-after field on edit")

    # Ownership: the same saddle's goal agent cannot touch another
    # goal's task; a plain session is not a goal-agent identity.
    other_goal = create_goal_row({"project_id": "1", "title": "Other goal",
                                  "brief": "b", "branch_name": "goal/other",
                                  "priority": "9"})
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token,
                              {"session_id": agent_session, "title": "x",
                               "mandate": "m"})
    scope_task = int(body["task_id"])
    check(int(task_state(auth, scope_task)["goal_id"] or 0) == goal,
          "the callback scopes task creation to its own goal")
    status, body = saddle_api("POST", "/api/saddle/task/delete", fake_token,
                              {"session_id": agent_session, "task_id": scope_task})
    check(status == 200, "the owned draft deletes")
    check(not any(t["id"] == scope_task
                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
          "the deleted task is gone from the queue")
    status, body = saddle_api("POST", "/api/saddle/task/enqueue", fake_token,
                              {"session_id": agent_session, "task_id": plain})
    check(status == 400 and "does not belong to your goal" in (body.get("error") or ""),
          "the goal agent cannot enqueue a standalone task")
    # The daemon's task CLI detail carries the goal context.
    status, body = saddle_api("GET", "/api/saddle/task/%d" % held, fake_token)
    check(status == 200
          and int(body["task"].get("goal_id") or 0) == goal
          and body["task"].get("goal_title") == "Sequencing goal",
          "the task CLI detail names the owning goal")

    print("== the goal-run block answers from the task page (C1) ==")
    # A question block raised in goal context lands on the run task's
    # own row; the TASK page's answer action funnels into the one
    # goal answer path and resumes the agent with the verdict riding
    # the allocation (the goal page shares the path).
    fake.goal_script = [("question", "Which harness: pytest or vitest?",
                         ["pytest", "vitest"])]
    conn = db()
    # Plan 15 C3: a terminal transition (a done task row newer than
    # the watermark) is the resume signal — no event rows exist.
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, completed_at, created_at) values (1, ?, 'task',"
                 " 'Nudge', 'done', 5,"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (goal,))
    conn.commit()
    conn.close()
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "blocked"
                   or False, timeout=60, what="goal block parks the run") is not None,
          "a goal question parks the run task on its own block row")
    run_detail = task_state(auth, run_task)
    question_block = [b for b in run_detail["blocks"]
                      if b["block_type"] == "question"][0]
    check(question_block.get("question") == "Which harness: pytest or vitest?",
          "the run task's block carries the goal question verbatim")
    goal_block_ids = [int(b["id"]) for b in goal_state(goal)["blocks"]
                      if b.get("block_type") == "question"]
    check(goal_block_ids == [int(question_block["id"])],
          "the goal page's card is the SAME row (no dual-row pair)")
    act(auth, question_block["answer_action"], {"answer": "pytest"})
    fake.goal_script = [("suspend",)]
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="answered block resumes the agent")
          is not None,
          "the answered block resumes the goal agent")
    answered = [b for b in task_state(auth, run_task)["blocks"]
                if b["block_type"] == "question"][0]
    check(answered.get("answer") == "pytest",
          "the verdict is stored on the single block row")

    print("== settle: release the hold and let the shard drain ==")
    # Drop the hold tag so the held worker can dispatch and complete;
    # the parked agent resumes on its completion, re-parks (script),
    # and the shard shuts down with an empty queue.
    act(auth, task_state(auth, held)["capabilities"][0]["detach"], {})
    check(wait_task_done(auth, held, timeout=90) is not None,
          "the released held worker completes")
    check(wait_for(lambda: any(e["kind"] == "task-done" and int(e["task_id"]) == held
                               for e in goal_state(goal)["events"]) or False,
                   timeout=30, what="narrative hears the held worker") is not None,
          "the narrative records the held worker's completion")
    act(auth, goal_state(goal)["cancel"], {})
    check(wait_for(lambda: goal_state(goal)["status"] == "cancelled" or False,
                   timeout=30, what="sequencing goal cancelled") is not None,
          "the sequencing goal cancels (the shard's cleanup path)")