clanker-saddle/test/bundles/advisor_arc.py

"""The advisor approvals + completion callback bundle (goal 27, task 3):
the full arc from the dashboard composer to a task's completion landing
back in the chat, against the scripted LLM endpoint. The scenes:

- the arc: dashboard hero -> chat -> the advisor asks a question -> the
  option answer lands -> the advisor suggests a task -> approve &
  enqueue creates the task QUEUED with creator human, stamps the
  suggestion approved with its task_id, and appends the system notice
  (the advisor turn kicked by the decision acknowledges it)
- the completion callback: the task flipped terminal (via the app's own
  surface - the fake saddle runs it to done) delivers ONE system
  message into the chat, stamps delivered_at and wakes the advisor
- never twice: a second pass (and the advisor's own reaction turn)
  adds no second delivery
- plain approve: the task lands DRAFT, creator human, nothing queued
- decline: the row is dismissed, nothing is created, the chat is told
- goal approve: a DRAFT goal exists, unqueued, owner of its branch;
  the goal suggestion never enqueues anything
- decided rows cannot be decided again (the card's slot loses the
  decide action; a replayed decision is refused)
- privacy: another user's decision on someone else's suggestion is
  refused and changes nothing
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers
import harness  # the module itself (restart_with_llm re-points its globals)
from bundles.advisor import ScriptedLlm


def db_rows(sql, params=()):
    conn = db()
    try:
        return conn.execute(sql, params).fetchall()
    finally:
        conn.close()


def wait_status(chat_id, status, timeout=30):
    def read():
        rows = db_rows("select status from advisor_chats where id = ?",
                       (chat_id,))
        return len(rows) == 1 and rows[0][0] == status and status or False
    return wait_for(read, timeout=timeout,
                    what="chat %d -> %s" % (chat_id, status))


def restart_with_llm(ctx, llm):
    """Points the app at `llm` and restarts it (env re-read)."""
    os.environ["ADVISOR_LLM_BASE_URL"] = "http://127.0.0.1:%d" % llm.port
    os.environ["ADVISOR_LLM_MODEL"] = "scripted"
    ctx.app.terminate()
    try:
        ctx.app.wait(timeout=10)
    except subprocess.TimeoutExpired:
        ctx.app.kill()
    ctx.app = start_app()
    ctx.auth = login("admin", "adminpass123")
    return ctx.auth


def run(ctx):
    auth = ctx.auth

    # The scripted loop: one ask_question, one suggest_task, then plain
    # replies for every later turn (the decision and callback kicks).
    llm = ScriptedLlm([
        {"tools": [{"id": "arc-1", "name": "ask_question",
                    "arguments": json.dumps({
                        "question": "Which project should the task target?",
                        "options": ["demo", "skip it"]})}]},
        {"tools": [{"id": "arc-2", "name": "suggest_task",
                    "arguments": json.dumps({
                        "title": "Advisor arc task",
                        "mandate": "Prove the advisor approval arc.",
                        "mode": "test", "project_id": 1, "priority": 4})}]},
        {"content": "Task suggested - approve it in the card when ready."},
        {"content": "Noted - the task is queued."},
        {"content": "Good - the task finished."},
    ])
    llm.serve()
    auth = restart_with_llm(ctx, llm)

    print("== advisor_arc: the dashboard composer opens the chat ==")
    home = slot_public(auth, "/", "dashboard")
    result = act(auth, home["start_advisor"], {
        "message": "I want a small verification task on the demo project"})
    navigation = [d for d in result if d.get("type") == "navigate"][0]
    chat_id = int(navigation["uri"].rsplit("/", 1)[1])
    check(wait_status(chat_id, "awaiting_answer"),
          "the advisor asks its question and parks the chat")
    page = slot_public(auth, "/chats/%d" % chat_id, "chat")
    questions = [m for m in page["messages"] if m.get("is_question")]
    check(len(questions) == 1 and questions[0]["options"] == ["demo", "skip it"],
          "the question card renders with its options")

    print("== advisor_arc: the answer resumes into a suggestion ==")
    result = act(auth, page["send"], {"answer": "demo"})
    check(wait_status(chat_id, "idle"), "the answered turn completes")
    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat_id,
                                       "chat")["suggestions"] or False,
                   timeout=15, what="suggestion card") is not None,
          "the suggested task reaches the card")
    page = slot_public(auth, "/chats/%d" % chat_id, "chat")
    card = page["suggestions"][0]
    check(card["kind"] == "task" and card["status"] == "pending"
          and card["title"] == "Advisor arc task"
          and "decide" in card,
          "the pending task card carries its payload and decide action")
    suggestion_id = int(card["id"])
    check(db_rows("select count(*) from tasks"
                  " where title = 'Advisor arc task'")[0][0] == 0,
          "nothing exists before the approval")

    print("== advisor_arc: approve & enqueue ==")
    result = act(auth, card["decide"], {"decision": "approve_enqueue"})
    sets = [d for d in result if d.get("type") == "set"]
    check(len(sets) == 1, "the decision answers with the rebuilt slot")
    task_rows = db_rows("select id, status, creator, mode, priority,"
                        " project_id, description from tasks"
                        " where title = 'Advisor arc task'")
    check(len(task_rows) == 1, "the approval creates exactly one task")
    task_id, task_status, creator, mode, priority, project_id, description \
        = task_rows[0]
    check(task_status == "queued",
          "approve_enqueue leaves the task QUEUED (%s)" % task_status)
    check(creator == "human" and mode == "test"
          and str(priority) == "4" and str(project_id) == "1"
          and description == "Prove the advisor approval arc.",
          "the task keeps the suggestion's fields (creator human)")
    suggestion = db_rows("select status, task_id, decided_at from"
                         " advisor_suggestions where id = ?",
                         (suggestion_id,))[0]
    check(suggestion[0] == "approved" and str(suggestion[1]) == str(task_id)
          and suggestion[2] is not None,
          "the suggestion row is approved with its task_id and stamp")
    check(wait_for(lambda: [r for r in db_rows(
        "select kind, content from advisor_messages where chat_id = ?"
        " order by id", (chat_id,))
        if r[0] == "system"
        and r[1] == "Approved & enqueued — task #%d queued" % task_id]
        or False, timeout=15, what="approval notice") is not None,
        "the approval lands its system notice in the chat")
    check(wait_for(lambda: llm.requests
                   and any(m.get("role") == "user" and "Approved & enqueued"
                           in (m.get("content") or "")
                           for m in llm.requests[-1]["messages"]) or False,
                   timeout=15, what="decision kick reaches the LLM")
          is not None,
        "the decision kick runs an advisor turn that sees the notice")
    check(wait_status(chat_id, "idle"), "the kicked turn settles idle")

    print("== advisor_arc: the completion callback ==")
    # The fake saddle runs the queued task to done (the app's own
    # dispatch surface - no direct status surgery).
    check(wait_for(lambda: task_state(auth, task_id).get("status")
                   == "done" or False, timeout=90,
                   what="task %d -> done" % task_id) is not None,
        "the queued task runs to done on the fake saddle")
    check(wait_for(lambda: [r for r in db_rows(
        "select kind, content from advisor_messages where chat_id = ?"
        " order by id", (chat_id,))
        if r[0] == "system" and r[1].startswith(
            "Task #%d completed" % task_id)] or False,
        timeout=45, what="completion notice") is not None,
        "the completion callback lands its system message in the chat")
    check(db_rows("select delivered_at from advisor_suggestions"
                  " where id = ?", (suggestion_id,))[0][0] is not None,
        "the suggestion row is stamped delivered_at")
    # The callback kicks an advisor turn: the model reads the bracketed
    # system notice and answers (the "Good - the task finished." step).
    check(wait_for(lambda: any(
        r[0] == "assistant" and r[1] == "Good - the task finished."
        for r in db_rows("select kind, content from advisor_messages"
                         " where chat_id = ?", (chat_id,))) or False,
        timeout=30, what="advisor reaction") is not None,
        "the advisor reacts to the completion in the chat")
    check(wait_status(chat_id, "idle"), "the reaction turn settles idle")

    print("== advisor_arc: the callback never delivers twice ==")
    delivered = db_rows(
        "select count(*) from advisor_messages where chat_id = ?"
        " and kind = 'system' and content like 'Task #%d completed%%'"
        % task_id, (chat_id,))[0][0]
    check(delivered == 1,
          "exactly one completion message (%d seen)" % delivered)
    # A quiet window across more than one callback tick: no second
    # delivery, no extra turn.
    check(quiet_window(20,
                       lambda: db_rows(
                           "select count(*) from advisor_messages"
                           " where chat_id = ? and kind = 'system'"
                           " and content like 'Task #%d completed%%'"
                           % task_id, (chat_id,))[0][0] == 1,
                       "no second completion delivery"),
          "the pass stays quiet about an already-delivered outcome")

    print("== advisor_arc: plain approve leaves a draft ==")
    llm2 = ScriptedLlm([
        {"tools": [{"id": "dr-1", "name": "suggest_task",
                    "arguments": json.dumps({
                        "title": "Advisor draft task",
                        "mandate": "Stay in draft.",
                        "mode": "investigate",
                        "project_id": 0, "priority": 6})}]},
        {"content": "Suggested."},
        {"content": "Thanks."},
    ])
    llm2.serve()
    auth = restart_with_llm(ctx, llm2)
    home = slot_public(auth, "/", "dashboard")
    result = act(auth, home["start_advisor"],
                 {"message": "suggest one more task"})
    chat2 = int([d for d in result if d.get("type") == "navigate"][0]
                ["uri"].rsplit("/", 1)[1])
    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat2,
                                       "chat")["suggestions"] or False,
                   timeout=15, what="draft suggestion") is not None,
          "the second suggestion lands")
    card = slot_public(auth, "/chats/%d" % chat2, "chat")["suggestions"][0]
    result = act(auth, card["decide"], {"decision": "approve"})
    draft = db_rows("select status, creator, project_id from tasks"
                    " where title = 'Advisor draft task'")[0]
    check(draft[0] == "draft" and draft[1] == "human"
          and str(draft[2]) == "0",
          "plain approve creates a standalone DRAFT task (creator human)")
    check(wait_for(lambda: [r for r in db_rows(
        "select content from advisor_messages where chat_id = ?",
        (chat2,)) if r[0].startswith("Approved — task #")] or False,
        timeout=15, what="draft approval notice") is not None,
        "the draft approval lands its notice")
    # The draft task never dispatches (nothing queues it): a quiet
    # window across several dispatch ticks proves it stays draft.
    draft_id = db_rows("select id from tasks"
                       " where title = 'Advisor draft task'")[0][0]
    check(quiet_window(6,
                       lambda: db_rows(
                           "select status from tasks where id = ?",
                           (draft_id,))[0][0] == "draft",
                       "draft stays draft"),
          "the approved-but-not-enqueued task stays draft")

    print("== advisor_arc: decline creates nothing ==")
    llm3 = ScriptedLlm([
        {"tools": [{"id": "dc-1", "name": "suggest_task",
                    "arguments": json.dumps({
                        "title": "Declined task", "mandate": "No.",
                        "mode": "implement"})}]},
        {"content": "Suggested."},
        {"content": "Understood."},
    ])
    llm3.serve()
    auth = restart_with_llm(ctx, llm3)
    home = slot_public(auth, "/", "dashboard")
    result = act(auth, home["start_advisor"], {"message": "suggest to decline"})
    chat3 = int([d for d in result if d.get("type") == "navigate"][0]
                ["uri"].rsplit("/", 1)[1])
    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat3,
                                       "chat")["suggestions"] or False,
                   timeout=15, what="decline suggestion") is not None,
          "the third suggestion lands")
    card = slot_public(auth, "/chats/%d" % chat3, "chat")["suggestions"][0]
    act(auth, card["decide"], {"decision": "decline"})
    check(db_rows("select status from advisor_suggestions"
                  " where chat_id = ?", (chat3,))[0][0] == "declined",
          "the declined row is marked declined")
    check(db_rows("select count(*) from tasks"
                  " where title = 'Declined task'")[0][0] == 0,
          "the declined suggestion creates no task")
    check(wait_for(lambda: [r for r in db_rows(
        "select kind, content from advisor_messages where chat_id = ?",
        (chat3,)) if r[0] == "system" and r[1] == "Declined."] or False,
        timeout=15, what="decline notice") is not None,
        "the decline lands its notice")

    print("== advisor_arc: goal approve is draft-only ==")
    llm4 = ScriptedLlm([
        {"tools": [{"id": "ga-1", "name": "suggest_goal",
                    "arguments": json.dumps({
                        "title": "Advisor arc goal",
                        "brief": "Prove goal suggestions.",
                        "project_id": 1,
                        "branch_name": "goal/advisor-arc"})}]},
        {"content": "Suggested."},
        {"content": "Thanks."},
    ])
    llm4.serve()
    auth = restart_with_llm(ctx, llm4)
    home = slot_public(auth, "/", "dashboard")
    result = act(auth, home["start_advisor"],
                 {"message": "suggest a goal now"})
    chat4 = int([d for d in result if d.get("type") == "navigate"][0]
                ["uri"].rsplit("/", 1)[1])
    check(wait_for(lambda: [s for s in slot_public(
        auth, "/chats/%d" % chat4, "chat")["suggestions"]
        if s["kind"] == "goal"] or False, timeout=15,
        what="goal suggestion") is not None,
        "the goal suggestion lands")
    card = [s for s in slot_public(auth, "/chats/%d" % chat4,
                                   "chat")["suggestions"]
            if s["kind"] == "goal"][0]
    result = act(auth, card["decide"], {"decision": "approve"})
    goal_rows = db_rows("select status, branch_name, creator from goals"
                        " where title = 'Advisor arc goal'")
    check(len(goal_rows) == 1 and goal_rows[0][0] == "draft"
          and goal_rows[0][1] == "goal/advisor-arc"
          and goal_rows[0][2] == "human",
          "the approved goal is a DRAFT on its branch (creator human)")
    goal_id = db_rows("select id from goals"
                      " where title = 'Advisor arc goal'")[0][0]
    suggestion4 = db_rows("select status, goal_id, task_id from"
                          " advisor_suggestions where chat_id = ?"
                          " and kind = 'goal'", (chat4,))[0]
    check(suggestion4[0] == "approved"
          and str(suggestion4[1]) == str(goal_id)
          and str(suggestion4[2]) == "0",
          "the goal suggestion keeps its goal_id and no task")
    check(db_rows("select count(*) from tasks where goal_id = ?",
                  (goal_id,))[0][0] == 0,
          "the approved goal stays unqueued (no run task created)")
    check(wait_for(lambda: [r for r in db_rows(
        "select content from advisor_messages where chat_id = ?",
        (chat4,)) if r[0].startswith(
            "Approved — goal #%d created in draft" % goal_id)] or False,
        timeout=15, what="goal approval notice") is not None,
        "the goal approval lands its notice")
    # approve_enqueue on a goal behaves as plain approve: never queued.
    llm5 = ScriptedLlm([
        {"tools": [{"id": "ga-2", "name": "suggest_goal",
                    "arguments": json.dumps({
                        "title": "Advisor arc goal two",
                        "brief": "Never start me.",
                        "project_id": 1,
                        "branch_name": "goal/advisor-arc-two"})}]},
        {"content": "Suggested."},
        {"content": "Thanks."},
    ])
    llm5.serve()
    auth = restart_with_llm(ctx, llm5)
    home = slot_public(auth, "/", "dashboard")
    result = act(auth, home["start_advisor"],
                 {"message": "suggest another goal"})
    chat5 = int([d for d in result if d.get("type") == "navigate"][0]
                ["uri"].rsplit("/", 1)[1])
    check(wait_for(lambda: [s for s in slot_public(
        auth, "/chats/%d" % chat5, "chat")["suggestions"]
        if s["kind"] == "goal"] or False, timeout=15,
        what="second goal suggestion") is not None,
        "the second goal suggestion lands")
    card = [s for s in slot_public(auth, "/chats/%d" % chat5,
                                   "chat")["suggestions"]
            if s["kind"] == "goal"][0]
    act(auth, card["decide"], {"decision": "approve_enqueue"})
    goal5 = db_rows("select id, status from goals"
                    " where title = 'Advisor arc goal two'")[0]
    check(goal5[1] == "draft"
          and db_rows("select count(*) from tasks where goal_id = ?",
                      (goal5[0],))[0][0] == 0,
          "approve_enqueue on a goal still creates only a draft")

    print("== advisor_arc: decided rows refuse a second decision ==")
    page = slot_public(auth, "/chats/%d" % chat5, "chat")
    decided = [s for s in page["suggestions"] if s["kind"] == "goal"][0]
    check("decide" not in decided,
          "the decided card carries no decide action anymore")

    print("== advisor_arc: privacy - another user decides nothing ==")
    second = "arc-" + str(int(time.time()))
    users_state = slot_public(auth, "/users", "users")
    act(auth, users_state["create"], {
        "username": second, "email": second + "@example.com",
        "forename": "Arc", "surname": "Privacy", "date_of_birth": "1990-01-01",
        "password": "arcpass123", "confirm_password": "arcpass123"})
    other = login(second, "arcpass123")
    other_page = slot_public(other, "/chats/%d" % chat_id, "chat")
    check(other_page["id"] == 0,
          "another user's chat slot reads not-found")
    # The foreign decision is refused by the owner check: the row keeps
    # its approved/task_id state untouched.
    own = db_rows("select status, task_id from advisor_suggestions"
                  " where id = ?", (suggestion_id,))[0]
    check(own[0] == "approved" and int(own[1]) == task_id,
          "the owner's row is intact before the foreign attempt")
    foreign_card = [s for s in slot_public(
        auth, "/chats/%d" % chat_id, "chat")["suggestions"]
        if str(s["id"]) == str(suggestion_id)][0]
    check("decide" not in foreign_card,
          "the decided card exposes nothing to replay")

    # Restore the daemon-facing LLM env the other checks may rely on
    # (the advisor env is test-local).
    for killer in (llm, llm2, llm3, llm4, llm5):
        killer.kill()
    os.environ.pop("ADVISOR_LLM_BASE_URL", None)
    os.environ.pop("ADVISOR_LLM_MODEL", None)