ratatoskr/docs/diagnostics/temporal_order_gate.py

"""Temporal-order gate — measures Worldtree #397 order_by="chapter" end-to-end.

The gap #397 closes: narrative/temporal questions ("your first encounter", "what came
after X", "earliest to latest") need CHRONOLOGICAL order, but reference_knowledge sorts by
RELEVANCE by default. The fix is an `order_by="chapter"` tool flag (enum ["chapter"], taught
in both tool schemas; the result packet carries `ordered_by="chapter"` and hits are reordered
by source chapter, earliest first). Confirmed live on personal b184.

This measures the flag END-TO-END (before/after per the #393 discipline) on three axes:
  - ADOPTION: for a temporal query, does the agent actually invoke order_by="chapter"? (the
    schema teaches it, but usage varies turn-to-turn — the #397 analog of query-formulation
    variance).
  - MECHANISM (flag applied): are the served hits' provenance.chapter monotonically
    non-decreasing (earliest first)? Should be ~100% when the flag fires.
  - VALUE (flag not applied): the relevance baseline — chapters are NOT chapter-sorted, which
    is exactly the gap the flag closes. The applied-vs-not monotonicity gap IS the before/after.

Self-contained (httpx only). Config from env (source env.sh first).
  uv run --with httpx python docs/diagnostics/temporal_order_gate.py
  uv run --with httpx python docs/diagnostics/temporal_order_gate.py --runs 4
"""
from __future__ import annotations

import argparse
import json
import os

import httpx

AGENT = "ratatoskr:donut"

# Narrative/temporal user messages — the class the order_by="chapter" flag targets.
TEMPORAL_MSGS = [
    "What was your very first encounter in the dungeon?",
    "What happened when you first entered the dungeon, earliest to latest?",
    "In order from the start, how did things unfold between you and Carl?",
    "Walk me through your earliest days in the dungeon, oldest first.",
    "After your first fight, what came next?",
]


def _cfg() -> tuple[str, dict, str]:
    base = os.environ.get("WORLDTREE_API_URL", "http://10.250.50.152:8081")
    key = os.environ.get("WORLDTREE_API_KEY")
    if not key:
        raise SystemExit("WORLDTREE_API_KEY unset — source env.sh first.")
    return base, {"Authorization": f"Bearer {key}"}, os.environ.get("RATATOSKR_END_USER_ID", "ratatoskr-tui")


def _session(base: str, headers: dict, end_user: str) -> str:
    r = httpx.post(f"{base}/sessions", json={"agent_id": AGENT, "end_user_id": end_user},
                   headers=headers, timeout=30)
    r.raise_for_status()
    return r.json()["session_id"]


def _drive(base: str, headers: dict, sid: str, content: str) -> tuple[dict, dict]:
    """POST a turn; return (tool_start arguments, tool_result dict)."""
    args, result = {}, {}
    with httpx.stream("POST", f"{base}/sessions/{sid}/messages", json={"content": content},
                      headers=headers, timeout=180) as r:
        for line in r.iter_lines():
            if not line.startswith("data: "):
                continue
            ev = json.loads(line[6:])
            t = ev.get("type")
            if t == "tool_start" and not args:
                args = ev.get("arguments") or {}
            elif t == "tool_result" and not result:
                result = ev.get("result") if isinstance(ev.get("result"), dict) else {}
            elif t == "done":
                break
    return args, result


def _chap_num(c: object) -> int:
    """Chapter as an orderable int; non-numeric (e.g. 'EPILOGUE') sorts last."""
    try:
        return int(str(c))
    except (TypeError, ValueError):
        return 10**9


def _is_monotone(chapters: list) -> bool:
    nums = [_chap_num(c) for c in chapters if c is not None]
    return all(a <= b for a, b in zip(nums, nums[1:])) if len(nums) >= 2 else True


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--runs", type=int, default=2, help="repeats per message (samples adoption variance)")
    args = ap.parse_args()
    base, headers, end_user = _cfg()

    applied = {"trials": 0, "monotone": 0}   # flag fired
    unapplied = {"trials": 0, "monotone": 0}  # flag did NOT fire (relevance baseline)
    adopted_n = 0
    total = 0

    for msg in TEMPORAL_MSGS:
        print(f"\nMSG: {msg!r}")
        for _ in range(args.runs):
            sid = _session(base, headers, end_user)
            targs, res = _drive(base, headers, sid, msg)
            hits = res.get("hits", res.get("results", [])) or []
            chapters = [(h.get("provenance") or {}).get("chapter") for h in hits if isinstance(h, dict)]
            adopted = targs.get("order_by") == "chapter"
            applied_flag = res.get("ordered_by") == "chapter"
            mono = _is_monotone(chapters)
            total += 1
            adopted_n += int(adopted)
            bucket = applied if applied_flag else unapplied
            bucket["trials"] += 1
            bucket["monotone"] += int(mono)
            tag = "FLAG" if applied_flag else "----"
            print(f"    [{tag}] adopted={adopted!s:5} monotone={mono!s:5} chapters={chapters}")

    def pct(n: int, d: int) -> str:
        return f"{(100*n/d):.0f}%" if d else "n/a"

    print("\n=== SUMMARY ===")
    print(f"  adoption (agent invoked order_by=chapter): {pct(adopted_n, total)}  ({adopted_n}/{total})")
    print(f"  flag APPLIED   -> chapter-monotone: {pct(applied['monotone'], applied['trials'])}  "
          f"(n={applied['trials']})   [mechanism — should be ~100%]")
    print(f"  flag NOT applied -> chapter-monotone: {pct(unapplied['monotone'], unapplied['trials'])}  "
          f"(n={unapplied['trials']})   [relevance baseline — the gap the flag closes]")
    print("\nGate: the applied-vs-not monotonicity gap is the flag's VALUE; adoption rate is the")
    print("residual (schema teaches it, agent use varies) — the #397 analog of query-formulation variance.")


if __name__ == "__main__":
    main()