clanker-saddle/test/bundles/lineage.py

"""The lineage 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: lineage — the plan-12 goal lineage: goal context on worker
# dispatches (read-only reference + ephemeral branch), the daemon's
# task-reading API (goal members, no parent/plan), and the goal
# agent's workspace affinity while parked.
# =============================================================================

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 exists only as a hold (no saddle carries it): the held goal
    # worker stays queued until the release loop lifts it.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "gpu"})

    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"])

    goal = create_goal_row({"project_id": "1", "title": "Lineage goal",
                            "brief": "Carry the goal lineage.",
                            "branch_name": "goal/lineage"})
    fake.goal_script = [("plan", "# Lineage plan\n\n1. Fan out\n2. Verify")]
    act(auth, goal_state(goal)["start"], {})
    check(wait_for(lambda: goal_state(goal)["status"] == "plan_review" or False,
                   timeout=60, what="lineage goal plans") is not None,
          "the lineage 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()
    auto_parent = 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",
        (auto_parent,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, auto_parent)["status"] == "suspended"
                   or False, timeout=60, what="lineage goal agent parks") is not None,
          "the lineage goal agent parks")

    # Two workers under the goal: one free, one held by the gpu tag.
    status, body = saddle_api("POST", "/api/saddle/task/create", r4_token, {
        "session_id": agent_session, "title": "Auto child",
        "mandate": "build the widget", "mode": "implement"})
    auto_child = int(body["task_id"])
    status, body = saddle_api("POST", "/api/saddle/task/create", r4_token, {
        "session_id": agent_session, "title": "Auto grandchild",
        "mandate": "verify the widget", "mode": "test",
        "capabilities": ["gpu"]})
    auto_grand = int(body["task_id"])
    status, body = saddle_api("POST", "/api/saddle/task/create", r4_token, {
        "session_id": agent_session, "title": "Wire the flange",
        "description": "torque to spec"})
    check(status == 400 and "mandate is required" in (body.get("error") or ""),
          "task create without a mandate errors verbatim (%s)" % body.get("error"))
    status, body = saddle_api("POST", "/api/saddle/task/create", r4_token, {
        "session_id": agent_session, "title": "Wire the flange",
        "mandate": "torque to spec"})
    flange_task = int(body["task_id"])
    for held in (auto_child, auto_grand, flange_task):
        saddle_api("POST", "/api/saddle/task/enqueue", r4_token,
                   {"session_id": agent_session, "task_id": held})
    check(wait_for(lambda: task_state(auth, auto_child)["status"] in
                   ("queued", "assigned", "working", "done") or False,
                   timeout=30, what="free goal worker dispatches") is not None,
          "the free goal worker dispatches")
    check(quiet_window(3, lambda: task_state(auth, auto_grand)["status"] == "queued",
                       "held goal worker stays queued"),
          "the gpu-held goal worker stays queued")

    # A goal worker's dispatch carries the read-only goal context.
    lineage_push = [p for p in fake.start_payloads
                    if int(p.get("task", {}).get("id") or 0) == auto_child][-1]
    goal_ctx = lineage_push.get("goal") or {}
    check(int(goal_ctx.get("id") or 0) == goal
          and goal_ctx.get("branch") == "goal/lineage"
          and "Carry the goal lineage." in goal_ctx.get("brief", "")
          and "# Lineage plan" in goal_ctx.get("plan", ""),
          "the worker dispatch carries the goal context (brief + approved plan)")
    check(lineage_push.get("ephemeral_branch") == "task/%d-%d" % (goal, auto_child)
          and lineage_push.get("kind") == "task"
          and lineage_push.get("mode") == "implement"
          and lineage_push.get("creator") == "goal",
          "the worker dispatch carries the ephemeral branch + identity triplet")
    check("parent" not in lineage_push and "subtasks" not in lineage_push,
          "the plan-12 payload drops the parent/subtasks members")

    print("== R5 B2: task-reading API ==")
    status, _ = saddle_api("GET", "/api/saddle/tasks", None)
    check(status == 401, "tasks endpoint rejects a missing token")
    status, _ = saddle_api("GET", "/api/saddle/task/%d" % auto_child, None)
    check(status == 401, "task detail endpoint rejects a missing token")

    # Fixture: a second project whose tasks all carry the "Diff " title
    # prefix for the project_id filter checks — created first so the
    # unfiltered list has the population the count check expects.
    state = slot_public(auth, "/projects", "projects")
    act(auth, state["create"], {"name": "diff-fabrica",
                                "git_url": "ssh://fabrica.unitatem.net:222/clanker/clanker-saddle.git"})
    conn = db()
    other_project = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                         if p["name"] == "diff-fabrica"][0])
    for suffix in ("fabrica", "scp"):
        conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                     " values (?, ?, 'done', 5, strftime('%s','now'))",
                     (other_project, "Diff %s" % suffix))
    conn.commit()
    conn.close()
    status, body = saddle_api("GET", "/api/saddle/tasks", r4_token)
    check(status == 200 and len(body.get("tasks", [])) > 5,
          "tasks endpoint lists tasks without filters")
    row = [t for t in body["tasks"] if int(t["id"]) == auto_child][0]
    check(row["title"] == "Auto child" and int(row["goal_id"] or 0) == goal
          and row["status"] in ("done", "queued", "working"),
          "compact rows carry id/title/status/goal_id")

    status, body = saddle_api("GET", "/api/saddle/tasks?project_id=1", r4_token)
    check(all(True for _ in [1])  # project 1 dominates the data set
          and not any(int(t["id"]) == 0 for t in body["tasks"]),
          "project_id filter accepted")
    status, body = saddle_api("GET",
                              "/api/saddle/tasks?project_id=%d" % other_project, r4_token)
    check(all(t["title"].startswith("Diff") for t in body["tasks"]),
          "other project lists only its own tasks")

    status, body = saddle_api("GET", "/api/saddle/tasks?q=FLANGE", r4_token)
    titles = [t["title"] for t in body["tasks"]]
    check("Wire the flange" in titles,
          "q matches titles case-insensitively")
    status, body = saddle_api("GET", "/api/saddle/tasks?q=torque%20to%20spec", r4_token)
    check(any(t["title"] == "Wire the flange" for t in body["tasks"]),
          "q matches mandate text")

    status, body = saddle_api("GET", "/api/saddle/task/%d" % auto_child, r4_token)
    task_view = body.get("task", {})
    check(status == 200 and int(task_view.get("id") or 0) == auto_child
          and task_view.get("goal_title") == "Lineage goal"
          and int(task_view.get("goal_id") or 0) == goal,
          "single-task view joins the goal title")
    check("status" in task_view and "description" in task_view
          and "summary" in task_view,
          "single-task view carries the full detail fields")
    check("plan" not in task_view and "parent_task_id" not in task_view,
          "the single-task view drops the deleted parent/plan members")
    status, body = saddle_api("GET", "/api/saddle/task/99999", r4_token)
    check(status == 404, "unknown task id returns 404")

    print("== settle: released workers resume the parked agent (coalesced) ==")
    # Release the held workers: their completions wake the parked goal
    # agent (affinity — the same fake hosts it) with the batched
    # notice; the free worker already completed.
    check(wait_task_done(auth, auto_child, timeout=90) is not None,
          "the free worker settles first")

    def agent_push_count():
        return len([p for p in fake.goal_assignments
                    if int(p.get("task", {}).get("id") or 0) == auto_parent])

    before_pushes = agent_push_count()
    for held_task in (auto_grand, flange_task):
        caps = task_state(auth, held_task).get("capabilities") or []
        if caps:
            act(auth, caps[0]["detach"], {})
        check(wait_task_done(auth, held_task, timeout=90) is not None,
              "released worker settles (%d)" % held_task)
    check(wait_for(lambda: agent_push_count() > before_pushes or False,
                   timeout=90, what="goal agent resumes") is not None,
          "the workers' completions resume the parked goal agent")
    notice = ""
    for p in fake.goal_assignments:
        if int(p.get("task", {}).get("id") or 0) == auto_parent:
            notice = p.get("resume_notice") or ""
    check("Auto child" in notice and "Auto grandchild" in notice
          and "Wire the flange" in notice,
          "the coalesced resume notice covers every plan task")
    check("New completion reports" in notice and "fake daemon finished" in notice,
          "the notice carries the new completion reports")

    act(auth, goal_state(goal)["cancel"], {})
    check(wait_for(lambda: goal_state(goal)["status"] == "cancelled" or False,
                   timeout=30, what="lineage goal cancelled") is not None,
          "the lineage goal cancels (drain)")