Blame for app.py

from typing import Any, Dict, List

import gradio as gr

from src.agent import run_agent, run_agent_iter
from src.audit import read_audit_records
from src.demo_presentation import (
    APP_CSS,
    ARCHITECTURE_MARKDOWN,
    AUTHORITY_HTML,
    BUSINESS_CASE_HTML,
    HERO_HTML,
    SECURITY_MARKDOWN,
    TOOL_BELT_HTML,
    idle_activity_html,
    render_activity,
    render_summary,
    starting_activity_html,
)
from src.schemas import AgentRunEvent, AgentRunResult


def get_run_audit_records(
    call_ids: List[str],
) -> List[Dict[str, Any]]:
    """Return audit records belonging only to the current agent run."""

    if not call_ids:
        return []

    records = read_audit_records()

    matching_records = [
        record
        for record in records
        if record.call.call_id in call_ids
    ]

    return [
        record.model_dump(mode="json")
        for record in matching_records
    ]


def handle_request(user_request: str):
    """Preserve the original programmatic callback contract."""

    if not user_request or not user_request.strip():
        return (
            "Please enter a request.",
            [],
            [],
            [],
        )

    try:
        result = run_agent(user_request.strip())

        tool_calls = [
            call.model_dump(mode="json")
            for call in result.tool_calls
        ]
        tool_results = [
            tool_result.model_dump(mode="json")
            for tool_result in result.tool_results
        ]
        call_ids = [
            call.call_id
            for call in result.tool_calls
        ]

        return (
            result.final_answer,
            tool_calls,
            tool_results,
            get_run_audit_records(call_ids),
        )

    except Exception:
        return (
            (
                "Agent execution failed. "
                "Please review the server logs for details."
            ),
            [],
            [],
            [],
        )


def _current_calls(events: list[AgentRunEvent]) -> list[Dict[str, Any]]:
    calls: dict[str, Dict[str, Any]] = {}

    for event in events:
        if event.call is not None:
            calls[event.call.call_id] = event.call.model_dump(mode="json")

    return list(calls.values())


def _current_results(events: list[AgentRunEvent]) -> list[Dict[str, Any]]:
    results: dict[str, Dict[str, Any]] = {}

    for event in events:
        if event.result is not None:
            results[event.result.call_id] = event.result.model_dump(mode="json")

    return list(results.values())


def _stream_outputs(
    events: list[AgentRunEvent],
    final_result: AgentRunResult | None = None,
    complete: bool = False,
):
    calls = (
        [call.model_dump(mode="json") for call in final_result.tool_calls]
        if final_result is not None
        else _current_calls(events)
    )
    results = (
        [result.model_dump(mode="json") for result in final_result.tool_results]
        if final_result is not None
        else _current_results(events)
    )
    call_ids = [call["call_id"] for call in calls]
    audits = get_run_audit_records(call_ids) if complete else []

    answer = (
        final_result.final_answer
        if final_result is not None
        else "*The final business answer will appear after controlled execution completes.*"
    )

    return (
        render_activity(events, complete=complete),
        answer,
        render_summary(final_result),
        calls,
        results,
        audits,
    )


def stream_request(user_request: str):
    """Stream real model/application boundary events into the Gradio demo."""

    if not user_request or not user_request.strip():
        yield (
            idle_activity_html(),
            "Please enter a request.",
            render_summary(None),
            [],
            [],
            [],
        )
        return

    events: list[AgentRunEvent] = []

    yield (
        starting_activity_html(),
        "*The final business answer will appear after controlled execution completes.*",
        render_summary(None),
        [],
        [],
        [],
    )

    try:
        for event in run_agent_iter(user_request.strip()):
            events.append(event)
            final_result = event.final_result
            complete = final_result is not None

            yield _stream_outputs(
                events=events,
                final_result=final_result,
                complete=complete,
            )

    except Exception:
        error_event = AgentRunEvent(
            event="max_rounds_reached",
            message=(
                "The request could not complete. The application stopped the run "
                "without exposing internal server details."
            ),
        )
        events.append(error_event)

        yield (
            render_activity(events, complete=True),
            (
                "Agent execution failed. Please review the server logs for details."
            ),
            render_summary(None),
            _current_calls(events),
            _current_results(events),
            [],
        )


FLAGSHIP_REQUEST = (
    "We may ship equipment to Japan. Find the Electronics items currently in "
    "inventory, calculate an accessory budget of $347 per matching item, and "
    "give me Japan's capital, region, and income classification."
)


with gr.Blocks(
    title="Governed Tool-Using Agent",
    css=APP_CSS,
) as demo:
    with gr.Column(elem_classes=["agent-shell"]):
        gr.HTML(HERO_HTML)
        gr.HTML(BUSINESS_CASE_HTML)

        gr.Markdown("## The approved tool belt")
        gr.HTML(TOOL_BELT_HTML)

        gr.Markdown("## Who controls what?")
        gr.HTML(AUTHORITY_HTML)

        gr.Markdown("## Try a controlled business request")
        request_box = gr.Textbox(
            label="Operations request",
            value=FLAGSHIP_REQUEST,
            placeholder="Ask for work that may require one or more approved tools.",
            lines=4,
        )

        gr.Examples(
            examples=[
                [FLAGSHIP_REQUEST],
                ["What is 347 multiplied by 29?"],
                ["What electronics are currently in inventory?"],
                ["What is the capital, region, and income level of Japan?"],
                [
                    (
                        "How many Electronics items are in inventory, "
                        "and what is 347 multiplied by that number?"
                    )
                ],
            ],
            inputs=request_box,
        )

        run_button = gr.Button(
            "Run controlled request",
            variant="primary",
        )

        activity_output = gr.HTML(idle_activity_html())

        with gr.Tabs():
            with gr.Tab("Business Result"):
                gr.Markdown(
                    "The answer below is produced only after requested capabilities "
                    "pass through the application-controlled boundary."
                )
                final_answer = gr.Markdown(
                    "*Run a request to produce a business answer.*"
                )
                summary_output = gr.HTML(render_summary(None))

            with gr.Tab("Engineering Audit"):
                gr.Markdown(
                    "These are the structured requests, normalized results, and "
                    "current-run audit records behind the business-facing view."
                )
                tool_calls_output = gr.JSON(label="Model-proposed Tool Calls")
                tool_results_output = gr.JSON(label="Controlled Tool Results")
                audit_output = gr.JSON(label="Current Run Audit Records")

            with gr.Tab("Security & Failure Semantics"):
                gr.Markdown(SECURITY_MARKDOWN)

            with gr.Tab("Architecture"):
                gr.Markdown(ARCHITECTURE_MARKDOWN)

        run_button.click(
            fn=stream_request,
            inputs=request_box,
            outputs=[
                activity_output,
                final_answer,
                summary_output,
                tool_calls_output,
                tool_results_output,
                audit_output,
            ],
            show_progress="hidden",
        )


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