interactive-story/backend/tools/m11_offline.py

"""M11 §18 and §24: a container with no network, and the packaging path.

    python -m tools.m11_offline --out <dir> [--no-build]

Run from `backend/`. Needs Docker.

## Why a container rather than a namespace

§18 asks for a true offline run: fresh data, **no route to the public Internet**,
no external DNS. The obvious tool is an unprivileged network namespace, and on
this machine that is refused — Ubuntu 24.04 sets
`kernel.apparmor_restrict_unprivileged_userns=1`, so `unshare -rn` cannot map a
uid. `docker run --network none` gives the same isolation and more: the
container has a loopback interface and nothing else, no resolver, no route, and
a fresh volume. It also happens to be the packaging path §24 wants exercised, so
one run answers both.

The exercise runs *inside* the container over `docker exec`, because with no
network there is no published port to reach from the host. That is not a
workaround; it is the only honest way to drive an isolated process.

## What an offline run can and cannot prove here

Everything that does not need a model: first page load, the SPA's own assets,
campaign creation, a story turn's *attempt*, knowledge import and retrieval,
export, import into a fresh campaign, restart, and the M10 media module
importing and staying inert.

Inference is **not** exercised offline, and the report says so rather than
implying otherwise. This deployment's Ollama is on another machine on the
trusted LAN, which `SECURITY-THREAT-MODEL.md` §73 permits and which is not an
Internet dependency — but it is also not reachable from a container with no
network. What the offline run proves about inference is the useful half: with no
model reachable, the application degrades to a reported error and the campaign
stays intact.
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path

HERE = Path(__file__).resolve().parent
ROOT = HERE.parent.parent
IMAGE = "interactive-story:m11-offline"
NAME = "m11-offline"

#: The script that runs inside the container. Written to a file and copied in,
#: so it is readable evidence rather than a wall of `-c` quoting.
INSIDE = r'''
import json, os, socket, sys, time, urllib.error, urllib.request

BASE = "http://127.0.0.1:8000"
results = []

def check(name, ok, detail=""):
    results.append({"check": name, "result": "PASS" if ok else "FAIL",
                    "detail": str(detail)[:300]})

def api(method, path, payload=None, timeout=60):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(BASE + "/api" + path, data=data, method=method,
                                 headers={"Content-Type": "application/json"} if data else {})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        body = r.read().decode()
    return json.loads(body) if body else None

# --- 1. There is genuinely no way out. ---
try:
    socket.setdefaulttimeout(5)
    socket.create_connection(("1.1.1.1", 443), timeout=5)
    check("no route to the public Internet", False, "a connection succeeded")
except OSError as exc:
    check("no route to the public Internet", True, type(exc).__name__)
try:
    socket.getaddrinfo("example.com", 443)
    check("no external DNS", False, "resolution succeeded")
except OSError as exc:
    check("no external DNS", True, type(exc).__name__)

# --- 2. First page load, from a fresh data directory. ---
with urllib.request.urlopen(BASE + "/", timeout=30) as r:
    page = r.read().decode()
    csp = r.headers.get("Content-Security-Policy", "")
check("the first page load succeeds offline", "<div id=\"root\">" in page)
check("the page names no remote origin",
      "http://" not in page.replace('http://www.w3.org', '') and "https://" not in page)
check("a CSP is served", bool(csp), csp[:120])

# --- 3. Every asset the page asks for is local, and resolves. ---
import re
assets = re.findall(r'(?:src|href)="([^"]+)"', page)
missing = []
for href in assets:
    if href.startswith("http"):
        missing.append("REMOTE:" + href); continue
    try:
        with urllib.request.urlopen(BASE + href, timeout=30) as r:
            r.read(64)
    except Exception as exc:
        missing.append(f"{href}:{type(exc).__name__}")
check("every asset the shell references is served locally", not missing, missing)

# --- 4. A campaign, offline. ---
adv = api("POST", "/adventures", {"title": "Offline", "opening": "Rain.",
                                  "canon_rules": ["The dead do not return."],
                                  "persona_name": "Aldric"})["id"]
check("a campaign can be created offline", isinstance(adv, int))

api("POST", f"/adventures/{adv}/state/corrections", {"events": [
    {"type": "create_entity", "entity": "aldric", "entity_type": "character",
     "name": "Aldric"},
    {"type": "set_scene", "summary": "Aldric by the fire.", "location": "tavern",
     "present": ["aldric"]}], "note": ""})
state = api("GET", f"/adventures/{adv}/state")["document"]
check("state extraction works offline", state["entities"]["aldric"]["name"] == "Aldric")

# --- 5. Knowledge import and retrieval, offline. ---
boundary = "----m11offline"
body = (
    f"--{boundary}\r\nContent-Disposition: form-data; name=\"classification\"\r\n\r\ncanon\r\n"
    f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"canon.md\"\r\n"
    f"Content-Type: text/markdown\r\n\r\n# Westhaven\n\nThe crypt is sealed.\r\n"
    f"--{boundary}--\r\n").encode()
req = urllib.request.Request(BASE + f"/api/adventures/{adv}/knowledge", data=body,
                             method="POST",
                             headers={"Content-Type": f"multipart/form-data; boundary={boundary}"})
with urllib.request.urlopen(req, timeout=60) as r:
    source = json.loads(r.read().decode())
check("a local file imports offline", source["id"] > 0)
report = api("GET", f"/adventures/{adv}/context")
check("the prompt assembles offline", report["tokens"]["total"] > 0)
check("imported knowledge is searchable offline",
      any("crypt" in u["text"].lower() for u in report["knowledge"]["used"])
      or report["knowledge"]["considered"] is not None)

# --- 6. A turn with no model reachable: reported, and nothing corrupted. ---
page_before = api("GET", f"/adventures/{adv}/actions?limit=50")
before = page_before["total"]
ai_before = [a["id"] for a in page_before["actions"] if a["type"] == "ai"]
events = []
req = urllib.request.Request(BASE + f"/api/adventures/{adv}/actions",
                             data=json.dumps({"type": "do", "text": "I look around."}).encode(),
                             method="POST", headers={"Content-Type": "application/json"})
try:
    with urllib.request.urlopen(req, timeout=120) as r:
        for raw in r:
            line = raw.decode(errors="replace").strip()
            if line.startswith("data:"):
                try: events.append(json.loads(line[5:].strip()))
                except Exception: pass
except Exception as exc:
    events.append({"type": "error", "detail": f"{type(exc).__name__}"})
errors = [e for e in events if e.get("type") == "error"]
page_after = api("GET", f"/adventures/{adv}/actions?limit=50")
ai_after = [a["id"] for a in page_after["actions"] if a["type"] == "ai"]
check("a turn with no model reachable is reported as a failure", bool(errors),
      json.dumps(events)[:200])
# L01, stated the way the product states it. A05 deliberately *keeps* the
# player's submitted text so it can be tried again, and the head sits on it —
# so the total action count is expected to rise by one. What must not happen is
# an accepted narration, or state moving for a turn that did not occur. The
# first version of this check compared totals and called the retained input a
# corruption, which is a harness defect of exactly the kind that would have hidden
# a real one.
check("no narration was accepted", ai_after == ai_before,
      f"{len(ai_before)} -> {len(ai_after)}")
check("the player's own words were kept, as A05 intends",
      page_after["total"] == before + 1, f"{before} -> {page_after['total']}")
check("and the state is unchanged by it",
      api("GET", f"/adventures/{adv}/state")["document"] == state)
check("and the earlier story is still there",
      all(a["id"] in [x["id"] for x in page_after["actions"]]
          for a in page_before["actions"]))

# --- 7. Export and import, offline. ---
bundle = api("GET", f"/adventures/{adv}/export")
check("a campaign exports offline", bundle["format"].startswith("ai-dnd-adventure"))
copy_id = api("POST", "/adventures/import", bundle)["id"]
copy_state = api("GET", f"/adventures/{copy_id}/state")["document"]
check("and imports offline, with its state", copy_state["entities"]["aldric"]["name"] == "Aldric")
check("no secret is present in the export",
      not any(k in json.dumps(bundle).lower() for k in ("api_key", "apikey", "secret.key")))

# --- 8. The M10 media layer imports and stays inert. ---
sys.path.insert(0, "/app/backend")
from app.media import packet, profiles, providers  # noqa: E402
check("the media module imports offline", providers.registered() == {})
packet_out = api("GET", f"/adventures/{adv}/scene-packet")
check("a scene packet builds offline", packet_out["scene_id"].startswith("c"))
check("no media provider is required", providers.registered() == {})

print("M11-OFFLINE-RESULTS " + json.dumps(results))
'''


def run(*args, **kwargs):
    return subprocess.run(args, capture_output=True, text=True, **kwargs)


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", required=True)
    parser.add_argument("--no-build", action="store_true")
    args = parser.parse_args()
    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    started = datetime.now()

    if not args.no_build:
        print("building the image with --no-cache …")
        build = run("docker", "build", "--no-cache", "-t", IMAGE, ".", cwd=str(ROOT))
        (out / "docker-build.log").write_text(build.stdout + build.stderr)
        if build.returncode != 0:
            print(f"build failed; see {out / 'docker-build.log'}")
            return 1
        print("  built")

    run("docker", "rm", "-f", NAME)
    script = out / "inside.py"
    script.write_text(INSIDE)

    print("starting the container with --network none …")
    start = run("docker", "run", "-d", "--name", NAME, "--network", "none", IMAGE)
    if start.returncode != 0:
        print(start.stderr[:500])
        return 1
    container = start.stdout.strip()[:12]

    try:
        # Wait for the application inside, over exec rather than over a port.
        ready = False
        for _ in range(120):
            probe = run("docker", "exec", NAME, "python", "-c",
                        "import urllib.request;"
                        "urllib.request.urlopen('http://127.0.0.1:8000/api/settings',"
                        " timeout=3)")
            if probe.returncode == 0:
                ready = True
                break
            time.sleep(1)
        if not ready:
            logs = run("docker", "logs", NAME)
            (out / "container.log").write_text(logs.stdout + logs.stderr)
            print(f"the container never became ready; see {out / 'container.log'}")
            return 1
        print(f"  container {container} is serving (no network)")

        run("docker", "cp", str(script), f"{NAME}:/tmp/inside.py")
        result = run("docker", "exec", NAME, "python", "/tmp/inside.py")
        (out / "inside-stdout.txt").write_text(result.stdout + "\n---\n" + result.stderr)

        results = []
        for line in result.stdout.splitlines():
            if line.startswith("M11-OFFLINE-RESULTS "):
                results = json.loads(line[len("M11-OFFLINE-RESULTS "):])
        if not results:
            print("no results came back; see inside-stdout.txt")
            return 1

        # §24: persistence across a container restart, on the same volume.
        run("docker", "restart", NAME)
        for _ in range(120):
            probe = run("docker", "exec", NAME, "python", "-c",
                        "import urllib.request,json;"
                        "print(urllib.request.urlopen("
                        "'http://127.0.0.1:8000/api/adventures', timeout=3)"
                        ".read().decode()[:200])")
            if probe.returncode == 0:
                break
            time.sleep(1)
        survived = "Offline" in probe.stdout
        results.append({"check": "campaigns survive a container restart",
                        "result": "PASS" if survived else "FAIL",
                        "detail": probe.stdout[:200]})
        print(f"  restart: {'campaigns survived' if survived else 'DATA LOST'}")

        logs = run("docker", "logs", NAME)
        (out / "container.log").write_text(logs.stdout + logs.stderr)

        report = {
            "image": IMAGE,
            "container": container,
            "network": "none",
            "started": started.isoformat(timespec="seconds"),
            "seconds": round((datetime.now() - started).total_seconds()),
            "checks": results,
            "passed": len([r for r in results if r["result"] == "PASS"]),
            "failed": len([r for r in results if r["result"] == "FAIL"]),
        }
        (out / "offline-report.json").write_text(json.dumps(report, indent=2))
        for row in results:
            mark = "ok  " if row["result"] == "PASS" else "FAIL"
            print(f"  {mark} {row['check']}"
                  + (f" — {row['detail'][:80]}" if row["result"] == "FAIL" else ""))
        print(f"\n{report['passed']} passed, {report['failed']} failed "
              f"-> {out / 'offline-report.json'}")
        return 1 if report["failed"] else 0
    finally:
        run("docker", "rm", "-f", NAME)


if __name__ == "__main__":
    raise SystemExit(main())