clanker-saddle/test/bundles/advisor_fix.py

"""The advisor review-fix bundle (goal 27, rejected-review follow-up):
one regression guard per defect the completion review found, each
driven end-to-end through the app and the scripted LLM endpoint.

- the tool schema (bug 1): the REQUEST's tools array carries no prose
  in a `type` field ("array of string" 400s strict providers),
  ask_question's options property is the items-object array shape,
  suggest_task's mode is an enum, and every schema has its `required`
  array
- the 400 body (bug 2): a provider 400 with a JSON body lands in the
  chat's error row verbatim enough to read the provider's reason
- the redirect loop (bug 3): the chat page's refresh action is SEALED
  (no form/chat_id round-trip) — a bare GET with no parameters answers
  a slot SET for the owner and navigates ONLY for a foreign or deleted
  chat
- the reply variants (bug 4): provider bodies shaped like real ones
  (tool_calls absent, an empty message object, an error object over
  HTTP 200, content null) parse without a json-glib CRITICAL and
  without wedging the turn
- the hero markup (bug 5): the dashboard's composer is its own block
  form (not form.inline/.form-row) with a full-width textarea
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers
import harness  # the module itself (global re-pointing on restarts)
from bundles import advisor as advisor_module
from bundles.advisor import ScriptedLlm, advisor_api, chat_rows


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 (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 schema_types(value, seen):
    """Collects every JSON-Schema `type` leaf in a parsed structure."""
    if isinstance(value, dict):
        for key, item in value.items():
            if key == "type" and isinstance(item, str):
                seen.add(item)
            schema_types(item, seen)
    elif isinstance(value, list):
        for item in value:
            schema_types(item, seen)


def run(ctx):
    auth = ctx.auth
    # The shared helpers imported from bundles.advisor (advisor_api
    # reads WEB) captured THIS module's star-import at load; in the
    # sequential full run those globals are the harness's re-pointed
    # ones while bundles.advisor's own copy stayed at ITS shard's
    # values — re-point them exactly as run_bundle does for us.
    for runtime_name in ("WORK", "DB_PATH", "APP_PORT", "DAEMON_PORT",
                         "WEB"):
        setattr(advisor_module, runtime_name,
                getattr(harness, runtime_name))

    print("== advisor_fix: the tool schema on the wire ==")
    llm = ScriptedLlm([
        {"tools": [{"id": "fx-1", "name": "ask_question",
                    "arguments": json.dumps({
                        "question": "Schema ok?",
                        "options": ["yes", "no"]})}]},
        {"content": "Schema verified."},
    ])
    llm.serve()
    auth = restart_with_llm(ctx, llm)
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "schema"})
    schema_chat = int(body["chat_id"])
    advisor_api(auth, "POST", "/api/advisor/turn",
                {"chat_id": schema_chat, "message": "check the schema"})
    check(wait_for(lambda: llm.requests or False, timeout=15,
                   what="first LLM call") is not None,
          "the turn reached the LLM")

    tools = llm.requests[0].get("tools") or []
    names = {t["function"]["name"]: t["function"] for t in tools}
    check(set(names) >= {"list_projects", "ask_question", "suggest_task",
                         "suggest_goal"},
          "the request declares the advisor tool set (%s)" % sorted(names))

    # Bug 1's regression: NO prose may ride a `type` field anywhere in
    # the tools array — strict providers validate it and 400 the whole
    # request before answering anything.
    types_seen = set()
    schema_types(tools, types_seen)
    check("array of string" not in types_seen
          and all(t in ("string", "integer", "array", "object", "boolean",
                        "number", "function")
                  for t in types_seen),
          "every schema type leaf is a real JSON-Schema type (%s)"
          % sorted(types_seen))

    question_params = names["ask_question"]["parameters"]
    options_property = question_params["properties"]["options"]
    check(options_property.get("type") == "array"
          and options_property.get("items", {}).get("type") == "string",
          "ask_question's options property is the items-object array shape"
          " (%s)" % options_property)
    check(question_params.get("required") == ["question"],
          "ask_question declares its required array (%s)"
          % question_params.get("required"))

    task_params = names["suggest_task"]["parameters"]
    mode_property = task_params["properties"]["mode"]
    check(mode_property.get("enum") == ["implement", "investigate",
                                        "review", "test"],
          "suggest_task's mode is the work-mode enum (%s)"
          % mode_property.get("enum"))
    check(task_params.get("required") == ["title", "mandate"],
          "suggest_task declares title+mandate required (%s)"
          % task_params.get("required"))

    goal_params = names["suggest_goal"]["parameters"]
    check(goal_params.get("required") == ["title", "brief", "project_id"],
          "suggest_goal declares title+brief+project_id required (%s)"
          % goal_params.get("required"))
    check("required" in names["list_projects"]["parameters"],
          "even the parameter-less read tools carry a required array")
    check(names["get_project"]["parameters"]["properties"]["project_id"]
          .get("type") == "integer",
          "integer params keep their integer type")

    # The question parks; answer it so the chat settles.
    check(wait_status(schema_chat, "awaiting_answer"),
          "the scripted question parks the chat")
    advisor_api(auth, "POST", "/api/advisor/answer",
                {"chat_id": schema_chat, "answer": "yes"})
    check(wait_status(schema_chat, "idle"), "the schema chat settles idle")

    print("== advisor_fix: the provider 400 body reaches the error row ==")
    # fail_next answers 500; the 400 scene needs a body the harness can
    # read back, so serve a raw 400 with a JSON error envelope.
    llm400 = ScriptedLlm([{"content": "never"}])
    llm400.serve()

    class Handler400(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, fmt, *args):
            pass

        def do_POST(self):
            length = int(self.headers.get("Content-Length") or 0)
            self.rfile.read(length)
            body = json.dumps({
                "error": {"message": "Invalid schema for tool"
                          " 'ask_question': options.type 'array of string'"
                          " is not a valid type",
                          "type": "invalid_request_error"}}).encode()
            self.send_response(400)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

    server400 = ThreadingHTTPServer(("127.0.0.1", 0), Handler400)
    threading.Thread(target=server400.serve_forever, daemon=True).start()
    auth = restart_with_llm(ctx, llm400)
    # Point the configuration at the raw 400 server AFTER the restart
    # read the (valid) llm400 for the configured check; the restart
    # below binds the app to the 400 endpoint.
    set_model_config(advisor_base_url=
        "http://127.0.0.1:%d" % server400.server_address[1])
    ctx.app.terminate()
    try:
        ctx.app.wait(timeout=10)
    except subprocess.TimeoutExpired:
        ctx.app.kill()
    ctx.app = start_app()
    ctx.auth = auth = login("admin", "adminpass123")

    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "errors"})
    err_chat = int(body["chat_id"])
    advisor_api(auth, "POST", "/api/advisor/turn",
                {"chat_id": err_chat, "message": "provoke the 400"})
    check(wait_status(err_chat, "idle"),
          "the 400 turn returns the chat to idle")
    error_rows = wait_for(lambda: [r for r in chat_rows(err_chat,
                                                        "kind, content")
                                   if r[0] == "error"] or False,
                          timeout=15, what="400 error row")
    check(error_rows is not None and len(error_rows) == 1,
          "the 400 lands as exactly one error row")
    content = error_rows[0][1] if error_rows else ""
    check("HTTP 400" in content,
          "the error row names the status (%s)" % content[:120])
    check("is not a valid type" in content and "Invalid schema" in content,
          "the error row CONTAINS the provider's own message text (%s)"
          % content[:200])
    check(len(content) <= 600,
          "the error row stays clamped (%d chars)" % len(content))
    server400.shutdown()
    server400.server_close()
    llm400.kill()

    print("== advisor_fix: the refresh action is sealed (no redirect) ==")
    # A quiet scripted LLM: the scene only needs a chat that exists.
    llmr = ScriptedLlm([{"content": "quiet"}])
    llmr.serve()
    auth = restart_with_llm(ctx, llmr)
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "refresh"})
    refresh_chat = int(body["chat_id"])
    advisor_api(auth, "POST", "/api/advisor/turn",
                {"chat_id": refresh_chat, "message": "say something"})
    check(wait_status(refresh_chat, "idle"), "the refresh chat settles")

    page = slot_public(auth, "/chats/%d" % refresh_chat, "chat")
    check("refresh" in page, "the chat slot carries the refresh action")
    refresh = page["refresh"]
    check(refresh.get("method") == "GET" and refresh.get("private"),
          "the refresh action is GET and SEALED (private blob present)")
    # The pulse listener calls statum.act(chat.refresh, {}) — a bare
    # GET with NO parameters. Before the fix, Forms.integer read 0 and
    # every bump navigated to /chats (the redirect loop).
    result = act(auth, refresh, {})
    sets = [d for d in result if d.get("type") == "set"]
    navigations = [d for d in result if d.get("type") == "navigate"]
    check(len(sets) == 1 and not navigations,
          "the bare GET refresh answers a slot SET, no navigate (%s)"
          % [d.get("type") for d in result])
    set_chat = json.loads(sets[0]["snapshot_frame"]["content"])
    check(set_chat["slot"]["type"] == "chat"
          and set_chat["public"]["id"] == refresh_chat,
          "the refreshed slot is the owner's own chat")

    # The chat page's script must no longer pass a chat_id (the param
    # that was silently lost in the query string).
    detail_html = http_get_page(auth, "/chats/%d" % refresh_chat)
    check("statum.act(chat.refresh, {})" in detail_html,
          "the pulse listener invokes the refresh with no parameters")

    # A foreign viewer's sealed refresh: they cannot author one (the
    # not-found slot carries no actions), and a deleted chat navigates.
    second = "fix-" + str(int(time.time()))
    users_state = slot_public(auth, "/users", "users")
    act(auth, users_state["create"], {
        "username": second, "email": second + "@example.com",
        "forename": "Fix", "surname": "Bundle",
        "date_of_birth": "1990-01-01",
        "password": "fixpass123", "confirm_password": "fixpass123"})
    other = login(second, "fixpass123")
    other_page = slot_public(other, "/chats/%d" % refresh_chat, "chat")
    check(other_page["id"] == 0 and "refresh" not in other_page,
          "a foreign viewer gets the not-found slot with no refresh")
    status, _ = advisor_api(auth, "POST", "/api/advisor/chat/delete",
                            {"chat_id": refresh_chat})
    doomed = slot_public(auth, "/chats", "chats")
    # Author a refresh from a LIVE chat, delete the chat, then fire the
    # action: only the genuine not-found path may navigate.
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "doomed"})
    doomed_chat = int(body["chat_id"])
    advisor_api(auth, "POST", "/api/advisor/turn",
                {"chat_id": doomed_chat, "message": "populate"})
    check(wait_status(doomed_chat, "idle"), "the doomed chat settles")
    doomed_page = slot_public(auth, "/chats/%d" % doomed_chat, "chat")
    advisor_api(auth, "POST", "/api/advisor/chat/delete",
                {"chat_id": doomed_chat})
    result = act(auth, doomed_page["refresh"], {})
    check(any(d.get("type") == "navigate" and d.get("uri") == "/chats"
              for d in result),
          "a refresh for a genuinely deleted chat navigates to /chats")

    print("== advisor_fix: reply variants parse without a CRITICAL ==")
    # Variants real providers send: no tool_calls member at all, an
    # empty message object, an error object over HTTP 200, and content
    # null. Each must parse (or throw a catchable Error) — never a
    # json-glib CRITICAL and never a wedged turn.
    variants = [
        ("no tool_calls member",
         {"choices": [{"index": 0, "finish_reason": "stop",
                       "message": {"role": "assistant",
                                   "content": "plain reply"}}],
          "model": "scripted"}),
        ("empty message object",
         {"choices": [{"index": 0, "finish_reason": "stop",
                       "message": {}}], "model": "scripted"}),
        ("null content",
         {"choices": [{"index": 0, "finish_reason": "stop",
                       "message": {"role": "assistant", "content": None}}],
          "model": "scripted"}),
        ("null message object",
         {"choices": [{"index": 0, "finish_reason": "stop",
                       "message": None}], "model": "scripted"}),
        ("first choice not an object",
         {"choices": [None], "model": "scripted"}),
        ("error object over HTTP 200",
         {"error": {"message": "quota exceeded", "type": "insufficient_quota"},
          "model": "scripted"}),
    ]
    variant_logs = []
    for label, envelope in variants:
        llmv = ScriptedLlm([{"content": "unused"}])
        llmv.serve()
        # Swap the served envelope by pointing the script at a raw
        # handler: simplest is a dedicated one-shot server.
        class HandlerVariant(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def log_message(self, fmt, *args):
                pass

            def do_POST(self):
                length = int(self.headers.get("Content-Length") or 0)
                self.rfile.read(length)
                served = json.dumps(envelope).encode()
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(served)))
                self.end_headers()
                self.wfile.write(served)

        serverv = ThreadingHTTPServer(("127.0.0.1", 0), HandlerVariant)
        threading.Thread(target=serverv.serve_forever, daemon=True).start()
        set_model_config(advisor_base_url=
            "http://127.0.0.1:%d" % serverv.server_address[1])
        ctx.app.terminate()
        try:
            ctx.app.wait(timeout=10)
        except subprocess.TimeoutExpired:
            ctx.app.kill()
        ctx.app = start_app()
        ctx.auth = auth = login("admin", "adminpass123")
        status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                                   {"title": label})
        variant_chat = int(body["chat_id"])
        advisor_api(auth, "POST", "/api/advisor/turn",
                    {"chat_id": variant_chat, "message": "parse this"})
        # The turn's TERMINAL ROW is the deterministic signal (a fresh
        # chat already reads idle) — an assistant reply, an error row,
        # or the no-choices throw landing as one.
        terminal = wait_for(lambda: [r for r in chat_rows(variant_chat,
                                                          "kind")
                                     if r[0] in ("assistant", "error")]
                            or False, timeout=20,
                            what="%s terminal row" % label)
        check(terminal is not None,
              "the %s variant lands a terminal row (%s)"
              % (label, [r[0] for r in chat_rows(variant_chat, "kind")]))
        check(wait_status(variant_chat, "idle", timeout=20),
              "the %s variant leaves the chat idle (no wedge)" % label)
        try:
            with open(os.path.join(WORK, "app.log"), "r",
                      errors="replace") as handle:
                variant_logs.append(handle.read())
        except OSError:
            pass
        serverv.shutdown()
        serverv.server_close()
        llmv.kill()

    # A CRITICAL in ANY variant's app log would name the json-glib
    # accessor (json_object_get_array_member, json_object_has_member,
    # …) — none may appear for any of the driven shapes.
    log_text = "\n".join(variant_logs)
    critical_lines = [line for line in log_text.splitlines()
                      if "Json-CRITICAL" in line]
    check(not critical_lines,
          "no json-glib CRITICAL lines across the variants (%d seen)%s"
          % (len(critical_lines),
             " e.g. %s" % critical_lines[0][:120] if critical_lines else ""))

    print("== advisor_fix: the hero composer markup ==")
    home_html = http_get_page(auth, "/")
    check('class="advisor-hero card"' in home_html,
          "the hero keeps its card")
    check('class="advisor-hero-form"' in home_html,
          "the hero form carries its own layout class")
    check('class="inline advisor-hero-form"' not in home_html,
          "the hero form is NOT form.inline")
    # The FORM tag onward (the explanatory comment above it names the
    # old classes in prose — it is not markup).
    hero_block = '<form class="advisor-hero-form"' + home_html.split(
        '<form class="advisor-hero-form"')[1].split("</form>")[0]
    check("form-row" not in hero_block,
          "the hero form no longer uses the 34rem .form-row column")
    check('id="advisor-composer"' in hero_block
          and "<textarea" in hero_block,
          "the hero textarea rides the form")
    css = ""
    for candidate in (os.path.join(os.path.dirname(os.path.abspath(
            harness.__file__)), "..", "src", "pages", "app.css"),):
        try:
            with open(candidate, "r", errors="replace") as handle:
                css += handle.read()
        except OSError:
            pass
    check(".advisor-hero-form textarea" in css
          and "min-height: 4.5rem" in css and "width: 100%" in css,
          "the CSS gives the hero textarea its full-width roomy layout")
    check(".advisor-hero .form-row" not in css,
          "the old squeezed .form-row hero rule is gone")

    llm.kill()
    llmr.kill()

    # Restore the daemon-facing model configuration the other checks
    # may rely on (the advisor override is test-local).
    set_model_config(advisor_base_url="", advisor_model="")