fix(web): unlock composer when a follow-up never terminates

A persisted FOLLOWUP_STARTED without COMPLETED/FAILED left followupPending
true across reload, so ask/writeup/resolve stayed disabled. Synthesize a
terminal failure from the standby wrapper and on fresh SSE replay, and
clear the leftover lock on run finish/reopen.

Author: cursoragent

apps/web/run_manager.py

@@ -3161,10 +3161,26 @@ def _ensure_standby(self, run_id: str, cmd: dict[str, Any]) -> bool:
         driver = build_standby_driver(cmd, mgr=self)
 
         async def _go() -> None:
+            followup_terminal = False
+
+            async def _note_followup_terminal(ev: Event) -> None:
+                nonlocal followup_terminal
+                if ev.event_type not in {
+                    EventType.FOLLOWUP_COMPLETED, EventType.FOLLOWUP_FAILED,
+                }:
+                    return
+                wanted = str(cmd.get("followup_id") or "")
+                ev_id = str((ev.payload or {}).get("followup_id") or "")
+                if not wanted or ev_id == wanted:
+                    followup_terminal = True
+
             async def _emit_followup_failed(detail: str) -> None:
-                if action not in {"ask", "writeup"}:
+                nonlocal followup_terminal
+                if action not in {"ask", "writeup"} or followup_terminal:
                     return
                 try:
+                    if bool(getattr(run.bus, "_closed", False)):
+                        self._fresh_bus(run)
                     await run.bus.emit(Event(
                         event_type=EventType.FOLLOWUP_FAILED,
                         run_id=run_id,
@@ -3174,9 +3190,11 @@ async def _emit_followup_failed(detail: str) -> None:
                             "detail": detail,
                         },
                     ))
+                    followup_terminal = True
                 except Exception:
                     pass
 
+            run.bus.add_sink(_note_followup_terminal)
             try:
                 LOG.info("standby worker starting for %s action=%s",
                          run_id, cmd.get("action"))
@@ -3210,6 +3228,11 @@ async def _emit_followup_failed(detail: str) -> None:
                 except Exception:
                     pass
             finally:
+                run.bus.remove_sink(_note_followup_terminal)
+                if action in {"ask", "writeup"} and not followup_terminal:
+                    # Crash, kill, or a failed start/complete emit must not leave
+                    # the finished composer locked on FOLLOWUP_STARTED.
+                    await _emit_followup_failed("后续操作已中断")
                 # Do not close the bus; retain the completed task as an observable
                 # receipt. `_ensure_standby` checks `.done()` and replaces it on the
                 # next command, so this does not block subsequent follow-ups.

apps/web/server.py

@@ -1482,6 +1482,9 @@ async def gen():
             replayed_seq = 0
             replayed_count = 0
             last_lifecycle = ""
+            last_followup = ""
+            last_followup_id = ""
+            last_followup_kind = "ask"
             async for ev in run.store.replay_monotonic(run_id, after_seq=last_id):
                 replayed_seq = ev.seq
                 replayed_count += 1
@@ -1490,6 +1493,14 @@ async def gen():
                                      EventType.RUN_FINISHED,
                                      EventType.RUN_REOPENED):
                     last_lifecycle = ev.event_type.value
+                if ev.event_type in (EventType.FOLLOWUP_STARTED,
+                                     EventType.FOLLOWUP_COMPLETED,
+                                     EventType.FOLLOWUP_FAILED):
+                    last_followup = ev.event_type.value
+                    last_followup_id = str(
+                        (ev.payload or {}).get("followup_id") or "")
+                    last_followup_kind = str(
+                        (ev.payload or {}).get("kind") or "ask")
                 yield {
                     "id": str(ev.seq),
                     "event": ev.event_type.value,
@@ -1524,7 +1535,26 @@ async def gen():
                     "event": synth.event_type.value,
                     "data": synth.model_dump_json(),
                 }
-            # live tail: everything after what we just replayed (or after the
+            # Ghost-followup guard: a persisted FOLLOWUP_STARTED without a
+            # terminal event and no live standby would lock ask/writeup/resolve
+            # across reload. Synthesize FOLLOWUP_FAILED so replay can settle.
+            if (fresh and not manager._standby_busy(run)
+                    and last_followup == EventType.FOLLOWUP_STARTED.value):
+                replayed_seq = max(replayed_seq, run.store.last_stream_seq(run_id)) + 1
+                synth = Event(
+                    event_type=EventType.FOLLOWUP_FAILED, run_id=run_id,
+                    seq=replayed_seq,
+                    payload={
+                        "followup_id": last_followup_id,
+                        "kind": last_followup_kind,
+                        "detail": "后续操作已中断",
+                    })
+                yield {
+                    "id": str(replayed_seq),
+                    "event": synth.event_type.value,
+                    "data": synth.model_dump_json(),
+                }
+            # live tail: everything after what we just replayed (or after the)
             # client's Last-Event-ID on a reconnect). A finished run's bus is
             # closed, so subscribe() returns after backlog replay. Do NOT let the
             # HTTP response EOF: browser EventSource treats EOF as an error and

apps/web/ui/lib/events.ts

@@ -2579,6 +2579,9 @@ export function reduce(prev: DeckState, ev: MutekiEvent): DeckState {
       }
       s.finished = true;
       s.preparing = false;
+      // A generation terminal is not a follow-up terminal. Drop a leftover
+      // ask/writeup lock so a later finished composer cannot stay disabled.
+      s.followupPending = false;
       // A hard stop can kill the coordinator before it ever emits race_concluded
       // (operator stop cancels the whole run task mid-race). Clear the pill here so
       // the UI never sticks on "racing" past the terminal event.
@@ -2791,6 +2794,7 @@ export function reduce(prev: DeckState, ev: MutekiEvent): DeckState {
       // false-positive flag invalidation. Keep the operator copy precise.
       s.finished = false;
       s.preparing = false;
+      s.followupPending = false;
       s.solved = false;
       s.acceptedOnly = false;
       s.finishedAt = undefined;

tests/test_standby_hitl.py

@@ -869,6 +869,52 @@ async def _run():
     assert lifecycle[-1].payload["detail"] == "后续操作已取消"
 
 
+def test_standby_wrapper_synthesizes_failed_when_start_has_no_terminal(
+        tmp_path, monkeypatch):
+    from apps.web import run_manager as rm
+    from muteki.core.events import Event, EventType
+    import apps.web.drivers as drivers
+
+    async def _started_then_exit(run):
+        await run.bus.emit(Event(
+            event_type=EventType.FOLLOWUP_STARTED,
+            run_id=run.run_id,
+            payload={
+                "followup_id": "followup-orphan", "kind": "ask",
+                "question": "证据来源是什么？",
+            },
+        ))
+
+    monkeypatch.setattr(
+        drivers, "build_standby_driver", lambda cmd, mgr=None: _started_then_exit,
+    )
+
+    async def _run():
+        mgr = rm.RunManager(sessions_root=tmp_path / "sessions")
+        run = mgr.create("run-x")
+        run.started = True
+        run.finished = True
+        run.solved = True
+        assert mgr._ensure_standby(run.run_id, {
+            "action": "ask", "text": "证据来源是什么？",
+            "followup_id": "followup-orphan",
+        })
+        await asyncio.gather(run.standby_task, return_exceptions=True)
+        return [event async for event in run.store.replay(run.run_id)]
+
+    events = asyncio.run(_run())
+    lifecycle = [
+        event for event in events if event.event_type in {
+            EventType.FOLLOWUP_STARTED, EventType.FOLLOWUP_FAILED,
+        }
+    ]
+    assert [event.event_type for event in lifecycle] == [
+        EventType.FOLLOWUP_STARTED, EventType.FOLLOWUP_FAILED,
+    ]
+    assert lifecycle[-1].payload["followup_id"] == "followup-orphan"
+    assert lifecycle[-1].payload["detail"] == "后续操作已中断"
+
+
 def test_standby_final_cancel_log_redacts_callback_exception(
         tmp_path, monkeypatch, caplog):
     from apps.web import run_manager as rm

tests/test_web_deck_ux.py

@@ -461,6 +461,21 @@ def test_followup_failure_replaces_its_correlated_pending_status():
         assert(s.chat.filter((m) => m.followupId === "F1"
           && m.content.includes("后续操作已取消")).length === 1,
           "terminal row remains idempotent after replay");
+
+        s = lib.reduce(s, {{ event_type: lib.EventType.FOLLOWUP_STARTED,
+          run_id: "run-followup", ts: 4, payload: {{ followup_id: "F2",
+            kind: "ask", question: "还卡着吗？" }} }});
+        assert(s.followupPending, "a new follow-up can pend again");
+        s = lib.reduce(s, {{ event_type: lib.EventType.RUN_FINISHED,
+          run_id: "run-followup", ts: 5, payload: {{ solved: true }} }});
+        assert(!s.followupPending, "run.finished clears a leftover follow-up lock");
+        s = lib.reduce(s, {{ event_type: lib.EventType.FOLLOWUP_STARTED,
+          run_id: "run-followup", ts: 6, payload: {{ followup_id: "F3",
+            kind: "writeup" }} }});
+        assert(s.followupPending, "writeup can pend after finish");
+        s = lib.reduce(s, {{ event_type: lib.EventType.RUN_REOPENED,
+          run_id: "run-followup", ts: 7, payload: {{ reason: "resolve" }} }});
+        assert(!s.followupPending, "run.reopened clears a leftover follow-up lock");
         """
     )
     _run_ui_node(script)

tests/test_web_server.py

@@ -731,6 +731,32 @@ async def test_events_injects_run_finished_for_ghost_run(server_mgr) -> None:
     assert EventType.RUN_FINISHED.value in seen  # the synthetic terminator
 
 
+async def test_events_injects_followup_failed_for_orphan_start(server_mgr) -> None:
+    from muteki.core.events import Event
+    s, mgr = server_mgr
+    rid = "ghost-followup-1"
+    run = mgr.create(rid)
+    await run.bus.emit(Event(event_type=EventType.RUN_STARTED, run_id=rid,
+                             payload={"challenge": {"name": "x"}}))
+    await run.bus.emit(Event(event_type=EventType.RUN_FINISHED, run_id=rid,
+                             payload={"solved": True}))
+    await run.bus.emit(Event(
+        event_type=EventType.FOLLOWUP_STARTED, run_id=rid,
+        payload={"followup_id": "F-orphan", "kind": "ask", "question": "why?"},
+    ))
+    run.started = True
+    run.finished = True
+    run.task = None
+    run.standby_task = None
+    async with httpx.AsyncClient(base_url=s.base, timeout=30, trust_env=False) as client:
+        seen: set = set()
+        await asyncio.wait_for(
+            _collect_sse(client, rid, seen, EventType.FOLLOWUP_FAILED.value),
+            timeout=15)
+    assert EventType.FOLLOWUP_STARTED.value in seen
+    assert EventType.FOLLOWUP_FAILED.value in seen
+
+
 async def test_events_do_not_inject_run_finished_for_protocol2_ghost(server_mgr) -> None:
     from muteki.core.events import Event