clanker-saddle/test/bundles/advisor_refine.py

"""The goal-27 review-refinement bundle (task 181): four deliverables
driven end-to-end.

1. Hidden tool-call bubbles: an assistant row that carries a tool call
   and said nothing worth a bubble (empty content or the bare
   "@tool call" marker) renders bubble_hidden — the chat page's slot
   hides its bubble while the paired tool row keeps its details
   expander. A row with real prose alongside the call keeps its bubble.

2. The chattier persona: the system prompt of every LLM request
   carries the conversational contract (address the user after tool
   calls, prefer ask_question, never open with a tool call).

3. Reports in reach: get_goal carries the completion report and
   completed_at, get_task the summary and completed_at, the list rows
   hint report_chars/summary_chars, and the new get_task_log returns
   the task's comments and progress/state-change entries (clamped).

4. The parking guard: a goal run pushing `suspended` while its goal
   owns NO wakeable task is refused — the park mechanics still run,
   but the run carries a one-shot pending_resume_notice, is
   immediately resume-offerable again, and the daemon's next ask
   hands it the refusal text as its resume_notice. With a live task
   under the goal the park lands normally (suspended, no notice).
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers
import harness  # the module itself (the re-pointing below reads it)
import bundles.advisor  # the imported helpers' module (see run())
from bundles.advisor import (ScriptedLlm, advisor_api, chat_rows,
                             wait_status)


def restart_with_llm(ctx, llm):
    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 db_rows(sql, params=()):
    conn = db()
    try:
        return conn.execute(sql, params).fetchall()
    finally:
        conn.close()


def run(ctx):
    # Sequential-suite trap (the shared wire helpers' known shape): the
    # port-dependent helpers imported from bundles.advisor read THAT
    # module's star-imported globals (WEB &c.), which the harness never
    # re-points for it — a solo run passes but the sequential full
    # suite dies with ConnectionRefused on the first call. Re-point.
    for name in ("WORK", "DB_PATH", "APP_PORT", "DAEMON_PORT", "WEB"):
        setattr(bundles.advisor, name, getattr(harness, name))

    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    # ------------------------------------------------------------------
    print("== refine 1+2: the prompt contract + the hidden call bubbles ==")
    llm = ScriptedLlm([
        # A silent tool call (no prose), then one WITH prose, then a
        # plain reply — three shapes of assistant row.
        {"tools": [{"id": "c1", "name": "list_projects",
                    "arguments": "{}"}]},
        {"content": "Let me look at the goals too.",
         "tools": [{"id": "c2", "name": "list_goals",
                    "arguments": "{}"}]},
        {"content": "All read."},
    ])
    llm.serve()
    auth = restart_with_llm(ctx, llm)

    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "Refinements"})
    chat_id = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": chat_id,
                             "message": "Take a look around."})
    check(status == 200, "the turn is accepted")
    check(wait_status(chat_id, "idle"), "the turn completes idle")
    check(wait_for(lambda: len(llm.requests) >= 3 or False, timeout=20,
                   what="three LLM calls") is not None,
          "the scripted rounds ran")

    # (2) the conversational contract rides every request's system prompt
    system = llm.requests[0]["messages"][0]
    check(system["role"] == "system", "the system prompt leads the wire")
    prompt = system["content"]
    for needle in ("CONVERSATION",
                   "ADDRESS THE USER",
                   "Never open with a",
                   "2-3 tool calls",
                   "PREFER ask_question"):
        check(needle in prompt,
              "the system prompt carries the contract language (%r)" % needle)
    for later in llm.requests[1:]:
        check(later["messages"][0]["content"] == prompt,
              "later rounds keep the same system prompt")

    # (1) the bubble gating: the slot's message rows flag the silent
    # call row hidden and the prose-alongside-call row visible.
    slot = slot_public(auth, "/chats/%d" % chat_id, "chat")
    rows = {(int(m["id"])): m for m in slot["messages"]}
    call_rows = [m for m in slot["messages"]
                 if m["kind"] == "assistant" and m.get("tool_name")]
    check(len(call_rows) == 2, "two tool-carrying assistant rows exist")
    silent = [m for m in call_rows if m.get("bubble_hidden")]
    spoken = [m for m in call_rows if not m.get("bubble_hidden")]
    check(len(silent) == 1 and len(spoken) == 1,
          "the silent call row hides its bubble, the spoken one keeps it")
    check(spoken[0]["content"] == "Let me look at the goals too.",
          "the kept bubble is the prose row")
    tool_details = [m for m in slot["messages"] if m["kind"] == "tool"]
    check(len(tool_details) == 2 and all(m.get("tool_name") for m in tool_details),
          "the tool rows (the details expanders) still render")

    # The page template itself gates on the flag.
    page = http_get_page(auth, "/chats/%d" % chat_id)
    check("message.bubble_hidden" in page,
          "the chat page's assistant bubble is gated on bubble_hidden")

    # ------------------------------------------------------------------
    print("== refine 3: completion reports within the advisor's reach ==")
    # A goal with a completion report + a task with a summary and log
    # rows, straight into the DB (the wire's own suites cover the
    # transitions; here the TOOLS are under test).
    conn = db()
    conn.execute(
        "insert into goals (project_id, title, brief, branch_name, status,"
        " plan, completion_report, completed_at, created_at)"
        " values (1, 'Reported goal', 'b', 'goal/reported', 'complete',"
        " 'plan text', 'The goal shipped.', strftime('%s','now'),"
        " strftime('%s','now'))")
    goal_id = conn.execute("select max(id) from goals").fetchone()[0]
    conn.execute(
        "insert into tasks (project_id, goal_id, kind, title, status,"
        " priority, mode, creator, summary, completed_at, created_at)"
        " values (1, ?, 'task', 'Reported task', 'done', 5, 'implement',"
        " 'human', 'Fixed the flange.', strftime('%s','now'),"
        " strftime('%s','now'))", (goal_id,))
    task_id = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute(
        "insert into task_comments (task_id, author, body, created_at)"
        " values (?, 'agent', 'The flange tests pass now.',"
        " strftime('%s','now'))", (task_id,))
    conn.execute(
        "insert into events (kind, actor, scope_type, scope_id, payload,"
        " created_at) values ('progress', 'agent', 'task', ?,"
        " '{\"text\": \"wrote the flange tests\"}', strftime('%s','now'))",
        (task_id,))
    conn.commit()
    conn.close()

    llm2 = ScriptedLlm([
        {"tools": [{"id": "g1", "name": "get_goal",
                    "arguments": json.dumps({"goal_id": goal_id})}]},
        {"tools": [{"id": "g2", "name": "get_task",
                    "arguments": json.dumps({"task_id": task_id})}]},
        {"tools": [{"id": "g3", "name": "get_task_log",
                    "arguments": json.dumps({"task_id": task_id})}]},
        {"content": "Read them all."},
    ])
    llm2.serve()
    auth = restart_with_llm(ctx, llm2)
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "Reports"})
    chat2 = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": chat2,
                             "message": "Pull the completion story."})
    check(status == 200, "the report-reading turn is accepted")
    # A fresh chat is already 'idle' — wait on the LLM traffic itself.
    check(wait_for(lambda: len(llm2.requests) >= 4 or False, timeout=25,
                   what="the report-reading rounds") is not None,
          "the report-reading turn completes")

    def tool_result(call_id):
        for request in llm2.requests:
            for message in request.get("messages") or []:
                if message.get("role") == "tool" and \
                        message.get("tool_call_id") == call_id:
                    return json.loads(message["content"])
        return None

    goal_view = tool_result("g1")
    check(goal_view is not None and goal_view.get("completion_report")
          == "The goal shipped." and goal_view.get("completed_at"),
          "get_goal returns the completion report + completed_at")
    task_view = tool_result("g2")
    check(task_view is not None and task_view.get("summary")
          == "Fixed the flange." and task_view.get("completed_at"),
          "get_task returns the final summary + completed_at")
    log_view = tool_result("g3")
    check(log_view is not None, "get_task_log answers")
    texts = [e.get("text", "") for e in (log_view.get("log") or [])]
    check(any("flange tests pass" in t for t in texts)
          and any("wrote the flange tests" in t for t in texts),
          "get_task_log carries the comment and the progress entry")

    # The list hints: a second turn lists and sees report_chars > 0.
    llm3 = ScriptedLlm([
        {"tools": [{"id": "l1", "name": "list_goals",
                    "arguments": "{}"}]},
        {"tools": [{"id": "l2", "name": "list_tasks",
                    "arguments": "{}"}]},
        {"content": "Listed."},
    ])
    llm3.serve()
    auth = restart_with_llm(ctx, llm3)
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "Lists"})
    chat3 = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": chat3, "message": "List things."})
    check(status == 200, "the listing turn is accepted")
    check(wait_for(lambda: len(llm3.requests) >= 3 or False, timeout=25,
                   what="the listing rounds") is not None,
          "the listing turn completes")

    def tool_result3(call_id):
        for request in llm3.requests:
            for message in request.get("messages") or []:
                if message.get("role") == "tool" and \
                        message.get("tool_call_id") == call_id:
                    return json.loads(message["content"])
        return None

    goals_list = tool_result3("l1") or {}
    row = next((g for g in goals_list.get("goals", [])
                if g.get("id") == goal_id), None)
    check(row is not None and row.get("report_chars") == len("The goal shipped."),
          "list_goals hints report_chars (%s)" % row)
    tasks_list = tool_result3("l2") or {}
    trow = next((t for t in tasks_list.get("tasks", [])
                 if t.get("id") == task_id), None)
    check(trow is not None and trow.get("summary_chars") == len("Fixed the flange."),
          "list_tasks hints summary_chars (%s)" % trow)

    # The new tool is declared on the wire.
    names = [t["function"]["name"] for t in llm.requests[0].get("tools") or []]
    check("get_task_log" in names, "get_task_log is declared to the model")

    # ------------------------------------------------------------------
    print("== refine 4: the parking guard ==")
    # Goal A: reaches active with NO owned task, then parks — refused.
    # After the refusal resume the script ENDS the run (the notice's
    # instruction: finish or cancel) so the churn cannot loop.
    fake.goal_script = [
        ("plan", "# Plan\n\n1. Park with nothing to wait on"),
        ("suspend",),
        ("done", "# Completion\n\nNothing to wait on; finished."),
    ]
    act(auth, slot_public(auth, "/goals", "goals")["create"],
        {"project_id": "1", "title": "Guard goal",
         "brief": "Parks with zero live tasks.",
         "branch_name": "goal/guard"})
    guard_goal = int([g for g in slot_public(auth, "/goals", "goals")["goals"]
                      if g["title"] == "Guard goal"][0]["id"])

    def run_task_of(goal_id):
        return int(db_rows(
            "select id from tasks where goal_id = ? and kind = 'goal_run'",
            (goal_id,))[0][0])

    def run_status(goal_id):
        return task_state(auth, run_task_of(goal_id))["status"]

    act(auth, goal_state(guard_goal)["start"], {})
    # The app restarts above orphan the fake's scheduled asks — the
    # goal run dispatches on an explicit ask (the daemon's own idle
    # ask after register).
    status, _ = fake.ask_now()
    check(wait_for(lambda: goal_state(guard_goal)["status"] == "plan_review"
                   or False, timeout=90, what="guard goal plans") is not None,
          "the guard goal reaches plan review")
    fake.goal_script = [("suspend",), ("done",
                        "# Completion\n\nNothing to wait on; finished.")]
    pending = goal_state(guard_goal)["plan_pending_block"]
    act(auth, pending["answer_action"], {"verdict": "accept"})
    check(wait_for(lambda: goal_state(guard_goal)["status"] == "active"
                   or False, timeout=60, what="plan approved") is not None,
          "approving the plan activates the goal")
    check(wait_for(lambda: run_status(guard_goal) == "suspended" or False,
                   timeout=60, what="the run parks") is not None,
          "the run parks (the mechanics still run)")

    guard_run = run_task_of(guard_goal)

    # The refusal is recorded durably: a system comment on the run.
    check(wait_for(lambda: db_rows(
        "select count(*) from task_comments where task_id = ?"
        " and body like 'Parking refused%'", (guard_run,))[0][0] > 0
        or False, timeout=15, what="the refusal comment") is not None,
        "the refused park leaves its instruction as a comment")

    # The refusal makes the run resume-offerable immediately — the
    # dispatcher's wake (or any ask) hands the daemon the notice as
    # its resume_notice.
    def refusal_resume():
        for assignment in fake.goal_assignments:
            notice = assignment.get("resume_notice") or ""
            if "Parking refused" in notice and "no live tasks" in notice:
                return notice
        return False
    notice = wait_for(refusal_resume, timeout=60, what="the refusal resume")
    check(notice is not None,
          "the resuming assignment carries the refusal notice")

    # One-shot: consumed by the resume that delivered it.
    check(wait_for(lambda: not db_rows(
        "select pending_resume_notice from tasks where id = ?",
        (guard_run,))[0][0] or False, timeout=15,
        what="the notice is consumed") is not None,
        "the notice is consumed by the resume (one-shot)")

    # The instructed ending: the agent finishes its run, the goal
    # enters completion review and completes.
    check(wait_for(lambda: goal_state(guard_goal)["status"]
                   == "completion_review" or False, timeout=60,
                   what="the refused run finishes") is not None,
          "the refused run finishes (submit its report)")
    pending = goal_state(guard_goal).get("completion_pending_block")
    if pending:
        act(auth, pending["answer_action"], {"verdict": "accept"})
    check(wait_for(lambda: goal_state(guard_goal)["status"] == "complete"
                   or False, timeout=60, what="the guard goal completes")
          is not None,
          "accepting the completion completes the guard goal")

    # Goal B: the SAME flow but with a live task under the goal — the
    # park must land normally (no notice). A BLOCKED live task: still
    # wakeable (not terminal), but never dispatched, so nothing fires
    # a terminal transition behind the test's back.
    fake.goal_script = [("plan", "# Plan\n\n1. Wait on a live task")]
    act(auth, slot_public(auth, "/goals", "goals")["create"],
        {"project_id": "1", "title": "Unguard goal",
         "brief": "Parks with a live task.",
         "branch_name": "goal/unguard"})
    unguard_goal = int([g for g
                        in slot_public(auth, "/goals", "goals")["goals"]
                        if g["title"] == "Unguard goal"][0]["id"])
    conn = db()
    conn.execute(
        "insert into tasks (project_id, goal_id, kind, title, status,"
        " priority, mode, creator, created_at)"
        " values (1, ?, 'task', 'Live worker', 'blocked', 5, 'implement',"
        " 'goal', strftime('%s','now'))", (unguard_goal,))
    conn.commit()
    conn.close()
    act(auth, goal_state(unguard_goal)["start"], {})
    status, _ = fake.ask_now()
    check(wait_for(lambda: goal_state(unguard_goal)["status"] == "plan_review"
                   or False, timeout=90, what="unguard goal plans") is not None,
          "the unguard goal reaches plan review")
    fake.goal_script = [("suspend",)]
    pending = goal_state(unguard_goal)["plan_pending_block"]
    act(auth, pending["answer_action"], {"verdict": "accept"})
    check(wait_for(lambda: goal_state(unguard_goal)["status"] == "active"
                   or False, timeout=60, what="unguard plan approved")
          is not None,
          "approving the unguard plan activates the goal")
    check(wait_for(lambda: run_status(unguard_goal) == "suspended" or False,
                   timeout=60, what="unguard run parks") is not None,
          "the run with a live task parks normally")
    conn = db()
    unguard_run = run_task_of(unguard_goal)
    plain_row = conn.execute(
        "select pending_resume_notice from tasks where id = ?",
        (unguard_run,)).fetchone()
    conn.close()
    check(plain_row is not None and not plain_row[0],
          "a park with something to wait on carries no refusal notice")

    llm.kill()
    llm2.kill()
    llm3.kill()