Resurrect the Live interface UI

composer/cli/live_autoprove.py

@@ -0,0 +1,39 @@
+"""Entry point for the auto-prove pipeline — live-display mode.
+
+Renders the pipeline as a single inline ``rich.live.Live`` region
+showing the tree of active agents (one root per phase task, sub-agents
+nested below). Scrollback above the Live region captures phase
+boundaries, prover lifecycle terminals, and per-rule analysis events.
+
+Sister to ``tui-autoprove`` (Textual full-screen TUI) and
+``console-autoprove`` (plain ``print``).
+"""
+
+import asyncio
+
+import composer.bind as _
+
+from composer.diagnostics.timing import RunSummary
+from composer.spec.source.autoprove_common import _entry_point
+from composer.ui.autoprove_live import AutoProveLiveHandler
+
+
+async def _main() -> int:
+    summary = RunSummary()
+    async with _entry_point(summary) as run:
+        async with AutoProveLiveHandler() as handler:
+            result = await run(handler.make_handler)
+        print(f"\n{'=' * 60}")
+        print(summary.format())
+        print(f"\n  Components:  {result.n_components}")
+        print(f"  Properties:  {result.n_properties}")
+        if result.failures:
+            print(f"  Failures:    {len(result.failures)}")
+            for f in result.failures:
+                print(f"    - {f}")
+        print(f"{'=' * 60}")
+        return 0
+
+
+def main() -> int:
+    return asyncio.run(_main())

composer/io/graph_runner.py

@@ -24,6 +24,51 @@
 from langchain_core.runnables import RunnableConfig
 
 
+def _normalize_updates_payload(payload: dict[str, Any]) -> dict:
+    """Flatten langgraph's mixed-mode tools-node payload to ``dict[node, dict]``.
+
+    When a ``ToolNode`` batch contains tools whose returns differ in shape
+    (some return raw ``ToolMessage`` / ``str`` / dicts; some return
+    ``Command(update=...)`` with extra state-channel deltas), langgraph's
+    ``_combine_tool_outputs`` emits the node's value as a *list* of separate
+    updates instead of a single merged dict — see
+    ``langgraph/prebuilt/tool_node.py``. Every downstream ``log_state_update``
+    handler in this codebase expects ``dict[node, dict_with_messages]`` and
+    silently drops the list shape (either via ``isinstance(update, dict)``
+    guards or ``"messages" in update`` membership checks, which return
+    ``False`` against a list of dicts/Commands). The result is that any
+    ToolMessage produced by a parallel batch with at least one
+    ``Command``-returning tool never reaches the renderer — the tool widgets
+    stay "pending" until something else forces a redraw.
+
+    Merge here so handlers stay simple. ``messages`` lists are concatenated;
+    other state keys retain last-write-wins semantics (matches what
+    langgraph's reducers would do for non-message channels, which the
+    handlers in this codebase don't introspect for ingest purposes).
+    """
+    out: dict[str, Any] = {}
+    for node_name, value in payload.items():
+        if isinstance(value, dict) or not isinstance(value, list):
+            out[node_name] = value
+            continue
+        merged: dict[str, Any] = {}
+        for item in value:
+            d = item.update if isinstance(item, Command) and isinstance(item.update, dict) else item
+            if not isinstance(d, dict):
+                continue
+            for k, v in d.items():
+                if (
+                    k == "messages"
+                    and isinstance(merged.get(k), list)
+                    and isinstance(v, list)
+                ):
+                    merged[k] = [*merged[k], *v]
+                else:
+                    merged[k] = v
+        out[node_name] = merged
+    return out
+
+
 class SinkProtocol(Protocol):
     """Write-only event sink.  Synchronous — must not block."""
     def __call__(self, event: GraphEvents) -> None:
@@ -111,29 +156,13 @@ async def run_graph[H, S: StateLike, I: StateLike, C: StateLike | None](
                             graph_input = Command(resume=human_response)
                             interrupted = True
                             break
-                        elif ty == "custom":
-                            event_sink(
-                                CustomUpdate(payload, thread_id=tid, checkpoint_id=curr_checkpoint) # pyright: ignore[reportPossiblyUnboundVariable]
-                            )
-                        else:
-                            assert ty == "updates"
-                            if "__interrupt__" in payload:
-                                assert human_handler is not None
-                                if "configurable" in curr_config and "checkpoint_id" in curr_config["configurable"]:
-                                    del curr_config["configurable"]["checkpoint_id"]
-                                interrupt_data = cast(H, payload["__interrupt__"][0].value)
-                                curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
-                                human_response = await human_handler(interrupt_data, curr_state)
-                                graph_input = Command(resume=human_response)
-                                interrupted = True
-                                break
-                            event_sink(
-                                StateUpdate(
-                                    payload, thread_id=tid
-                                )
+                        event_sink(
+                            StateUpdate(
+                                _normalize_updates_payload(payload), thread_id=tid
                             )
-                    if interrupted:
-                        continue
+                        )
+                if interrupted:
+                    continue
 
                 result_state = (await graph.aget_state({"configurable": {"thread_id": tid}})).values
                 return cast(S, result_state)

composer/ui/autoprove_live.py

@@ -0,0 +1,205 @@
+"""Live-display handler for the auto-prove pipeline.
+
+Drives an inline ``rich.live.Live`` region that shows the tree of
+active agents (one root per phase task; sub-agents nest below).
+Scrollback above the Live region accumulates phase headers, prover
+lifecycle terminals, and errors. Per-line subprocess output streams
+(``prover_output``, ``cloud_polling``, ``auto_setup_output``) are
+dropped — there's no real way to surface a continuous stdout stream
+in a Live region without fighting its redraw cycle.
+
+Sister to:
+
+  - ``AutoProveApp`` (Textual full-screen TUI) — for users who want
+    per-task panels and scrollable log widgets.
+  - ``AutoProveConsoleHandler`` (plain ``print``) — for users who
+    want raw log output (CI, redirected pipes, etc.).
+
+Pattern follows ``AutoProveConsoleHandler``: one handler instance per
+pipeline run, ``make_handler`` returns a ``TaskHandle`` that wraps the
+same instance for every phase so all roots feed into a single Live
+region.
+"""
+
+from contextlib import asynccontextmanager
+from typing import AsyncIterator, Callable, cast, override
+
+from rich.console import Console, RenderableType
+from rich.text import Text
+
+from composer.io.conversation import ConversationClient
+from composer.io.event_handler import NullEventHandler
+from composer.io.multi_job import TaskHandle, TaskInfo
+from composer.spec.source.prover import ProverEvents
+from composer.spec.source.autosetup import AutoSetupEvents
+from composer.ui.autoprove_app import AUTOPROVE_PHASE_LABELS, AutoProvePhase
+from composer.ui.conversation_client import ConsoleConversationClient
+from composer.ui.live_display import LiveDisplayHandler
+from composer.ui.tool_display import ToolDisplayConfig
+
+
+class AutoProveLiveHandler(LiveDisplayHandler[None], NullEventHandler):
+    """``IOHandler[None]`` + ``EventHandler`` + ``HandlerFactory`` for
+    the auto-prove pipeline, rendered as a single rich.live region.
+
+    Use as::
+
+        async with AutoProveLiveHandler() as handler:
+            await executor(handler.make_handler)
+    """
+
+    def __init__(
+        self,
+        *,
+        tool_display_config: ToolDisplayConfig | None = None,
+        console: Console | None = None,
+    ) -> None:
+        super().__init__(
+            tool_display_config=tool_display_config or ToolDisplayConfig(),
+            console=console,
+        )
+        # Tracks which phases have already produced a header. The first
+        # task within a phase to fire ``on_start`` emits the banner;
+        # subsequent parallel tasks in the same phase don't repeat it.
+        # Safe without a lock: ``on_start`` is sync and asyncio is
+        # single-threaded, so the check-and-add doesn't race.
+        self._phases_seen: set[AutoProvePhase] = set()
+
+    # ── IOHandler hook: HITL — autoprove has none ────────────────────
+
+    @override
+    async def handle_human_interaction(
+        self, ty: None, debug_thunk: Callable[[], None]
+    ) -> str:
+        raise RuntimeError(
+            "Unexpected HITL interrupt in auto-prove live handler"
+        )
+
+    # ── EventHandler hooks ──────────────────────────────────────────
+
+    @override
+    async def handle_event(
+        self, payload: dict, path: list[str], checkpoint_id: str
+    ) -> None:
+        evt = cast(ProverEvents, payload)
+        thread_id = path[-1]
+        match evt["type"]:
+            case "prover_run":
+                self._update_progress(thread_id, "running prover")
+            case "prover_link":
+                self._scrollback(
+                    Text(
+                        f"[{self._label(path)}] prover link → {evt['link']}",
+                        style="dim",
+                    )
+                )
+            case "prover_result":
+                self._update_progress(thread_id, None)
+                self._scrollback(
+                    Text(f"[{self._label(path)}] prover complete", style="green")
+                )
+            case "cex_analysis":
+                self._update_progress(
+                    thread_id, f"analyzing CEX: {evt['rule_name']}"
+                )
+            case "rule_analysis":
+                self._scrollback(
+                    Text(
+                        f"[{self._label(path)}] rule analysis → {evt['rule']}",
+                        style="dim",
+                    )
+                )
+            case "prover_output" | "cloud_polling":
+                # Per-line subprocess stream — no good way to surface a
+                # continuous stdout flow in a Live region. Use AutoProveApp
+                # or AutoProveConsoleHandler for raw prover output.
+                pass
+
+    @override
+    async def handle_progress_event(self, payload: dict) -> None:
+        evt = cast(AutoSetupEvents, payload)
+        match evt["type"]:
+            case "auto_setup_start":
+                self._scrollback(Text("AutoSetup: started", style="dim"))
+            case "auto_setup_complete":
+                self._scrollback(
+                    Text(
+                        f"AutoSetup: complete (rc={evt['return_code']})",
+                        style="dim",
+                    )
+                )
+            case "auto_setup_output":
+                # Per-line subprocess stream — same reasoning as
+                # prover_output / cloud_polling above.
+                pass
+
+    # ── ConversationProvider ────────────────────────────────────────
+
+    @asynccontextmanager
+    async def _start_conversation(
+        self, initial: RenderableType
+    ) -> AsyncIterator[ConversationClient]:
+        # Pause the Live region for the duration of the conversation so
+        # ``prompt_toolkit`` owns the terminal; restore on exit. The
+        # client itself is shared with ``AutoProveConsoleHandler`` —
+        # same prompt-toolkit prompt, same progress-event drainer.
+        #
+        # The status provider closes over our live agent table so the
+        # prompt's bottom toolbar shows "Background: N agents running"
+        # while the user composes input. prompt-toolkit's
+        # ``refresh_interval`` keeps it ticking.
+        async with self._paused_live():
+            client = ConsoleConversationClient(
+                initial
+            )
+            async with client:
+                yield client
+
+    def _background_status(self) -> str | None:
+        n = len(self._agents)
+        if n == 0:
+            return None
+        return f"Background: {n} agent{'' if n == 1 else 's'} running"
+
+    # ── HandlerFactory ──────────────────────────────────────────────
+
+    async def make_handler(
+        self, info: TaskInfo[AutoProvePhase]
+    ) -> TaskHandle[None]:
+        # ``run_task`` fires ``on_start`` / ``on_done`` per-task, not
+        # per-phase — a single phase like CVL_GEN spawns one
+        # "Invariant CVL" task plus one per-component batch, all
+        # sharing the same ``AutoProvePhase``. Use ``_phases_seen`` to
+        # collapse the per-task callback into a single phase header
+        # emitted the first time any task of that phase actually starts
+        # work. Subsequent tasks in the same phase don't repeat it.
+        # No ``on_done`` — top-level task completion is already
+        # captured by ``format_end``.
+
+        def _on_start() -> None:
+            if info.phase in self._phases_seen:
+                return
+            self._phases_seen.add(info.phase)
+            label = AUTOPROVE_PHASE_LABELS.get(info.phase, info.phase.name)
+            self._scrollback(
+                Text(
+                    "─" * 60 + f"\nPhase: {label}\n" + "─" * 60,
+                    style="bold",
+                )
+            )
+
+        async def _on_error(exc: Exception, tb: str) -> None:
+            self._scrollback(
+                Text(
+                    f"[{info.label}] ERROR — {type(exc).__name__}: {exc}\n{tb}",
+                    style="red",
+                )
+            )
+
+        return TaskHandle(
+            handler=self,
+            event_handler=self,
+            on_start=_on_start,
+            on_error=_on_error,
+            conversation_provider=self._start_conversation,
+        )

composer/ui/conversation_client.py

@@ -102,4 +102,3 @@ async def __aexit__(self, exc_type, exc, tb):
             await self.drain_task
         except Exception:
             print("Conversation cleanup failed")
-