app.py

from __future__ import annotations

import json
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from secrets import token_hex
from typing import Any
from uuid import uuid4

import gradio as gr

from verifiable_tool_invocation_flow.guarded_tool_call import guarded_tool_call
from verifiable_tool_invocation_flow.models import ExecutionRequest, PolicySnapshot, ToolManifest
from verifiable_tool_invocation_flow.resources import load_example_json
from verifiable_tool_invocation_flow.signer import ReceiptSigner
from verifiable_tool_invocation_flow.tools.demo_metadata_lookup_tool import demo_metadata_lookup_tool
from verifiable_tool_invocation_flow.validator import validate_receipt

DEFAULT_AUDIENCE = "demo-validator"
CHECK_FIELDS = [
    "schema_valid",
    "input_hash_match",
    "policy_hash_match",
    "tool_manifest_hash_match",
    "tool_input_hash_match",
    "tool_output_hash_match",
    "result_hash_match",
    "pre_execution_commitment_match",
    "policy_decision_valid",
    "signature_valid",
    "time_window_valid",
    "replay_check_performed",
    "replay_detected",
    "audience_match",
    "request_binding_match",
]


def load_json_upload(file_path: str | None) -> dict[str, Any]:
    if not file_path:
        raise gr.Error("Missing JSON file upload.")
    try:
        payload = json.loads(Path(file_path).read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise gr.Error(f"Invalid JSON upload: {exc.msg}") from exc
    if not isinstance(payload, dict):
        raise gr.Error("Uploaded JSON must be an object.")
    return payload


def load_text_upload(file_path: str | None) -> bytes:
    if not file_path:
        raise gr.Error("Missing text file upload.")
    return Path(file_path).read_bytes()


def summarize_report(report: dict[str, Any]) -> list[list[str]]:
    rows: list[list[str]] = []
    for field in CHECK_FIELDS:
        value = report.get(field)
        if isinstance(value, bool):
            status = "pass" if value else "fail"
        else:
            status = "n/a"
        rows.append([field, status])
    return rows


def validate_uploaded_receipt(
    receipt_file: str | None,
    evidence_file: str | None,
    public_key_file: str | None,
    audience: str | None,
) -> tuple[str, list[list[str]], list[str], list[str], dict[str, Any], str]:
    receipt = load_json_upload(receipt_file)
    evidence = load_json_upload(evidence_file)
    public_key = load_text_upload(public_key_file)
    report = validate_receipt(
        receipt=receipt,
        evidence_bundle=evidence,
        public_key_pem=public_key,
        audience=(audience or DEFAULT_AUDIENCE).strip() or DEFAULT_AUDIENCE,
        replay_cache_path=None,
        update_replay_cache=False,
    )
    report_path = _write_temp_json("verification_report.json", report)
    return (
        str(report["verdict"]),
        summarize_report(report),
        report.get("errors", []),
        report.get("warnings", []),
        report,
        report_path,
    )


def run_builtin_demo() -> tuple[str, list[list[str]], dict[str, Any], str, str, str, str]:
    request = _build_demo_request()
    policy = PolicySnapshot.model_validate(load_example_json("policy_snapshot.json"))
    tool_manifest = ToolManifest.model_validate(load_example_json("tool_manifest.json"))
    tool_input = load_example_json("tool_input.json")
    signer = ReceiptSigner.generate_demo()

    result = guarded_tool_call(
        request,
        policy,
        tool_manifest,
        tool_input,
        demo_metadata_lookup_tool,
        signer,
        audience=DEFAULT_AUDIENCE,
        replay_cache_path=None,
        update_replay_cache=False,
    )

    receipt_path = _write_temp_json("execution_receipt.json", result.receipt)
    evidence_path = _write_temp_json("evidence_bundle.json", result.evidence_bundle)
    report_path = _write_temp_json("verification_report.json", result.verification_report)
    public_key_path = _write_temp_bytes("demo_public_key.pem", signer.public_key_pem())

    return (
        str(result.verification_report["verdict"]),
        summarize_report(result.verification_report),
        result.verification_report,
        receipt_path,
        evidence_path,
        public_key_path,
        report_path,
    )


def _build_demo_request() -> ExecutionRequest:
    payload = load_example_json("input_request.json")
    now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
    payload.update(
        {
            "request_id": f"req-{uuid4()}",
            "execution_id": f"exec-{uuid4()}",
            "nonce": token_hex(16),
            "requested_at": now,
        }
    )
    return ExecutionRequest.model_validate(payload)


def _write_temp_json(filename: str, payload: dict[str, Any]) -> str:
    temp_dir = Path(tempfile.mkdtemp(prefix="agent-receipt-validator-"))
    target = temp_dir / filename
    target.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return str(target)


def _write_temp_bytes(filename: str, payload: bytes) -> str:
    temp_dir = Path(tempfile.mkdtemp(prefix="agent-receipt-validator-"))
    target = temp_dir / filename
    target.write_bytes(payload)
    return str(target)


with gr.Blocks() as demo:
    gr.Markdown(
        """
        # Agent Receipt Validator

        Validate signed agent execution receipts against evidence bundles. This is a demo for verifiable execution
        evidence, not a compliance certification system.
        """
    )

    with gr.Tab("Validate uploaded receipt"):
        uploaded_receipt = gr.File(label="execution_receipt.json", type="filepath")
        uploaded_evidence = gr.File(label="evidence_bundle.json", type="filepath")
        uploaded_public_key = gr.File(label="public key PEM", type="filepath")
        uploaded_audience = gr.Textbox(label="Expected audience", value=DEFAULT_AUDIENCE)
        uploaded_button = gr.Button("Validate")
        uploaded_verdict = gr.Label(label="Verdict")
        uploaded_checks = gr.Dataframe(label="Validation checks", headers=["check", "status"], interactive=False)
        uploaded_errors = gr.JSON(label="Errors")
        uploaded_warnings = gr.JSON(label="Warnings")
        uploaded_report = gr.JSON(label="Verification report")
        uploaded_report_file = gr.File(label="Download verification_report.json")

        uploaded_button.click(
            validate_uploaded_receipt,
            inputs=[uploaded_receipt, uploaded_evidence, uploaded_public_key, uploaded_audience],
            outputs=[
                uploaded_verdict,
                uploaded_checks,
                uploaded_errors,
                uploaded_warnings,
                uploaded_report,
                uploaded_report_file,
            ],
        )

    with gr.Tab("Run built-in demo"):
        demo_button = gr.Button("Generate and validate fresh demo artifacts")
        demo_verdict = gr.Label(label="Verdict")
        demo_checks = gr.Dataframe(label="Validation checks", headers=["check", "status"], interactive=False)
        demo_report = gr.JSON(label="Verification report")
        demo_receipt_file = gr.File(label="Download execution_receipt.json")
        demo_evidence_file = gr.File(label="Download evidence_bundle.json")
        demo_public_key_file = gr.File(label="Download demo_public_key.pem")
        demo_report_file = gr.File(label="Download verification_report.json")

        demo_button.click(
            run_builtin_demo,
            inputs=[],
            outputs=[
                demo_verdict,
                demo_checks,
                demo_report,
                demo_receipt_file,
                demo_evidence_file,
                demo_public_key_file,
                demo_report_file,
            ],
        )

    gr.Markdown(
        "This Space does not store uploaded files permanently. Avoid uploading confidential production receipts, "
        "private keys, API tokens, or sensitive evidence bundles."
    )


if __name__ == "__main__":
    demo.launch()