feat: replace polling with WS subscriptions + wire workflow approvals (WF-08)

Replace REST polling with WebSocket subscriptions across the desktop app
for real-time reactivity. Wire the workflow approval infrastructure so
runs suspend at approval gates instead of failing.

Desktop subscriptions:
- Forum posts/threads: subscribeLive for kinds 45001/45003 (was 15s/10s poll)
- Workflows: subscribeLive for kind 46010 (was 1s/10s poll)
- Pulse timeline: subscribeLive for contact note events (was 30s poll)
- Channels list: reduced backstop to 5min (live updates already wired)
- Agents/personas/teams: removed polling, CRUD mutations invalidate cache
- Home feed: kept at 30s poll (relay scoping prevents global subscriptions)

Shared useReactiveSubscription hook extracts common subscribe/reconnect/
cleanup lifecycle across forum, workflow, and pulse hooks.

WF-08 workflow approvals:
- executor.rs: create_approval() called with token, expiry, approver spec
- lib.rs: finalize_run sets WaitingApproval instead of Failed stub
- Trace entry pushed for suspended steps (UI can render approval card)
- Approver spec validated at creation (rejects unresolvable role specs)
- e2e test: full round-trip trigger → waiting_approval → grant → completed

Docs: added "Subscriptions, Not Polling" principle to AGENTS.md.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

AGENTS.md

@@ -92,6 +92,18 @@ simple and testable.
 thread root events. Any code that inserts replies must update these counters —
 check existing reply handlers for the pattern.
 
+### Subscriptions, Not Polling
+
+The desktop app uses WebSocket subscriptions (NIP-29 relay protocol) instead of polling for real-time data. When adding new data-fetching hooks:
+
+- **Subscribe to events** via `relayClient.subscribeLive()` or existing `subscribe*` methods
+- **Update TanStack Query cache** directly with `setQueriesData` (targeted) or `invalidateQueries` (broad)
+- **Handle reconnects** via `relayClient.subscribeToReconnects()` to recover missed events
+- **Backstop polling** (60s+) is acceptable ONLY for data written by non-WS clients (e.g., ACP agents via REST)
+- **CRUD-driven data** (agents, personas, teams) should invalidate cache in mutations, not poll
+
+Reference: `usePresenceSubscription()` in `features/presence/hooks.ts`
+
 ---
 
 ## Testing

crates/sprout-test-client/tests/e2e_workflows.rs

@@ -614,17 +614,13 @@ async fn test_workflow_update_and_delete() {
     );
 }
 
-// ── Test 7: Approval gate (WF-08 stub) ────────────────────────────────────────
+// ── Test 7: Approval gate round-trip (WF-08) ──────────────────────────────────
 
-/// Create a workflow with a `request_approval` step, trigger it, and verify
-/// the run fails with the "approval gates not yet implemented" message.
-///
-/// This test documents the current stub behavior. When WF-08 is implemented,
-/// this test should be updated to verify the full approval round-trip:
-/// create → trigger → poll for waiting_approval → grant → verify completed.
+/// Full approval round-trip: create workflow with `request_approval` step,
+/// trigger, poll for `waiting_approval`, fetch the approval token, grant it,
+/// and verify the run completes.
 #[tokio::test]
-#[ignore]
-async fn test_approval_gate_stub_fails_gracefully() {
+async fn test_approval_gate_round_trip() {
     let client = http_client();
     let pubkey_hex: &str = SEEDED_PUBKEY;
     let base = relay_http_url();
@@ -683,10 +679,10 @@ steps:
         .expect("trigger response must include 'run_id'")
         .to_string();
 
-    // ── Step 3: Poll until the run reaches a terminal status ──────────────────
+    // ── Step 3: Poll until the run reaches waiting_approval ──────────────────
     let runs_url = format!("{base}/api/workflows/{workflow_id}/runs");
-    let mut final_run: Option<serde_json::Value> = None;
-    for _ in 0..10 {
+    let mut waiting_run: Option<serde_json::Value> = None;
+    for _ in 0..20 {
         tokio::time::sleep(std::time::Duration::from_millis(100)).await;
         let runs_resp = client
             .get(&runs_url)
@@ -696,6 +692,78 @@ steps:
             .expect("GET runs failed");
         assert_eq!(runs_resp.status(), 200, "GET runs must return 200");
         let runs: Vec<serde_json::Value> = runs_resp.json().await.expect("runs must be JSON array");
+        if let Some(run) = runs.iter().find(|r| r["id"].as_str() == Some(&run_id)) {
+            let status = run["status"].as_str().unwrap_or("");
+            if status == "waiting_approval" {
+                waiting_run = Some(run.clone());
+                break;
+            }
+            if matches!(status, "completed" | "failed" | "cancelled") {
+                panic!("run reached terminal status '{status}' instead of waiting_approval");
+            }
+        }
+    }
+
+    let _run = waiting_run.expect("run must reach waiting_approval within 2 seconds");
+
+    // ── Step 4: Fetch the approval token ─────────────────────────────────────
+    let approvals_url = format!("{base}/api/workflows/{workflow_id}/runs/{run_id}/approvals");
+    let approvals_resp = client
+        .get(&approvals_url)
+        .header("X-Pubkey", pubkey_hex)
+        .send()
+        .await
+        .expect("GET approvals failed");
+    assert_eq!(
+        approvals_resp.status(),
+        200,
+        "GET approvals must return 200"
+    );
+    let approvals: Vec<serde_json::Value> = approvals_resp
+        .json()
+        .await
+        .expect("approvals must be JSON array");
+    assert!(
+        !approvals.is_empty(),
+        "there must be at least one pending approval"
+    );
+    let approval = &approvals[0];
+    assert_eq!(
+        approval["status"].as_str().unwrap_or(""),
+        "pending",
+        "approval must be pending"
+    );
+    let approval_token_hash = approval["token"]
+        .as_str()
+        .expect("approval must have a token hash");
+
+    // ── Step 5: Grant the approval (using by-hash endpoint since the listing
+    // returns the stored hash, not the raw token) ─────────────────────────────
+    let grant_url = format!("{base}/api/approvals/by-hash/{approval_token_hash}/grant");
+    let grant_resp = client
+        .post(&grant_url)
+        .header("X-Pubkey", pubkey_hex)
+        .send()
+        .await
+        .expect("POST grant failed");
+    assert!(
+        grant_resp.status().is_success(),
+        "grant must succeed, got {}",
+        grant_resp.status()
+    );
+
+    // ── Step 6: Poll until the run completes ─────────────────────────────────
+    let mut final_run: Option<serde_json::Value> = None;
+    for _ in 0..20 {
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        let runs_resp = client
+            .get(&runs_url)
+            .header("X-Pubkey", pubkey_hex)
+            .send()
+            .await
+            .expect("GET runs failed");
+        assert_eq!(runs_resp.status(), 200);
+        let runs: Vec<serde_json::Value> = runs_resp.json().await.expect("runs JSON");
         if let Some(run) = runs.iter().find(|r| r["id"].as_str() == Some(&run_id)) {
             let status = run["status"].as_str().unwrap_or("");
             if matches!(status, "completed" | "failed" | "cancelled") {
@@ -705,22 +773,14 @@ steps:
         }
     }
 
-    // ── Step 4: Assert the run failed with the expected stub error ────────────
-    let run = final_run.expect("run must reach a terminal status within 1 second");
-
+    let run = final_run.expect("run must reach terminal status after approval");
     assert_eq!(
         run["status"].as_str().unwrap_or(""),
-        "failed",
-        "approval gate stub must cause the run to fail"
+        "completed",
+        "run must complete after approval grant"
     );
 
-    let error_msg = run["error_message"].as_str().unwrap_or("");
-    assert!(
-        error_msg.contains("approval gates not yet implemented"),
-        "run error must contain 'approval gates not yet implemented', got: {error_msg:?}"
-    );
-
-    // ── Step 5: Clean up ──────────────────────────────────────────────────────
+    // ── Step 7: Clean up ──────────────────────────────────────────────────────
     let del_status = delete_workflow(&client, &base, pubkey_hex, &workflow_id).await;
     assert_eq!(del_status, 204, "cleanup DELETE should return 204");
 }

crates/sprout-workflow/src/executor.rs

@@ -552,6 +552,7 @@ pub async fn dispatch_action(
     engine: &WorkflowEngine,
     run_id: Uuid,
     trigger_ctx: &TriggerContext,
+    step_index: usize,
 ) -> Result<StepResult, WorkflowError> {
     use ActionDef::*;
 
@@ -680,10 +681,58 @@ pub async fn dispatch_action(
                 "RequestApproval from={from} timeout={timeout_str}: {message}"
             );
 
+            // Validate the approver spec: only "any" or a 64-char hex pubkey
+            // are supported today. Role-based specs (e.g. "@engineering-lead")
+            // will need relay-side group role resolution — reject them early so
+            // workflow authors get a clear error at trigger time.
+            let is_valid_approver =
+                from == "any" || (from.len() == 64 && from.chars().all(|c| c.is_ascii_hexdigit()));
+            if !is_valid_approver {
+                return Err(WorkflowError::InvalidDefinition(format!(
+                    "unsupported approver spec \"{from}\" — only \"any\" or a 64-char hex pubkey are supported"
+                )));
+            }
+
             let token = generate_approval_token(run_id, step_id);
 
-            // TODO (WF-08): create approval record in DB, emit kind:46010.
-            // For now, return Suspended with the token so the caller can persist state.
+            // Parse timeout duration and compute expiry.
+            // TODO: expires_at is stored but not yet enforced — there is no
+            // background sweep that auto-denies approvals past their deadline.
+            // Until that is added, expired approvals remain in "pending" state
+            // and the run stays in WaitingApproval indefinitely.
+            let timeout_secs = parse_duration_secs(timeout_str)?;
+            let expires_at = chrono::Utc::now() + chrono::Duration::seconds(timeout_secs as i64);
+
+            // Look up workflow metadata for the approval record.
+            let wf_run = engine.db.get_workflow_run(run_id).await.map_err(|e| {
+                WorkflowError::WebhookError(format!(
+                    "RequestApproval: failed to load workflow run {run_id}: {e}"
+                ))
+            })?;
+
+            // Persist the approval record in the database.
+            engine
+                .db
+                .create_approval(sprout_db::workflow::CreateApprovalParams {
+                    token: &token,
+                    workflow_id: wf_run.workflow_id,
+                    run_id,
+                    step_id,
+                    step_index: step_index as i32,
+                    approver_spec: from,
+                    expires_at,
+                })
+                .await
+                .map_err(|e| {
+                    WorkflowError::WebhookError(format!(
+                        "RequestApproval: failed to create approval record: {e}"
+                    ))
+                })?;
+
+            info!(
+                run_id = %run_id, step = step_id,
+                "Approval record created — token persisted, awaiting approval"
+            );
 
             Ok(StepResult::Suspended {
                 approval_token: token,
@@ -1162,7 +1211,7 @@ async fn execute_steps(
             .unwrap_or(engine.config.default_timeout_secs);
         let dispatch_result = tokio::time::timeout(
             std::time::Duration::from_secs(timeout_secs),
-            dispatch_action(&step.id, &resolved_action, engine, run_id, trigger_ctx),
+            dispatch_action(&step.id, &resolved_action, engine, run_id, trigger_ctx, i),
         )
         .await;
 
@@ -1205,6 +1254,10 @@ async fn execute_steps(
                     run_id = %run_id, step = %step.id,
                     "Step suspended — awaiting approval (token: <redacted>)"
                 );
+                trace.push(serde_json::json!({
+                    "step_id": step.id,
+                    "status": "waiting_approval",
+                }));
                 // Return the token and current state so the caller can persist the
                 // approval record and update the run's execution trace.
                 return Ok(ExecutionResult {

crates/sprout-workflow/src/lib.rs

@@ -156,27 +156,29 @@ impl WorkflowEngine {
                 let step_count = result.step_index as i32;
 
                 if result.approval_token.is_some() {
-                    // Approval gates are not yet implemented (WF-08).
-                    // Fail explicitly rather than creating unreachable WaitingApproval rows.
-                    tracing::warn!(
+                    // WF-08: Approval gate reached — suspend the run.
+                    // The approval record was already created by the executor's
+                    // dispatch_action. Mark the run as WaitingApproval so the
+                    // resume path can pick up at step_index + 1 after grant.
+                    tracing::info!(
                         run_id = %run_id,
                         step_index = result.step_index,
-                        "Workflow hit approval gate — not yet implemented, marking as failed"
+                        "Workflow suspended — waiting for approval"
                     );
                     if let Err(e) = self
                         .db
                         .update_workflow_run(
                             run_id,
-                            RunStatus::Failed,
+                            RunStatus::WaitingApproval,
                             step_count,
                             &trace_json,
-                            Some("approval gates not yet implemented — see WF-08"),
+                            None,
                         )
                         .await
                     {
                         tracing::error!(
                             run_id = %run_id,
-                            "Failed to update run to Failed (approval gate): {e}"
+                            "Failed to update run to WaitingApproval: {e}"
                         );
                     }
                 } else {

desktop/src/features/agents/hooks.ts

@@ -119,7 +119,7 @@ export function usePersonasQuery() {
     queryKey: personasQueryKey,
     queryFn: listPersonas,
     staleTime: 30_000,
-    refetchInterval: 30_000,
+    // CRUD-driven: mutations invalidate cache directly, no polling needed.
   });
 }
 
@@ -150,7 +150,7 @@ export function useRelayAgentsQuery(options?: { enabled?: boolean }) {
     queryKey: relayAgentsQueryKey,
     queryFn: listRelayAgents,
     staleTime: 30_000,
-    refetchInterval: 30_000,
+    // CRUD-driven: mutations invalidate cache directly, no polling needed.
     enabled: options?.enabled,
   });
 }
@@ -166,9 +166,10 @@ export function useManagedAgentsQuery(options?: { enabled?: boolean }) {
       // Only local "running" agents need fast polling (process state can
       // change). "deployed" is static control-plane state — presence polling
       // handles the live signal for remote agents separately.
+      // CRUD-driven: mutations invalidate cache directly, no idle polling.
       return agents?.some((agent) => agent.status === "running")
         ? 5_000
-        : 30_000;
+        : false;
     },
   });
 }
@@ -484,7 +485,8 @@ export function useManagedAgentLogQuery(
     enabled: pubkey !== null,
     retry: false,
     staleTime: 3_000,
-    refetchInterval: pubkey ? 30_000 : false,
+    // Logs are local process state; keep polling while viewing.
+    refetchInterval: pubkey ? 5_000 : false,
   });
 }
 
@@ -493,7 +495,7 @@ export function useTeamsQuery() {
     queryKey: teamsQueryKey,
     queryFn: listTeams,
     staleTime: 30_000,
-    refetchInterval: 30_000,
+    // CRUD-driven: mutations invalidate cache directly, no polling needed.
   });
 }
 

desktop/src/features/channels/hooks.ts

@@ -110,7 +110,9 @@ export function useChannelsQuery() {
     queryKey: channelsQueryKey,
     queryFn: async () => sortChannels(await getChannels()),
     staleTime: 60_000,
-    refetchInterval: 60_000,
+    // Live updates handled by useLiveChannelUpdates (subscribes to channel
+    // events and invalidates on reconnect). 5-minute backstop for edge cases.
+    refetchInterval: 5 * 60_000,
     refetchIntervalInBackground: false,
   });
 }

desktop/src/features/forum/hooks.ts

@@ -8,6 +8,7 @@ import type {
   ForumThreadResponse,
 } from "@/shared/api/types";
 import { KIND_FORUM_COMMENT, KIND_FORUM_POST } from "@/shared/constants/kinds";
+import { useReactiveSubscription } from "@/shared/hooks/useReactiveSubscription";
 
 export function forumPostsQueryKey(channelId: string) {
   return ["forum-posts", channelId] as const;
@@ -25,7 +26,8 @@ export function useForumPostsQuery(channel: Channel | null) {
     queryKey: forumPostsQueryKey(channelId),
     queryFn: () => getForumPosts(channelId, 50),
     staleTime: 15_000,
-    refetchInterval: 15_000,
+    // Live updates via useForumSubscription; 60s backstop for edge cases.
+    refetchInterval: 60_000,
   });
 }
 
@@ -38,10 +40,34 @@ export function useForumThreadQuery(
     queryKey: forumThreadQueryKey(channelId ?? "", eventId ?? ""),
     queryFn: () => getForumThread(channelId ?? "", eventId ?? ""),
     staleTime: 10_000,
-    refetchInterval: 10_000,
+    // Live updates via useForumSubscription; 60s backstop for edge cases.
+    refetchInterval: 60_000,
   });
 }
 
+/**
+ * Subscribe to forum post and comment events for a channel.
+ * Invalidates forum queries on incoming events and reconnects.
+ */
+export function useForumSubscription(channelId: string | null) {
+  const queryClient = useQueryClient();
+
+  useReactiveSubscription(
+    channelId
+      ? { kinds: [KIND_FORUM_POST, KIND_FORUM_COMMENT], "#h": [channelId] }
+      : null,
+    () => {
+      void queryClient.invalidateQueries({
+        predicate: (query) =>
+          (query.queryKey[0] === "forum-posts" ||
+            query.queryKey[0] === "forum-thread") &&
+          query.queryKey[1] === channelId,
+      });
+    },
+    "forum",
+  );
+}
+
 export function useCreateForumPostMutation(channel: Channel | null) {
   const queryClient = useQueryClient();
 

desktop/src/features/forum/ui/ForumView.tsx

@@ -12,6 +12,7 @@ import {
   useDeleteForumPostMutation,
   useDeleteForumReplyMutation,
   useForumPostsQuery,
+  useForumSubscription,
   useForumThreadQuery,
 } from "../hooks";
 import { ForumComposer } from "./ForumComposer";
@@ -47,6 +48,7 @@ export function ForumView({
   const [isComposerOpen, setIsComposerOpen] = React.useState(false);
 
   const profileQuery = useProfileQuery();
+  useForumSubscription(channel.id);
   const postsQuery = useForumPostsQuery(channel);
   const threadQuery = useForumThreadQuery(
     selectedPostId ? channel.id : null,

desktop/src/features/home/hooks.ts

@@ -2,9 +2,11 @@ import { useQuery } from "@tanstack/react-query";
 
 import { getHomeFeed } from "@/shared/api/tauri";
 
+const homeFeedQueryKey = ["home-feed"] as const;
+
 export function useHomeFeedQuery() {
   return useQuery({
-    queryKey: ["home-feed"],
+    queryKey: homeFeedQueryKey,
     queryFn: () =>
       getHomeFeed({
         limit: 12,

desktop/src/features/pulse/hooks.ts

@@ -1,5 +1,7 @@
+import * as React from "react";
 import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
 
+import { useReactiveSubscription } from "@/shared/hooks/useReactiveSubscription";
 import {
   getContactList,
   getNotesTimeline,
@@ -46,7 +48,8 @@ export function useMyNotesQuery(pubkey?: string) {
     enabled: !!pubkey,
     staleTime: 15_000,
     gcTime: 5 * 60_000,
-    refetchInterval: 30_000,
+    // Invalidated by usePublishNoteMutation; 120s backstop.
+    refetchInterval: 120_000,
   });
 }
 
@@ -59,10 +62,46 @@ export function useTimelineQuery(contactPubkeys: string[], enabled: boolean) {
     enabled: enabled && contactPubkeys.length > 0,
     staleTime: 15_000,
     gcTime: 5 * 60_000,
-    refetchInterval: 30_000,
+    // Live updates via usePulseSubscription; 120s backstop.
+    refetchInterval: 120_000,
   });
 }
 
+/**
+ * Subscribe to note events (kind:1) from contacts.
+ * Invalidates timeline and my-notes queries on incoming events and reconnects.
+ */
+export function usePulseSubscription(
+  contactPubkeys: string[],
+  currentPubkey: string | undefined,
+) {
+  const queryClient = useQueryClient();
+  const normalizedPubkey = currentPubkey?.trim().toLowerCase() ?? "";
+
+  const authors = React.useMemo(() => {
+    const set = new Set(contactPubkeys);
+    if (normalizedPubkey.length > 0) {
+      set.add(normalizedPubkey);
+    }
+    return [...set].sort();
+  }, [contactPubkeys, normalizedPubkey]);
+
+  useReactiveSubscription(
+    authors.length > 0 ? { kinds: [1], authors } : null,
+    () => {
+      void queryClient.invalidateQueries({
+        queryKey: pulseQueryKeys.allTimelines,
+      });
+      if (normalizedPubkey) {
+        void queryClient.invalidateQueries({
+          queryKey: pulseQueryKeys.myNotes(normalizedPubkey),
+        });
+      }
+    },
+    "pulse",
+  );
+}
+
 // ── Publish note mutation ───────────────────────────────────────────────────
 
 export function usePublishNoteMutation(currentPubkey?: string) {

desktop/src/features/pulse/ui/PulseView.tsx

@@ -8,6 +8,7 @@ import {
   useFollowMutation,
   useMyNotesQuery,
   usePublishNoteMutation,
+  usePulseSubscription,
   useTimelineQuery,
   useUnfollowMutation,
 } from "@/features/pulse/hooks";
@@ -182,6 +183,9 @@ export function PulseView({ currentPubkey }: PulseViewProps) {
     [peoplePubkeys, agentPubkeys],
   );
 
+  // ── Live subscription for note events ───────────────────────────────
+  usePulseSubscription(forYouPubkeys, currentPubkey);
+
   // ── Queries per tab ────────────────────────────────────────────────────
   const forYouQuery = useTimelineQuery(forYouPubkeys, activeTab === "foryou");
   const peopleQuery = useTimelineQuery(peoplePubkeys, activeTab === "people");

desktop/src/features/workflows/hooks.ts

@@ -1,6 +1,8 @@
 import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
 
 import type { WorkflowRun, WorkflowRunStatus } from "@/shared/api/types";
+import { KIND_APPROVAL_REQUEST } from "@/shared/constants/kinds";
+import { useReactiveSubscription } from "@/shared/hooks/useReactiveSubscription";
 import {
   createWorkflow,
   deleteWorkflow,
@@ -71,10 +73,12 @@ export function useWorkflowRunsQuery(workflowId: string | null) {
       getWorkflowRuns(resolvedWorkflowId),
     enabled: workflowId !== null,
     staleTime: 10_000,
+    // Live updates via useWorkflowSubscription (kind:46010).
+    // 10s backstop only when a workflow is actively running.
     refetchInterval: (query) => {
       const runs = query.state.data as WorkflowRun[] | undefined;
       return runs?.some((run) => isActiveWorkflowRunStatus(run.status))
-        ? 1_000
+        ? 10_000
         : false;
     },
   });
@@ -90,10 +94,37 @@ export function useRunApprovalsQuery(
       getRunApprovals(resolvedWorkflowId, resolvedRunId),
     enabled: workflowId !== null && runId !== null,
     staleTime: 10_000,
-    refetchInterval: 10_000,
+    // Live updates via useWorkflowSubscription (kind:46010); 30s backstop.
+    refetchInterval: 30_000,
   });
 }
 
+/**
+ * Subscribe to workflow status change events (kind:46010).
+ * Invalidates workflow run and approval queries on incoming events.
+ *
+ * NOTE: This subscription is forward-looking — the executor does not emit
+ * kind:46010 events yet.  The subscription is wired up now so the UI will
+ * react automatically once the relay-side publish is implemented.  Until then,
+ * refetchInterval backstops in the queries above drive updates.
+ */
+export function useWorkflowSubscription() {
+  const queryClient = useQueryClient();
+
+  useReactiveSubscription(
+    { kinds: [KIND_APPROVAL_REQUEST] },
+    () => {
+      void queryClient.invalidateQueries({
+        predicate: (query) =>
+          query.queryKey[0] === "workflow-runs" ||
+          query.queryKey[0] === "run-approvals" ||
+          query.queryKey[0] === "workflow",
+      });
+    },
+    "workflow",
+  );
+}
+
 export function useCreateWorkflowMutation(channelId: string) {
   const queryClient = useQueryClient();
 

desktop/src/features/workflows/ui/WorkflowDetailPanel.tsx

@@ -6,6 +6,7 @@ import {
   useTriggerWorkflowMutation,
   useWorkflowQuery,
   useWorkflowRunsQuery,
+  useWorkflowSubscription,
 } from "@/features/workflows/hooks";
 import { WorkflowRunTrace } from "@/features/workflows/ui/WorkflowRunTrace";
 import type { Workflow } from "@/shared/api/types";
@@ -29,6 +30,7 @@ export function WorkflowDetailPanel({
   onClose,
   onEdit,
 }: WorkflowDetailPanelProps) {
+  useWorkflowSubscription();
   const workflowQuery = useWorkflowQuery(workflowId);
   const runsQuery = useWorkflowRunsQuery(workflowId);
   const triggerMutation = useTriggerWorkflowMutation(workflowId);

desktop/src/shared/hooks/useReactiveSubscription.ts

@@ -0,0 +1,89 @@
+import * as React from "react";
+
+import { relayClient } from "@/shared/api/relayClient";
+import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared";
+
+type SubscriptionFilter = Omit<RelaySubscriptionFilter, "limit" | "since">;
+
+/**
+ * Subscribe to live relay events matching `filter` and invalidate React Query
+ * cache via `onInvalidate` whenever a new event arrives or the WebSocket
+ * reconnects.
+ *
+ * Pass `null` as the filter to skip subscribing (e.g. when a required ID isn't
+ * available yet). The reconnect listener is still registered so cache is
+ * refreshed after connection recovery even when the subscription is skipped.
+ *
+ * @param filter       Nostr subscription filter, or `null` to disable.
+ * @param onInvalidate Called on each incoming event and on reconnect.
+ * @param label        Human-readable label for error logging (e.g. "forum").
+ */
+export function useReactiveSubscription(
+  filter: SubscriptionFilter | null,
+  onInvalidate: () => void,
+  label: string,
+) {
+  // Stabilise the invalidation callback so the effect doesn't re-run when
+  // the caller passes an inline arrow.
+  const onInvalidateRef = React.useRef(onInvalidate);
+  React.useLayoutEffect(() => {
+    onInvalidateRef.current = onInvalidate;
+  });
+
+  // Memoize the filter by its serialized content so callers can pass object
+  // literals without churning the subscription effect.
+  const filterKey = filter === null ? "null" : stableFilterKey(filter);
+  // biome-ignore lint/correctness/useExhaustiveDependencies: filterKey is the stable serialization of filter
+  const stableFilter = React.useMemo(() => filter, [filterKey]);
+
+  React.useEffect(() => {
+    let isCancelled = false;
+    let cleanup: (() => Promise<void>) | undefined;
+
+    const invalidate = () => {
+      onInvalidateRef.current();
+    };
+
+    const disposeReconnect = relayClient.subscribeToReconnects(invalidate);
+
+    if (stableFilter === null) {
+      return () => {
+        disposeReconnect();
+      };
+    }
+
+    relayClient
+      .subscribeLive(
+        { ...stableFilter, limit: 0, since: Math.floor(Date.now() / 1_000) },
+        () => {
+          if (!isCancelled) {
+            invalidate();
+          }
+        },
+      )
+      .then((dispose) => {
+        if (isCancelled) {
+          void dispose();
+          return;
+        }
+        cleanup = dispose;
+      })
+      .catch((error) => {
+        console.error(`Failed to subscribe to ${label} events`, error);
+      });
+
+    return () => {
+      isCancelled = true;
+      disposeReconnect();
+      if (cleanup) void cleanup();
+    };
+  }, [label, stableFilter]);
+}
+
+/**
+ * Produce a stable string key for a filter object so the effect only re-runs
+ * when the filter semantically changes.
+ */
+function stableFilterKey(filter: SubscriptionFilter): string {
+  return JSON.stringify(filter, Object.keys(filter).sort());
+}