app.py

from __future__ import annotations

import json
import os
from pathlib import Path

import gradio as gr

try:
    import spaces
except ImportError:  # Local validation outside Hugging Face ZeroGPU.
    class _SpacesShim:
        @staticmethod
        def GPU(*_args, **_kwargs):
            def decorate(function):
                return function
            return decorate

    spaces = _SpacesShim()

from seed_company.engine import AutonomousCompanyEngine
from seed_company.models import CompanyEvent, utc_now
from seed_company.projections import (
    agents_frame,
    events_frame,
    opportunities_frame,
    opportunities_plot,
    org_html,
    project_markdown,
    summary,
)
from seed_company.research import collect_public_evidence
from seed_company.runtime import infer_json, initialize_model, model_status
from seed_company.store import HubStateStore

BASE_DIR = Path(__file__).parent
CSS = (BASE_DIR / "assets" / "style.css").read_text(encoding="utf-8")
LOCAL_ROOT = Path(os.getenv("SEED_LOCAL_ROOT", "/tmp/seed-autonomous-company"))
MEMORY_REPO = os.getenv("SEED_MEMORY_REPO", "Italianhype/seed-company-memory")
TRIGGER_SECRET = os.getenv("SEED_TRIGGER_SECRET", "")
INTERVAL_MINUTES = int(os.getenv("SEED_INTERVAL_MINUTES", "120"))

# ZeroGPU's CUDA emulation allows model placement at module startup. Real GPU is
# allocated only while a @spaces.GPU function is executing.
initialize_model()

store = HubStateStore(
    local_root=LOCAL_ROOT,
    hub_enabled=True,
    repo_id=MEMORY_REPO,
    token=os.getenv("HF_TOKEN"),
)
engine = AutonomousCompanyEngine(
    store=store,
    llm=infer_json,
    research=collect_public_evidence,
    trigger_secret=TRIGGER_SECRET,
    interval_minutes=INTERVAL_MINUTES,
)


def _fmt_time(value) -> str:
    return value.strftime("%Y-%m-%d %H:%M UTC") if value else "Not yet"


def _hero(state) -> str:
    active = state.status == "running"
    runtime = "AUTONOMOUS OPERATIONS ACTIVE" if active else "ORGANIZATION PAUSED"
    return f"""
    <section class="seed-hero">
      <div class="seed-kicker">OPEN-SOURCE AUTONOMOUS SOFTWARE COMPANY</div>
      <h1>{state.organization_name}</h1>
      <p>{state.mission}</p>
      <div class="seed-live"><span class="seed-dot"></span>{runtime}</div>
      <div class="seed-subline">Real evidence · open-weight inference · candidate code · independent gates · Hub persistence</div>
    </section>
    """


def _kpis(state) -> str:
    cards = summary(state)
    values = (
        ("Status", cards["status"]),
        ("Autonomous cycle", cards["cycle"]),
        ("Active roles", cards["roles"]),
        ("Organization", cards["version"]),
        ("Last cycle", cards["last_cycle"]),
        ("Next due", cards["next_due"]),
    )
    return '<div class="kpi-grid">' + "".join(
        f'<div class="kpi-card"><div class="kpi-label">{label}</div><div class="kpi-value">{value}</div></div>'
        for label, value in values
    ) + "</div>"


def _command(state) -> str:
    project = state.active_project
    product = project.name if project else "First autonomous cycle pending"
    artifact = project.latest_artifact if project and project.latest_artifact else "None yet"
    return (
        f"### Current strategic focus\n{state.current_focus}\n\n"
        f"### Operating state\n"
        f"- **Model:** `{state.model_id}`\n"
        f"- **Model runtime:** `{model_status()}`\n"
        f"- **Memory repository:** `{MEMORY_REPO}`\n"
        f"- **Persistence:** `{state.last_persistence_status}`\n"
        f"- **Last trigger:** `{state.last_trigger}`\n"
        f"- **Active product:** {product}\n"
        f"- **Latest generated artifact:** `{artifact}`"
    )


def _governance(state) -> str:
    secret_status = "configured" if TRIGGER_SECRET else "not configured"
    token_status = "configured" if os.getenv("HF_TOKEN") else "missing — Hub persistence will use local fallback"
    return f"""
### Immutable operating boundaries

1. Generated candidates never overwrite the stable Space runtime.
2. Every cycle writes a structured audit event and stores its artifacts.
3. Candidate Python must compile and cannot import blocked high-risk modules.
4. Tester and Compliance Reviewer must approve a candidate before promotion.
5. New positions require a mission, reporting line and finite probation period.
6. The company may pause, recover state and reject its own work.
7. The runtime does not control payments, contracts or external credentials.

### Configuration

- Trigger secret: **{secret_status}**
- Hugging Face token: **{token_status}**
- Autonomous interval target: **{INTERVAL_MINUTES} minutes**
- Last cycle: **{_fmt_time(state.last_cycle_at)}**
- Next due: **{_fmt_time(state.next_due_at)}**

### No paid Scheduled Jobs

This package contains no Hugging Face Scheduled Job and requires no prepaid Job credit. To work while nobody is viewing the Space, a free external HTTP wake-up must call the `autonomous_cycle` Gradio API endpoint. The wake-up stores no company data; all state and artifacts remain on Hugging Face.
"""


def render_all():
    state = engine.get_state()
    return (
        _hero(state),
        _kpis(state),
        _command(state),
        events_frame(state, 30),
        org_html(state),
        agents_frame(state),
        opportunities_plot(state),
        opportunities_frame(state),
        project_markdown(state),
        "\n".join(f"- {lesson}" for lesson in reversed(state.lessons[-20:])) or "No lessons recorded yet.",
        events_frame(state, 300),
        _governance(state),
    )


@spaces.GPU(duration=240)
def autonomous_cycle(secret: str = "", force: bool = False) -> str:
    """Public API used by the UI and a free external wake-up.

    Scheduled wake-ups run only when a cycle is due and recover at most two missed
    intervals. The UI can force one immediate cycle for testing or supervision.
    """
    if TRIGGER_SECRET and secret != TRIGGER_SECRET:
        return json.dumps({"status": "rejected", "reason": "invalid trigger secret"}, indent=2)
    state = engine.get_state()
    cycles_to_run = 1 if force else state.due_cycles(INTERVAL_MINUTES, max_catch_up=2)
    if cycles_to_run == 0:
        return json.dumps(
            {
                "status": "not_due",
                "cycle": state.cycle,
                "next_due_at": state.next_due_at.isoformat() if state.next_due_at else None,
            },
            indent=2,
        )
    results = [
        engine.run_cycle(trigger="manual" if force else "external_wake_up", supplied_secret=secret)
        for _ in range(cycles_to_run)
    ]
    return json.dumps({"status": "completed", "cycles_run": cycles_to_run, "results": results}, indent=2, ensure_ascii=False)


def pause_company() -> str:
    state = engine.get_state()
    state.status = "paused"
    state.events.append(
        CompanyEvent(
            cycle=state.cycle,
            actor="human_control",
            event_type="organization_paused",
            summary="External control paused new autonomous cycles.",
            severity="warning",
        )
    )
    store.save(state)
    return "Organization paused."


def resume_company() -> str:
    state = engine.get_state()
    state.status = "running"
    state.events.append(
        CompanyEvent(
            cycle=state.cycle,
            actor="human_control",
            event_type="organization_resumed",
            summary="External control resumed autonomous cycles.",
            severity="success",
        )
    )
    state.updated_at = utc_now()
    store.save(state)
    return "Organization resumed."


def reset_company(confirm: str) -> str:
    if confirm.strip().upper() != "RESET SEED":
        return "Reset rejected. Type RESET SEED exactly."
    store.reset()
    return "Company state reset."


with gr.Blocks(title="SEED Autonomous Unicorn Company") as demo:
    gr.HTML(f"<style>{CSS}</style>")
    hero = gr.HTML()
    kpis = gr.HTML()

    with gr.Row():
        trigger_secret_input = gr.Textbox(
            label="Trigger secret",
            type="password",
            placeholder="Required only when SEED_TRIGGER_SECRET is configured",
            scale=2,
        )
        force_cycle = gr.Checkbox(label="Force immediate cycle", value=True, scale=1)
        run_button = gr.Button("Run real autonomous cycle", variant="primary", scale=1)
        refresh_button = gr.Button("Refresh", scale=1)

    cycle_result = gr.Code(label="Latest cycle result", language="json", interactive=False)

    with gr.Tabs():
        with gr.Tab("Command Center"):
            command = gr.Markdown()
            gr.Markdown("### Latest real operating events")
            recent_events = gr.Dataframe(interactive=False, wrap=True)

        with gr.Tab("Organization"):
            org = gr.HTML()
            agents = gr.Dataframe(interactive=False, wrap=True)

        with gr.Tab("Opportunity Portfolio"):
            opportunity_plot_component = gr.Plot()
            opportunities = gr.Dataframe(interactive=False, wrap=True)

        with gr.Tab("Product & Code"):
            project = gr.Markdown()
            gr.Markdown(
                f"Generated product files and review reports are persisted under `artifacts/cycle-XXXX/` "
                f"inside the Dataset repository `{MEMORY_REPO}`."
            )

        with gr.Tab("Self-Improvement"):
            gr.Markdown("### Lessons retained across cycles")
            lessons = gr.Markdown()

        with gr.Tab("Audit Log"):
            audit = gr.Dataframe(interactive=False, wrap=True)

        with gr.Tab("Governance & Setup"):
            governance = gr.Markdown()
            with gr.Row():
                pause_button = gr.Button("Pause company")
                resume_button = gr.Button("Resume company")
            reset_text = gr.Textbox(label="Type RESET SEED to erase the current company state")
            reset_button = gr.Button("Reset company", variant="stop")
            control_status = gr.Textbox(label="Control status", interactive=False)

    outputs = [
        hero,
        kpis,
        command,
        recent_events,
        org,
        agents,
        opportunity_plot_component,
        opportunities,
        project,
        lessons,
        audit,
        governance,
    ]

    demo.load(render_all, outputs=outputs)
    refresh_button.click(render_all, outputs=outputs, show_progress="hidden")
    run_button.click(
        autonomous_cycle,
        inputs=[trigger_secret_input, force_cycle],
        outputs=cycle_result,
        api_name="autonomous_cycle",
        concurrency_limit=1,
    ).then(render_all, outputs=outputs, show_progress="hidden")
    pause_button.click(pause_company, outputs=control_status).then(render_all, outputs=outputs)
    resume_button.click(resume_company, outputs=control_status).then(render_all, outputs=outputs)
    reset_button.click(reset_company, inputs=reset_text, outputs=control_status).then(render_all, outputs=outputs)

    timer = gr.Timer(value=30, active=True)
    timer.tick(render_all, outputs=outputs, show_progress="hidden")


if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch(server_name="0.0.0.0", server_port=7860)