test(13-08): failing tests for the D-07/D-11 confirm gate (RED)

- confirm.rs: ConfirmGate/PendingConfirmation/Confirmed/PendingSnapshot/
  ResolveRefusal API skeleton (request/resolve/mint_nonce/build_description
  still todo!()) plus the five named confirm tests: S-02 nonce binding,
  S-03 no-model-text, S-08 distinct resources, S-09 restart drops pending,
  timeout declines, and the no-shared-lock-across-the-wait case
- mod.rs: ToolExecCtx gains the confirm gate (global by default, injectable
  for tests) and the S-01 destructive_tool_requires_confirm test with a
  seeded installed-app snapshot
- verified RED: 7 new tests fail (todo! cores + unfilled destructive branch)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

core/archipelago/src/assistant/mod.rs

@@ -11,6 +11,7 @@
 //! authenticated caller uses. See `13-01-PLAN.md` for the full spine.
 
 pub mod backends;
+pub mod confirm;
 pub mod grants;
 pub mod loop_;
 pub mod tools;
@@ -129,15 +130,31 @@ pub struct ToolExecCtx {
     pub registry: tools::ToolRegistry,
     pub caller: CallerScope,
     pub handler: Arc<RpcHandler>,
+    /// D-07/D-11: the confirm gate every destructive tool suspends on.
+    /// Defaults to the process-wide gate (`confirm::global()`) so the loop
+    /// and the `assistant.confirm-tool`/`assistant.pending` RPC handlers
+    /// share one pending queue; tests inject a fresh gate per test via
+    /// [`ToolExecCtx::with_confirm_gate`] for isolation.
+    pub confirm: Arc<confirm::ConfirmGate>,
     validation_failures: Mutex<HashMap<String, u32>>,
 }
 
 impl ToolExecCtx {
     pub fn new(registry: tools::ToolRegistry, caller: CallerScope, handler: Arc<RpcHandler>) -> Self {
+        Self::with_confirm_gate(registry, caller, handler, confirm::global())
+    }
+
+    pub fn with_confirm_gate(
+        registry: tools::ToolRegistry,
+        caller: CallerScope,
+        handler: Arc<RpcHandler>,
+        confirm: Arc<confirm::ConfirmGate>,
+    ) -> Self {
         Self {
             registry,
             caller,
             handler,
+            confirm,
             validation_failures: Mutex::new(HashMap::new()),
         }
     }
@@ -264,6 +281,181 @@ mod tests {
         (Arc::new(handler), tmp)
     }
 
+    /// A minimal installed-and-running app entry, seeded into the scanner
+    /// snapshot so the app-lifecycle tools' business-rule id resolution
+    /// (`container-list`) finds it — the same cached-state path the real
+    /// node serves, no orchestrator or podman needed.
+    fn installed_entry(app_id: &str) -> crate::data_model::PackageDataEntry {
+        use crate::data_model::{Description, Manifest, PackageDataEntry, PackageState, StaticFiles};
+        PackageDataEntry {
+            state: PackageState::Running,
+            health: None,
+            exit_code: None,
+            static_files: StaticFiles {
+                license: String::new(),
+                instructions: String::new(),
+                icon: String::new(),
+            },
+            manifest: Manifest {
+                id: app_id.to_string(),
+                title: app_id.to_string(),
+                version: String::new(),
+                description: Description {
+                    short: String::new(),
+                    long: String::new(),
+                },
+                release_notes: String::new(),
+                license: String::new(),
+                wrapper_repo: String::new(),
+                upstream_repo: String::new(),
+                support_site: String::new(),
+                marketing_site: String::new(),
+                donation_url: None,
+                author: None,
+                website: None,
+                interfaces: None,
+                tier: None,
+            },
+            installed: None,
+            install_progress: None,
+            uninstall_stage: None,
+            available_update: None,
+        }
+    }
+
+    /// Like `test_rpc_handler`, but with one installed, running app seeded
+    /// into the state snapshot so destructive app-lifecycle calls pass
+    /// business-rule validation and actually reach the confirm gate.
+    async fn test_rpc_handler_with_app(app_id: &str) -> (Arc<RpcHandler>, tempfile::TempDir) {
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let mut config = crate::config::Config::default();
+        config.data_dir = tmp.path().to_path_buf();
+        let state_manager = Arc::new(crate::state::StateManager::new());
+
+        let mut data = crate::data_model::DataModel::new();
+        data.server_info.status_info.containers_scanned = true;
+        data.package_data
+            .insert(app_id.to_string(), installed_entry(app_id));
+        state_manager.update_data(data).await;
+
+        let metrics_store = Arc::new(crate::monitoring::MetricsStore::new());
+        let session_store =
+            crate::session::SessionStore::new_for_tests(tmp.path().join("sessions.json"));
+        let handler = RpcHandler::new(
+            config,
+            state_manager,
+            metrics_store,
+            session_store,
+            None,
+            None,
+        )
+        .await
+        .expect("RpcHandler::new");
+        (Arc::new(handler), tmp)
+    }
+
+    async fn grant(handler: &Arc<RpcHandler>, category: PermissionCategory) {
+        let mut g = grants::Grants::load(handler.data_dir()).await;
+        g.set(category, true);
+        g.save(handler.data_dir()).await.expect("save grants");
+    }
+
+    async fn wait_pending(gate: &confirm::ConfirmGate) -> confirm::PendingSnapshot {
+        for _ in 0..500 {
+            if let Some(snap) = gate.peek() {
+                return snap;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+        }
+        panic!("no pending confirmation appeared — the destructive branch did not suspend on the gate");
+    }
+
+    /// S-01 / D-07: a destructive tool call suspends before any execution
+    /// and produces a pending confirmation whose node-authored description
+    /// names the resource; nothing runs until a human resolves it, and a
+    /// decline returns an error result without executing.
+    #[tokio::test]
+    async fn destructive_tool_requires_confirm() {
+        let (handler, _tmp) = test_rpc_handler_with_app("demo-app").await;
+        grant(&handler, PermissionCategory::Apps).await;
+
+        // Part 1 — the choke point directly: execute_tool suspends on the
+        // gate, and a decline returns a declined error result (dispatch is
+        // never reached — nothing was changed).
+        let gate = Arc::new(confirm::ConfirmGate::new());
+        let ctx = Arc::new(ToolExecCtx::with_confirm_gate(
+            tools::registry(),
+            CallerScope::LocalOperator {
+                session_id: "s".to_string(),
+            },
+            handler.clone(),
+            gate.clone(),
+        ));
+        let call = tools::ToolCall {
+            id: "call-1".to_string(),
+            name: "app_restart".to_string(),
+            arguments: serde_json::json!({ "app_id": "demo-app" }),
+        };
+        let (ctx_task, call_task) = (ctx.clone(), call.clone());
+        let suspended =
+            tokio::spawn(async move { loop_::execute_tool(&call_task, &ctx_task).await });
+
+        let snap = wait_pending(&gate).await;
+        assert!(
+            snap.description.contains("demo-app"),
+            "the node-authored description must name the resource: {}",
+            snap.description
+        );
+        assert!(
+            !suspended.is_finished(),
+            "execute_tool must stay suspended until the human decides"
+        );
+
+        gate.resolve(&snap.req_id, &snap.nonce, false)
+            .expect("decline resolves");
+        let result = suspended.await.expect("join");
+        assert!(result.is_error, "a declined action must not execute");
+        assert!(
+            result.content.to_lowercase().contains("declined"),
+            "the tool result must say the user declined: {}",
+            result.content
+        );
+
+        // Part 2 — through the real loop: a scripted backend proposes the
+        // destructive call, the loop suspends on the gate, and only after
+        // the human answers does the turn finish.
+        let gate2 = Arc::new(confirm::ConfirmGate::new());
+        let ctx2 = ToolExecCtx::with_confirm_gate(
+            tools::registry(),
+            CallerScope::LocalOperator {
+                session_id: "s".to_string(),
+            },
+            handler.clone(),
+            gate2.clone(),
+        );
+        let backend = crate::assistant::backends::scripted::ScriptedBackend::new(vec![
+            crate::assistant::backends::BackendTurn::ToolCalls(vec![call.clone()]),
+            crate::assistant::backends::BackendTurn::Text(
+                "Understood — I did not restart it.".to_string(),
+            ),
+        ]);
+        let tools_list = vec![tools::app_restart_tool()];
+        let loop_task = tokio::spawn(async move {
+            loop_::run_loop(&backend, "sys", &tools_list, vec![], &ctx2).await
+        });
+
+        let snap2 = wait_pending(&gate2).await;
+        assert!(
+            !loop_task.is_finished(),
+            "run_loop must stay suspended while the confirmation is pending"
+        );
+        gate2
+            .resolve(&snap2.req_id, &snap2.nonce, false)
+            .expect("decline resolves");
+        let answer = loop_task.await.expect("join").expect("run_loop");
+        assert_eq!(answer, "Understood — I did not restart it.");
+    }
+
     /// S-06 / D-16: a fresh node's `LocalOperator` resolves to no granted
     /// categories at all.
     #[tokio::test]