clanker-saddle/test/bundles/advisor_utf8.py

"""The advisor UTF-8 wire bundle (invalid-UTF-8 payload fix): a
regression guard per entry point the defect travelled, each driven
end-to-end through the app and the scripted LLM endpoint.

The defect: the advisor's request body (and the stored advisor_messages
payload, re-read on the next turn) carried structurally invalid UTF-8
whenever a multi-byte character straddled a byte-count clamp — every
clamp in AdvisorService was a raw byte cut:

- get_task/get_goal/get_task_log clamp long text to 2000 BYTES; a
  completion report whose 2000th byte is mid-character (an em dash —
  routine in these reports) emitted an orphaned lead byte. json-glib
  embeds it verbatim, the provider 400s the body, and our own parser
  rejects the stored row on read-back ("JSON data must be UTF-8
  encoded"), silently dropping the turn's content.
- query_canonical_text's 90000-byte head/tail cut had the same shape.
- a provider error body (latin-1 HTML from proxies) rode the chat's
  error row unsanitised, with a raw 500-byte cut.

The fix mirrors the daemon's Wire/Canon idiom: cuts land on UTF-8
character boundaries, and every string that reaches a serialisation
boundary is sanitised (invalid bytes -> U+FFFD).

- the completion-report cut: a goal whose completion_report places a
  3-byte em dash at bytes 1999..2001 reaches the LLM as a VALID UTF-8
  JSON string, ends on a whole character, and the stored advisor
  message payload re-parses
- the task summary cut: the same straddle through get_task
- the canon query middle cut: a canon text over the 90000-byte budget
  cut on boundaries on BOTH sides (checked through a fresh canon dir)
- legacy rescue: a stored advisor payload that already holds invalid
  bytes (written before the fix) is recovered, not silently dropped
"""
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
from bundles.advisor_fix import restart_with_llm, wait_status


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


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))

    # The clamp constant under test (TEXT_CLAMP_CHARS in
    # AdvisorService): the straddle is arranged so the em dash's lead
    # byte sits exactly AT byte 1999, i.e. byte 2000 lands mid-sequence.
    CLAMP = 2000
    EM = "\u2014"          # — (3 bytes: e2 80 94)
    DOT = "\u2026"         # … (3 bytes: e2 80 a6)

    def straddling(pad_to, tail=" and the closing thought of the report"):
        """Text whose multi-byte character straddles byte `pad_to`."""
        return "a" * (pad_to - 1) + EM + tail

    report = straddling(CLAMP)                     # em dash at 1999..2001
    summary = straddling(CLAMP - 400)              # well inside the clamp
    long_summary = straddling(CLAMP)               # the cut case
    print("== advisor_utf8: the completion-report cut ==")
    conn = db()
    conn.execute(
        "insert into goals (project_id, title, brief, branch_name, status,"
        " plan, completion_report, completed_at, created_at)"
        " values (1, 'UTF8 goal', 'b', 'goal/utf8', 'complete',"
        " 'plan text', ?, strftime('%s','now'), strftime('%s','now'))",
        (report,))
    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', 'UTF8 task', 'done', 5, 'implement',"
        " 'human', ?, strftime('%s','now'), strftime('%s','now'))",
        (goal_id, long_summary))
    task_id = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute(
        "insert into tasks (project_id, kind, title, status, priority,"
        " mode, creator, summary, created_at)"
        " values (1, 'task', 'Short task', 'done', 5, 'implement',"
        " 'human', ?, strftime('%s','now'))",
        (summary,))
    short_id = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.commit()
    conn.close()

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

    llm = ScriptedLlm([
        {"tools": [{"id": "u1", "name": "get_goal",
                    "arguments": json.dumps({"goal_id": goal_id})}]},
        {"tools": [{"id": "u2", "name": "get_task",
                    "arguments": json.dumps({"task_id": task_id})}]},
        {"tools": [{"id": "u3", "name": "get_task",
                    "arguments": json.dumps({"task_id": short_id})}]},
        {"content": "Read them all."},
    ])
    llm.serve()
    auth = restart_with_llm(ctx, llm)
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "utf8"})
    chat_id = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": chat_id,
                             "message": "Pull the completion story."})
    check(status == 200, "the report-reading turn is accepted")
    check(wait_for(lambda: len(llm.requests) >= 4 or False, timeout=25,
                   what="the report-reading rounds") is not None,
          "the report-reading turn completes")

    # EVERY request body the LLM received must be valid UTF-8 JSON. The
    # ScriptedLlm already json.loads each body (a raise would 500 the
    # turn before any of the checks below), so this inspects the wire
    # strings themselves.
    def wire_ok():
        for index, request in enumerate(llm.requests):
            for message in request.get("messages") or []:
                content = message.get("content")
                if not isinstance(content, str):
                    continue
                try:
                    content.encode("utf-8", "strict")
                except UnicodeEncodeError:
                    return index, message.get("role")
        return None

    check(wire_ok() is None,
          "every message content on the wire is encodable UTF-8")

    # The completion report: cut on a character boundary, never halved.
    goal_view = tool_result("u1", llm.requests)
    check(goal_view is not None, "get_goal answered")
    parsed = json.loads(goal_view) if goal_view else {}
    seen = parsed.get("completion_report", "")
    check(seen.endswith(DOT) and not seen.endswith(EM[0]),
          "the cut completion report ends on the ellipsis marker")
    check(EM not in seen,
          "the halved em dash is gone (the straddling char was moved out)")
    check(len(seen.encode("utf-8")) <= CLAMP + len(DOT.encode()),
          "the clamped report respects the byte budget (got %d bytes)"
          % len(seen.encode("utf-8")))

    # The task summary: same shape through get_task.
    task_view = tool_result("u2", llm.requests)
    parsed = json.loads(task_view) if task_view else {}
    seen = parsed.get("summary", "")
    check(seen.endswith(DOT) and EM not in seen,
          "the cut task summary ends on the ellipsis marker, char whole")

    # A short summary passes through untouched (boundary cut is a no-op).
    short_view = tool_result("u3", llm.requests)
    parsed = json.loads(short_view) if short_view else {}
    check(parsed.get("summary") == summary,
          "an under-budget summary rides verbatim")

    # The stored advisor_messages payload must re-parse — the row the
    # next turn's wire rebuild reads (the pre-fix bug poisoned it).
    tool_payloads = [r[0] for r in chat_rows(chat_id, "payload")
                     if r[0]]
    reparsed = 0
    for payload in tool_payloads:
        doc = json.loads(payload)          # raises on invalid UTF-8 JSON
        reparsed += 1
        inner = doc.get("result")
        if isinstance(inner, str):
            json.loads(inner)              # the double-encoded result
    check(reparsed >= 3,
          "every stored tool payload re-parses cleanly (%d rows)" % reparsed)

    print("== advisor_utf8: the canon query middle cut ==")
    # A canon dir with one text over the 90000-byte budget, its cut
    # point arranged mid-character, pointed at the app through
    # CLANKER_CANON_DIR... which the app reads at startup: restart.
    canon_dir = os.path.join(WORK, "canon-utf8")
    os.makedirs(canon_dir, exist_ok=True)
    BUDGET = 90000
    body_text = ("word — dash … ellipsis ✓ check " * 3000)[:BUDGET + 700]
    # place a 4-byte emoji so the head cut at BUDGET/2 lands mid-char
    body_bytes = body_text.encode("utf-8")
    half = BUDGET // 2
    body_bytes = (body_bytes[:half] + "\U0001F600".encode("utf-8")
                  + body_bytes[half:])
    with open(os.path.join(canon_dir, "utf8probe.md"), "wb") as handle:
        handle.write(
            b"---\nslug: utf8probe\ntitle: UTF8 Probe\nauthor: anon\n"
            b"source: test\nlicence: pd\nsummary: probe\n---\n")
        handle.write(body_bytes)

    llm2 = ScriptedLlm([
        {"tools": [{"id": "c1", "name": "query_canonical_text",
                    "arguments": json.dumps({"slug": "utf8probe",
                                             "question": "anything"})}]},
        {"content": "Queried."},
    ])
    llm2.serve()
    set_model_config(advisor_base_url="http://127.0.0.1:%d" % llm2.port,
                     advisor_model="scripted")
    ctx.app.terminate()
    try:
        ctx.app.wait(timeout=10)
    except subprocess.TimeoutExpired:
        ctx.app.kill()
    ctx.app = start_app(extra_env={"CLANKER_CANON_DIR": canon_dir})
    ctx.auth = auth = login("admin", "adminpass123")

    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "canon utf8"})
    canon_chat = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": canon_chat,
                             "message": "query the canon"})
    check(status == 200, "the canon-query turn is accepted")
    # TWO requests: the main loop's tool round, then the side call the
    # tool itself makes (the text never enters the conversation).
    check(wait_for(lambda: len(llm2.requests) >= 2 or False, timeout=25,
                   what="the canon query side call") is not None,
          "the canon query reaches the LLM")
    # The side call's user message carries the clamped body: it must be
    # encodable UTF-8 (the last ScriptedLlm json.loads proves decodable;
    # this pins the encoding side of the contract).
    side_bodies = [m.get("content", "") for r in llm2.requests
                   for m in (r.get("messages") or [])
                   if m.get("role") == "user" and "Text:" in
                   (m.get("content") or "")]
    check(len(side_bodies) >= 1, "the canon side call carried the text")
    for text in side_bodies:
        try:
            text.encode("utf-8", "strict")
            ok = True
        except UnicodeEncodeError:
            ok = False
        check(ok, "the canon side-call text is encodable UTF-8")
        check("[output truncated:" in text,
              "the over-budget canon body states its truncation")

    print("== advisor_utf8: legacy stored payload rescue ==")
    # A stored advisor_messages row holding invalid UTF-8 (written
    # before the fix): the next turn must recover its members through
    # the sanitising loader instead of silently dropping them.
    legacy_chat_title = "legacy"
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": legacy_chat_title})
    legacy_chat = int(body["chat_id"])
    conn = db()
    dirty_payload = ('{"call_id": "legacy-1", "name": "get_task",'
                     ' "result": "{\\"note\\": \\"broken \xff bytes\\"}"}')
    dirty = dirty_payload.encode("utf-8", "surrogateescape")
    # CAST keeps the row TEXT-typed (the shape the pre-fix Vala writer
    # stored: sqlite3_bind_text keeps raw bytes); a bare bytes param
    # would type it BLOB and the ORM's string mapping would reject the
    # ROW — a different failure this bundle does not claim to guard.
    conn.execute(
        "insert into advisor_messages (chat_id, kind, content, payload,"
        " created_at) values (?, 'tool', '@tool result', CAST(? AS TEXT),"
        " strftime('%s','now'))", (legacy_chat, dirty))
    conn.commit()
    conn.close()

    llm3 = ScriptedLlm([{"content": "Recovered."}])
    llm3.serve()
    auth = restart_with_llm(ctx, llm3)
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": legacy_chat,
                             "message": "carry on"})
    check(status == 200, "the legacy-row turn is accepted")
    check(wait_for(lambda: len(llm3.requests) >= 1 or False, timeout=25,
                   what="the legacy-row LLM call") is not None,
          "the legacy-row turn reaches the LLM")
    legacy_tool = [m for r in llm3.requests for m in (r.get("messages") or [])
                   if m.get("role") == "tool"
                   and m.get("tool_call_id") == "legacy-1"]
    check(len(legacy_tool) == 1
          and "broken" in legacy_tool[0].get("content", ""),
          "the legacy dirty row's result survives to the wire (was silently"
          " dropped before the sanitising loader)")

    print("== advisor_utf8: a provider 400 body with non-UTF-8 bytes ==")
    # The provider's error body rides the chat's error row: raw bytes
    # from a latin-1 proxy page must not poison the row or the next
    # turn's wire rebuild.
    class DirtyErrorLlm(ScriptedLlm):
        def serve(self):
            class Handler(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)
                    # cp1252: the single-byte codepage a real proxy
                    # error page carries (latin-1 proper cannot hold
                    # the em dash — cp1252's 0x97 is exactly the
                    # byte soup this scenario wants).
                    body = ("<html>Erreur — café naïve</html>"
                            ).encode("cp1252")
                    self.send_response(400)
                    self.send_header("Content-Type", "text/html")
                    self.send_header("Content-Length", str(len(body)))
                    self.end_headers()
                    self.wfile.write(body)

            self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
            self.port = self.server.server_address[1]
            threading.Thread(target=self.server.serve_forever,
                             daemon=True).start()
            return self.server

    llm4 = DirtyErrorLlm([])
    llm4.serve()
    auth = restart_with_llm(ctx, llm4)
    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "dirty 400"})
    dirty_chat = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": dirty_chat, "message": "go"})
    check(status == 200, "the dirty-400 turn is accepted")
    # A fresh chat is already 'idle' — wait on the error row itself.
    def error_row():
        rows = [r for r in chat_rows(dirty_chat, "kind, content")
                if r[0] == "error"]
        return rows or False
    rows = wait_for(error_row, timeout=25, what="the 400 error row")
    check(rows is not None and "HTTP 400" in rows[0][1],
          "the provider 400 lands in the chat's error row")
    try:
        rows[0][1].encode("utf-8", "strict")
        ok = True
    except UnicodeEncodeError:
        ok = False
    check(ok, "the error row content is encodable UTF-8")

    # And the poisoned-error-row chat must still hold a next turn: its
    # wire rebuild sanitises the notice instead of failing the turn.
    llm5 = ScriptedLlm([{"content": "Still alive."}])
    llm5.serve()
    auth = restart_with_llm(ctx, llm5)
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": dirty_chat,
                             "message": "continue after the error"})
    check(status == 200, "the post-error turn is accepted")
    check(wait_for(lambda: len(llm5.requests) >= 1 or False, timeout=25,
                   what="the post-error LLM call") is not None,
          "the post-error turn reaches the LLM")
    notices = [m.get("content", "") for r in llm5.requests
               for m in (r.get("messages") or [])
               if m.get("role") == "user" and "notice]" in
               (m.get("content") or "")]
    check(len(notices) >= 1, "the error notice rides the rebuilt wire")
    for notice in notices:
        try:
            notice.encode("utf-8", "strict")
            ok = True
        except UnicodeEncodeError:
            ok = False
        check(ok, "the rebuilt error notice is encodable UTF-8")