feat: recover detached agent runs and interactions

crates/agent-core/src/interaction.rs

@@ -258,6 +258,21 @@ impl InteractionBroker {
         self.entries.lock().await.len()
     }
 
+    /// Return a snapshot of unresolved interactions for one run. The full
+    /// binding must match, so callers cannot inspect another actor's prompts.
+    pub async fn pending_for_binding(
+        &self,
+        binding: &InteractionBinding,
+    ) -> Vec<InteractionRequest> {
+        self.entries
+            .lock()
+            .await
+            .values()
+            .filter(|entry| &entry.request.binding == binding)
+            .map(|entry| entry.request.clone())
+            .collect()
+    }
+
     async fn expire(&self, interaction_id: &str, generation: uuid::Uuid) -> bool {
         let mut entries = self.entries.lock().await;
         let matches_generation = entries
@@ -470,4 +485,20 @@ mod tests {
             Err(BrokerError::Cancelled(_))
         ));
     }
+
+    #[tokio::test]
+    async fn pending_snapshot_is_scoped_to_the_full_binding() {
+        let broker = InteractionBroker::default();
+        let pending = broker
+            .register(request("approval-6"), CancellationToken::new())
+            .await
+            .unwrap();
+        assert_eq!(broker.pending_for_binding(&binding()).await.len(), 1);
+        assert!(broker
+            .pending_for_binding(&InteractionBinding::new("other", "session", "run"))
+            .await
+            .is_empty());
+        broker.cancel_run(&binding()).await;
+        assert!(matches!(pending.wait().await, Err(BrokerError::Cancelled(_))));
+    }
 }

src/routes/agent.rs

@@ -80,6 +80,7 @@ pub fn routes() -> Router<AppState> {
         .route("/api/agent/attachments", post(upload_attachment_handler))
         .route("/api/agent/approve", post(approve_handler))
         .route("/api/agent/answer", post(answer_handler))
+        .route("/api/agent/runs/{run_id}", get(get_run_handler))
         .route("/api/agent/runs/{run_id}/cancel", post(cancel_run_handler))
         .route("/api/agent/models", get(models_handler))
         .route("/api/agent/sessions", get(list_sessions_handler))
@@ -1771,6 +1772,30 @@ async fn cancel_run_handler(
     }
 }
 
+async fn get_run_handler(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Path(run_id): Path<String>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let auth = resolve_auth(&headers, &state).await?;
+    let run = session_store::get_run(&state.db, &run_id, &auth.actor_key)
+        .await
+        .map_err(internal_db_error)?
+        .ok_or_else(|| {
+            (
+                StatusCode::NOT_FOUND,
+                Json(json!({ "ok": false, "error": "Agent run not found" })),
+            )
+        })?;
+    let binding = InteractionBinding::new(&auth.actor_key, &run.session_id, &run.id);
+    let interactions = INTERACTIONS.pending_for_binding(&binding).await;
+    Ok(Json(json!({
+        "ok": true,
+        "run": run,
+        "interactions": interactions,
+    })))
+}
+
 async fn list_sessions_handler(
     State(state): State<AppState>,
     headers: HeaderMap,

src/services/agent_workbench/session_store.rs

@@ -72,6 +72,22 @@ pub struct RunClaim {
     pub revision: i64,
 }
 
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct AgentRunSummary {
+    pub id: String,
+    pub session_id: String,
+    pub user_id: Option<i64>,
+    pub status: String,
+    pub model: String,
+    pub revision: i64,
+    pub error_message: Option<String>,
+    pub cancel_requested_at: Option<String>,
+    pub started_at: String,
+    pub finished_at: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum ClaimRunError {
     #[error("agent session belongs to another actor")]
@@ -385,6 +401,23 @@ pub async fn active_run_session(
         .map_err(AppError::Database)
 }
 
+pub async fn get_run(
+    db: &AnyPool,
+    run_id: &str,
+    actor_key: &str,
+) -> AppResult<Option<AgentRunSummary>> {
+    crate::db::sql::query_as::<_, AgentRunSummary>(
+        "SELECT id, session_id, user_id, status, model, revision, error_message,
+                cancel_requested_at, started_at, finished_at, created_at, updated_at
+         FROM agent_runs WHERE id = ? AND actor_key = ?",
+    )
+    .bind(run_id)
+    .bind(actor_key)
+    .fetch_optional(db)
+    .await
+    .map_err(AppError::Database)
+}
+
 /// Settle a run exactly once and release its session lease only after the
 /// worker has stopped. The returned boolean is the canonical terminal-event
 /// ownership flag.
@@ -1015,6 +1048,35 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn run_status_is_scoped_to_the_actor() {
+        let db = pool().await;
+        let claim = claim_run(
+            &db,
+            &Uuid::new_v4().to_string(),
+            &Uuid::new_v4().to_string(),
+            &Uuid::new_v4().to_string(),
+            Some(6),
+            "user:6",
+            "m",
+            "default",
+            2,
+        )
+        .await
+        .unwrap();
+
+        assert!(get_run(&db, &claim.run_id, "user:5")
+            .await
+            .unwrap()
+            .is_none());
+        let run = get_run(&db, &claim.run_id, "user:6")
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(run.session_id, claim.session_id);
+        assert_eq!(run.status, "starting");
+    }
+
     #[tokio::test]
     async fn cancelling_run_keeps_lease_until_worker_finishes() {
         let db = pool().await;

src/web/pages/AgentWorkbench.test.ts

@@ -84,3 +84,18 @@ describe('AgentWorkbench session restoration', () => {
     expect(normalizePermissionMode('unexpected')).toBe('default');
   });
 });
+
+describe('AgentWorkbench detached run recovery', () => {
+  it('recovers run and interaction state after the SSE transport disconnects', () => {
+    expect(agentWorkbenchSource).toContain('/api/agent/runs/${encodeURIComponent(runId)}');
+    expect(agentWorkbenchSource).toContain('ACTIVE_AGENT_RUN_STATUSES.has(recovered.status)');
+    expect(agentWorkbenchSource).toContain("pending?.kind === 'approval'");
+    expect(agentWorkbenchSource).toContain("pending?.kind === 'input'");
+    expect(agentWorkbenchSource).toContain('window.setTimeout(poll, 2000)');
+  });
+
+  it('allows cancelling an active run after its original request controller is gone', () => {
+    expect(agentWorkbenchSource).toContain('if (cancellingRun || (!controller && !runId)) return;');
+    expect(agentWorkbenchSource).toContain('{(running || activeRunId) && (');
+  });
+});

src/web/pages/AgentWorkbench.tsx

@@ -100,6 +100,27 @@ type AgentSession = {
   updatedAt?: string;
 };
 
+type AgentRunStatus = {
+  id: string;
+  session_id: string;
+  status: string;
+  error_message?: string | null;
+};
+
+type PendingAgentInteraction = {
+  interaction_id: string;
+  kind: 'approval' | 'input';
+  payload?: Record<string, unknown>;
+};
+
+const ACTIVE_AGENT_RUN_STATUSES = new Set([
+  'starting',
+  'running',
+  'waiting_approval',
+  'waiting_input',
+  'cancelling',
+]);
+
 type CreationModeOption = {
   value: CreationMode;
   label: string;
@@ -1660,6 +1681,8 @@ export default function AgentWorkbench() {
         case 'result':
           flushAgentDeltas();
           setCancellingRun(false);
+          activeRunIdRef.current = null;
+          setActiveRunId(null);
           {
             const eventUsage = d.usage && typeof d.usage === 'object'
               ? d.usage as Record<string, unknown>
@@ -1676,6 +1699,8 @@ export default function AgentWorkbench() {
         case 'error':
           flushAgentDeltas();
           setCancellingRun(false);
+          activeRunIdRef.current = null;
+          setActiveRunId(null);
           toast.error(String(d.message ?? d.error ?? '执行出错'));
           updateAssistant((m) => {
             m.text += `\n\n[错误: ${String(d.message ?? d.error ?? '未知错误')}]`;
@@ -1684,6 +1709,8 @@ export default function AgentWorkbench() {
         case 'cancelled':
           flushAgentDeltas();
           setCancellingRun(false);
+          activeRunIdRef.current = null;
+          setActiveRunId(null);
           updateAssistant((m) => {
             if (!m.text.includes('[已取消]')) m.text += '\n\n[已取消]';
           });
@@ -1691,6 +1718,8 @@ export default function AgentWorkbench() {
         case 'interrupted':
           flushAgentDeltas();
           setCancellingRun(false);
+          activeRunIdRef.current = null;
+          setActiveRunId(null);
           updateAssistant((m) => {
             if (!m.text.includes('[运行已中断]')) m.text += '\n\n[运行已中断]';
           });
@@ -1708,6 +1737,49 @@ export default function AgentWorkbench() {
     [flushAgentDeltas, queueAgentDelta, toast, updateAssistant],
   );
 
+  const recoverRunState = useCallback(async (runId: string): Promise<AgentRunStatus> => {
+    const payload = await fetchJson(`/api/agent/runs/${encodeURIComponent(runId)}`);
+    const run = payload?.run as AgentRunStatus | undefined;
+    if (!run?.id || !run.session_id || !run.status) {
+      throw new Error('Agent 运行状态响应无效');
+    }
+
+    const active = ACTIVE_AGENT_RUN_STATUSES.has(run.status);
+    activeRunIdRef.current = active ? run.id : null;
+    setActiveRunId(active ? run.id : null);
+    setSessionId(run.session_id);
+    setCancellingRun(run.status === 'cancelling');
+
+    const interactions = Array.isArray(payload?.interactions)
+      ? payload.interactions as PendingAgentInteraction[]
+      : [];
+    const pending = interactions[0];
+    if (pending?.kind === 'approval') {
+      const data = pending.payload ?? {};
+      setApproval({
+        requestId: pending.interaction_id,
+        runId: run.id,
+        name: String(data.name ?? ''),
+        input: data.input,
+        risk: String(data.risk ?? ''),
+      });
+      setAsk(null);
+    } else if (pending?.kind === 'input') {
+      const data = pending.payload ?? {};
+      setAsk({
+        requestId: pending.interaction_id,
+        runId: run.id,
+        question: String(data.question ?? ''),
+        options: Array.isArray(data.options) ? data.options.map(String) : [],
+      });
+      setApproval(null);
+    } else if (!active) {
+      setApproval(null);
+      setAsk(null);
+    }
+    return run;
+  }, [fetchJson]);
+
   const send = useCallback(async () => {
     const rawPrompt = input.trim();
     const attachmentText = attachments.length > 0
@@ -1869,12 +1941,28 @@ export default function AgentWorkbench() {
         updateAssistant((m) => {
           m.text += '\n\n[已停止]';
         });
+      } else {
+        const runId = activeRunIdRef.current;
+        let recovered: AgentRunStatus | null = null;
+        if (runId) {
+          try {
+            recovered = await recoverRunState(runId);
+          } catch {
+            // Keep the original transport error as the user-facing failure.
+          }
+        }
+        if (recovered && ACTIVE_AGENT_RUN_STATUSES.has(recovered.status)) {
+          toast.info('连接已中断，Agent 仍在后台运行，正在自动恢复状态');
+          updateAssistant((m) => {
+            m.text += '\n\n[连接中断，任务仍在后台运行]';
+          });
         } else {
           toast.error(`请求失败: ${err instanceof Error ? err.message : String(err)}`);
           updateAssistant((m) => {
             m.text += '\n\n[连接中断]';
           });
         }
+      }
     } finally {
       flushAgentDeltas();
       abortRef.current = null;
@@ -1882,14 +1970,14 @@ export default function AgentWorkbench() {
       setCancellingRun(false);
       void loadSessions();
     }
-  }, [activeCanvasAction, agentModel, attachments, authToken, clearAttachments, createJimengJob, createSeedanceJob, createVoiceover, creationMode, customAgentModel, customJimengModel, customSeedanceModel, currentCreationMode.label, flushAgentDeltas, handleEvent, input, jimengImageModel, jimengReferenceCount, jimengVideoModel, loadSessions, mode, parsedJimengEndFrameUrls.length, parsedJimengImageUrls.length, running, seedanceEnabled, seedanceModel, seedanceStatusError, selectedAgentModel, sessionId, toast, updateAssistant, videoProvider]);
+  }, [activeCanvasAction, agentModel, attachments, authToken, clearAttachments, createJimengJob, createSeedanceJob, createVoiceover, creationMode, customAgentModel, customJimengModel, customSeedanceModel, currentCreationMode.label, flushAgentDeltas, handleEvent, input, jimengImageModel, jimengReferenceCount, jimengVideoModel, loadSessions, mode, parsedJimengEndFrameUrls.length, parsedJimengImageUrls.length, recoverRunState, running, seedanceEnabled, seedanceModel, seedanceStatusError, selectedAgentModel, sessionId, toast, updateAssistant, videoProvider]);
 
   const stop = useCallback(() => {
     const controller = abortRef.current;
     const runId = activeRunIdRef.current;
-    if (!controller || cancellingRun) return;
+    if (cancellingRun || (!controller && !runId)) return;
     if (!runId) {
-      controller.abort();
+      controller?.abort();
       return;
     }
     setCancellingRun(true);
@@ -2008,6 +2096,36 @@ export default function AgentWorkbench() {
     [agentModels, fetchJson, leaveCreativeView, running, toast],
   );
 
+  useEffect(() => {
+    if (running || !activeRunId) return undefined;
+    let disposed = false;
+    let timer: number | undefined;
+
+    const poll = async () => {
+      try {
+        const run = await recoverRunState(activeRunId);
+        if (disposed) return;
+        if (ACTIVE_AGENT_RUN_STATUSES.has(run.status)) {
+          timer = window.setTimeout(poll, 2000);
+          return;
+        }
+        await restoreSession(run.session_id);
+        if (!disposed) {
+          void loadSessions();
+          toast.info(run.status === 'succeeded' ? '后台 Agent 运行已完成' : `后台 Agent 运行已${run.status}`);
+        }
+      } catch {
+        if (!disposed) timer = window.setTimeout(poll, 4000);
+      }
+    };
+
+    void poll();
+    return () => {
+      disposed = true;
+      if (timer !== undefined) window.clearTimeout(timer);
+    };
+  }, [activeRunId, loadSessions, recoverRunState, restoreSession, running, toast]);
+
   const isFreshSession = messages.length === 0;
   const activeModelLabel = useMemo(() => {
     if (creationMode === 'agent') {
@@ -2328,7 +2446,7 @@ export default function AgentWorkbench() {
                 <em>模型与参数</em>
               </span>
             </div>
-            {running && (
+            {(running || activeRunId) && (
               <button className="agent-workbench-stop-top" type="button" onClick={stop} disabled={cancellingRun}>
                 {cancellingRun ? '取消中...' : '停止'}
               </button>