ContainNode action + resolver escalation + honest proposal surface (ADR-0040, shadow-complete)

engine/src/engine/adj_pass.rs

@@ -33,7 +33,7 @@
 //! model-chosen cuts, not a deterministic insertion.
 
 use futures::StreamExt;
-use std::collections::{BTreeMap, HashMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 
 use super::{
     Engine, PendingEntry, RestoredDecision, adj_gate, churn_diag, graph, journal, model, notify,
@@ -118,6 +118,15 @@ impl Engine {
         // mid-pass swaps the next pass's snapshot, never this one — so every entry judged this
         // pass sees a consistent provider table (mirrors the KEV/EPSS per-pass snapshot).
         let asn = self.asn.snapshot();
+        // ADR-0040 §3(d): the LIVE cross-entry model-attack set — every workload named in a
+        // decisive `assessment=attack`'s `contain`, across EVERY entry's current standing
+        // decision (not just the one whose menu is being built; a co-resident pair can be
+        // decided across two different entries). Computed ONCE before the per-entry loop from
+        // `self.decisions` as it stands at the START of this pass (last pass's decisive calls,
+        // or a carried-forward decision — ADR-0034 D7), so `boundary_break`'s trigger (d) sees
+        // exactly the decisions the live `IncidentDecision`/judge pipeline has already made,
+        // never a caller-supplied stand-in.
+        let model_attack = model_attack_set(&self.decisions);
         for (entry_key, idxs) in &by_entry {
             let entry = chains[idxs[0]].entry.clone();
             // The (objective, technique) set this entry reaches — what the model judges.
@@ -137,7 +146,7 @@ impl Engine {
             // across every one of its objective-chains (see `entry_menu`) — the SAME menu the
             // prompt's containment-options section renders and the model's `contain` reply
             // resolves against.
-            let menu = entry_menu(idxs, chains, graph, health);
+            let menu = entry_menu(idxs, chains, graph, health, &model_attack);
 
             // Build the entry's delta-aware pending record (prompt + fingerprint + projected
             // surface) and read its baseline — see [`Engine::prepare_pending`] (ADR-0023
@@ -583,11 +592,12 @@ fn entry_menu(
     chains: &[reason::proof::ProvenChain],
     graph: &graph::SecurityGraph,
     health: &observe::health::HealthReport,
+    model_attack: &BTreeSet<graph::NodeKey>,
 ) -> incident::Menu {
     let mut selectable = Vec::new();
     let mut uncontainable = Vec::new();
     for &i in idxs {
-        let m = incident::build_menu(&chains[i], graph, health);
+        let m = incident::build_menu(&chains[i], graph, health, model_attack);
         selectable.extend(m.selectable);
         uncontainable.extend(m.uncontainable);
     }
@@ -598,6 +608,22 @@ fn entry_menu(
     }
 }
 
+/// The LIVE cross-entry model-attack set (ADR-0040 §3(d)): every workload named in
+/// `contain` by a decisive `assessment=attack` decision, across every entry `decisions`
+/// currently holds — the set [`crate::engine::reason::proof::boundary_break`]'s trigger (d)
+/// composes two SEPARATE decisive calls over (co-resident workloads decided by different
+/// entries still compose). Pure over `Engine::decisions` so it is directly testable without
+/// spinning up a whole pass.
+fn model_attack_set(
+    decisions: &BTreeMap<String, incident::IncidentDecision>,
+) -> BTreeSet<graph::NodeKey> {
+    decisions
+        .values()
+        .filter(|d| d.assessment == incident::Assessment::Attack)
+        .flat_map(|d| d.cuts.iter().map(|c| c.node.clone()))
+        .collect()
+}
+
 #[cfg(test)]
 #[path = "adj_pass_tests.rs"]
 mod tests;

engine/src/engine/adj_pass_tests.rs

@@ -168,3 +168,109 @@ fn one_failing_cut_drops_the_whole_multi_cut_decision() {
         "one cut failing its lock must drop the entire decision, not just that cut"
     );
 }
+
+// --- ADR-0040: the escalated `ContainNode` cut needs NO special case in the replay lock ---
+
+/// A hand-built menu line resolving to `ContainNode` (a self-reference on a `Host`, not a
+/// workload) — mirrors [`quarantine_line`] but for the node-scoped mechanism.
+fn contain_node_line(node: &str, host: &str) -> incident::MenuLine {
+    let host_key = NodeKey(host.to_string());
+    let cut = Link {
+        from: host_key.clone(),
+        to: host_key,
+        relation: "contain-node".to_string(),
+        technique: None,
+        from_labels: BTreeMap::new(),
+        to_labels: BTreeMap::new(),
+    };
+    incident::MenuLine {
+        node: NodeKey(node.to_string()),
+        action: ProposedAction::ContainNode,
+        cut_signature: crate::engine::respond::cut_signature(&cut),
+        cut,
+        blast_note: "damage-limitation, not a clean sever: ...".to_string(),
+    }
+}
+
+/// A `ContainNode` decision re-arms and re-resolves through the SAME lock as any other
+/// mechanism — the lock only ever compares node/cut_signature strings, never the
+/// `ProposedAction` itself, so the escalated action needs no special case.
+#[test]
+fn a_contain_node_decision_rearms_through_the_same_lock() {
+    let line = contain_node_line("workload/app/Pod/store", "host/node-1");
+    let sig = line.cut_signature.clone();
+    let menu = incident::Menu {
+        selectable: vec![line],
+        uncontainable: Vec::new(),
+    };
+    let r = restored(
+        "fp-1",
+        vec![journal::JournaledCut {
+            node: "workload/app/Pod/store".into(),
+            cut_signature: sig.clone(),
+        }],
+    );
+    let decision = rearm_restored_decision(&r, "fp-1", &menu).expect("both locks hold");
+    assert_eq!(decision.cuts[0].action, ProposedAction::ContainNode);
+    assert_eq!(decision.cuts[0].cut_signature, sig);
+}
+
+/// A `boundary_break` flip between passes changes a node's mechanism resolution
+/// (`ContainNode` ⇄ its ordinary pod-scoped cut) — a DIFFERENT `cut_signature` for the SAME
+/// node key — so lock 2 fails closed: cold re-judge, never a silent repoint from a node cut
+/// to a pod cut or back.
+#[test]
+fn a_boundary_break_flip_fails_the_replay_lock_rather_than_repointing_the_cut() {
+    let r = restored(
+        "fp-1",
+        vec![journal::JournaledCut {
+            node: "workload/app/Pod/store".into(),
+            cut_signature: "host/node-1 -[contain-node]-> host/node-1".into(),
+        }],
+    );
+    // `boundary_break` flipped OFF since the decision was journaled: the current menu now
+    // resolves `store` back to its ordinary pod-scoped quarantine cut.
+    let menu = menu_with(&["workload/app/Pod/store"]);
+    assert!(
+        rearm_restored_decision(&r, "fp-1", &menu).is_none(),
+        "a boundary_break flip must fail the replay-lock, never silently repoint the cut"
+    );
+}
+
+// --- ADR-0040 §3(d): the LIVE cross-entry model-attack set ---
+
+/// [`model_attack_set`] collects every `contain`-named node from a decisive `Attack`
+/// decision, across every entry, and nothing from a `NoAttack`/`Uncertain` one.
+#[test]
+fn model_attack_set_collects_named_nodes_from_every_decisive_attack_entry() {
+    use crate::engine::graph::NodeKey as GraphNodeKey;
+
+    let mut decisions = BTreeMap::new();
+    decisions.insert(
+        "entry-a".to_string(),
+        incident::IncidentDecision {
+            assessment: incident::Assessment::Attack,
+            reason: "x".into(),
+            cuts: vec![incident::ChosenCut {
+                node: GraphNodeKey("workload/app/Pod/a".into()),
+                action: ProposedAction::QuarantineWorkload,
+                cut: Link {
+                    from: GraphNodeKey("workload/app/Pod/a".into()),
+                    to: GraphNodeKey("workload/app/Pod/a".into()),
+                    relation: "quarantine-workload".into(),
+                    technique: None,
+                    from_labels: BTreeMap::new(),
+                    to_labels: BTreeMap::new(),
+                },
+                cut_signature: "sig-a".into(),
+            }],
+        },
+    );
+    decisions.insert(
+        "entry-b".to_string(),
+        incident::IncidentDecision::uncertain("model unavailable"),
+    );
+    let set = model_attack_set(&decisions);
+    assert_eq!(set.len(), 1);
+    assert!(set.contains(&GraphNodeKey("workload/app/Pod/a".into())));
+}

engine/src/engine/cut_divergence_tests.rs

@@ -1,4 +1,4 @@
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 
 use serde_json::json;
 
@@ -102,7 +102,7 @@ fn decisive_attack(
     graph: &crate::engine::graph::SecurityGraph,
     nodes: &[&str],
 ) -> BTreeMap<String, IncidentDecision> {
-    let menu = build_menu(chain, graph, &HealthReport::default());
+    let menu = build_menu(chain, graph, &HealthReport::default(), &BTreeSet::new());
     let cuts = nodes
         .iter()
         .map(|n| {

engine/src/engine/journal/tests.rs

@@ -158,6 +158,48 @@ fn incident_decision_round_trips_across_a_reopen() {
     cleanup(&path);
 }
 
+/// ADR-0040: the escalated `ContainNode` cut needs no schema change — `JournaledCut` is
+/// already action-agnostic (only `node` + `cut_signature`, never the `ProposedAction` itself,
+/// per the type's own docs), so a node-scoped cut signature (`host/... -[contain-node]->
+/// host/...`, distinct in SHAPE from a workload self-reference) round-trips byte-for-byte
+/// exactly like any other mechanism's.
+#[test]
+fn a_contain_node_decision_round_trips_across_a_reopen() {
+    let path = temp_path("contain-node-roundtrip");
+    {
+        let journal = DecisionJournal::open(&path);
+        journal.record(Decision::Incident {
+            entry: "workload/app/Pod/web".into(),
+            objectives: 1,
+            assessment: crate::engine::reason::adjudicate::incident::Assessment::Attack,
+            reason: "kernel tamper on a proven pod-boundary break".into(),
+            cuts: vec![JournaledCut {
+                node: "workload/app/Pod/store".into(),
+                cut_signature: "host/node-1 -[contain-node]-> host/node-1".into(),
+            }],
+            fingerprint: "cves=|rt=ptrace-attach|objs=secret|findings=".into(),
+        });
+    }
+    let reopened = DecisionJournal::open(&path);
+    let entries = reopened.replay();
+    assert_eq!(entries.len(), 1);
+    match &entries[0].decision {
+        Decision::Incident {
+            cuts, fingerprint, ..
+        } => {
+            assert_eq!(cuts.len(), 1);
+            assert_eq!(cuts[0].node, "workload/app/Pod/store");
+            assert_eq!(
+                cuts[0].cut_signature, "host/node-1 -[contain-node]-> host/node-1",
+                "the node-keyed cut signature survives the reopen byte-for-byte"
+            );
+            assert_eq!(fingerprint, "cves=|rt=ptrace-attach|objs=secret|findings=");
+        }
+        other => panic!("expected an Incident, got {other:?}"),
+    }
+    cleanup(&path);
+}
+
 /// ADR-0035's shadow-bake step: a `CutDivergence` line — the model-vs-deterministic cut
 /// comparator's classification for one entry — round-trips a "restart" byte-for-byte, so the
 /// bake history a human reads for the arm-readiness review survives across a restart instead of

engine/src/engine/reason/adjudicate/incident/fixtures.rs

@@ -145,6 +145,93 @@ pub(super) fn web_reaches_pivot_store_with_image(
     (graph, chains)
 }
 
+/// As [`web_reaches_pivot_store`], but `store` is SCHEDULED on a `Host` (`nodeName:
+/// node-1`) and carries a `PtraceAttach` signal — `boundary_break(store)` trigger (c)
+/// (kernel tamper, ADR-0040 §3), the pod-boundary-break shape the `ContainNode` escalation
+/// exists for. `web` stays unscheduled and untampered — the negative contrast within the
+/// SAME chain (its own line must NOT escalate).
+pub(super) fn web_reaches_boundary_broken_store() -> (SecurityGraph, Vec<ProvenChain>) {
+    let web = pod(json!({
+        "apiVersion": "v1", "kind": "Pod",
+        "metadata": {"name": "web", "namespace": "app", "labels": {"role": "web"}},
+        "spec": {"containers": [{"name": "web", "image": "web:1"}]}
+    }));
+    let lb = service(json!({
+        "apiVersion": "v1", "kind": "Service",
+        "metadata": {"name": "web-lb", "namespace": "app"},
+        "spec": {"type": "LoadBalancer", "selector": {"role": "web"}}
+    }));
+    let store = pod(json!({
+        "apiVersion": "v1", "kind": "Pod",
+        "metadata": {"name": "store", "namespace": "app", "labels": {"role": "store"}},
+        "spec": {
+            "nodeName": "node-1",
+            "containers": [{
+                "name": "store", "image": "store:1",
+                "envFrom": [{"secretRef": {"name": "store-creds"}}]
+            }]
+        }
+    }));
+    let policy = netpol(json!({
+        "apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy",
+        "metadata": {"name": "store-ingress", "namespace": "app"},
+        "spec": {
+            "podSelector": {},
+            "policyTypes": ["Ingress"],
+            "ingress": [{"from": [{"podSelector": {"matchLabels": {"role": "web"}}}]}]
+        }
+    }));
+    let ptrace = RuntimeObservation {
+        attribution: Attribution::by_namespaced_name("app", "store"),
+        source: None,
+        observed_at_ms: None,
+        node: None,
+        behavior: Behavior::PtraceAttach,
+    };
+    let snap = Snapshot {
+        pods: vec![web, store],
+        services: vec![lb],
+        network_policies: vec![policy],
+        image_vulns: vec![critical_image("store:1")],
+        runtime_events: vec![ptrace],
+        ..Default::default()
+    };
+    let graph = build_graph(&snap, &default_adapters());
+    let chains = prove(&graph);
+    (graph, chains)
+}
+
+/// As [`web_reaches_boundary_broken_store`], but a THIRD pod — labeled like the eBPF
+/// agent DaemonSet's own pods (`app.kubernetes.io/component: agent`) — is ALSO scheduled
+/// on `node-1`, so cordoning it would collaterally sever protector's own sensor there
+/// (the self-severance warning, ADR-0040 "New failure/interaction surfaces").
+pub(super) fn boundary_broken_store_node_hosts_protector_agent() -> (SecurityGraph, Vec<ProvenChain>)
+{
+    let (mut graph, chains) = web_reaches_boundary_broken_store();
+    let agent = pod(json!({
+        "apiVersion": "v1", "kind": "Pod",
+        "metadata": {
+            "name": "protector-agent-xyz", "namespace": "protector-system",
+            "labels": {"app.kubernetes.io/component": "agent"}
+        },
+        "spec": {"nodeName": "node-1", "containers": [{"name": "agent", "image": "agent:1"}]}
+    }));
+    // Fold the agent pod into the SAME graph the chain was proven over (rather than
+    // rebuilding from a snapshot, which would also re-run `prove` and could renumber node
+    // indices out from under `chains`): `WorkloadAdapter` sets its labels (the fact
+    // `self_severance` reads), `PlacementAdapter` adds its `ScheduledOn` edge — neither
+    // needs any cross-object context, so running just these two over a snapshot holding
+    // only the agent pod is a faithful incremental fold.
+    use crate::engine::observe::adapter::{Adapter, PlacementAdapter, WorkloadAdapter};
+    let snap = Snapshot {
+        pods: vec![agent],
+        ..Default::default()
+    };
+    WorkloadAdapter.contribute(&snap, &mut graph);
+    PlacementAdapter.contribute(&snap, &mut graph);
+    (graph, chains)
+}
+
 pub(super) fn web_to_store_chain(chains: &[ProvenChain]) -> &ProvenChain {
     chains
         .iter()