destroying an agent clears its per-agent archetype-flow switch-offs

swarm-controller/src/notify_flows.rs

@@ -229,6 +229,80 @@ async fn archetype_flows(client: &async_nats::Client) -> Vec<ArchetypeFlowView>
     }
 }
 
+/// `disabled` less the switch-offs for `agent` alone (`archetype:…/…@agent`; no
+/// built-in id has an `@`),
+/// or `None` when it has none. The per-archetype keys stay.
+fn without_agent_keys(disabled: &[String], agent: &str) -> Option<Vec<String>> {
+    let suffix = format!("@{agent}");
+    let kept: Vec<String> = disabled
+        .iter()
+        .filter(|k| !k.ends_with(&suffix))
+        .cloned()
+        .collect();
+    (kept.len() != disabled.len()).then_some(kept)
+}
+
+/// Remove the keys that switch archetype flows off for `agent` from the stored
+/// document, in the compare-and-set loop a `PUT` uses, so a concurrent edit
+/// is not lost. Answers the keys removed. No write when there are none.
+async fn clear_agent_keys(client: &async_nats::Client, agent: &str) -> anyhow::Result<Vec<String>> {
+    let store = open(client).await?;
+    for _ in 0..MAX_ATTEMPTS {
+        let (current, revision) = stored(&store).await?;
+        let Stored::Document(mut doc) = current else {
+            return Ok(Vec::new());
+        };
+        let Some(kept) = without_agent_keys(&doc.disabled_builtins, agent) else {
+            return Ok(Vec::new());
+        };
+        let removed: Vec<String> = doc
+            .disabled_builtins
+            .iter()
+            .filter(|k| !kept.contains(k))
+            .cloned()
+            .collect();
+        doc.disabled_builtins = kept;
+        if write_revision(&store, notify_flows::KEY, &doc, revision)
+            .await?
+            .is_some()
+        {
+            return Ok(removed);
+        }
+    }
+    anyhow::bail!("gave up after {MAX_ATTEMPTS} conflicting writes")
+}
+
+/// Clear a destroyed agent's per-agent switch-offs, so a new agent of that
+/// name starts with every archetype flow on. A failure is logged with the keys
+/// left behind and does not stop the destroy.
+pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) {
+    let cleared = async {
+        let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
+        clear_agent_keys(&client, agent).await
+    }
+    .await;
+    match cleared {
+        Ok(removed) if removed.is_empty() => {}
+        Ok(removed) => {
+            tracing::info!(
+                hive,
+                agent,
+                ?removed,
+                "destroyed agent's flow switch-offs cleared"
+            );
+        }
+        Err(e) => tracing::error!(
+            hive,
+            agent,
+            error = %format!("{e:#}"),
+            "clearing this agent's flow switch-offs (`archetype:…@<agent>` in the flows \
+             document's disabled list) FAILED; it was destroyed anyway, so a new agent of the \
+             same name starts with those flows off. Declaring the agent destroyed again \
+             re-runs this."
+        ),
+    }
+}
+
 pub(crate) fn unreadable(why: &DecodeError) -> ProblemDetails {
     error_problem(
         StatusCode::CONFLICT,
@@ -653,4 +727,42 @@ mod tests {
             assert_eq!(refused.status, Some(StatusCode::BAD_REQUEST), "{bad}");
         }
     }
+
+    /// Destroying an agent removes its own switch-offs from the document and
+    /// nothing else; a document without any is not written.
+    #[tokio::test]
+    async fn a_destroyed_agents_switch_offs_are_cleared_and_nothing_else() {
+        let nats = TestNats::start();
+        let client = nats.client().await;
+        let state = crate::agent_limits::tests::state(&client);
+
+        let keys = [
+            "archetype:a/f",
+            "archetype:a/f@alice",
+            "archetype:a/f@bob",
+            "archetype:a/g@alice",
+            "closed",
+        ];
+        let Json(saved) = put_flows(State(state.clone()), Ok(body(0, &keys, vec![])))
+            .await
+            .expect("put");
+
+        let removed = clear_agent_keys(&client, "alice").await.expect("clear");
+        assert_eq!(removed, ["archetype:a/f@alice", "archetype:a/g@alice"]);
+        let Json(after) = get_flows(State(state.clone())).await.expect("get");
+        assert_eq!(
+            after.disabled_builtins,
+            ["archetype:a/f", "archetype:a/f@bob", "closed"]
+        );
+        assert!(after.revision > saved.revision);
+
+        // Nothing of alice's left: no write.
+        assert!(clear_agent_keys(&client, "alice").await.unwrap().is_empty());
+        let Json(same) = get_flows(State(state.clone())).await.expect("get");
+        assert_eq!(same.revision, after.revision);
+        // A name that is the end of another's is not it.
+        assert!(clear_agent_keys(&client, "ob").await.unwrap().is_empty());
+        let Json(same) = get_flows(State(state)).await.expect("get");
+        assert_eq!(same.disabled_builtins, after.disabled_builtins);
+    }
 }

swarm-controller/src/routes/agents.rs

@@ -215,6 +215,7 @@ pub(crate) async fn set_agent_state(
         revoke_or_complain(&hive, &agent).await;
         agent_permissions::clear_on_destroy(&state, &hive, &agent).await;
         agent_archetype::clear_on_destroy(&state, &hive, &agent).await;
+        crate::notify_flows::clear_on_destroy(&state, &hive, &agent).await;
         agents_md_initial::clear_on_destroy(&state, &hive, &agent).await;
         egress_allow::clear_on_destroy(&state, &hive, &agent).await;
         egress_deny::clear_on_destroy(&state, &hive, &agent).await;