from __future__ import annotations import json from pathlib import Path import gradio as gr from lib.packet_builder import build_packet from lib.metric_analysis import METRIC_HEADERS from lib.holographic_compare import compare_packets from lib.continuity_audit import audit_packet from lib.source_lineage import lineage_summary ROOT = Path(__file__).resolve().parent SCHEMA_DIR = ROOT / "schemas" EVENT_HEADERS = [ "sequence_index", "event_type", "event_time", "observed_time", "description", "state", "evidence_basis (; separated)", "corrects_event_id", "uncertainty", "event_id (optional)", "predecessor_event_ids (; separated)", ] CLAIM_HEADERS = [ "claim_level", "statement", "basis (; separated)", "state", "uncertainty", "human_review_required", "claim_id (optional)", ] def generate_packet( created_at, display_name, platform, handle_or_pseudonym, account_url, account_created, followers, followers_observed_at, attribution_preference, consent_state, consent_scope, attribution_name, creator_exact_description, recurring_subjects, criticized_institutions_or_conduct, modes, formats, posting_cadence, subject_change_near_event, baseline_start, baseline_end, typical_impressions, typical_engagement_rate, prior_high_reach_examples, follower_delivery_baseline, notification_baseline, verification_state, subscription_state, events, metric_receipts, uploaded_files, asset_source_surface, asset_source_url, asset_creator_description, asset_date_represented, asset_redaction_state, asset_public_status, asset_linked_event_ids, asset_metadata_json, claims, creator_reported_context, controls, source_return_requests, loop_state, closure_blockers, evidence_to_reduce_uncertainty, repair_deltas, next_reviewer_targets, ): try: result = build_packet( { "created_at": created_at, "display_name": display_name, "platform": platform, "handle_or_pseudonym": handle_or_pseudonym, "account_url": account_url, "account_created": account_created, "followers": followers, "followers_observed_at": followers_observed_at, "attribution_preference": attribution_preference, "consent_state": consent_state, "consent_scope": consent_scope, "attribution_name": attribution_name, "creator_exact_description": creator_exact_description, "recurring_subjects": recurring_subjects, "criticized_institutions_or_conduct": criticized_institutions_or_conduct, "modes": modes, "formats": formats, "posting_cadence": posting_cadence, "subject_change_near_event": subject_change_near_event, "baseline_start": baseline_start, "baseline_end": baseline_end, "typical_impressions": typical_impressions, "typical_engagement_rate": typical_engagement_rate, "prior_high_reach_examples": prior_high_reach_examples, "follower_delivery_baseline": follower_delivery_baseline, "notification_baseline": notification_baseline, "verification_state": verification_state, "subscription_state": subscription_state, "events": events, "metric_receipts": metric_receipts, "uploaded_files": uploaded_files, "asset_source_surface": asset_source_surface, "asset_source_url": asset_source_url, "asset_creator_description": asset_creator_description, "asset_date_represented": asset_date_represented, "asset_redaction_state": asset_redaction_state, "asset_public_status": asset_public_status, "asset_linked_event_ids": asset_linked_event_ids, "asset_metadata_json": asset_metadata_json, "claims": claims, "creator_reported_context": creator_reported_context, "controls": controls, "source_return_requests": source_return_requests, "loop_state": loop_state, "closure_blockers": closure_blockers, "evidence_to_reduce_uncertainty": evidence_to_reduce_uncertainty, "repair_deltas": repair_deltas, "next_reviewer_targets": next_reviewer_targets, }, schema_dir=SCHEMA_DIR, ) manifest = result["run_manifest"] warnings = manifest.get("sensitive_data_warnings", []) warning_text = "\n".join(f"- {w}" for w in warnings) if warnings else "- None detected by the limited text/filename scanner." status = f""" ## Packet generated — `{result['capsule']['capsule_id']}` - ZIP SHA-256: `{result['zip_sha256']}` - Loop state: `{result['capsule']['loop_state']}` - Schema validation: **PASS** - Semantic gates: **PASS** - Cross-account comparison now: **BLOCKED / HUMAN REVIEW REQUIRED** - Public exhibit now: **BLOCKED / SEPARATE REVIEW REQUIRED** ### Sensitive-data scan {warning_text} The creator controls whether this packet is shared. This output does not prove suppression, targeting, motive, intent, or executive direction. """ preview = { "capsule_id": result["capsule"]["capsule_id"], "loop_state": result["capsule"]["loop_state"], "observed_facts": result["dpio_read"]["observed_facts"], "execution_order_receipt": result["dpio_read"]["execution_order_receipt"], "causal_families": result["dpio_read"]["causal_families"], "frozen_discriminator_predictions": result["dpio_read"]["frozen_discriminator_predictions"], "competing_hypotheses": result["dpio_read"]["competing_hypotheses"], "minimum_cut_candidates": result["dpio_read"]["minimum_cut_candidates"], "source_return_requests": result["dpio_read"]["source_return_requests"], "claim_ceiling": result["dpio_read"]["claim_ceiling"], } return status, preview, result["zip_path"], result["sidecar_path"] except Exception as exc: return f"## MUST STOP\n\nPacket generation failed closed:\n\n```text\n{exc}\n```", None, None, None def run_holographic_comparison(packet_files, human_review_approved): try: paths = packet_files if isinstance(packet_files, list) else ([packet_files] if packet_files else []) result = compare_packets(paths, bool(human_review_approved)) c = result["comparison"] status = f"""## Holographic comparison generated - Cases: **{c['case_count']}** - Median post/baseline ratio: `{c['normalized_distribution']['median_post_to_baseline_ratio']}` - Loop state: `{c['loop_state']}` - SHA-256: `{result['sha256']}` H3 attribution, executive knowledge, and intent remain **BLOCKED_PENDING_SOURCE_RETURN**. Publication requires a separate review. """ return status, c, result["zip_path"], result["sidecar_path"] except Exception as exc: return f"## MUST STOP\n\n```text\n{exc}\n```", None, None, None def run_continuity_audit(packet_file): try: if not packet_file: raise ValueError("Upload a creator packet ZIP.") result = audit_packet(packet_file) return f"## Continuity state: `{result['state']}`", result except Exception as exc: return f"## MUST STOP\n\n```text\n{exc}\n```", None CSS = """ .gradio-container {max-width: 1320px !important;} .boundary {border: 1px solid #5d6673; border-radius: 12px; padding: 14px;} """ with gr.Blocks(title="Substrate Creator Distribution Evidence Intake", delete_cache=(3600, 3600)) as demo: gr.Markdown( """ # Substrate Creator Distribution Evidence Intake — DPIO + Holographic Prototype v0.2.0 **What pressure concealed, we make legible.** Generate a source-bound creator trace capsule, platform-displayed metric receipt ledger, bounded causal-arc packet, deterministic DPIO procedural read, minimum-cut worksheet, continuity audit, governed holographic comparison, and deterministic ZIP export. > **Boundary:** This prototype preserves receipts and causal order. It does not scrape platforms, call external AI, retain a database, publish submissions, or automatically prove suppression, targeting, theft, motive, intent, or executive direction. """, elem_classes=["boundary"], ) with gr.Accordion("Required privacy warning", open=True): gr.Markdown( """ Do not upload passwords, API keys, access tokens, exact private addresses, unredacted private messages, minors' personal data, or confidential legal/medical records. Images and videos are **not OCR-scanned**; creator redaction remains mandatory. Maximum: **100 MB per file / 500 MB total**. """ ) with gr.Tabs(): with gr.Tab("1 — Identity & Consent"): with gr.Row(): created_at = gr.Textbox(label="Run timestamp (ISO-8601; blank = current UTC)", placeholder="2026-08-02T19:15:00Z") platform = gr.Textbox(label="Platform", value="X") handle_or_pseudonym = gr.Textbox(label="Handle or pseudonym") with gr.Row(): display_name = gr.Textbox(label="Display name") account_url = gr.Textbox(label="Account URL (optional)") account_created = gr.Textbox(label="Account created / age") with gr.Row(): followers = gr.Number(label="Follower count", minimum=0, precision=0) followers_observed_at = gr.Textbox(label="Follower count observed at") attribution_preference = gr.Dropdown( ["ATTRIBUTED", "PSEUDONYMOUS", "ANONYMOUS"], value="PSEUDONYMOUS", label="Attribution preference" ) with gr.Row(): consent_state = gr.Dropdown(["PENDING", "GRANTED", "WITHDRAWN", "NOT_REQUESTED"], value="PENDING", label="Consent state") consent_scope = gr.Dropdown( ["PRIVATE_PACKET_ONLY", "ATTRIBUTED_COMPARATIVE_REVIEW", "ANONYMOUS_AGGREGATE_REVIEW", "PUBLIC_EXHIBIT_CANDIDATE"], value="PRIVATE_PACKET_ONLY", label="Consent scope", ) attribution_name = gr.Textbox(label="Attribution name, when applicable") with gr.Tab("2 — Creator Topology & Baseline"): creator_exact_description = gr.Textbox( label="Creator's exact description of their work (required; preserved verbatim)", lines=5 ) with gr.Row(): recurring_subjects = gr.Textbox(label="Recurring subjects — one per line", lines=5) criticized_institutions_or_conduct = gr.Textbox(label="Institutions, systems, or conduct discussed — one per line", lines=5) with gr.Row(): modes = gr.Textbox(label="Modes — reporting, commentary, art, technical analysis, etc.", lines=4) formats = gr.Textbox(label="Formats — posts, images, video, music, software, etc.", lines=4) with gr.Row(): posting_cadence = gr.Textbox(label="Posting cadence") subject_change_near_event = gr.Textbox(label="Did subject matter change near the event?") gr.Markdown("### Distribution baseline") with gr.Row(): baseline_start = gr.Textbox(label="Baseline start") baseline_end = gr.Textbox(label="Baseline end") typical_impressions = gr.Number(label="Typical displayed impressions/views") typical_engagement_rate = gr.Number(label="Typical engagement rate") prior_high_reach_examples = gr.Textbox(label="Prior high-reach examples — one per line", lines=4) with gr.Row(): follower_delivery_baseline = gr.Textbox(label="Follower delivery baseline", lines=3) notification_baseline = gr.Textbox(label="Notification baseline", lines=3) with gr.Row(): verification_state = gr.Textbox(label="Verification state") subscription_state = gr.Textbox(label="Subscription state") with gr.Tab("3 — Temporal Events & Evidence"): gr.Markdown( """ Events are append-only. A correction must be a new `CREATOR_CORRECTION` event with `corrects_event_id`; do not replace the ancestor event. Valid states: `OBSERVED`, `CREATOR_REPORTED`, `PROVISIONAL`, `CORRECTED`, `DISPUTED`. """ ) events = gr.Dataframe( headers=EVENT_HEADERS, datatype=["number"] + ["str"] * 10, value=[ [0, "BASELINE", "", "", "Platform-displayed baseline documented.", "OBSERVED", "", "", "", "EVENT_0000", ""], [1, "REACH_CHANGE", "", "", "Platform-displayed distribution change documented.", "OBSERVED", "", "", "", "EVENT_0001", "EVENT_0000"], ], row_count=(2, "dynamic"), column_count=(11, "fixed"), label="Append-only temporal chain", ) gr.Markdown("### Platform-displayed metric receipts") gr.Markdown("Each row preserves a witnessed display state. It establishes what the platform displayed at the observation time, not unique-human count, hidden mechanism, authorization, or intent.") metric_receipts = gr.Dataframe( headers=METRIC_HEADERS, datatype=["str", "str", "str", "str", "number", "str", "str", "str", "str"], value=[ ["METRIC_0001", "POST_OR_OBJECT_1", "", "views", 1400000, "BASELINE", "ASSET_0001", "1.4M views", "Documented platform-displayed baseline"], ["METRIC_0002", "POST_OR_OBJECT_2", "", "views", 700000, "POST_BREAK", "ASSET_0002", "700K views", "Documented post-break display state"], ], row_count=(2, "dynamic"), column_count=(9, "fixed"), label="Immutable metric receipt ledger", ) uploaded_files = gr.File( label="Evidence assets (inert; no execution)", file_count="multiple", type="filepath", file_types=[".png", ".jpg", ".jpeg", ".webp", ".gif", ".mp4", ".mov", ".webm", ".csv", ".json", ".txt", ".md", ".pdf", ".zip"], ) gr.Markdown("### Common asset metadata") with gr.Row(): asset_source_surface = gr.Textbox(label="Source surface", value="Creator-provided platform screenshot or export") asset_source_url = gr.Textbox(label="Source URL (optional)") asset_date_represented = gr.Textbox(label="Date represented") asset_creator_description = gr.Textbox(label="Creator description of uploaded assets", value="Creator-provided evidence asset") with gr.Row(): asset_redaction_state = gr.Dropdown(["NOT_REQUIRED", "REDACTED", "REDACTION_REQUIRED", "BLOCKED_SENSITIVE"], value="REDACTION_REQUIRED", label="Redaction state") asset_public_status = gr.Dropdown(["PUBLIC", "PRIVATE", "UNKNOWN"], value="UNKNOWN", label="Public/private status") asset_linked_event_ids = gr.Textbox(label="Linked event IDs — one per line") asset_metadata_json = gr.Code( label="Optional per-file metadata JSON keyed by original filename", language="json", value="{}", lines=8, ) with gr.Tab("4 — Claims, DPIO & Review"): gr.Markdown( """ Automatic authority ends at L1. L2-L4 require human confirmation. L5 attributed cause and L6 intent/motive are always exported as `BLOCKED` by this prototype. """ ) claims = gr.Dataframe( headers=CLAIM_HEADERS, datatype=["str"] * 7, value=[ ["L1_DIRECT_OBSERVATION", "The platform displayed the documented metric at the recorded observation time.", "EVENT_0000", "SUPPORTED", "The receipt does not establish unique humans or internal counting method.", "false", "CLAIM_0001"], ], row_count=(1, "dynamic"), column_count=(7, "fixed"), label="Bounded claim ladder", ) with gr.Row(): creator_reported_context = gr.Textbox(label="Creator-reported context — one item per line", lines=6) controls = gr.Textbox(label="Controls and competing conditions — one item per line", lines=6) source_return_requests = gr.Textbox( label="Platform-controlled source-return requests — one per line", lines=6, value="Account-level recommendation eligibility and distribution-state history\nNotification generation, suppression, deduplication, and delivery logs\nBefore/after enforcement-state diff and restoration execution receipt", ) with gr.Row(): loop_state = gr.Dropdown( ["OPEN", "ACTIVE_REVIEW", "HELD", "STRAINED", "PARTIAL_CLOSURE", "ROUTED", "ENCAPSULATED", "REPAIRING", "REOPENED", "FALSE_CLOSURE_RISK", "CLOSED_FOR_CURRENT_SCOPE", "MUST_STOP"], value="ACTIVE_REVIEW", label="Provisional loop state", ) closure_blockers = gr.Textbox(label="Closure blockers — one per line", lines=4) with gr.Row(): evidence_to_reduce_uncertainty = gr.Textbox(label="Evidence needed to reduce uncertainty", lines=5) repair_deltas = gr.Textbox(label="Repair deltas", lines=5) next_reviewer_targets = gr.Textbox(label="Next reviewer targets", lines=5) with gr.Tab("5 — Generate Governed Packet"): gr.Markdown( """ Generation validates schemas, temporal ancestry, claim ceilings, consent, false closure, source immutability, upload limits, and a limited sensitive-data scan. Failure stops packet emission. """ ) generate = gr.Button("Generate DPIO creator packet", variant="primary", size="lg") status = gr.Markdown() preview = gr.JSON(label="DPIO read preview") with gr.Row(): zip_output = gr.File(label="Creator packet ZIP") sidecar_output = gr.File(label="Detached SHA-256 sidecar") with gr.Tab("6 — Holographic Comparison"): gr.Markdown(""" Upload at least two independently generated creator packets. The route verifies archive safety, checksum continuity, consent scope, loop state, and explicit human-review approval before normalizing cross-case phenotypes. **No motive or executive attribution is generated.** """) comparison_packets = gr.File(label="Eligible creator packet ZIPs", file_count="multiple", type="filepath", file_types=[".zip"]) comparison_human_review = gr.Checkbox(label="I have completed individual trace review and approve this bounded comparison route", value=False) compare_button = gr.Button("Generate governed holographic comparison", variant="primary") comparison_status = gr.Markdown() comparison_preview = gr.JSON(label="Comparison preview") with gr.Row(): comparison_zip = gr.File(label="Holographic comparison ZIP") comparison_sidecar = gr.File(label="Detached SHA-256 sidecar") compare_button.click(run_holographic_comparison, inputs=[comparison_packets, comparison_human_review], outputs=[comparison_status, comparison_preview, comparison_zip, comparison_sidecar]) with gr.Tab("7 — Continuity Auditor"): gr.Markdown("The auditor checks archive safety, required organs, checksum continuity, DPIO execution order, consent locks, and false-closure surfaces. It emits repair deltas without mutating the submitted packet.") audit_upload = gr.File(label="Creator packet ZIP", type="filepath", file_types=[".zip"]) audit_button = gr.Button("Run full-stack continuity audit") audit_status = gr.Markdown() audit_preview = gr.JSON(label="Continuity receipt") audit_button.click(run_continuity_audit, inputs=[audit_upload], outputs=[audit_status, audit_preview]) with gr.Tab("8 — Source Lineage"): gr.Markdown(lineage_summary(ROOT)) inputs = [ created_at, display_name, platform, handle_or_pseudonym, account_url, account_created, followers, followers_observed_at, attribution_preference, consent_state, consent_scope, attribution_name, creator_exact_description, recurring_subjects, criticized_institutions_or_conduct, modes, formats, posting_cadence, subject_change_near_event, baseline_start, baseline_end, typical_impressions, typical_engagement_rate, prior_high_reach_examples, follower_delivery_baseline, notification_baseline, verification_state, subscription_state, events, metric_receipts, uploaded_files, asset_source_surface, asset_source_url, asset_creator_description, asset_date_represented, asset_redaction_state, asset_public_status, asset_linked_event_ids, asset_metadata_json, claims, creator_reported_context, controls, source_return_requests, loop_state, closure_blockers, evidence_to_reduce_uncertainty, repair_deltas, next_reviewer_targets, ] generate.click(generate_packet, inputs=inputs, outputs=[status, preview, zip_output, sidecar_output]) if __name__ == "__main__": demo.queue(max_size=8).launch(css=CSS)