Update app.py

app.py

@@ -25,6 +25,7 @@ RECEIPTS_DIR = pathlib.Path("/tmp/receipts"); RECEIPTS_DIR.mkdir(exist_ok=True)
 EVENTS_PATH  = "/tmp/events.json"
 REQS_PATH    = "/tmp/requests.jsonl"
 BUDGET_FILE  = "/tmp/nim_calls_today.json"
+SESSIONS_DIR = pathlib.Path("/tmp/sessions"); SESSIONS_DIR.mkdir(exist_ok=True)
 
 # ========= Helpers: canonicalization & hashing =========
 def canonicalize(text: str) -> str:
@@ -257,14 +258,12 @@ def completions(request: Request, body: ChatRequest):
     }
 
 # ========= Session & Merkle helpers =========
-SESSIONS_DIR = pathlib.Path("/tmp/sessions"); SESSIONS_DIR.mkdir(exist_ok=True)
-
 def _session_path(sid: str) -> pathlib.Path:
     return SESSIONS_DIR / f"{sid}.json"
 
 def merkle_root_from_leaves(leaves: list[str]) -> str:
     """Compute a simple Merkle root from hex-string leaves.
-    Parent = sha256_hex(left + right) (deterministic demo)."""
+    Parent = sha256_hex(left + right) (deterministic demo). Duplicate last if odd per level."""
     if not leaves:
         return ""
     level = leaves[:]
@@ -342,17 +341,83 @@ def api_session_finish(sid: str):
 def api_session_get(sid: str):
     return session_load(sid)
 
-# ========= UI helpers =========
-def _load_events_df():
-    if not os.path.exists(EVENTS_PATH):
-        return pd.DataFrame({"info": ["No alerts yet"]})
-    try:
-        data = json.loads(open(EVENTS_PATH).read())
-        for d in data:
-            d.setdefault("prompt", ""); d.setdefault("snippet", "")
-        return pd.DataFrame(data)[["ts", "type", "hits", "prompt", "snippet"]].iloc[::-1]
-    except Exception:
-        return pd.DataFrame({"info": ["Could not parse events"]})
+# ========= Verify-All helpers (session + per-turn text) =========
+def _list_sessions():
+    if not SESSIONS_DIR.exists():
+        return []
+    files = []
+    for p in SESSIONS_DIR.glob("*.json"):
+        try:
+            files.append((p, p.stat().st_mtime))
+        except Exception:
+            pass
+    files.sort(key=lambda x: x[1], reverse=True)
+    return [p.name[:-5] for p, _ in files]
+
+def latest_session_id() -> str:
+    ids = _list_sessions()
+    return ids[0] if ids else ""
+
+def recompute_session_root_and_table(sid: str):
+    """Return (computed_root, match_str, rows_for_table, proof_link_md)."""
+    doc = session_load(sid)
+    # Recompute each leaf from receipts
+    rows = []
+    leaves = []
+    for idx, rid in enumerate(doc.get("receipts", [])):
+        rp = RECEIPTS_DIR / f"{rid}.json"
+        if not rp.exists():
+            rows.append([rid, "(missing receipt)", "", False])
+            leaves.append("")
+            continue
+        rj = json.loads(rp.read_text())
+        recomputed_leaf = leaf_from_receipt(rj)
+        session_leaf = doc.get("leaf_hashes", [])[idx] if idx < len(doc.get("leaf_hashes", [])) else ""
+        ok = (session_leaf == recomputed_leaf) if session_leaf else True
+        rows.append([rid, session_leaf, recomputed_leaf, ok])
+        leaves.append(recomputed_leaf)
+    # Recompute Merkle root from recomputed leaves
+    comp_root = merkle_root_from_leaves(leaves)
+    ok_root = (comp_root == doc.get("merkle_root", ""))
+    return comp_root, f"Merkle root match: {ok_root}", rows, f"[Open proof](/session/{sid})"
+
+def make_text_rows_for_session(sid: str):
+    """Return rows the user can paste prompts/answers into: [receipt_id, prompt, answer, sha_ok, hmac_ok]."""
+    doc = session_load(sid)
+    rows = []
+    for rid in doc.get("receipts", []):
+        rows.append([rid, "", "", False, (True if not AUDIT_PEPPER else False)])
+    return rows
+
+def verify_text_rows(sid: str, rows: list, pepper: str):
+    """Rows: [receipt_id, prompt, answer, sha_ok, hmac_ok] -> fill sha_ok/hmac_ok."""
+    out = []
+    for rec in rows or []:
+        if len(rec) < 5:
+            rec = list(rec) + ["", "", False, False]
+        rid, prompt, answer, sha_ok, hmac_ok = rec[:5]
+        try:
+            rj = json.loads((RECEIPTS_DIR / f"{rid}.json").read_text())
+        except Exception:
+            out.append([rid, prompt, answer, False, False])
+            continue
+
+        # Canonicalize like the receipt did
+        def C(t):
+            t = unicodedata.normalize("NFKC", t or "").replace("\r\n", "\n").replace("\r", "\n")
+            t = "\n".join([ln.rstrip() for ln in t.split("\n")]); t = t.rstrip() + "\n"; return t
+        def s256(t): return hashlib.sha256(C(t).encode()).hexdigest()
+        def h256(k,t):
+            if not k: return ""
+            return hmac.new(k.encode(), C(t).encode(), hashlib.sha256).hexdigest()
+
+        sha_ok  = (s256(prompt)  == rj.get("prompt_sha256")) and (s256(answer) == rj.get("response_sha256"))
+        if pepper:
+            hmac_ok = (h256(pepper, prompt) == rj.get("prompt_hmac","")) and (h256(pepper, answer) == rj.get("response_hmac",""))
+        else:
+            hmac_ok = True  # if no pepper is supplied, treat HMAC check as N/A/True
+        out.append([rid, prompt, answer, bool(sha_ok), bool(hmac_ok)])
+    return out
 
 def verify_local(prompt_text, response_text, receipt_json, pepper):
     try:
@@ -436,6 +501,7 @@ api_ui = gr.Blocks(title="IoM-TR Receipts — Portable Governance")
 with api_ui:
     gr.Markdown("# IoM-TR Receipts — Portable Governance (NIM + HF Space)")
 
+    # ---- Ask & Verify ----
     with gr.Tab("Ask & Verify (in this Space)"):
         gr.Markdown("Type a prompt, click **Ask & Get Receipt**. The UI shows the answer, a Receipt ID, the full receipt JSON, and local verification (SHA/HMAC).")
         p2 = gr.Textbox(label="Prompt", value="In one sentence, what are portable AI governance receipts?")
@@ -448,11 +514,13 @@ with api_ui:
         link = gr.Markdown()
         ask.click(ui_chat_with_receipt, inputs=p2, outputs=[ans, st2, rid, rjson, verdict, link])
 
+    # ---- Dashboard ----
     with gr.Tab("Dashboard"):
         refresh = gr.Button("Refresh")
         grid = gr.Dataframe(interactive=False)
         refresh.click(lambda: _load_events_df(), outputs=grid)
 
+    # ---- Manual Verify ----
     with gr.Tab("Manual Verify (paste receipt)"):
         gr.Markdown("Paste the exact prompt & response you saw, and the receipt JSON from `/receipts/{id}`.")
         p = gr.Textbox(label="Prompt used")
@@ -463,6 +531,7 @@ with api_ui:
         verdict2 = gr.Textbox(label="Result")
         check.click(verify_local, inputs=[p, a, rj, pepper], outputs=verdict2)
 
+    # ---- Session Root ----
     with gr.Tab("Session Root"):
         gr.Markdown("**Create a session → add last receipt(s) → finish → get one Merkle root + proof JSON.**")
 
@@ -502,6 +571,56 @@ with api_ui:
         add_btn.click(ui_session_add_last, inputs=[sid_tb, rid], outputs=[sid_tb, sess_status, count_tb, root_tb, proof_code, proof_link])
         finish_btn.click(ui_session_finish, inputs=[sid_tb], outputs=[sid_tb, sess_status, count_tb, root_tb, proof_code, proof_link])
 
+    # ---- Verify All (one place) ----
+    with gr.Tab("Verify All (one place)"):
+        gr.Markdown(
+            "Load a session (or leave blank to use the latest), recompute the Merkle root and leaf hashes, "
+            "and optionally paste the exact prompts/answers to verify every receipt’s SHA/HMAC in one click."
+        )
+
+        sid_in   = gr.Textbox(label="Session ID (leave blank = latest)")
+        load_btn = gr.Button("Load & Recompute")
+        comp_root = gr.Textbox(label="Recomputed Merkle root", interactive=False)
+        root_match = gr.Textbox(label="Root match", interactive=False)
+        table = gr.Dataframe(headers=["receipt_id", "session_leaf", "recomputed_leaf", "match"],
+                             interactive=False, wrap=True)
+
+        proof_link2 = gr.Markdown()
+
+        def ui_load_and_recompute(sid_text: str):
+            sid = sid_text.strip() or latest_session_id()
+            if not sid:
+                return "", "No sessions found", [], ""
+            comp, msg, rows, link = recompute_session_root_and_table(sid)
+            return comp, msg, rows, link
+
+        load_btn.click(ui_load_and_recompute, inputs=sid_in, outputs=[comp_root, root_match, table, proof_link2])
+
+        gr.Markdown("### Per-turn text verification (paste the exact prompts & answers)")
+        pepper_tb = gr.Textbox(label="AUDIT_PEPPER (optional; if set in Space secrets, paste the same value)")
+        prep_btn  = gr.Button("Prepare rows for this session")
+        rows_df   = gr.Dataframe(headers=["receipt_id","prompt","answer","sha_ok","hmac_ok"],
+                                 row_count=(0, "dynamic"), datatype=["str","str","str","bool","bool"])
+
+        verify_btn = gr.Button("Verify pasted texts for all receipts")
+        rows_out   = gr.Dataframe(headers=["receipt_id","prompt","answer","sha_ok","hmac_ok"],
+                                  row_count=(0, "dynamic"), datatype=["str","str","str","bool","bool"])
+
+        def ui_prep_rows(sid_text: str):
+            sid = sid_text.strip() or latest_session_id()
+            if not sid:
+                return []
+            return make_text_rows_for_session(sid)
+
+        def ui_verify_rows(sid_text: str, rows, pepper):
+            sid = sid_text.strip() or latest_session_id()
+            if not sid:
+                return rows or []
+            return verify_text_rows(sid, rows, pepper or "")
+
+        prep_btn.click(ui_prep_rows, inputs=sid_in, outputs=rows_df)
+        verify_btn.click(ui_verify_rows, inputs=[sid_in, rows_df, pepper_tb], outputs=rows_out)
+
 # Expose the ASGI app for Spaces (Docker or Gradio SDK)
 demo = api_ui
 app  = gr.mount_gradio_app(api, api_ui, path="/")