Merge branch 'feat/thread-running-state-event-driven'

Author: Pyiner

desktop/garyx-desktop/src/main/gary-client.ts

@@ -402,6 +402,9 @@ interface ThreadSummaryPayload {
   teamDisplayName?: string | null;
   team?: ThreadTeamBlockPayload | null;
   recent_run_id?: string | null;
+  recentRunId?: string | null;
+  run_state?: string | null;
+  runState?: string | null;
   sdk_session_id?: string | null;
   worktree?: unknown;
 }
@@ -2098,9 +2101,8 @@ function mapThreadSummary(value: ThreadSummaryPayload): DesktopThreadSummary {
           : null),
     team,
     recentRunId:
-      typeof (value as { recent_run_id?: unknown }).recent_run_id === "string"
-        ? ((value as { recent_run_id?: string }).recent_run_id ?? null)
-        : null,
+      asString(value.recent_run_id) || asString(value.recentRunId) || null,
+    runState: asString(value.run_state) || asString(value.runState) || null,
     worktree: mapThreadWorktreeInfo(value.worktree),
   };
 }

desktop/garyx-desktop/src/renderer/src/app-shell/AppShell.tsx

@@ -1241,8 +1241,32 @@ const TRANSIENT_STATUS_MS = 3200;
 const ERROR_TOAST_MS = 4400;
 const GATEWAY_HEALTHY_POLL_MS = 12000;
 const SILENT_DESKTOP_STATE_REFRESH_MS = 60000;
+const RUN_STATE_LIST_REFRESH_DEBOUNCE_MS = 350;
+const GATEWAY_READY_STATE_REFRESH_THROTTLE_MS = 12000;
 const GATEWAY_RETRY_BACKOFF_MS = [2500, 4000, 6500, 10000, 15000];
 
+function threadRunStateIsRunning(thread: DesktopThreadSummary): boolean {
+  return (thread.runState || "").trim().toLowerCase() === "running";
+}
+
+function chatStreamEventHasRunLifecycle(event: DesktopChatStreamEvent): boolean {
+  const events =
+    event.type === "thread_render_frame"
+      ? event.events
+      : event.type === "committed_message"
+        ? [event]
+        : [];
+  return events.some((committed) => {
+    const controlKind = transcriptControlKind(committed.message);
+    return (
+      controlKind === "run_start" ||
+      controlKind === "run_complete" ||
+      controlKind === "run_interrupted" ||
+      controlKind === "interrupt_confirmed"
+    );
+  });
+}
+
 function savedContentView(): ContentView {
   const saved = sessionStorage.getItem("gary-content-view");
   const valid: ContentView[] = [
@@ -1770,6 +1794,8 @@ export function AppShell() {
   const gatewaySetupSavedConnectionRef = useRef<ConnectionStatus | null>(null);
   const botBindingRequestSequenceRef = useRef(0);
   const previousConnectionOkRef = useRef<boolean | null>(null);
+  const desktopStateRefreshTimeoutRef = useRef<number | null>(null);
+  const lastGatewayReadyStateRefreshAtRef = useRef(0);
   const lastRemoteStateWarningKeyRef = useRef<string | null>(null);
   const threadLogsPanelWidthRef = useRef(
     DEFAULT_DESKTOP_SETTINGS.threadLogsPanelWidth,
@@ -3110,11 +3136,10 @@ export function AppShell() {
         isActive:
           visibleThreadEntrySelectionSource === "recent" &&
           visibleSelectedThreadId === thread.id,
-        isBusy: isRuntimeBusy(selectThreadRuntime(messageState, thread.id)?.state),
+        isBusy: threadRunStateIsRunning(thread),
       })),
     [
       desktopState?.threads,
-      messageState,
       visibleSelectedThreadId,
       visibleThreadEntrySelectionSource,
     ],
@@ -3130,10 +3155,9 @@ export function AppShell() {
           isActive:
             visibleThreadEntrySelectionSource === "pinned" &&
             visibleSelectedThreadId === thread.id,
-          isBusy: isRuntimeBusy(selectThreadRuntime(messageState, thread.id)?.state),
+          isBusy: threadRunStateIsRunning(thread),
         })),
     [
-      messageState,
       pinnedThreadIds,
       threadAvatarCatalog,
       threadSummaryById,
@@ -3654,6 +3678,30 @@ export function AppShell() {
     return nextState;
   }
 
+  function scheduleDesktopStateRefresh(delayMs = RUN_STATE_LIST_REFRESH_DEBOUNCE_MS) {
+    if (desktopStateRefreshTimeoutRef.current !== null) {
+      window.clearTimeout(desktopStateRefreshTimeoutRef.current);
+    }
+    desktopStateRefreshTimeoutRef.current = window.setTimeout(() => {
+      desktopStateRefreshTimeoutRef.current = null;
+      void refreshDesktopState().catch((refreshError) => {
+        console.debug("Desktop state refresh failed.", refreshError);
+      });
+    }, delayMs);
+  }
+
+  function scheduleGatewayReadyStateRefresh() {
+    const now = Date.now();
+    if (
+      now - lastGatewayReadyStateRefreshAtRef.current <
+      GATEWAY_READY_STATE_REFRESH_THROTTLE_MS
+    ) {
+      return;
+    }
+    lastGatewayReadyStateRefreshAtRef.current = now;
+    scheduleDesktopStateRefresh(0);
+  }
+
   async function refreshAgentTargets() {
     const [nextAgents, nextTeams] = await Promise.all([
       window.garyxDesktop
@@ -4363,6 +4411,9 @@ export function AppShell() {
   useEffect(() => {
     const listener = (event: DesktopChatStreamEvent) => {
       enqueueClientLogEvent(event);
+      if (chatStreamEventHasRunLifecycle(event)) {
+        scheduleDesktopStateRefresh();
+      }
       streamEventHandlerRef.current(event);
     };
     window.garyxDesktop.subscribeChatStream(listener);
@@ -4381,6 +4432,15 @@ export function AppShell() {
     };
   }, []);
 
+  useEffect(() => {
+    return () => {
+      if (desktopStateRefreshTimeoutRef.current !== null) {
+        window.clearTimeout(desktopStateRefreshTimeoutRef.current);
+        desktopStateRefreshTimeoutRef.current = null;
+      }
+    };
+  }, []);
+
   useEffect(() => {
     const handlePointerDown = (event: MouseEvent) => {
       const target = event.target;
@@ -4809,6 +4869,7 @@ export function AppShell() {
         }
         if (nextOk) {
           gatewayRetryStepRef.current = 0;
+          scheduleGatewayReadyStateRefresh();
         } else {
           gatewayRetryStepRef.current = Math.min(
             gatewayRetryStepRef.current + 1,

desktop/garyx-desktop/src/shared/contracts.ts

@@ -883,6 +883,7 @@ export interface DesktopThreadSummary {
   teamId?: string | null;
   teamName?: string | null;
   recentRunId?: string | null;
+  runState?: string | null;
   worktree?: ThreadWorktreeInfo | null;
   /**
    * Full team block when this thread is bound to a Team. It is filled by

desktop/garyx-desktop/src/web/web-api.ts

@@ -31,6 +31,10 @@ type ThreadSummaryPayload = {
   message_count?: number;
   last_user_message?: string | null;
   last_assistant_message?: string | null;
+  recent_run_id?: string | null;
+  recentRunId?: string | null;
+  run_state?: string | null;
+  runState?: string | null;
 };
 
 type ThreadsPayload = {
@@ -262,6 +266,8 @@ function mapThreadSummary(value: ThreadSummaryPayload): DesktopThreadSummary {
     lastMessagePreview: preview,
     workspacePath: value.workspace_dir || null,
     messageCount: value.message_count,
+    recentRunId: value.recent_run_id || value.recentRunId || null,
+    runState: value.run_state || value.runState || null,
   };
 }
 

docs/design/thread-running-state.md

@@ -0,0 +1,148 @@
+# Thread Running-State Projection — Event-Driven, Backend Source of Truth
+
+Status: approved for implementation (user-confirmed direction "按 v2 干").
+
+## Goal
+
+The thread list (pinned + recent) shows a per-row "running" badge (the
+three-dot typing indicator). Today each client decides it differently:
+
+- iOS `homeThreadRunningThreadIds` mixes three sources (local run tracker,
+  per-thread committed run-state, server `run_state` fallback).
+- Desktop ignores the server field entirely and infers busy purely from its
+  local `messageState` stream machine, so runs started elsewhere never light up.
+- Widget reads the server snapshot only.
+
+We converge on a single rule: **the backend owns one `run_state` field per
+thread; all clients dumb-read it.** No client recomputes "is it running".
+
+## Source Of Truth
+
+The authoritative per-thread value is `recent_threads.run_state` (and its
+mirror `active_run_id`), which already exists in the SQLite projection
+(`garyx_db.rs`). What changes is *how it is decided*:
+
+> A thread is `running` iff its transcript's last run-lifecycle control is an
+> open (`run_start` with no paired close) **AND** the bridge's in-memory run
+> index still holds that run as active.
+
+The two-condition AND is deliberate:
+
+- **transcript event** drives the steady state immediately. `run_complete` /
+  `run_interrupted` / `interrupt_confirmed` flips the reduced state to
+  not-busy the moment it is committed, independent of any in-memory cleanup
+  ordering.
+- **in-memory confirmation** vetoes orphans. After a crash / SIGKILL the
+  transcript may keep a dangling `run_start` (the close was never written), but
+  the bridge run index is rebuilt empty on restart, so the orphan resolves to
+  `idle`.
+
+This is still event-driven (transcript run-lifecycle controls are the events);
+the in-memory index only removes the one failure mode events cannot cover
+(process death with no close event).
+
+### Why not "pure in-memory truth"
+
+`active_runs.remove` happens in the run task's cleanup block *after*
+`run_complete` is committed (`run_management.rs:2092-2095`). A projection that
+read in-memory state on the `run_complete`-triggered re-projection would still
+see the run present and fail to clear the badge. Gating on
+`transcript_busy AND memory_active` avoids touching bridge ordering: once
+`run_complete` is committed, `transcript_busy` is already false, so the result
+is `idle` regardless of remove timing.
+
+## State Machine (`idle` ⇄ `running`)
+
+We only care about the **thread itself**, not sub-tasks. A thread is serial:
+sending a new message first `abort_thread_runs` the in-flight run
+(`run_management.rs:1451`), and `max_concurrent_runs` is a global cap, not
+per-thread. So a thread has at most one of its own runs at a time → the field
+is a boolean (has its own active run / not). Sub-agent / workflow child runs
+live on their own child threads (already `exclude_from_recent`) and never
+aggregate onto the parent row.
+
+- **Open → running:** `run_start` (your message, automation, another device).
+- **Close → idle:** `run_complete` (status completed / error / interrupted /
+  aborted), `run_interrupted`, `interrupt_confirmed`.
+
+### Completeness
+
+| End scenario | Event | Covered by |
+| --- | --- | --- |
+| Normal finish | `run_complete` (completed) | transcript event |
+| Error / provider failure | `run_complete` (error) | transcript event |
+| User stop | `run_interrupted` / `interrupt_confirmed` | transcript event |
+| New message preempts in-flight | `run_complete` (interrupted) via `abort_thread_runs` | transcript event |
+| Program/system abort | `run_complete` (aborted) | transcript event |
+| Same thread concurrency | n/a (serial) | only ever 0 or 1 |
+| **Process crash / SIGKILL / power loss** | **no event** | in-memory confirm + startup reconcile + SIGTERM abort-all |
+
+Events cover every process-alive end path. The single blind spot (process death
+with no close event) is closed by the in-memory AND + the startup reconcile +
+graceful-shutdown abort.
+
+## Backend Changes (garyx-gateway, garyx-bridge)
+
+1. **bridge**: add `MultiProviderBridge::abort_all_active_runs()` — iterate
+   `get_active_runs()` and `abort_run` each. `is_run_active(run_id)` already
+   exists (`run_management.rs:2825`) and is reused as the in-memory probe.
+2. **gateway**: introduce an `ActiveRunProbe` trait and a
+   `BridgeActiveRunProbe(Weak<MultiProviderBridge>)` adapter. **Weak** is
+   required: the bridge holds `Arc<thread_store>` (the projecting store) via
+   `set_thread_store_blocking`, so the projecting store must not hold an
+   `Arc<MultiProviderBridge>` back (would leak/cycle).
+3. **gateway**: `RecentThreadProjectingStore` gains the probe. In
+   `project_thread`, after computing the transcript `active_run_id`, drop it to
+   `None` when `!probe.is_run_active(id)`. Same gate in
+   `reconcile_active_recent_thread_projection` (startup orphan clear — at boot
+   the index is empty, so every stale `running` row clears to `idle`).
+4. **gateway**: `server.rs` `serve` — after graceful shutdown returns, call
+   `bridge.abort_all_active_runs()` with a bounded timeout so a clean restart
+   writes close events and leaves no orphan; the startup reconcile only backs
+   up true hard crashes.
+
+The steady-state write path (`project_thread` on every `thread_store.set`) is
+unchanged in shape; only the `active_run_id` value gains the memory veto.
+Contract preserved: recent_threads stays write-time maintained, read routes do
+not repair (`docs/agents/repository-contracts.md`).
+
+## Client Changes (dumb-read the one field)
+
+- **Desktop**: `DesktopThreadSummary` gains `runState`; `gary-client.ts`
+  `mapThreadSummary` parses `run_state`; `AppShell.tsx` `recentThreadRows` /
+  `pinnedThreadRows` `isBusy` reads `thread.runState === "running"` instead of
+  `isRuntimeBusy(selectThreadRuntime(messageState, …))`. Keep the local stream
+  machine for the open thread's transcript; only the list badge switches source.
+  A light list refresh reuses existing poll/stream hooks.
+- **iOS**: `homeThreadRunningThreadIds` collapses to
+  `threads.filter { $0.run_state == "running" }`. Remove the runTracker /
+  runStateByThread contributions to the *list badge* (they remain for the open
+  thread's composer/steer state). Drop the
+  `GaryxThreadSummaryRunStateResolver` local override of `thread.runState`.
+  Keep the background refresh loop alive on home-visible+ready (decouple it from
+  the local-candidate gate).
+- **Widget**: already dumb-reads `run_state`/`active_run_id`; converge to
+  `run_state == "running"` only; verify against the host snapshot.
+
+### Optional immediacy (not required)
+
+Pure polling means a locally-sent message lights the badge on the next poll
+(≤1.5s on iOS). Optional: the locally-interacting thread may optimistically
+light its badge as a display-only override that yields to the backend field —
+never delays the off. Ship without it first.
+
+## Migration / Validation
+
+- Backend lands first (steady state unchanged; only orphan handling + the AND
+  gate change behavior). Each client is independent and ships separately because
+  the field already exists in the payload.
+- Tests: `cargo test -p garyx-gateway --all-targets`,
+  `cargo test -p garyx-bridge --all-targets`. Add: project clears `running` when
+  probe reports inactive; reconcile clears all `running` on empty index;
+  `abort_all_active_runs` writes close controls.
+- Orphan e2e: start a run, `kill -9` gateway, restart, assert
+  `/api/recent-threads` `run_state != "running"`.
+- Cross-end: one running thread, assert badge parity across desktop + iOS +
+  widget + `/api/recent-threads`.
+- Review: Claude implements → Codex adversarial review (`garyx task create
+  --assignee codex --workspace-dir <worktree> --notify current-thread`).

garyx-bridge/src/multi_provider/run_management.rs

@@ -2809,6 +2809,21 @@ impl MultiProviderBridge {
         (!aborted.is_empty(), aborted)
     }
 
+    /// Abort every active run across all threads. Called on graceful shutdown
+    /// so a clean restart writes close controls and leaves no orphaned
+    /// `running` projection behind; the startup reconcile only needs to back up
+    /// true hard crashes (SIGKILL / power loss).
+    pub async fn abort_all_active_runs(&self) -> Vec<String> {
+        let run_ids = self.get_active_runs().await;
+        let mut aborted = Vec::new();
+        for run_id in &run_ids {
+            if self.abort_run(run_id).await {
+                aborted.push(run_id.clone());
+            }
+        }
+        aborted
+    }
+
     /// Get list of active run IDs.
     pub async fn get_active_runs(&self) -> Vec<String> {
         self.inner