advance 1-8 post-mvp execution batches

rust/crates/mnote-web/src/routes/hermes_client.rs

@@ -29,6 +29,7 @@ const HEADER_MNOTE_WEB_OWNER: &str = "x-mnote-web-owner";
 const HEADER_HERMES_CLIENT_OWNER: &str = "x-mnote-hermes-client-owner";
 const ACP_RUNTIME_RUN_MUTATION: &str = "aiSessions:upsertRuntimeRun";
 const ACP_RUNTIME_EVENT_MUTATION: &str = "aiSessions:appendRuntimeEvent";
+const ACP_ABORT_NOTIFICATION_TIMEOUT_MS: u64 = 2_500;
 const LOCAL_SHARE_GRANTS_JSON: &str = "/mnt/Data1T/Mnote_data/control-plane/share-grants.json";
 const ENV_LOCAL_SHARE_GRANTS_FILE: &str = "MNOTE_SHARE_GRANTS_FILE";
 
@@ -90,6 +91,7 @@ struct AcpActiveRun {
     manager: Arc<crate::acp_session_manager::AcpSessionManager>,
     mnote_session_id: String,
     acp_session_id: String,
+    event_tx: broadcast::Sender<crate::acp_bridge::SseEvent>,
 }
 
 #[derive(Debug, Clone)]
@@ -1484,6 +1486,7 @@ async fn acp_stream_events(
             manager: Arc::clone(&mgr),
             mnote_session_id: mnote_session_id.clone(),
             acp_session_id,
+            event_tx: event_tx.clone(),
         },
     );
 
@@ -1567,6 +1570,12 @@ async fn acp_stream_events(
         {
             Ok(result) => {
                 info!("ACP prompt completed: stop_reason={:?}", result.stop_reason);
+                let prompt_cancelled =
+                    matches!(result.stop_reason, crate::acp_types::StopReason::Cancelled);
+                let runtime_aborted = matches!(
+                    runtime_status_for_run(&run_id_owned).as_deref(),
+                    Some("aborting" | "aborted")
+                );
                 let agent_audit = if audit_payload
                     .get("sourceKind")
                     .and_then(Value::as_str)
@@ -1590,7 +1599,11 @@ async fn acp_stream_events(
                     Value::Null
                 };
                 let _ = event_tx_prompt.send(crate::acp_bridge::SseEvent {
-                    event: "run.completed".into(),
+                    event: if prompt_cancelled || runtime_aborted {
+                        "run.aborted".into()
+                    } else {
+                        "run.completed".into()
+                    },
                     data: json!({
                         "stopReason": format!("{:?}", result.stop_reason),
                         "agentAudit": agent_audit,
@@ -1627,15 +1640,16 @@ async fn acp_stream_events(
                 });
             }
         }
+        let was_aborted = matches!(
+            runtime_status_for_run(&run_id_owned).as_deref(),
+            Some("aborting" | "aborted")
+        );
         mgr_clone.close().await;
         ACP_ACTIVE_RUNS
             .lock()
             .expect("acp active runs")
             .remove(&run_id_owned);
-        if !matches!(
-            runtime_status_for_run(&run_id_owned).as_deref(),
-            Some("aborting" | "aborted")
-        ) {
+        if !was_aborted {
             update_runtime_by_run_id(&run_id_owned, "completed", Some("acp.prompt.done"), None);
         }
     });
@@ -1763,11 +1777,37 @@ pub async fn abort_run(
                 })),
             ));
         };
-        active.manager.cancel().await.map_err(|error| {
-            WebError::bad_gateway_code("acp_abort_failed", format!("ACP abort failed: {error}"))
-                .with_context(&context)
-        })?;
+        let cancel_result = tokio::time::timeout(
+            Duration::from_millis(ACP_ABORT_NOTIFICATION_TIMEOUT_MS),
+            active.manager.cancel(),
+        )
+        .await;
+        let cancel_status = match cancel_result {
+            Ok(Ok(())) => json!({"ok": true}),
+            Ok(Err(error)) => {
+                warn!(error = ?error, run_id = %run_id, "ACP abort notification failed; marking run aborted");
+                json!({"ok": false, "error": error.to_string()})
+            }
+            Err(_) => {
+                warn!(
+                    run_id = %run_id,
+                    timeout_ms = ACP_ABORT_NOTIFICATION_TIMEOUT_MS,
+                    "ACP abort notification timed out; marking run aborted"
+                );
+                json!({"ok": false, "error": "abort notification timed out"})
+            }
+        };
         update_runtime_by_run_id(&run_id, "aborted", Some("abort.completed"), None);
+        let _ = active.event_tx.send(crate::acp_bridge::SseEvent {
+            event: "run.aborted".into(),
+            data: json!({
+                "reason": payload
+                    .get("reason")
+                    .and_then(Value::as_str)
+                    .unwrap_or("client_abort"),
+                "cancel": cancel_status.clone(),
+            }),
+        });
         return Ok((
             StatusCode::OK,
             stamp_client_headers(),
@@ -1778,6 +1818,7 @@ pub async fn abort_run(
                 "acpSessionId": active.acp_session_id,
                 "status": "aborted",
                 "runtime": runtime_state_for_run(&run_id).unwrap_or(Value::Null),
+                "cancel": cancel_status,
                 "events": [
                     {"event": "abort.started", "runId": run_id},
                     {"event": "abort.completed", "runId": run_id}