"""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 max(id) from sessions where task_id = ?", (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] 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") conn = db() agent_session = conn.execute( "select max(id) from sessions where task_id = ?", (run_task,)).fetchone()[0] conn.close() # 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)")