Create app.py

app.py

@@ -0,0 +1,262 @@
+import os
+import json
+from difflib import unified_diff
+
+import gradio as gr
+
+import rft_flightrecorder as fr
+
+LOG_PATH = fr.DEFAULT_LOG_PATH
+
+
+def text_diff(before_text: str, after_text: str):
+    a = (before_text or "").splitlines()
+    b = (after_text or "").splitlines()
+    diff = list(unified_diff(a, b, fromfile="before", tofile="after", lineterm=""))
+    return "\n".join(diff) if diff else "(no differences)"
+
+
+def download_log():
+    return LOG_PATH if os.path.exists(LOG_PATH) else None
+
+
+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
+    return sid, sid, sid, sid, sid, msg
+
+
+def ui_append_event(session_id, event_type, parent_hash, payload_text, sign_event, sk_hex, model_id, run_mode):
+    ev, msg = fr.append_event(
+        log_path=LOG_PATH,
+        session_id=session_id,
+        event_type=event_type,
+        payload_text=payload_text,
+        parent_event_hash=parent_hash,
+        sign_event=sign_event,
+        sk_hex=sk_hex,
+        model_id=model_id,
+        run_mode=run_mode,
+    )
+    return ev, msg
+
+
+def ui_timeline(session_id):
+    rows, msg = fr.session_timeline_rows(LOG_PATH, session_id)
+    return rows, msg
+
+
+def ui_verify(session_id, pk_hex, require_sigs):
+    return fr.verify_session(LOG_PATH, session_id, pk_hex, require_sigs)
+
+
+def ui_finalise(session_id, sign_anchor, sk_hex, model_id, run_mode):
+    return fr.finalise_session(LOG_PATH, session_id, sign_anchor, sk_hex, model_id, run_mode)
+
+
+def ui_export(session_id):
+    return fr.export_session_bundle(LOG_PATH, 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
+
+
+def ui_pick_session(sid):
+    sid = (sid or "").strip()
+    return sid, sid, sid, sid, sid
+
+
+def ui_get_event(session_id, ev_hash):
+    return fr.get_event_by_hash(LOG_PATH, session_id, ev_hash)
+
+
+def ui_import_bundle(bundle_file, pk_hex, require_sigs, store_into_log):
+    # gr.File returns a path string
+    status, ok, report, stored_msg = fr.import_bundle_verify(
+        bundle_path=bundle_file,
+        pk_hex=pk_hex,
+        require_signatures=require_sigs,
+        store_into_log=store_into_log,
+        log_path=LOG_PATH,
+    )
+    extra = (stored_msg or "")
+    return status, ok, report + (("\n\n" + extra) if extra else "")
+
+
+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"
+        "This Space records a **tamper-evident, hash-chained event timeline** for AI/agent runs.\n\n"
+        "**Core deliverable:** export a ZIP bundle that **anyone can verify**.\n\n"
+        "**Key safety note:** Public demo. Do not paste production private keys here."
+    )
+
+    with gr.Tab("Keys"):
+        sk = gr.Textbox(label="Ed25519 Private Key (hex)", lines=2)
+        pk = gr.Textbox(label="Ed25519 Public Key (hex)", lines=2)
+        gen = gr.Button("Generate Keypair")
+        gen.click(fn=fr.gen_keys, outputs=[sk, pk])
+
+    with gr.Tab("Sessions"):
+        with gr.Row():
+            list_btn = gr.Button("Refresh session list")
+            sessions_dd = gr.Dropdown(label="Existing sessions", choices=[], value=None)
+        sessions_msg = gr.Textbox(label="Status", lines=1)
+
+        sid_start = gr.Textbox(label="session_id (Start/Record)", lines=1)
+        sid_tl = gr.Textbox(label="session_id (Timeline)", lines=1)
+        sid_verify = gr.Textbox(label="session_id (Verify)", lines=1)
+        sid_final = gr.Textbox(label="session_id (Finalise/Export)", lines=1)
+        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])
+
+    with gr.Tab("Start Session"):
+        model_id = gr.Textbox(label="Model ID", value="audit-demo")
+        run_mode = gr.Radio(["deterministic", "creative"], label="Run mode", value="deterministic")
+        notes = gr.Textbox(label="Notes (optional)", lines=3)
+        sign_start = gr.Checkbox(label="Sign session_start event", value=False)
+        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],
+            outputs=[sid_start, sid_record, sid_tl, sid_verify, sid_final, start_status],
+        )
+
+    with gr.Tab("Record Event"):
+        event_type = gr.Dropdown(
+            choices=[
+                "prompt",
+                "output",
+                "tool_call",
+                "tool_result",
+                "memory_read",
+                "memory_write",
+                "retrieval",
+                "policy_block",
+                "error",
+                "note",
+            ],
+            value="note",
+            label="event_type",
+        )
+        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,
+            placeholder='Example JSON:\n{\n  "tool":"search",\n  "input":{"q":"..."},\n  "output":{"items":[...]}\n}\n',
+        )
+        sign_event = gr.Checkbox(label="Sign this event (Ed25519)", value=False)
+        append_btn = gr.Button("Append Event")
+        event_out = gr.JSON(label="event.json")
+        append_status = gr.Textbox(label="Status", lines=2)
+
+        append_btn.click(
+            fn=ui_append_event,
+            inputs=[sid_record, event_type, parent_hash, payload_text, sign_event, sk, model_id, run_mode],
+            outputs=[event_out, append_status],
+        )
+
+        flightlog_file = gr.File(label="flightlog.jsonl (download)")
+        gr.Button("Download flightlog.jsonl").click(fn=download_log, outputs=[flightlog_file])
+
+    with gr.Tab("Timeline"):
+        refresh = gr.Button("Load timeline")
+        tl_status = gr.Textbox(label="Status", lines=1)
+        tl = gr.Dataframe(
+            headers=[
+                "seq",
+                "ts_utc",
+                "event_type",
+                "model_id",
+                "run_mode",
+                "parent_hash",
+                "prev_hash",
+                "event_hash",
+                "signed",
+            ],
+            datatype=["number", "str", "str", "str", "str", "str", "str", "str", "str"],
+            row_count=10,
+            col_count=(9, "fixed"),
+            label="Event timeline",
+            wrap=True,
+        )
+
+        refresh.click(fn=ui_timeline, inputs=[sid_tl], outputs=[tl, tl_status])
+
+        with gr.Accordion("View event by hash", open=False):
+            ev_hash_in = gr.Textbox(label="event_hash_sha256", lines=1)
+            ev_get = gr.Button("Get event")
+            ev_status = gr.Textbox(label="Status", lines=1)
+            ev_json = gr.JSON(label="event.json")
+            ev_get.click(fn=ui_get_event, inputs=[sid_tl, ev_hash_in], outputs=[ev_json, ev_status])
+
+    with gr.Tab("Verify Session"):
+        require_sigs = gr.Checkbox(label="Require signatures on every event", value=False)
+        verify_btn = gr.Button("Verify")
+        verify_msg = gr.Textbox(label="Result", lines=1)
+        verify_ok = gr.Checkbox(label="Valid", value=False)
+        verify_report = gr.Textbox(label="Report", lines=14)
+
+        verify_btn.click(
+            fn=ui_verify,
+            inputs=[sid_verify, pk, require_sigs],
+            outputs=[verify_msg, verify_ok, verify_report],
+        )
+
+    with gr.Tab("Finalise + Export"):
+        sign_anchor = gr.Checkbox(label="Sign session anchor + session_end event", value=False)
+        fin_btn = gr.Button("Finalise session")
+        anchor_out = gr.JSON(label="session_anchor.json")
+        fin_status = gr.Textbox(label="Status", lines=2)
+
+        fin_btn.click(
+            fn=ui_finalise,
+            inputs=[sid_final, sign_anchor, sk, model_id, run_mode],
+            outputs=[anchor_out, fin_status],
+        )
+
+        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_btn.click(
+            fn=ui_export,
+            inputs=[sid_final],
+            outputs=[export_file, export_status],
+        )
+
+    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)
+        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)
+        import_ok = gr.Checkbox(label="Valid", value=False)
+        import_report = gr.Textbox(label="Report", lines=14)
+
+        import_btn.click(
+            fn=ui_import_bundle,
+            inputs=[bundle, pk, import_require_sigs, store_into_log],
+            outputs=[import_status, import_ok, import_report],
+        )
+
+    with gr.Tab("Diagnostics"):
+        diag_btn = gr.Button("Run diagnostics")
+        diag_out = gr.JSON(label="diagnostics.json")
+        diag_btn.click(fn=lambda: fr.diagnostics(LOG_PATH), outputs=[diag_out])
+
+    with gr.Tab("Diff Helper"):
+        before = gr.Textbox(label="Before (text/JSON)", lines=8)
+        after = gr.Textbox(label="After (text/JSON)", lines=8)
+        diff_btn = gr.Button("Generate unified diff")
+        diff_out = gr.Textbox(label="Diff", lines=14)
+        diff_btn.click(fn=text_diff, inputs=[before, after], outputs=[diff_out])
+
+demo.launch()