Update rft_flightrecorder.py

rft_flightrecorder.py

@@ -1,837 +1,144 @@
-import json
 import os
-import uuid
-import hashlib
+import json
+import threading
+import tempfile
 import zipfile
-from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional, Tuple
 
-from filelock import FileLock, Timeout
+import rft_flightrecorder as fr
 
-from cryptography.hazmat.primitives.asymmetric.ed25519 import (
-    Ed25519PrivateKey,
-    Ed25519PublicKey,
-)
-from cryptography.hazmat.primitives import serialization
 
+def two_tab_spam_test(log_path: str, threads: int = 6, events_per_thread: int = 80):
+    sk_hex, pk_hex = fr.gen_keys()
+    sid, msg = fr.start_session(log_path, "brutal-test", "deterministic", "spam test", False, sk_hex)
+    assert sid, f"Failed to start session: {msg}"
 
-EVENT_SPEC = "rft-flight-event-v0"
-ROOT_SPEC = "rft-flight-session-root-v0"
-DEFAULT_LOG_PATH = "flightlog.jsonl"
+    errors = []
 
-# ---------- Locking ----------
-LOCK_TIMEOUT_S = 10  # short on purpose; UI should not hang
+    def worker(tid: int):
+        for i in range(events_per_thread):
+            payload = json.dumps({"thread": tid, "i": i}, ensure_ascii=False)
+            ev, m = fr.append_event(
+                log_path=log_path,
+                session_id=sid,
+                event_type="note",
+                payload_text=payload,
+                parent_event_hash="",
+                sign_event=False,
+                sk_hex=sk_hex,
+                model_id="brutal-test",
+                run_mode="deterministic",
+            )
+            if not ev:
+                errors.append(m)
 
-# ---------- Public abuse limits (Import Bundle) ----------
-MAX_BUNDLE_BYTES = 15 * 1024 * 1024          # 15 MB zip upload cap
-MAX_ZIP_MEMBER_BYTES = 25 * 1024 * 1024      # 25 MB decompressed member cap
-MAX_EVENTS_PER_BUNDLE = 50_000
-MAX_EVENT_LINE_BYTES = 200_000               # 200 KB per JSONL line (prevents huge single lines)
-MAX_ZIP_RATIO = 200.0                        # zip-bomb heuristic (file_size / compress_size)
+    ts = [threading.Thread(target=worker, args=(t,)) for t in range(threads)]
+    for t in ts:
+        t.start()
+    for t in ts:
+        t.join()
 
+    assert not errors, f"Append errors occurred (first 5): {errors[:5]}"
 
-# ============================================================
-# Canonical JSON + hashing
-# ============================================================
+    status, ok, report = fr.verify_session(log_path, sid, pk_hex="", require_signatures=False)
+    assert ok, f"Session failed verification after spam.\n{status}\n{report}"
 
-def canon(obj) -> bytes:
-    """Canonical JSON encoding (stable for hashing/verifying across machines)."""
-    return json.dumps(
-        obj,
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-    ).encode("utf-8")
+    anchor, m = fr.finalise_session(log_path, sid, False, sk_hex, "brutal-test", "deterministic")
+    assert anchor, f"Finalise failed: {m}"
 
+    # Must refuse writes after end
+    ev, m = fr.append_event(
+        log_path=log_path,
+        session_id=sid,
+        event_type="note",
+        payload_text='{"post_end": true}',
+        parent_event_hash="",
+        sign_event=False,
+        sk_hex=sk_hex,
+        model_id="brutal-test",
+        run_mode="deterministic",
+    )
+    assert ev is None and "ended" in m.lower(), f"Expected append to be refused after session_end, got: {m}"
 
-def sha256_hex(b: bytes) -> str:
-    return hashlib.sha256(b).hexdigest()
+    status, ok, report = fr.verify_session(log_path, sid, pk_hex="", require_signatures=False)
+    assert ok, f"Session failed verification after finalise.\n{status}\n{report}"
 
+    return sid
 
-def now_utc_iso() -> str:
-    return datetime.now(timezone.utc).isoformat()
 
+def tamper_zip_test(log_path: str):
+    sk_hex, _pk_hex = fr.gen_keys()
+    sid, msg = fr.start_session(log_path, "brutal-test", "deterministic", "tamper test", False, sk_hex)
+    assert sid, f"Failed to start session: {msg}"
 
-def short(h: str, n: int = 12) -> str:
-    h = (h or "").strip()
-    return h[:n] if len(h) > n else h
+    for i in range(25):
+        fr.append_event(
+            log_path=log_path,
+            session_id=sid,
+            event_type="note",
+            payload_text=json.dumps({"i": i}),
+            parent_event_hash="",
+            sign_event=False,
+            sk_hex=sk_hex,
+            model_id="brutal-test",
+            run_mode="deterministic",
+        )
 
+    fr.finalise_session(log_path, sid, False, sk_hex, "brutal-test", "deterministic")
 
-# ============================================================
-# File locking helpers
-# ============================================================
+    zip_name, msg = fr.export_session_bundle(log_path, sid)
+    assert zip_name and os.path.exists(zip_name), f"Export failed: {msg}"
 
-def _lock_for(path: str) -> FileLock:
-    return FileLock(path + ".lock", timeout=LOCK_TIMEOUT_S)
+    with tempfile.TemporaryDirectory() as td:
+        with zipfile.ZipFile(zip_name, "r") as z:
+            z.extractall(td)
 
+        events_file = None
+        for fn in os.listdir(td):
+            if fn.endswith("_events.jsonl"):
+                events_file = os.path.join(td, fn)
+                break
+        assert events_file, "No events jsonl found inside exported zip"
 
-def _read_jsonl_locked(path: str) -> Tuple[List[Dict[str, Any]], int]:
-    with _lock_for(path):
-        return read_jsonl(path)
+        lines = open(events_file, "r", encoding="utf-8").read().splitlines()
+        assert len(lines) > 5, "Not enough events in bundle to tamper"
 
+        # Tamper: modify payload of an early event but keep stored hash/signature -> MUST FAIL verification
+        obj = json.loads(lines[4])
+        obj.setdefault("payload", {})
+        if isinstance(obj["payload"], dict):
+            obj["payload"]["tampered"] = True
+        lines[4] = json.dumps(obj, ensure_ascii=False)
 
-# ============================================================
-# Key management (Ed25519)
-# ============================================================
+        open(events_file, "w", encoding="utf-8").write("\n".join(lines) + "\n")
 
-def gen_keys() -> Tuple[str, str]:
-    sk = Ed25519PrivateKey.generate()
-    pk = sk.public_key()
+        tampered_zip = os.path.join(td, "tampered.zip")
+        with zipfile.ZipFile(tampered_zip, "w", compression=zipfile.ZIP_DEFLATED) as z:
+            z.write(events_file, arcname=os.path.basename(events_file))
 
-    sk_b = sk.private_bytes(
-        encoding=serialization.Encoding.Raw,
-        format=serialization.PrivateFormat.Raw,
-        encryption_algorithm=serialization.NoEncryption(),
-    )
-    pk_b = pk.public_bytes(
-        encoding=serialization.Encoding.Raw,
-        format=serialization.PublicFormat.Raw,
-    )
-    return sk_b.hex(), pk_b.hex()
-
-
-def load_sk(sk_hex: str) -> Ed25519PrivateKey:
-    return Ed25519PrivateKey.from_private_bytes(bytes.fromhex(sk_hex.strip()))
-
-
-def load_pk(pk_hex: str) -> Ed25519PublicKey:
-    return Ed25519PublicKey.from_public_bytes(bytes.fromhex(pk_hex.strip()))
-
-
-# ============================================================
-# Payload parsing
-# ============================================================
-
-def parse_payload_text(payload_text: str) -> Dict[str, Any]:
-    """Accept JSON or plain text; store as structured payload either way."""
-    txt = (payload_text or "").strip()
-    if not txt:
-        return {}
-    try:
-        v = json.loads(txt)
-        if isinstance(v, dict):
-            return v
-        return {"_value": v}
-    except Exception:
-        return {"_text": txt}
-
-
-# ============================================================
-# Log IO
-# ============================================================
-
-def read_jsonl(path: str) -> Tuple[List[Dict[str, Any]], int]:
-    """Read JSONL. Returns (records, corrupt_lines_count)."""
-    if not os.path.exists(path):
-        return [], 0
-
-    out: List[Dict[str, Any]] = []
-    corrupt = 0
-    with open(path, "r", encoding="utf-8") as f:
-        for ln in f:
-            ln = ln.strip()
-            if not ln:
-                continue
-            try:
-                obj = json.loads(ln)
-                if isinstance(obj, dict):
-                    out.append(obj)
-                else:
-                    corrupt += 1
-            except Exception:
-                corrupt += 1
-    return out, corrupt
-
-
-def append_jsonl(path: str, obj: Dict[str, Any]) -> None:
-    with open(path, "a", encoding="utf-8") as f:
-        f.write(json.dumps(obj, ensure_ascii=False) + "\n")
-
-
-# ============================================================
-# Core event model
-# ============================================================
-
-def make_event_core(
-    session_id: str,
-    seq: int,
-    event_type: str,
-    payload: Dict[str, Any],
-    parent_event_hash: str,
-    prev_event_hash: str,
-    meta: Dict[str, Any],
-) -> Dict[str, Any]:
-    return {
-        "spec": EVENT_SPEC,
-        "ts_utc": now_utc_iso(),
-        "session_id": session_id,
-        "seq": int(seq),
-        "event_type": (event_type or "note").strip(),
-        "parent_event_hash_sha256": (parent_event_hash or "").strip(),
-        "prev_event_hash_sha256": (prev_event_hash or "").strip() if prev_event_hash else ("0" * 64),
-        "payload": payload if isinstance(payload, dict) else {"_payload": payload},
-        "meta": meta if isinstance(meta, dict) else {},
-    }
-
-
-def events_for_session(all_events: List[Dict[str, Any]], session_id: str) -> List[Dict[str, Any]]:
-    sid = (session_id or "").strip()
-    if not sid:
-        return []
-    return [
-        e for e in all_events
-        if isinstance(e, dict)
-        and e.get("session_id") == sid
-        and e.get("spec") == EVENT_SPEC
-        and "event_hash_sha256" in e
-    ]
-
-
-def last_event_for_session(all_events: List[Dict[str, Any]], session_id: str) -> Optional[Dict[str, Any]]:
-    evs = events_for_session(all_events, session_id)
-    if not evs:
-        return None
-    evs.sort(key=lambda x: int(x.get("seq", 0)))
-    return evs[-1]
-
-
-def next_seq(all_events: List[Dict[str, Any]], session_id: str) -> int:
-    last = last_event_for_session(all_events, session_id)
-    if not last:
-        return 1
-    try:
-        return int(last.get("seq", 0)) + 1
-    except Exception:
-        return 1
-
-
-# ============================================================
-# Append events (hash-chained; optional signature)
-# ============================================================
-
-def append_event(
-    log_path: str,
-    session_id: str,
-    event_type: str,
-    payload_text: str,
-    parent_event_hash: str,
-    sign_event: bool,
-    sk_hex: str,
-    model_id: str,
-    run_mode: str,
-) -> Tuple[Optional[Dict[str, Any]], str]:
-    sid = (session_id or "").strip()
-    if not sid:
-        return None, "Missing session_id."
-
-    # IMPORTANT: Atomic critical section:
-    # read -> compute seq/prev -> build event -> append
-    try:
-        with _lock_for(log_path):
-            all_events, _corrupt = read_jsonl(log_path)
-
-            payload = parse_payload_text(payload_text)
-            meta = {
-                "model_id": (model_id or "unknown").strip(),
-                "run_mode": (run_mode or "unknown").strip(),
-            }
-
-            last = last_event_for_session(all_events, sid)
-            prev_hash = last.get("event_hash_sha256") if last else ("0" * 64)
-            seq = next_seq(all_events, sid)
-
-            # If parent hash not provided, default to prev hash (simple linear causality)
-            parent = (parent_event_hash or "").strip()
-            if not parent and seq > 1:
-                parent = prev_hash
-
-            core = make_event_core(
-                session_id=sid,
-                seq=seq,
-                event_type=event_type,
-                payload=payload,
-                parent_event_hash=parent,
-                prev_event_hash=prev_hash,
-                meta=meta,
-            )
+        status, ok, report, _ = fr.import_bundle_verify(
+            bundle_path=tampered_zip,
+            pk_hex="",
+            require_signatures=False,
+            store_into_log=False,
+            log_path=log_path,
+        )
+        assert not ok, f"Tampered bundle incorrectly verified PASS.\n{status}\n{report}"
 
-            event_hash = sha256_hex(canon(core))
-            event = dict(core)
-            event["event_hash_sha256"] = event_hash
-
-            if sign_event:
-                if not sk_hex or not sk_hex.strip():
-                    return None, "Signing enabled but private key is missing."
-                try:
-                    sk = load_sk(sk_hex)
-                    sig = sk.sign(bytes.fromhex(event_hash))
-                    event["signature_ed25519"] = sig.hex()
-                except Exception:
-                    return None, "Failed to sign event (invalid private key?)."
-
-            append_jsonl(log_path, event)
-
-        return event, f"OK. Appended event #{seq} ({event_type}). Hash: {event_hash}"
-
-    except Timeout:
-        return None, "Busy: log is locked (try again)."
-    except Exception:
-        return None, "Failed to append event (unexpected error)."
-
-
-def start_session(
-    log_path: str,
-    model_id: str,
-    run_mode: str,
-    notes: str,
-    sign_start: bool,
-    sk_hex: str,
-) -> Tuple[str, str]:
-    sid = uuid.uuid4().hex
-    payload = {"notes": (notes or "").strip()}
-
-    ev, msg = append_event(
-        log_path=log_path,
-        session_id=sid,
-        event_type="session_start",
-        payload_text=json.dumps(payload, ensure_ascii=False),
-        parent_event_hash="",
-        sign_event=sign_start,
-        sk_hex=sk_hex,
-        model_id=model_id,
-        run_mode=run_mode,
-    )
-    if not ev:
-        return "", msg
-    return sid, f"OK. Session started: {sid} (first hash: {ev['event_hash_sha256']})"
-
-
-def finalise_session(
-    log_path: str,
-    session_id: str,
-    sign_anchor: bool,
-    sk_hex: str,
-    model_id: str,
-    run_mode: str,
-) -> Tuple[Optional[Dict[str, Any]], str]:
-    sid = (session_id or "").strip()
-    if not sid:
-        return None, "Missing session_id."
-
-    try:
-        # Lock the entire finalise operation so nothing can append between:
-        # anchor computation -> session_end append.
-        with _lock_for(log_path):
-            all_events, _corrupt = read_jsonl(log_path)
-            evs = events_for_session(all_events, sid)
-            if not evs:
-                return None, "No events found for this session."
-
-            evs.sort(key=lambda x: int(x.get("seq", 0)))
-
-            # Anchor must NOT include the session_end event itself (avoids circularity).
-            first_hash = evs[0]["event_hash_sha256"]
-            last_hash = evs[-1]["event_hash_sha256"]
-            count = len(evs)
-
-            root_core = {
-                "spec": ROOT_SPEC,
-                "session_id": sid,
-                "first_event_hash_sha256": first_hash,
-                "last_event_hash_sha256": last_hash,
-                "event_count": count,
-            }
-            root_hash = sha256_hex(canon(root_core))
-
-            anchor = dict(root_core)
-            anchor["root_hash_sha256"] = root_hash
-            anchor["created_utc"] = now_utc_iso()
-            anchor["model_id"] = (model_id or "unknown").strip()
-            anchor["run_mode"] = (run_mode or "unknown").strip()
-
-            if sign_anchor:
-                if not sk_hex or not sk_hex.strip():
-                    return None, "Anchor signing enabled but private key is missing."
-                try:
-                    sk = load_sk(sk_hex)
-                    sig = sk.sign(bytes.fromhex(root_hash))
-                    anchor["signature_ed25519"] = sig.hex()
-                except Exception:
-                    return None, "Failed to sign anchor (invalid private key?)."
-
-            # Append session_end event (manually, inside the same lock)
-            payload_text = json.dumps({"anchor": anchor}, ensure_ascii=False)
-            meta = {
-                "model_id": (model_id or "unknown").strip(),
-                "run_mode": (run_mode or "unknown").strip(),
-            }
-
-            seq = int(evs[-1].get("seq", 0)) + 1
-            prev_hash = evs[-1]["event_hash_sha256"]
-            parent = prev_hash
-
-            core = make_event_core(
-                session_id=sid,
-                seq=seq,
-                event_type="session_end",
-                payload=parse_payload_text(payload_text),
-                parent_event_hash=parent,
-                prev_event_hash=prev_hash,
-                meta=meta,
-            )
-            event_hash = sha256_hex(canon(core))
-            event = dict(core)
-            event["event_hash_sha256"] = event_hash
-
-            if sign_anchor:
-                try:
-                    sk = load_sk(sk_hex)
-                    sig = sk.sign(bytes.fromhex(event_hash))
-                    event["signature_ed25519"] = sig.hex()
-                except Exception:
-                    return None, "Failed to sign session_end event."
-
-            append_jsonl(log_path, event)
-
-        return anchor, f"OK. Session finalised. Root hash: {root_hash} (session_end hash: {event_hash})"
-
-    except Timeout:
-        return None, "Busy: log is locked (try again)."
-    except Exception:
-        return None, "Failed to finalise session (unexpected error)."
-
-
-# ============================================================
-# Verification
-# ============================================================
-
-def _anchor_expected_from_events(evs_sorted: List[Dict[str, Any]], session_end_index: int) -> str:
-    """
-    Compute the expected anchor root_hash for the chain *before* the session_end event at session_end_index.
-    This avoids circularity (session_end payload contains anchor).
-    """
-    chain = evs_sorted[:session_end_index]  # exclude the session_end itself
-    first_hash = chain[0]["event_hash_sha256"]
-    last_hash = chain[-1]["event_hash_sha256"]
-    count = len(chain)
-
-    root_core = {
-        "spec": ROOT_SPEC,
-        "session_id": chain[0]["session_id"],
-        "first_event_hash_sha256": first_hash,
-        "last_event_hash_sha256": last_hash,
-        "event_count": count,
-    }
-    return sha256_hex(canon(root_core))
-
-
-def verify_session_from_events(
-    evs: List[Dict[str, Any]],
-    session_id: str,
-    pk_hex: str = "",
-    require_signatures: bool = False,
-) -> Tuple[str, bool, str]:
-    sid = (session_id or "").strip()
-    if not sid:
-        return "Missing session_id.", False, ""
-
-    if not evs:
-        return "No events found for this session.", False, ""
-
-    report: List[str] = []
-    ok = True
-
-    pk = None
-    if require_signatures:
-        if not pk_hex or not pk_hex.strip():
-            return "Public key required to verify signatures.", False, ""
-        try:
-            pk = load_pk(pk_hex)
-        except Exception:
-            return "Invalid public key.", False, ""
-
-    # Ensure order by seq
-    evs = [e for e in evs if isinstance(e, dict)]
-    evs.sort(key=lambda x: int(x.get("seq", 0)))
-
-    expected_prev = "0" * 64
-    expected_seq = 1
-
-    for i, e in enumerate(evs):
-        for k in ("spec", "ts_utc", "session_id", "seq", "event_type", "prev_event_hash_sha256", "payload", "meta", "event_hash_sha256"):
-            if k not in e:
-                ok = False
-                report.append(f"[FAIL] Event {i+1}: missing field '{k}'.")
-                continue
-
-        if e.get("spec") != EVENT_SPEC:
-            ok = False
-            report.append(f"[FAIL] Bad spec at seq {e.get('seq')}.")
-            continue
-
-        if e.get("session_id") != sid:
-            ok = False
-            report.append(f"[FAIL] session_id mismatch at seq {e.get('seq')}.")
-            continue
-
-        if int(e.get("seq", -1)) != expected_seq:
-            ok = False
-            report.append(f"[FAIL] Seq mismatch: got {e.get('seq')} expected {expected_seq}.")
-        expected_seq += 1
-
-        if e.get("prev_event_hash_sha256") != expected_prev:
-            ok = False
-            report.append(
-                f"[FAIL] Chain broken at seq {e.get('seq')}: prev {short(e.get('prev_event_hash_sha256'))} expected {short(expected_prev)}."
-            )
+    return True
 
-        core = {
-            "spec": e["spec"],
-            "ts_utc": e["ts_utc"],
-            "session_id": e["session_id"],
-            "seq": int(e["seq"]),
-            "event_type": e["event_type"],
-            "parent_event_hash_sha256": e.get("parent_event_hash_sha256", ""),
-            "prev_event_hash_sha256": e["prev_event_hash_sha256"],
-            "payload": e["payload"],
-            "meta": e["meta"],
-        }
-        h = sha256_hex(canon(core))
-        if h != e["event_hash_sha256"]:
-            ok = False
-            report.append(f"[FAIL] Hash mismatch at seq {e.get('seq')}: stored {short(e['event_hash_sha256'])} recomputed {short(h)}.")
-
-        if require_signatures:
-            sig_hex = (e.get("signature_ed25519") or "").strip()
-            if not sig_hex:
-                ok = False
-                report.append(f"[FAIL] Missing signature at seq {e.get('seq')}.")
-            else:
-                try:
-                    pk.verify(bytes.fromhex(sig_hex), bytes.fromhex(e["event_hash_sha256"]))
-                except Exception:
-                    ok = False
-                    report.append(f"[FAIL] Bad signature at seq {e.get('seq')}.")
-
-        expected_prev = e["event_hash_sha256"]
-
-    # Anchor check: validate the LAST session_end event (if present)
-    end_idxs = [i for i, e in enumerate(evs) if e.get("event_type") == "session_end"]
-    if end_idxs:
-        se_idx = end_idxs[-1]
-        se = evs[se_idx]
-        anchor = (se.get("payload") or {}).get("anchor")
-
-        if isinstance(anchor, dict) and se_idx > 0:
-            expected_root = _anchor_expected_from_events(evs, se_idx)
-            if anchor.get("root_hash_sha256") != expected_root:
-                ok = False
-                report.append("[FAIL] Session anchor root hash does not match the pre-session_end event chain.")
-            else:
-                report.append("[OK] Session anchor matches the pre-session_end event chain.")
-        else:
-            report.append("[WARN] session_end found but anchor payload is missing/invalid.")
-
-    if ok:
-        report.insert(0, f"[PASS] Session verified: {len(evs)} events, chain intact.")
-    else:
-        report.insert(0, f"[FAIL] Session verification failed: {len(evs)} events checked.")
-
-    return ("PASS" if ok else "FAIL"), ok, "\n".join(report)
-
-
-def verify_session(log_path: str, session_id: str, pk_hex: str, require_signatures: bool) -> Tuple[str, bool, str]:
-    try:
-        all_events, _corrupt = _read_jsonl_locked(log_path)
-    except Timeout:
-        return "FAIL", False, "Busy: log is locked (try again)."
-
-    evs = events_for_session(all_events, session_id)
-    return verify_session_from_events(evs, session_id, pk_hex=pk_hex, require_signatures=require_signatures)
-
-
-# ============================================================
-# Timeline + listing
-# ============================================================
-
-def session_timeline_rows(log_path: str, session_id: str) -> Tuple[List[List[Any]], str]:
-    sid = (session_id or "").strip()
-    if not sid:
-        return [], "Missing session_id."
-
-    try:
-        all_events, _corrupt = _read_jsonl_locked(log_path)
-    except Timeout:
-        return [], "Busy: log is locked (try again)."
-
-    evs = events_for_session(all_events, sid)
-    if not evs:
-        return [], "No events found."
-
-    evs.sort(key=lambda x: int(x.get("seq", 0)))
-
-    rows: List[List[Any]] = []
-    for e in evs:
-        rows.append([
-            int(e.get("seq", 0)),
-            e.get("ts_utc", ""),
-            e.get("event_type", ""),
-            e.get("meta", {}).get("model_id", ""),
-            e.get("meta", {}).get("run_mode", ""),
-            e.get("parent_event_hash_sha256", ""),
-            e.get("prev_event_hash_sha256", ""),
-            e.get("event_hash_sha256", ""),
-            "yes" if e.get("signature_ed25519") else "no",
-        ])
-    return rows, f"Loaded {len(rows)} events."
-
-
-def get_event_by_hash(log_path: str, session_id: str, event_hash: str) -> Tuple[Optional[Dict[str, Any]], str]:
-    sid = (session_id or "").strip()
-    h = (event_hash or "").strip()
-    if not sid or not h:
-        return None, "Missing session_id or event hash."
-
-    try:
-        all_events, _corrupt = _read_jsonl_locked(log_path)
-    except Timeout:
-        return None, "Busy: log is locked (try again)."
-
-    evs = events_for_session(all_events, sid)
-    for e in evs:
-        if e.get("event_hash_sha256") == h:
-            return e, "OK."
-    return None, "Not found."
-
-
-def list_sessions(log_path: str) -> Tuple[List[str], str]:
-    try:
-        all_events, corrupt = _read_jsonl_locked(log_path)
-    except Timeout:
-        return [], "Busy: log is locked (try again)."
-
-    counts: Dict[str, int] = {}
-    for e in all_events:
-        if isinstance(e, dict) and e.get("spec") == EVENT_SPEC:
-            sid = e.get("session_id")
-            if sid:
-                counts[sid] = counts.get(sid, 0) + 1
-    sessions = sorted(counts.keys())
-    msg = f"Found {len(sessions)} sessions. Corrupt lines ignored: {corrupt}."
-    return sessions, msg
-
-
-def diagnostics(log_path: str) -> Dict[str, Any]:
-    try:
-        all_events, corrupt = _read_jsonl_locked(log_path)
-    except Timeout:
-        return {
-            "log_path": log_path,
-            "exists": os.path.exists(log_path),
-            "bytes": os.path.getsize(log_path) if os.path.exists(log_path) else 0,
-            "total_events": 0,
-            "sessions": 0,
-            "signed_events": 0,
-            "corrupt_lines_ignored": 0,
-            "error": "Busy: log is locked (try again).",
-        }
-
-    size = os.path.getsize(log_path) if os.path.exists(log_path) else 0
-    sessions = set()
-    signed = 0
-    total = 0
-    for e in all_events:
-        if isinstance(e, dict) and e.get("spec") == EVENT_SPEC:
-            total += 1
-            if e.get("session_id"):
-                sessions.add(e["session_id"])
-            if e.get("signature_ed25519"):
-                signed += 1
-    return {
-        "log_path": log_path,
-        "exists": os.path.exists(log_path),
-        "bytes": size,
-        "total_events": total,
-        "sessions": len(sessions),
-        "signed_events": signed,
-        "corrupt_lines_ignored": corrupt,
-    }
-
-
-# ============================================================
-# Export bundle
-# ============================================================
-
-def export_session_bundle(log_path: str, session_id: str) -> Tuple[Optional[str], str]:
-    sid = (session_id or "").strip()
-    if not sid:
-        return None, "Missing session_id."
-
-    try:
-        all_events, _corrupt = _read_jsonl_locked(log_path)
-    except Timeout:
-        return None, "Busy: log is locked (try again)."
-
-    evs = events_for_session(all_events, sid)
-    if not evs:
-        return None, "No events found."
-
-    evs.sort(key=lambda x: int(x.get("seq", 0)))
-    status, ok, report = verify_session_from_events(evs, sid, pk_hex="", require_signatures=False)
-
-    zip_name = f"rft_flight_bundle_{sid}.zip"
-    tmp_dir = "tmp_export"
-    os.makedirs(tmp_dir, exist_ok=True)
-
-    events_path = os.path.join(tmp_dir, f"{sid}_events.jsonl")
-    report_path = os.path.join(tmp_dir, f"{sid}_verify_report.txt")
-
-    with open(events_path, "w", encoding="utf-8") as f:
-        for e in evs:
-            f.write(json.dumps(e, ensure_ascii=False) + "\n")
-
-    with open(report_path, "w", encoding="utf-8") as f:
-        f.write(f"session_id: {sid}\n")
-        f.write(f"status: {status}\n")
-        f.write(f"ok: {ok}\n\n")
-        f.write(report + "\n")
-
-    with zipfile.ZipFile(zip_name, "w", compression=zipfile.ZIP_DEFLATED) as z:
-        z.write(events_path, arcname=os.path.basename(events_path))
-        z.write(report_path, arcname=os.path.basename(report_path))
-
-    return zip_name, f"OK. Exported {zip_name} ({len(evs)} events)."
-
-
-# ============================================================
-# Import bundle (third-party verification)
-# ============================================================
-
-def _zip_member_safe(z: zipfile.ZipFile, member: str) -> Tuple[bool, str]:
-    try:
-        info = z.getinfo(member)
-    except Exception:
-        return False, "Missing events member in zip."
-
-    # Decompressed size cap
-    if info.file_size > MAX_ZIP_MEMBER_BYTES:
-        return False, "Events file too large (decompressed)."
-
-    # Zip-bomb heuristic: insane compression ratio
-    if info.compress_size and info.compress_size > 0:
-        ratio = float(info.file_size) / float(info.compress_size)
-        if ratio > MAX_ZIP_RATIO:
-            return False, "Suspicious zip compression ratio (possible zip bomb)."
-
-    return True, "OK"
-
-
-def _read_jsonl_from_zip(z: zipfile.ZipFile, member: str) -> List[Dict[str, Any]]:
-    out: List[Dict[str, Any]] = []
-    raw_text = z.read(member).decode("utf-8", errors="replace")
-
-    for raw in raw_text.splitlines():
-        raw = raw.strip()
-        if not raw:
-            continue
-
-        # single-line cap
-        if len(raw.encode("utf-8", errors="ignore")) > MAX_EVENT_LINE_BYTES:
-            continue
-
-        try:
-            obj = json.loads(raw)
-            if isinstance(obj, dict):
-                out.append(obj)
-        except Exception:
-            continue
-
-        if len(out) > MAX_EVENTS_PER_BUNDLE:
-            break
-
-    return out
-
-
-def import_bundle_verify(
-    bundle_path: str,
-    pk_hex: str = "",
-    require_signatures: bool = False,
-    store_into_log: bool = False,
-    log_path: str = DEFAULT_LOG_PATH,
-) -> Tuple[str, bool, str, Optional[str]]:
-    if not bundle_path or not os.path.exists(bundle_path):
-        return "Missing bundle file.", False, "", None
-
-    # zip upload cap
-    try:
-        if os.path.getsize(bundle_path) > MAX_BUNDLE_BYTES:
-            return "FAIL", False, "Bundle too large.", None
-    except Exception:
-        return "FAIL", False, "Unable to read bundle size.", None
-
-    try:
-        with zipfile.ZipFile(bundle_path, "r") as z:
-            members = z.namelist()
-
-            events_member = None
-            for m in members:
-                if m.endswith("_events.jsonl"):
-                    events_member = m
-                    break
-            if not events_member:
-                for m in members:
-                    if m.endswith(".jsonl"):
-                        events_member = m
-                        break
-            if not events_member:
-                return "No .jsonl events file found in bundle.", False, "", None
-
-            safe, why = _zip_member_safe(z, events_member)
-            if not safe:
-                return "FAIL", False, why, None
-
-            evs = _read_jsonl_from_zip(z, events_member)
-
-    except Exception:
-        return "Failed to read bundle (invalid zip?).", False, "", None
-
-    if not evs:
-        return "Bundle contains no usable events.", False, "", None
-
-    if len(evs) > MAX_EVENTS_PER_BUNDLE:
-        return "FAIL", False, "Too many events in bundle.", None
-
-    # Determine session_id
-    sid = ""
-    for e in evs:
-        if e.get("spec") == EVENT_SPEC and e.get("session_id"):
-            sid = e["session_id"]
-            break
-    if not sid:
-        return "Bundle contains no valid flight events.", False, "", None
-
-    evs = [e for e in evs if e.get("spec") == EVENT_SPEC and e.get("session_id") == sid]
-    evs.sort(key=lambda x: int(x.get("seq", 0)))
-
-    status, ok, report = verify_session_from_events(
-        evs, sid, pk_hex=pk_hex, require_signatures=require_signatures
-    )
 
-    if store_into_log and ok:
-        # Store under a lock + dedupe by event_hash_sha256 to avoid log pollution
-        try:
-            with _lock_for(log_path):
-                existing, _ = read_jsonl(log_path)
-                seen = {e.get("event_hash_sha256") for e in existing if isinstance(e, dict)}
+def main():
+    with tempfile.TemporaryDirectory() as td:
+        log_path = os.path.join(td, "flightlog.jsonl")
+
+        sid1 = two_tab_spam_test(log_path, threads=8, events_per_thread=60)
+        print(f"[PASS] two_tab_spam_test: session_id={sid1}")
+
+        ok = tamper_zip_test(log_path)
+        print(f"[PASS] tamper_zip_test: {ok}")
 
-                for e in evs:
-                    h = e.get("event_hash_sha256")
-                    if h and h not in seen:
-                        append_jsonl(log_path, e)
-                        seen.add(h)
+    print("[ALL PASS]")
 
-        except Timeout:
-            return "FAIL", False, "Busy: log is locked (try again).", None
 
-    stored_msg = "Stored into local flightlog.jsonl." if (store_into_log and ok) else None
-    return status, ok, report, stored_msg
+if __name__ == "__main__":
+    main()