from collections.abc import Callable import os import gradio as gr from pydantic import JsonValue try: import spaces except ModuleNotFoundError: class _LocalSpaces: @staticmethod def GPU(**_kwargs: object): def decorate(function): return function return decorate spaces = _LocalSpaces() from space.contracts import MAX_BYOK_KEY_LENGTH, ProviderChoiceError, ProviderSelection from space.providers import SpaceProviderError from space.runtime import JsonRecord, SpaceRuntimeSession, create_runtime_session @spaces.GPU(duration=120) def _run_bonsai_prompt(session: SpaceRuntimeSession, prompt: str): status, outputs = submit_prompt(session, prompt) return status, *_output_values(outputs) def select_provider(raw_provider: str | None, raw_key: str) -> ProviderSelection: normalized_provider = (raw_provider or "").strip().casefold() match normalized_provider: case "byok": if not raw_key or not raw_key.strip(): raise ProviderChoiceError("A BYOK key is required.") if raw_key != raw_key.strip(): raise ProviderChoiceError( "The BYOK key cannot start or end with whitespace." ) if len(raw_key) > MAX_BYOK_KEY_LENGTH: raise ProviderChoiceError("The BYOK key exceeds the maximum length.") return ProviderSelection(provider="byok", session_key=raw_key) case "bonsai": return ProviderSelection(provider="bonsai", session_key=None) case "fixture" | "demo" | "fixture demo": return ProviderSelection(provider="fixture", session_key=None) case _: raise ProviderChoiceError("Choose Fixture demo, BYOK, or Bonsai.") def _status(selection: ProviderSelection) -> str: provider_label = { "byok": "BYOK", "bonsai": "Bonsai", "fixture": "Fixture demo", }[selection.provider] mode = "fixture" if selection.provider == "fixture" else "live" status = ( "configured live model" if mode == "live" and selection.provider != "fixture" else "local fixture runtime" ) return f"**{status} · provider: {provider_label} · ready**" def accept_provider( raw_provider: str, raw_key: str, on_provider_ready: Callable[[ProviderSelection], None] | None = None, ) -> str: selection = select_provider(raw_provider, raw_key) if on_provider_ready is not None: on_provider_ready(selection) return _status(selection) def _empty_outputs() -> JsonRecord: return { "provider": "", "model": "", "mode": "", "closed": False, "events": [], "notifications": [], "plans": [], "agent_statuses": [], "agents": [], "workflows": [], "mcp": [], "plugins": [], "artifacts": [], "evolution": [], "orchestration_markdown": "## Jane's orchestration\n\nNo delegation events yet.", "souls_markdown": "## Voice SOUL changes\n\nNo SOUL files generated yet.", } def _snapshot_outputs(session: SpaceRuntimeSession | None) -> JsonRecord: if session is None: return _empty_outputs() snapshot = session.snapshot() return snapshot.as_outputs() def submit_prompt( session: SpaceRuntimeSession | None, prompt: str ) -> tuple[str, JsonRecord]: if session is None: return "**Runtime unavailable:** choose Fixture demo, BYOK, or Bonsai first.", _empty_outputs() previous_notification_count = len(session.snapshot().notifications) try: snapshot = session.prompt_sync(prompt) except (OSError, RuntimeError, SpaceProviderError, ValueError) as error: return f"**Prompt error:** {error}", _snapshot_outputs(session) if snapshot.notifications[previous_notification_count:]: return ( "**Runtime degraded:** provider error recorded; inspect runtime state.", snapshot.as_outputs(), ) return ( "**Runtime ready:** prompt processed through the event bus.", snapshot.as_outputs(), ) def _cleanup_session(session: SpaceRuntimeSession | None) -> None: if session is not None: session.close_sync() def _output_values( value: JsonRecord, ) -> tuple[ JsonRecord, list[JsonValue], list[JsonValue], list[JsonValue], list[JsonValue], list[JsonValue], JsonRecord, list[JsonValue], str, str, str, str, str, str, ]: artifacts = _list_value(value.get("artifacts")) files = { str(item.get("name")): str(item.get("content", "")) for item in artifacts if isinstance(item, dict) } return ( { "provider": value.get("provider", ""), "model": value.get("model", ""), "mode": value.get("mode", ""), }, _list_value(value.get("events")), _list_value(value.get("plans")), _list_value(value.get("agents")), _list_value(value.get("workflows")), _list_value(value.get("artifacts")), {"mcp": value.get("mcp", []), "plugins": value.get("plugins", [])}, _list_value(value.get("evolution")), str(value.get("orchestration_markdown", "")), files.get("PLAN.md", "### PLAN.md\n\nNo PLAN.md generated."), files.get("BACKLOG.md", "### BACKLOG.md\n\nNo BACKLOG.md generated."), files.get("TODO.md", "### TODO.md\n\nNo TODO.md generated."), files.get("workflows/space-plan/workflow.md", "### workflow.md\n\nNo workflow generated."), str(value.get("souls_markdown", "## Voice SOUL changes\n\nNo SOUL files generated.")), ) def create_app( on_provider_ready: Callable[[ProviderSelection], None] | None = None, ) -> gr.Blocks: with gr.Blocks(title="Kateto Space") as app: _ = gr.Markdown( "# Kateto\nChoose Fixture demo for a local, no-network run, or connect a provider." ) _ = gr.Markdown( "**Status:** choose Fixture demo for local evidence, or BYOK/Bonsai for a real model connection." ) provider = gr.Radio( choices=["Fixture demo", "BYOK", "Bonsai"], value="Fixture demo", label="Provider", info="Fixture demo runs locally. BYOK and Bonsai remain available for live mode.", elem_id="provider-choice", ) key = gr.Textbox( label="OpenRouter key", info=f"Session-only; maximum {MAX_BYOK_KEY_LENGTH} characters.", type="password", max_length=MAX_BYOK_KEY_LENGTH, visible=False, elem_id="byok-key", ) submit = gr.Button("Continue", variant="primary", elem_id="provider-submit") status = gr.Markdown( "Select Fixture demo, BYOK, or Bonsai to continue.", elem_id="provider-status" ) session_state = gr.State(value=None, delete_callback=_cleanup_session) prompt = gr.Textbox( label="Prompt", placeholder="Ask the team to plan work", visible=False, elem_id="prompt", ) prompt_submit = gr.Button("Send", visible=False, elem_id="prompt-submit") bonsai_submit = gr.Button( "Send · ZeroGPU", visible=False, elem_id="bonsai-prompt-submit" ) _ = gr.Markdown("## Live orchestration evidence") provider_model = gr.JSON( value={}, label="Provider / model", elem_id="provider-model" ) timeline = gr.JSON( value=[], label="Event timeline · name + source", elem_id="event-timeline" ) plans = gr.JSON(value=[], label="Plans produced", elem_id="plans") agents = gr.JSON( value=[], label="Agents / voices · status + actions", elem_id="agents" ) workflows = gr.JSON( value=[], label="Workflow tree · phase + task + checkpoints", elem_id="workflows", ) artifacts = gr.JSON( value=[], label="Created files · local storage", elem_id="artifacts" ) integrations = gr.JSON( value={}, label="MCP / plugin status", elem_id="integrations" ) evolution = gr.JSON( value=[], label="Evolution / work ledger", elem_id="evolution" ) orchestration = gr.Markdown("## Jane's orchestration\n\nRun a prompt to see delegation events.") with gr.Accordion("Generated files · rendered Markdown", open=True): plan_file = gr.Markdown("### PLAN.md\n\nRun a prompt to generate the plan.") backlog_file = gr.Markdown("### BACKLOG.md\n\nWaiting for triangulation.") todo_file = gr.Markdown("### TODO.md\n\nWaiting for update.") workflow_file = gr.Markdown("### workflow.md\n\nWaiting for workflow execution.") souls = gr.Markdown("## Voice SOUL changes\n\nWaiting for generated personality files.") outputs = ( provider_model, timeline, plans, agents, workflows, artifacts, integrations, evolution, orchestration, plan_file, backlog_file, todo_file, workflow_file, souls, ) def reveal_key(choice: str | None) -> gr.Textbox: return gr.Textbox(visible=choice == "BYOK") def submit_choice( choice: str | None, session_key: str, previous_session: SpaceRuntimeSession | None, ): try: selection = select_provider(choice or "", session_key) except ProviderChoiceError as error: return ( f"**Provider selection error:** {error}", gr.Textbox(visible=choice == "BYOK"), previous_session, gr.Textbox(visible=False), gr.Button(visible=False), gr.Button(visible=False), *_output_values(_empty_outputs()), ) try: selected_session = create_runtime_session(selection) except SpaceProviderError as error: return ( f"**Provider error:** {error}", gr.Textbox(visible=choice == "BYOK"), previous_session, gr.Textbox(visible=False), gr.Button(visible=False), gr.Button(visible=False), *_output_values(_empty_outputs()), ) _cleanup_session(previous_session) return ( accept_provider( selection.provider, selection.session_key or "", on_provider_ready ), gr.Textbox(value="", visible=False), selected_session, gr.Textbox(visible=True), gr.Button(visible=selection.provider != "bonsai"), gr.Button(visible=selection.provider == "bonsai"), *_output_values(_snapshot_outputs(selected_session)), ) def submit_prompt_callback( session: SpaceRuntimeSession | None, value: str, ): status_value, state = submit_prompt(session, value) return status_value, *_output_values(state) _ = provider.change(reveal_key, inputs=provider, outputs=key) _ = submit.click( submit_choice, inputs=[provider, key, session_state], outputs=[status, key, session_state, prompt, prompt_submit, bonsai_submit, *outputs], ) _ = prompt_submit.click( submit_prompt_callback, inputs=[session_state, prompt], outputs=[status, *outputs], ) _ = bonsai_submit.click( _run_bonsai_prompt, inputs=[session_state, prompt], outputs=[status, *outputs], ) return app app = create_app() def _list_value(value: JsonValue) -> list[JsonValue]: return value if isinstance(value, list) else [] if __name__ == "__main__": _ = app.launch()