clanker-saddle/test/bundles/canon.py

"""The canon/ethos centralisation bundle (goal 27, review refinement):
the WEB becomes the single owner of the canonical texts and the ethos —
served over the per-saddle-token API the daemon already speaks, carried
inline on every assignment, and surfaced to the advisor through the
same three tools the daemon's ToolLoop exposes. The scenarios:

- GET /api/saddle/canon lists the library (15 texts, headers parsed:
  slug/title/author/source/licence/summary plus words and bytes)
- GET /api/saddle/canon/{slug} returns one full document; an unknown
  slug answers 404
- every canon/ethos route requires the bearer token (401 without)
- GET /api/saddle/ethos serves the one web-side ethos text, which
  carries the BINDING operative clause and the goal-agent
  canon-grounding tier (goal 32): dig past a bare "not prohibited"
  and re-ground on every course shift — named markers on the served
  text, on the assignment payload member, and on the advisor system
  prompt's tail, so a softened edit fails loudly
- a fresh assignment payload (the fake saddle's next-task ask) carries
  the same text as its `ethos` member
- the advisor's tool set declares list/read/query_canonical_text, its
  system prompt carries the ethos section verbatim, a scripted turn
  lists then reads a canon text, and query_canonical_text's side call
  hits the same scripted endpoint
"""

from harness import *  # noqa: F401,F403 — the shared wire helpers
from bundles.advisor import ScriptedLlm  # port-independent helper


def advisor_api(auth, method, path, payload=None):
    """A login-gated advisor API call with the statum slot header."""
    request = urllib.request.Request(
        WEB + path, data=json.dumps(payload or {}).encode(), method=method)
    request.add_header("Content-Type", "application/json")
    request.add_header("X-Statum-Slot",
                       "key=%s; as_at=%s" % (auth["key"], auth["as_at"]))
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, json.loads(response.read().decode() or "{}")
    except urllib.error.HTTPError as error:
        body = error.read().decode()
        try:
            return error.code, json.loads(body)
        except ValueError:
            return error.code, {"raw": body}


def chat_rows(chat_id, what="*"):
    conn = db()
    try:
        return conn.execute(
            "select %s from advisor_messages where chat_id = ? order by id"
            % what, (chat_id,)).fetchall()
    finally:
        conn.close()


def wait_status(chat_id, status, timeout=30):
    def read():
        conn = db()
        try:
            row = conn.execute(
                "select status from advisor_chats where id = ?",
                (chat_id,)).fetchone()
            return row and row[0] == status and status or False
        finally:
            conn.close()
    return wait_for(read, timeout=timeout,
                    what="chat %d -> %s" % (chat_id, status))


def run(ctx):
    auth, fake, fake_token = ctx.auth, ctx.fake, ctx.fake_token

    print("== canon: the listing (bearer-gated) ==")
    status, body = saddle_api("GET", "/api/saddle/canon", fake_token)
    check(status == 200 and body.get("ok") is True,
          "GET /api/saddle/canon answers ok with the saddle token")
    texts = body.get("texts") or []
    check(body.get("count") == 15 and len(texts) == 15,
          "the listing carries all 15 canon texts (%s)" % body.get("count"))
    slugs = {t.get("slug") for t in texts}
    check({"gnu-manifesto", "free-software-definition", "plato-republic-vi",
           "magnifica-humanitas-summary"} <= slugs,
          "the expected slugs are present")
    gnu = [t for t in texts if t.get("slug") == "gnu-manifesto"][0]
    check(gnu.get("title") == "The GNU Manifesto"
          and gnu.get("author") == "Richard Stallman"
          and len(gnu.get("licence") or "") > 5
          and len(gnu.get("source") or "") > 5
          and int(gnu.get("words", 0)) > 100
          and int(gnu.get("bytes", 0)) > 1000
          and len(gnu.get("summary") or "") > 10,
          "the parsed header of gnu-manifesto is complete (%s)"
          % sorted(gnu.keys()))

    print("== canon: one document by slug ==")
    status, doc = saddle_api("GET", "/api/saddle/canon/free-software-definition",
                             fake_token)
    check(status == 200 and doc.get("ok") is True
          and doc.get("slug") == "free-software-definition"
          and "Four Essential Freedoms".lower() in (doc.get("body") or "").lower(),
          "GET /api/saddle/canon/{slug} returns the full document")
    check(doc.get("title") and doc.get("author") and doc.get("summary")
          and int(doc.get("bytes", 0)) > 1000,
          "the document carries the header fields and size")
    status, missing = saddle_api("GET", "/api/saddle/canon/no-such-text",
                                 fake_token)
    check(status == 404 and missing.get("ok") is False,
          "an unknown slug answers 404")

    print("== canon + ethos: the bearer token gates every route ==")
    for path in ("/api/saddle/canon", "/api/saddle/canon/plato-apology",
                 "/api/saddle/ethos"):
        status, _ = saddle_api("GET", path, None)
        check(status == 401, "%s without a token answers 401" % path)
    status, _ = saddle_api("GET", "/api/saddle/canon", "not-a-token")
    check(status == 401, "a bogus token answers 401")

    print("== ethos: the served text and the assignment member ==")
    status, ethos_body = saddle_api("GET", "/api/saddle/ethos", fake_token)
    check(status == 200 and ethos_body.get("ok") is True,
          "GET /api/saddle/ethos answers ok")
    ethos = ethos_body.get("ethos") or ""
    check(ethos.startswith("ETHOS") and "cosmic order" in ethos
          and "order of goods" in ethos and "servant, not a oracle" in ethos,
          "the ethos text carries the daemon's section verbatim")
    # The operative duty (goal 32): the order of goods is BINDING, with a
    # once-and-ask conflict clause and the goal-agent canon-grounding tier
    # — dig past a bare "not prohibited", re-ground on every course shift.
    # Every marker is asserted by name so a softened edit fails loudly.
    for marker in ("BINDING, not advisory",
                   "state the conflict plainly ONCE",
                   "ask the human before proceeding",
                   "ask_question where available",
                   "do not nag",
                   "their judgement is senior",
                   "CANON GROUNDING FOR A GOAL AGENT",
                   "is not clearance",
                   "not convenient",
                   "RE-GROUND ON EVERY COURSE SHIFT",
                   "after each",
                   "covers only what was examined then",
                   "BEFORE any task queues",
                   "Goal-initiated TASKS carry only the binding duty"):
        check(marker in ethos,
              "the ethos carries the operative marker %r" % marker)
    # The assignment the fake saddle receives must carry the same text
    # inline (no second fetch needed): a queued task dispatches to it.
    fake.settle_asks(timeout=10)
    payload_count = len(fake.start_payloads)
    create_queued(auth, {"project_id": "1",
                         "title": "Canon ethos carrier",
                         "priority": "2"})
    check(wait_for(lambda: len(fake.start_payloads) > payload_count or False,
                   timeout=30, what="the canon probe assignment") is not None,
          "the fake saddle adopted a fresh assignment")
    payload = fake.start_payloads[-1]
    check(payload.get("ethos") == ethos,
          "a fresh assignment payload carries the ethos verbatim")
    check("BINDING, not advisory" in (payload.get("ethos") or ""),
          "the assignment payload's ethos member carries the binding clause")

    print("== advisor: the canon tool surface ==")
    llm = ScriptedLlm([
        {"tools": [{"id": "call-c1", "name": "list_canonical_texts",
                    "arguments": "{}"}]},
        {"tools": [{"id": "call-c2", "name": "read_canonical_text",
                    "arguments": json.dumps(
                        {"slug": "gnu-manifesto", "offset": 1, "limit": 20})}]},
        {"tools": [{"id": "call-c3", "name": "query_canonical_text",
                    "arguments": json.dumps(
                        {"slug": "free-software-definition",
                         "question": "What are the four freedoms?"})}]},
        {"content": "The four freedoms are use, study, share and improve."},
    ])
    llm.serve()
    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 = auth = login("admin", "adminpass123")

    status, body = advisor_api(auth, "POST", "/api/advisor/chat",
                               {"title": "Canon grounding"})
    check(status == 200 and body.get("ok"), "the chat is created")
    chat_id = int(body["chat_id"])
    status, _ = advisor_api(auth, "POST", "/api/advisor/turn",
                            {"chat_id": chat_id,
                             "message": "Ground your advice in the canon."})
    check(status == 200, "the turn is accepted")

    check(wait_for(lambda: len(llm.requests) >= 1 or False, timeout=30,
                   what="first advisor LLM call") is not None,
          "the advisor calls the LLM")
    first_request = llm.requests[0]
    tool_names = [t["function"]["name"] for t in first_request.get("tools") or []]
    check("list_canonical_texts" in tool_names
          and "read_canonical_text" in tool_names
          and "query_canonical_text" in tool_names,
          "the advisor declares the three canon tools")
    system = first_request["messages"][0]
    check(system.get("role") == "system"
          and "cosmic order" in system.get("content", "")
          and "order of goods" in system.get("content", ""),
          "the advisor system prompt carries the ethos section")
    # The strengthened ethos rides LAST on the advisor system prompt: the
    # binding clause and the goal-agent canon tier (dig + re-ground) must
    # both be present, and the prompt must END with the served text
    # (content is the prompt itself with the ethos appended, so the whole
    # ethos is the tail — equal to the served text modulo its leading/
    # trailing whitespace).
    system_content = system.get("content", "")
    check("CANON GROUNDING FOR A GOAL AGENT" in system_content
          and "BINDING, not advisory" in system_content,
          "the advisor system prompt carries the strengthened ethos")
    check(system_content.strip().endswith(ethos.strip()),
          "the advisor system prompt ends with the served ethos")

    # The list tool result rides back to the next request.
    check(wait_for(lambda: len(llm.requests) >= 2 or False, timeout=30,
                   what="the list result round") is not None,
          "list_canonical_texts triggers the next round")
    second_request = llm.requests[1]
    tool_rows = [m for m in second_request["messages"] if m.get("role") == "tool"]
    listed = json.loads(tool_rows[0]["content"]) if tool_rows else {}
    check(tool_rows and tool_rows[0].get("tool_call_id") == "call-c1"
          and listed.get("count") == 15
          and any(t.get("slug") == "gnu-manifesto"
                  for t in listed.get("texts") or []),
          "list_canonical_texts returns the 15-text library")

    # The read tool result: a numbered window with the title riding along.
    check(wait_for(lambda: len(llm.requests) >= 3 or False, timeout=30,
                   what="the read result round") is not None,
          "read_canonical_text triggers the next round")
    third_request = llm.requests[2]
    read_rows = [m for m in third_request["messages"] if m.get("role") == "tool"]
    read = None
    for row in read_rows:
        parsed = json.loads(row["content"])
        if parsed.get("slug") == "gnu-manifesto" and "content" in parsed:
            read = parsed
    check(read is not None
          and read.get("title") == "The GNU Manifesto"
          and read.get("total_lines", 0) > 50
          and read.get("lines_read") == 20
          and "GNU" in (read.get("content") or ""),
          "read_canonical_text returns a numbered window with the header")

    # The query tool's side call: a FRESH single-turn request (two
    # messages, no tools array) hitting the same scripted endpoint;
    # only the answer enters the conversation. Tool calls execute in
    # order, so the request sequence is: list round, read round, query
    # round, the SIDE single-turn call, then the final round.
    check(wait_for(lambda: len(llm.requests) >= 5 or False, timeout=30,
                   what="the query side call") is not None,
          "query_canonical_text's side call reaches the endpoint")
    side_call = llm.requests[3]
    side_messages = side_call.get("messages") or []
    check(len(side_messages) == 2
          and side_messages[0].get("role") == "system"
          and "helpful assistant" in side_messages[0].get("content", "")
          and side_messages[1].get("role") == "user"
          and "four freedoms" in side_messages[1].get("content", "")
          and "free software" in side_messages[1].get("content", "").lower(),
          "the side call is single-turn and carries the text, not the chat")
    check(not side_call.get("tools"), "the side call declares no tools")

    check(wait_status(chat_id, "idle", timeout=30) is not None,
          "the turn completes to idle")
    final_rows = chat_rows(chat_id, "kind, content, payload")
    answers = []
    for row in final_rows:
        if row[0] != "tool":
            continue
        result = json.loads(row[2]).get("result") or ""
        try:
            answers.append(json.loads(result).get("answer", ""))
        except ValueError:
            pass
    check(any("four freedoms" in a or "freedom" in a.lower() for a in answers),
          "the query answer is stored as the tool result")

    llm.kill()