Update app.py

app.py

@@ -1,5 +1,11 @@
 import os
 import json
+import time
+import random
+import shutil
+import zipfile
+import tempfile
+import threading
 from difflib import unified_diff
 
 import gradio as gr
@@ -9,6 +15,10 @@ import rft_flightrecorder as fr
 LOG_PATH = fr.DEFAULT_LOG_PATH
 
 
+# ---------------------------
+# Helpers
+# ---------------------------
+
 def text_diff(before_text: str, after_text: str):
     a = (before_text or "").splitlines()
     b = (after_text or "").splitlines()
@@ -20,6 +30,31 @@ def download_log():
     return LOG_PATH if os.path.exists(LOG_PATH) else None
 
 
+def _filepath(file_obj) -> str:
+    """
+    Gradio file inputs can be:
+      - str path
+      - dict-like with 'path'
+      - object with .name or .path
+    This normalises to a filesystem path string.
+    """
+    if not file_obj:
+        return ""
+    if isinstance(file_obj, str):
+        return file_obj
+    if isinstance(file_obj, dict) and file_obj.get("path"):
+        return file_obj["path"]
+    for attr in ("path", "name"):
+        v = getattr(file_obj, attr, None)
+        if isinstance(v, str) and v:
+            return v
+    return ""
+
+
+# ---------------------------
+# UI wrappers
+# ---------------------------
+
 def ui_start_session(model_id, run_mode, notes, sign_start, sk_hex):
     sid, msg = fr.start_session(LOG_PATH, model_id, run_mode, notes, sign_start, sk_hex)
     # fan-out the session id into all tabs
@@ -60,7 +95,8 @@ def ui_export(session_id):
 
 def ui_list_sessions():
     sessions, msg = fr.list_sessions(LOG_PATH)
-    return gr.Dropdown(choices=sessions, value=(sessions[-1] if sessions else None)), msg
+    val = sessions[-1] if sessions else None
+    return gr.update(choices=sessions, value=val), msg
 
 
 def ui_pick_session(sid):
@@ -73,9 +109,12 @@ def ui_get_event(session_id, ev_hash):
 
 
 def ui_import_bundle(bundle_file, pk_hex, require_sigs, store_into_log):
-    # gr.File returns a path string
+    bundle_path = _filepath(bundle_file)
+    if not bundle_path:
+        return "Missing bundle file.", False, "Upload a ZIP bundle first."
+
     status, ok, report, stored_msg = fr.import_bundle_verify(
-        bundle_path=bundle_file,
+        bundle_path=bundle_path,
         pk_hex=pk_hex,
         require_signatures=require_sigs,
         store_into_log=store_into_log,
@@ -85,6 +124,240 @@ def ui_import_bundle(bundle_file, pk_hex, require_sigs, store_into_log):
     return status, ok, report + (("\n\n" + extra) if extra else "")
 
 
+# ---------------------------
+# Self-tests (do NOT touch main LOG_PATH)
+# ---------------------------
+
+def _two_tab_spam_test(n_per_thread: int = 120) -> dict:
+    tmpdir = tempfile.mkdtemp(prefix="rft_selftest_spam_")
+    log_path = os.path.join(tmpdir, "flightlog.jsonl")
+
+    sk_hex, _pk_hex = fr.gen_keys()
+    sid, msg = fr.start_session(
+        log_path=log_path,
+        model_id="self-test",
+        run_mode="deterministic",
+        notes="two-tab spam test (temp log)",
+        sign_start=False,
+        sk_hex=sk_hex,
+    )
+    if not sid:
+        return {"ok": False, "error": msg, "tmpdir": tmpdir}
+
+    errors = []
+
+    def worker(tid: int):
+        for i in range(n_per_thread):
+            payload = {"thread": tid, "i": i, "rand": random.random()}
+            payload_text = json.dumps(payload, ensure_ascii=False)
+
+            # Retry on lock contention
+            ok_write = False
+            for attempt in range(60):
+                ev, m = fr.append_event(
+                    log_path=log_path,
+                    session_id=sid,
+                    event_type="note",
+                    payload_text=payload_text,
+                    parent_event_hash="",
+                    sign_event=False,
+                    sk_hex=sk_hex,
+                    model_id="self-test",
+                    run_mode="deterministic",
+                )
+                if ev is not None:
+                    ok_write = True
+                    break
+                if "Busy" in (m or ""):
+                    time.sleep(0.001 * (attempt + 1))
+                    continue
+                errors.append({"thread": tid, "i": i, "msg": m})
+                break
+
+            if not ok_write:
+                errors.append({"thread": tid, "i": i, "msg": "Too many retries (lock contention)"})
+
+            if i % 40 == 0:
+                time.sleep(0.001)
+
+    t1 = threading.Thread(target=worker, args=(1,))
+    t2 = threading.Thread(target=worker, args=(2,))
+    t0 = time.time()
+    t1.start(); t2.start()
+    t1.join(); t2.join()
+    dt = time.time() - t0
+
+    anchor, fin_msg = fr.finalise_session(
+        log_path=log_path,
+        session_id=sid,
+        sign_anchor=False,
+        sk_hex=sk_hex,
+        model_id="self-test",
+        run_mode="deterministic",
+    )
+    if anchor is None:
+        return {"ok": False, "error": fin_msg, "tmpdir": tmpdir}
+
+    status, ok, report = fr.verify_session(log_path, sid, pk_hex="", require_signatures=False)
+
+    all_events, corrupt = fr.read_jsonl(log_path)
+    evs = fr.events_for_session(all_events, sid)
+    evs.sort(key=lambda x: int(x.get("seq", 0)))
+
+    expected = 2 * n_per_thread + 2  # session_start + events + session_end
+    counts_ok = (len(evs) == expected)
+    seq_ok = all(int(evs[i].get("seq", 0)) == i + 1 for i in range(len(evs)))
+
+    return {
+        "ok": bool(ok and counts_ok and seq_ok and len(errors) == 0 and corrupt == 0),
+        "verify_status": status,
+        "verify_ok": ok,
+        "event_count": len(evs),
+        "expected_count": expected,
+        "counts_ok": counts_ok,
+        "seq_ok": seq_ok,
+        "corrupt_lines": corrupt,
+        "append_error_count": len(errors),
+        "duration_s": round(dt, 4),
+        "report_tail": "\n".join(report.splitlines()[-8:]),
+    }
+
+
+def _tamper_zip_test() -> dict:
+    tmpdir = tempfile.mkdtemp(prefix="rft_selftest_tamper_")
+    log_path = os.path.join(tmpdir, "flightlog.jsonl")
+
+    sk_hex, pk_hex = fr.gen_keys()
+
+    sid, msg = fr.start_session(
+        log_path=log_path,
+        model_id="self-test",
+        run_mode="deterministic",
+        notes="tamper zip test (temp log)",
+        sign_start=True,
+        sk_hex=sk_hex,
+    )
+    if not sid:
+        return {"ok": False, "error": msg}
+
+    for i in range(5):
+        ev, m = fr.append_event(
+            log_path=log_path,
+            session_id=sid,
+            event_type="note",
+            payload_text=json.dumps({"i": i, "msg": "hello"}, ensure_ascii=False),
+            parent_event_hash="",
+            sign_event=True,
+            sk_hex=sk_hex,
+            model_id="self-test",
+            run_mode="deterministic",
+        )
+        if ev is None:
+            return {"ok": False, "error": m}
+
+    anchor, m = fr.finalise_session(
+        log_path=log_path,
+        session_id=sid,
+        sign_anchor=True,
+        sk_hex=sk_hex,
+        model_id="self-test",
+        run_mode="deterministic",
+    )
+    if anchor is None:
+        return {"ok": False, "error": m}
+
+    zip_name, m = fr.export_session_bundle(log_path, sid)
+    if zip_name is None:
+        return {"ok": False, "error": m}
+
+    # Move bundle into tempdir to keep repo tidy
+    zip_full = os.path.join(tmpdir, os.path.basename(zip_name))
+    try:
+        shutil.move(zip_name, zip_full)
+    except Exception:
+        zip_full = zip_name
+
+    # Untampered must PASS
+    status1, ok1, report1, _ = fr.import_bundle_verify(
+        bundle_path=zip_full,
+        pk_hex=pk_hex,
+        require_signatures=True,
+        store_into_log=False,
+        log_path=log_path,
+    )
+
+    # Tamper: edit seq=3 payload inside zip; verification must FAIL
+    tampered_zip = os.path.join(tmpdir, "tampered_" + os.path.basename(zip_full))
+    with zipfile.ZipFile(zip_full, "r") as z:
+        members = z.namelist()
+        events_member = next((m for m in members if m.endswith("_events.jsonl")), None)
+        if not events_member:
+            return {"ok": False, "error": "No _events.jsonl in bundle."}
+
+        lines = z.read(events_member).decode("utf-8", errors="replace").splitlines()
+        new_lines = []
+        changed = False
+
+        for ln in lines:
+            try:
+                obj = json.loads(ln)
+            except Exception:
+                new_lines.append(ln)
+                continue
+
+            if obj.get("spec") == fr.EVENT_SPEC and int(obj.get("seq", 0)) == 3 and not changed:
+                payload = obj.get("payload")
+                if not isinstance(payload, dict):
+                    payload = {}
+                payload["msg"] = "hacked"
+                obj["payload"] = payload
+                new_lines.append(json.dumps(obj, ensure_ascii=False))
+                changed = True
+            else:
+                new_lines.append(ln)
+
+        if not changed:
+            return {"ok": False, "error": "Could not locate seq=3 to tamper."}
+
+        new_events_bytes = ("\n".join(new_lines) + "\n").encode("utf-8")
+
+        with zipfile.ZipFile(tampered_zip, "w", compression=zipfile.ZIP_DEFLATED) as outz:
+            for mname in members:
+                if mname == events_member:
+                    outz.writestr(mname, new_events_bytes)
+                else:
+                    outz.writestr(mname, z.read(mname))
+
+    status2, ok2, report2, _ = fr.import_bundle_verify(
+        bundle_path=tampered_zip,
+        pk_hex=pk_hex,
+        require_signatures=True,
+        store_into_log=False,
+        log_path=log_path,
+    )
+
+    return {
+        "ok": bool(ok1 is True and ok2 is False),
+        "original_ok": ok1,
+        "original_status": status1,
+        "tampered_ok": ok2,
+        "tampered_status": status2,
+        "original_report_tail": "\n".join(report1.splitlines()[-8:]),
+        "tampered_report_head": "\n".join(report2.splitlines()[:12]),
+    }
+
+
+def ui_run_selftests(n_events_each: int):
+    spam = _two_tab_spam_test(n_per_thread=int(n_events_each))
+    tamper = _tamper_zip_test()
+    overall = bool(spam.get("ok")) and bool(tamper.get("ok"))
+    return ("PASS" if overall else "FAIL"), overall, {"two_tab_spam": spam, "tamper_zip": tamper}
+
+
+# ---------------------------
+# UI
+# ---------------------------
+
 with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Party Verification") as demo:
     gr.Markdown(
         "# RFT Agent Flight Recorder — Black Box Trace + Third-Party Verification\n"
@@ -112,7 +385,11 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
         sid_record = gr.Textbox(label="session_id (Record Event)", lines=1)
 
         list_btn.click(fn=ui_list_sessions, outputs=[sessions_dd, sessions_msg])
-        sessions_dd.change(fn=ui_pick_session, inputs=[sessions_dd], outputs=[sid_start, sid_record, sid_tl, sid_verify, sid_final])
+        sessions_dd.change(
+            fn=ui_pick_session,
+            inputs=[sessions_dd],
+            outputs=[sid_start, sid_record, sid_tl, sid_verify, sid_final],
+        )
 
     with gr.Tab("Start Session"):
         model_id = gr.Textbox(label="Model ID", value="audit-demo")
@@ -122,7 +399,6 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
         start_btn = gr.Button("Start New Session")
         start_status = gr.Textbox(label="Status", lines=2)
 
-        # fan-out session id to all places
         start_btn.click(
             fn=ui_start_session,
             inputs=[model_id, run_mode, notes, sign_start, sk],
@@ -146,7 +422,10 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
             value="note",
             label="event_type",
         )
-        parent_hash = gr.Textbox(label="parent_event_hash_sha256 (optional). If empty, defaults to previous event.", lines=1)
+        parent_hash = gr.Textbox(
+            label="parent_event_hash_sha256 (optional). If empty, defaults to previous event.",
+            lines=1,
+        )
         payload_text = gr.Textbox(
             label="payload (JSON or plain text)",
             lines=10,
@@ -163,7 +442,7 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
             outputs=[event_out, append_status],
         )
 
-        flightlog_file = gr.File(label="flightlog.jsonl (download)")
+        flightlog_file = gr.File(label="flightlog.jsonl (download)", type="filepath")
         gr.Button("Download flightlog.jsonl").click(fn=download_log, outputs=[flightlog_file])
 
     with gr.Tab("Timeline"):
@@ -224,7 +503,7 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
 
         export_btn = gr.Button("Export session bundle (ZIP)")
         export_status = gr.Textbox(label="Export status", lines=1)
-        export_file = gr.File(label="Bundle download")
+        export_file = gr.File(label="Bundle download", type="filepath")
 
         export_btn.click(
             fn=ui_export,
@@ -233,8 +512,11 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
         )
 
     with gr.Tab("Import Bundle"):
-        bundle = gr.File(label="Upload rft_flight_bundle_*.zip")
-        store_into_log = gr.Checkbox(label="Store imported events into local flightlog.jsonl (only if PASS)", value=False)
+        bundle = gr.File(label="Upload rft_flight_bundle_*.zip", type="filepath")
+        store_into_log = gr.Checkbox(
+            label="Store imported events into local flightlog.jsonl (only if PASS)",
+            value=False,
+        )
         import_require_sigs = gr.Checkbox(label="Require signatures on every imported event", value=False)
         import_btn = gr.Button("Verify bundle")
         import_status = gr.Textbox(label="Result", lines=1)
@@ -247,6 +529,20 @@ with gr.Blocks(title="RFT Agent Flight Recorder — Black Box Trace + Third-Part
             outputs=[import_status, import_ok, import_report],
         )
 
+    with gr.Tab("Self Tests"):
+        gr.Markdown(
+            "Runs **two brutal checks** against a **temporary log** (does not touch your real `flightlog.jsonl`).\n\n"
+            "- **Two-tab spam:** concurrent appends -> verifies seq + hash chain\n"
+            "- **Tamper ZIP:** export -> edit bundle -> verify should FAIL"
+        )
+        n_each = gr.Slider(30, 250, value=120, step=10, label="Events per thread (spam test)")
+        run_tests = gr.Button("Run self-tests")
+        tests_status = gr.Textbox(label="Overall result", lines=1)
+        tests_ok = gr.Checkbox(label="PASS", value=False)
+        tests_json = gr.JSON(label="selftest_report.json")
+
+        run_tests.click(fn=ui_run_selftests, inputs=[n_each], outputs=[tests_status, tests_ok, tests_json])
+
     with gr.Tab("Diagnostics"):
         diag_btn = gr.Button("Run diagnostics")
         diag_out = gr.JSON(label="diagnostics.json")