"""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).""" set_model_config(advisor_base_url="http://127.0.0.1:%d" % llm.port, advisor_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() set_model_config(advisor_base_url="", advisor_model="")