clanker-saddle/test/bundles/chatui.py

"""The advisor chat UI bundle (goal 27, task 2): the /chats list, the
/chats/{id} session page, the dashboard hero composer and the statum
actions under them, driven end-to-end against the scripted LLM
endpoint (the same ScriptedLlm the backend bundle uses). The scenes:

- the pages render: /chats lists the user's chats (title, status
  chip, delete action), the dashboard carries the hero composer
- the hero: StartAdvisorChatAction creates the chat titled from the
  first line (clamped), records the message, kicks the turn and
  navigates to /chats/{id}
- a chat page: the transcript slot carries the user message, the
  scripted tool call round (name/arguments/result parsed from the
  payload), the parked question with its options, and the send
  action is sealed against the chat
- answering: the question card's answer rides SendAdvisorMessage's
  answer path -> "Human selected: ..." lands, the turn resumes and
  the final assistant reply appears in the slot
- the refresh action: RefreshAdvisorChatAction rebuilds the slot
- suggestions: the card state (title/mandate/mode/priority + the
  sealed decide action) renders from the inserted row
- delete: the chat leaves the list and its page state reads as
  not-found (id 0)
- privacy: a second user's /chats slot sees none of the first
  user's chats and their /chats/{id} slot renders not-found
- the unconfigured dashboard: the hero renders with
  advisor_configured false
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers
import harness  # the module itself (the unconfigured scene re-points its globals)
from bundles.advisor import ScriptedLlm


def restart_with_llm(ctx, llm):
    """Points the app at `llm` and restarts it (the model_config row
    is the configuration source — set_model_config writes it)."""
    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 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 run(ctx):
    auth = ctx.auth

    print("== chatui: the pages render ==")
    chats = slot_public(auth, "/chats", "chats")
    check("chats" in chats, "the /chats slot carries the chat rows")
    check(chats.get("advisor_configured") is True,
          "the /chats slot reports the advisor configured")
    home = slot_public(auth, "/", "dashboard")
    check("start_advisor" in home, "the dashboard carries the hero action")
    home_html = http_get_page(auth, "/")
    check("advisor-composer" in home_html and "Ask the advisor" in home_html,
          "the dashboard renders the hero composer")
    check('stm-attribute.disabled="!dashboard.advisor_configured"'
          in home_html,
          "the hero composer disables itself on the configured flag")
    chats_html = http_get_page(auth, "/chats")
    check("Your advisor conversations" in chats_html
          and "dashboard" in chats_html,
          "the /chats page renders its copy and the empty-state pointer")

    print("== chatui: the hero starts a chat ==")
    llm = ScriptedLlm([
        {"tools": [{"id": "ui-1", "name": "list_projects",
                    "arguments": "{}"}]},
        {"tools": [{"id": "ui-2", "name": "ask_question",
                    "arguments": json.dumps({
                        "question": "Which project is the priority?",
                        "options": ["demo", "later"]})}]},
        {"content": "Targeting demo, then."},  # after the answer
    ])
    llm.serve()
    auth = restart_with_llm(ctx, llm)

    # A first line well past the clamp: the title is its head, on a
    # word boundary, with the ellipsis.
    long_opening = ("Please review the whole fleet and tell me what to "
                    "work on next, with reasoning and references")
    home = slot_public(auth, "/", "dashboard")
    result = act(auth, home["start_advisor"], {"message": long_opening})
    navigations = [d for d in result if d.get("type") == "navigate"]
    check(len(navigations) == 1
          and navigations[0].get("uri", "").startswith("/chats/"),
          "the hero action navigates to the new chat (%s)" % navigations)
    chat_id = int(navigations[0]["uri"].rsplit("/", 1)[1])
    title = db_rows("select title from advisor_chats where id = ?",
                    (chat_id,))[0][0]
    check(len(title) <= 81 and title.endswith("…"),
          "the title is clamped from the first line (%d chars)" % len(title))
    check(title.startswith("Please review"),
          "the title leads with the message's first words (%s)" % title)
    kinds = [r[0] for r in db_rows(
        "select kind from advisor_messages where chat_id = ? order by id",
        (chat_id,))]
    check(kinds == ["user"], "the opening message is recorded as a user row")

    print("== chatui: the chat page state (tool round + question) ==")
    check(wait_status(chat_id, "awaiting_answer"),
          "the scripted turn parks on ask_question")
    page = slot_public(auth, "/chats/%d" % chat_id, "chat")
    check(page["id"] == chat_id and page["status"] == "awaiting_answer"
          and page["status_display"] == "awaiting answer",
          "the chat slot carries the chat row and its status label")
    check("send" in page and "delete" in page and "refresh" in page,
          "the chat slot carries the send/delete/refresh actions")
    messages = page["messages"]
    check(any(m["kind"] == "user" and long_opening in m["content"]
              for m in messages), "the user message rides the slot")
    check(any(m["kind"] == "tool" and m.get("tool_name") == "list_projects"
              and "projects" in (m.get("result") or "")
              for m in messages),
          "the tool call renders with name, arguments and result")
    questions = [m for m in messages if m.get("is_question")]
    check(len(questions) == 1
          and questions[0]["content"] == "Which project is the priority?"
          and questions[0]["options"] == ["demo", "later"]
          and not questions[0].get("answered"),
          "the parked question renders with its options, unanswered")
    detail_html = http_get_page(auth, "/chats/%d" % chat_id)
    check("chat-thread" in detail_html and "chat-composer" in detail_html
          and "chat-options" in detail_html,
          "the chat page renders the thread, composer and options")

    print("== chatui: answering the question via the action ==")
    # The question card's option button rides the send action's
    # answer path (the same statum.act the page script issues).
    result = act(auth, page["send"], {"answer": "demo"})
    sets = [d for d in result if d.get("type") == "set"]
    check(len(sets) == 1, "the answer action answers with the rebuilt slot")
    check(wait_status(chat_id, "idle"), "the answered turn ends idle")
    check(any(m["kind"] == "user" and m["content"] == "Human selected: demo"
              for m in slot_public(auth, "/chats/%d" % chat_id,
                                   "chat")["messages"]),
          'the option lands as "Human selected: demo" in the slot')
    check(wait_for(lambda: [m for m in slot_public(
        auth, "/chats/%d" % chat_id, "chat")["messages"]
        if m["kind"] == "assistant"
        and m["content"] == "Targeting demo, then."] or False,
        timeout=15, what="final reply in slot") is not None,
        "the resumed turn's reply reaches the slot")

    print("== chatui: the composer sends a follow-up ==")
    result = act(auth, page["send"], {"message": "thanks, that is all"})
    check(wait_status(chat_id, "idle"), "the follow-up turn ends idle")
    check(any(m["kind"] == "user" and m["content"] == "thanks, that is all"
              for m in slot_public(auth, "/chats/%d" % chat_id,
                                   "chat")["messages"]),
          "the composer message lands in the transcript")
    check(len(llm.requests) >= 3, "the follow-up reached the LLM")

    print("== chatui: the list page and the refresh action ==")
    chats = slot_public(auth, "/chats", "chats")
    row = [c for c in chats["chats"] if str(c["id"]) == str(chat_id)]
    check(len(row) == 1 and row[0]["status"] == "idle"
          and "delete" in row[0],
          "the list slot carries the chat with its status and delete")
    result = act(auth, page["refresh"], {"chat_id": chat_id})
    sets = [d for d in result if d.get("type") == "set"]
    check(len(sets) == 1, "the refresh action rebuilds the slot")

    print("== chatui: a suggestion card renders ==")
    llm2 = ScriptedLlm([
        {"tools": [{"id": "sug-1", "name": "suggest_task",
                    "arguments": json.dumps({
                        "title": "UI-driven task",
                        "mandate": "Verify the chat UI suggestion card.",
                        "mode": "test", "project_id": 1, "priority": 3})}]},
        {"content": "Suggested."},
    ])
    llm2.serve()
    auth = restart_with_llm(ctx, llm2)
    page = slot_public(auth, "/chats/%d" % chat_id, "chat")
    act(auth, page["send"], {"message": "suggest one task"})
    check(wait_status(chat_id, "idle"), "the suggestion turn ends idle")
    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat_id,
                                       "chat")["suggestions"] or False,
                   timeout=15, what="suggestion in slot") is not None,
          "the suggestion reaches the slot")
    card = slot_public(auth, "/chats/%d" % chat_id, "chat")["suggestions"][0]
    check(card["kind"] == "task" and card["status"] == "pending"
          and card["title"] == "UI-driven task"
          and card["body"] == "Verify the chat UI suggestion card."
          and card["mode"] == "test" and str(card["priority"]) == "3",
          "the card carries the payload fields")
    check("decide" in card,
          "the pending card carries the sealed decide action")

    print("== chatui: per-user privacy ==")
    second = "chatui-" + str(int(time.time()))
    users_state = slot_public(auth, "/users", "users")
    act(auth, users_state["create"], {
        "username": second, "email": second + "@example.com",
        "forename": "Chat", "surname": "Ui", "date_of_birth": "1990-01-01",
        "password": "chatuipass1", "confirm_password": "chatuipass1"})
    other = login(second, "chatuipass1")
    other_chats = slot_public(other, "/chats", "chats")
    check(other_chats["chats"] == [],
          "the second user's list is empty (chats are private)")
    other_page = slot_public(other, "/chats/%d" % chat_id, "chat")
    check(other_page["id"] == 0,
          "the second user's chat slot reads not-found (id 0)")
    # their delete of someone else's chat: the slot carries no delete
    # action (id 0 not-found state), so there is nothing to invoke —
    # the row must survive untouched.
    check("delete" not in other_page,
          "the second user's chat slot carries no delete action")
    check(db_rows("select count(*) from advisor_chats where id = ?",
                  (chat_id,))[0][0] == 1,
          "the second user cannot delete the first user's chat")

    print("== chatui: delete the chat ==")
    own_page = slot_public(auth, "/chats/%d" % chat_id, "chat")
    result = act(auth, own_page["delete"], {})
    check(any(d.get("type") == "navigate" and d.get("uri") == "/chats"
              for d in result), "delete navigates to /chats")
    check(db_rows("select count(*) from advisor_chats where id = ?",
                  (chat_id,))[0][0] == 0, "the chat row is gone")
    check(db_rows("select count(*) from advisor_messages where chat_id = ?",
                  (chat_id,))[0][0] == 0, "the messages cascade")
    check(db_rows("select count(*) from advisor_suggestions"
                  " where chat_id = ?", (chat_id,))[0][0] == 0,
          "the suggestions cascade")
    gone = slot_public(auth, "/chats/%d" % chat_id, "chat")
    check(gone["id"] == 0, "the deleted chat's slot reads not-found")
    chats = slot_public(auth, "/chats", "chats")
    check(all(c["id"] != chat_id for c in chats["chats"]),
          "the deleted chat left the list")

    print("== chatui: the unconfigured hero ==")
    # The configuration lives in the model_config row, so the honest
    # unconfigured scene empties it (both the main and the advisor
    # override) — the hand-spawned app below then reads nothing.
    set_model_config(base_url="", model="", advisor_base_url="",
                     advisor_model="")
    llm_env = {}
    ctx.app.terminate()
    try:
        ctx.app.wait(timeout=10)
    except subprocess.TimeoutExpired:
        ctx.app.kill()

    saved_web, saved_port = harness.WEB, harness.APP_PORT
    harness.APP_PORT = free_port()
    harness.WEB = "http://127.0.0.1:%d" % harness.APP_PORT
    env = dict(os.environ)
    env.update({
        "SPRY_DB_PATH": DB_PATH,
        "SPRY_SEED_ADMIN_USERNAME": "admin",
        "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
        "SADDLE_PROJECTS_ROOT": os.path.join(WORK, "workspace"),
        "SADDLE_LLM_BASE_URL": "",
        "ADVISOR_LLM_BASE_URL": "",
        "LD_LIBRARY_PATH": "",
    })
    with open(os.path.join(WORK, "app.log"), "a") as log:
        ctx.app = subprocess.Popen([APP, str(harness.APP_PORT)], cwd=WORK,
                                   env=env, stdout=log,
                                   stderr=subprocess.STDOUT)
    if wait_for(app_ready, timeout=30, what="unconfigured app ready") is None:
        raise SystemExit("unconfigured app did not come up")
    ctx.auth = login("admin", "adminpass123")
    home = slot_public(ctx.auth, "/", "dashboard")
    check(home.get("advisor_configured") is False,
          "the dashboard reports the advisor unconfigured")
    home_html = http_get_page(ctx.auth, "/")
    check("advisor is not" in home_html.lower()
          or "/models" in home_html,
          "the unconfigured hero renders its notice")
    unconfigured_chats = slot_public(ctx.auth, "/chats", "chats")
    check(unconfigured_chats.get("advisor_configured") is False,
          "the /chats slot reports the advisor unconfigured too")
    ctx.app.terminate()
    try:
        ctx.app.wait(timeout=10)
    except subprocess.TimeoutExpired:
        ctx.app.kill()
    harness.WEB, harness.APP_PORT = saved_web, saved_port
    # Restore whatever the caller had (other bundles share the env).
    for name, value in llm_env.items():
        if value is not None:
            os.environ[name] = value
    llm2.kill()
    llm.kill()