diff --git a/crates/agent-spec/src/kdl_format.rs b/crates/agent-spec/src/kdl_format.rs
index 6e1a0c21..1a407922 100644
--- a/crates/agent-spec/src/kdl_format.rs
+++ b/crates/agent-spec/src/kdl_format.rs
@@ -3,14 +3,14 @@
 //! serde-native) and fills a [`RawSpec`], which then flows through the shared identity/host
 //! resolution. A file may hold more than one `agent` node.
 //!
-//! Only runner-normative fields plus metadata `role` are read; render-only fields (`harness`,
-//! `model`, `persona`, `permissions`, `transport`, `strategy`) and the inert `meta{}` block are
-//! ignored.
+//! Runner-normative fields, managed delivery readiness, and metadata `role` are read. Other
+//! render-only fields (`harness`, `model`, `persona`, `permissions`, `transport`, `strategy`) and
+//! the inert `meta{}` block are ignored.
 
 use crate::declared::{DeclaredDocument, DeclaredNode, DeclaredValue};
 use crate::spec::{
-    ClaudeDriver, CodexDriver, OmpDriver, OpenCodeDriver, PiDriver, RawResource, RawRestart,
-    RawSpec, RawTask,
+    ClaudeDriver, CodexDriver, DeliveryReadiness, OmpDriver, OpenCodeDriver, PiDriver, RawResource,
+    RawRestart, RawSpec, RawTask, SessionDriver,
 };
 
 /// Lower an already parsed declaration document into the runner's raw representation.
@@ -162,6 +162,13 @@ fn agent_node_to_raw(node: &DeclaredNode) -> anyhow::Result<RawSpec> {
                     anyhow::anyhow!("agent `session-driver` value must be a string")
                 })?));
             }
+            "delivery-readiness" => {
+                anyhow::ensure!(
+                    raw.delivery_readiness.is_none(),
+                    "agent declares `delivery-readiness` more than once"
+                );
+                raw.delivery_readiness = Some(delivery_readiness_node_to_raw(child)?);
+            }
             "claude" => {
                 anyhow::ensure!(
                     raw.driver.claude.is_none(),
@@ -512,6 +519,68 @@ fn restart_node_to_raw(node: &DeclaredNode) -> RawRestart {
     r
 }
 
+fn delivery_readiness_node_to_raw(node: &DeclaredNode) -> anyhow::Result<DeliveryReadiness> {
+    anyhow::ensure!(
+        node.type_name.is_none() && node.children.is_empty(),
+        "agent `delivery-readiness` cannot have a type annotation or children"
+    );
+    let kind = node
+        .argument(0)
+        .and_then(DeclaredValue::as_str)
+        .ok_or_else(|| anyhow::anyhow!("agent `delivery-readiness` needs a kind string"))?;
+    match kind {
+        "credential" => {
+            anyhow::ensure!(
+                node.arguments().count() == 1
+                    && node.properties_named("account-id").count() <= 1
+                    && node.entries.len() <= 2,
+                "credential delivery-readiness must be `delivery-readiness \"credential\"` with at most one string `account-id`"
+            );
+            let account_id = node
+                .property("account-id")
+                .map(|value| {
+                    value.as_str().map(String::from).ok_or_else(|| {
+                        anyhow::anyhow!(
+                            "credential delivery-readiness `account-id` must be a string"
+                        )
+                    })
+                })
+                .transpose()?;
+            Ok(DeliveryReadiness::Credential { account_id })
+        }
+        "anonymous" => {
+            anyhow::ensure!(
+                node.arguments().count() >= 2
+                    && node.properties_named("harness").count() == 1
+                    && node.entries.len() == node.arguments().count() + 1,
+                "anonymous delivery-readiness must be `delivery-readiness \"anonymous\" \"<model>\"… harness=\"<driver>\"`"
+            );
+            let harness = node
+                .property("harness")
+                .and_then(DeclaredValue::as_str)
+                .ok_or_else(|| {
+                    anyhow::anyhow!("anonymous delivery-readiness `harness` must be a string")
+                })
+                .and_then(SessionDriver::from_name)?;
+            let models = node
+                .arguments()
+                .skip(1)
+                .map(|value| {
+                    value.as_str().map(String::from).ok_or_else(|| {
+                        anyhow::anyhow!(
+                            "anonymous delivery-readiness accepts only string model arguments"
+                        )
+                    })
+                })
+                .collect::<anyhow::Result<Vec<_>>>()?;
+            Ok(DeliveryReadiness::Anonymous { harness, models })
+        }
+        other => anyhow::bail!(
+            "unsupported delivery-readiness kind '{other}' (expected `credential` or `anonymous`)"
+        ),
+    }
+}
+
 fn task_node_to_raw(node: &DeclaredNode) -> anyhow::Result<RawTask> {
     let mut t = RawTask::default();
     for child in &node.children {
diff --git a/crates/agent-spec/src/lib.rs b/crates/agent-spec/src/lib.rs
index bc357333..050b5c13 100644
--- a/crates/agent-spec/src/lib.rs
+++ b/crates/agent-spec/src/lib.rs
@@ -46,10 +46,10 @@ pub use discovery::{
     parse_declared, path_defaults,
 };
 pub use spec::{
-    AgentDesiredState, AgentSpec, ClaudeDriver, CodexDriver, DeliveryTransport, Driver, JobType,
-    OpenCodeDriver, PiDriver, Resource, Restart, RestartMode, STREAM_TASK_PREFIX, SessionDriver,
-    Stream, StreamLaunch, Task, TaskKind, TaskLifecycle, parse_duration, stream_name_of_task,
-    validate_desired_state_reason,
+    AgentDesiredState, AgentSpec, ClaudeDriver, CodexDriver, DeliveryReadiness, DeliveryTransport,
+    Driver, JobType, OpenCodeDriver, PiDriver, Resource, Restart, RestartMode, STREAM_TASK_PREFIX,
+    SessionDriver, Stream, StreamLaunch, Task, TaskKind, TaskLifecycle, parse_duration,
+    stream_name_of_task, validate_desired_state_reason,
 };
 pub use profile::{
     DEFAULT_SELECTOR_LIMIT_BYTES, DescriptorValidationError, PROFILE_DESCRIPTOR_ABI_VERSION,
diff --git a/crates/agent-spec/src/spec.rs b/crates/agent-spec/src/spec.rs
index 1e850fa9..f8e8cd35 100644
--- a/crates/agent-spec/src/spec.rs
+++ b/crates/agent-spec/src/spec.rs
@@ -68,12 +68,21 @@ impl DeliveryTransport {
             ),
         }
     }
+
+    pub fn session_driver(self) -> SessionDriver {
+        match self {
+            Self::Mcp => SessionDriver::Claude,
+            Self::AppServer => SessionDriver::Codex,
+            Self::PiChannel => SessionDriver::Pi,
+        }
+    }
 }
 /// The native session driver entered by an otherwise opaque launch.
 ///
 /// This is an ownership assertion only. It does not render a provider launch or select a message
 /// delivery transport.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
 pub enum SessionDriver {
     Claude,
     Codex,
@@ -105,6 +114,71 @@ impl SessionDriver {
             ),
         }
     }
+
+    /// Parse the canonical driver name carried by `session-driver`.
+    pub fn from_name(value: &str) -> anyhow::Result<Self> {
+        Self::parse(value)
+    }
+}
+
+/// Non-secret facts that prove how a managed session can be admitted for delivery.
+///
+/// This is deliberately separate from activity and runtime health. A credential-backed seat names
+/// only its opaque account identifier; an anonymous seat names the exact harness and model allowlist
+/// it can launch without credentials.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", tag = "kind", deny_unknown_fields)]
+pub enum DeliveryReadiness {
+    Credential {
+        account_id: Option<String>,
+    },
+    Anonymous {
+        harness: SessionDriver,
+        models: Vec<String>,
+    },
+}
+
+impl DeliveryReadiness {
+    pub fn validate(&mut self) -> anyhow::Result<()> {
+        match self {
+            Self::Credential { account_id } => {
+                if let Some(account_id) = account_id {
+                    validate_delivery_readiness_value("account-id", account_id)?;
+                }
+            }
+            Self::Anonymous { harness: _, models } => {
+                anyhow::ensure!(
+                    !models.is_empty(),
+                    "anonymous delivery-readiness requires at least one model"
+                );
+                anyhow::ensure!(
+                    models.len() <= 32,
+                    "anonymous delivery-readiness accepts at most 32 models"
+                );
+                for model in models.iter() {
+                    validate_delivery_readiness_value("model", model)?;
+                }
+                models.sort();
+                models.dedup();
+            }
+        }
+        Ok(())
+    }
+}
+
+fn validate_delivery_readiness_value(field: &str, value: &str) -> anyhow::Result<()> {
+    anyhow::ensure!(
+        !value.is_empty() && value.len() <= 200,
+        "delivery-readiness {field} must be 1..=200 UTF-8 bytes"
+    );
+    anyhow::ensure!(
+        value.trim() == value
+            && !value
+                .chars()
+                .any(|character| character.is_control() || matches!(character, '\u{2028}' | '\u{2029}')),
+        "delivery-readiness {field} must have no surrounding whitespace, controls, or line separators"
+    );
+    Ok(())
 }
 
 
@@ -130,6 +204,16 @@ impl Driver {
             Self::Omp(_) => "omp",
         }
     }
+
+    pub fn session_driver(&self) -> SessionDriver {
+        match self {
+            Self::Claude(_) => SessionDriver::Claude,
+            Self::Codex(_) => SessionDriver::Codex,
+            Self::Pi(_) => SessionDriver::Pi,
+            Self::OpenCode(_) => SessionDriver::OpenCode,
+            Self::Omp(_) => SessionDriver::Omp,
+        }
+    }
 }
 
 /// Typed fields accepted by a `claude {}` driver block.
@@ -258,6 +342,8 @@ pub struct AgentSpec {
     pub session_driver: Option<SessionDriver>,
     /// Typed harness declaration used by task and render compilation.
     pub driver: Option<Driver>,
+    /// Non-secret admission facts for the managed delivery path.
+    pub delivery_readiness: Option<DeliveryReadiness>,
     /// Named typed references used by the agent. st2 preserves these for readers but does not
     /// resolve them or assign launch, readiness, access, or lifecycle semantics.
     pub resources: Vec<Resource>,
@@ -270,6 +356,14 @@ pub struct AgentSpec {
     pub path: PathBuf,
 }
 
+impl AgentSpec {
+    /// The explicit native session owner after typed-driver normalization.
+    pub fn effective_session_driver(&self) -> Option<SessionDriver> {
+        self.session_driver
+            .or_else(|| self.driver.as_ref().map(Driver::session_driver))
+    }
+}
+
 fn deserialize_optional_selector<'de, D>(
     deserializer: D,
 ) -> Result<Option<serde_json::Value>, D::Error>
@@ -603,6 +697,8 @@ pub(crate) struct RawSpec {
     /// Native session ownership asserted for an otherwise opaque launch.
     #[serde(default, deserialize_with = "deserialize_explicit_optional")]
     pub session_driver: Option<Option<String>>,
+    /// Non-secret facts used to admit the managed delivery path.
+    pub delivery_readiness: Option<DeliveryReadiness>,
     /// Direct typed provider driver block.
     #[serde(flatten)]
     pub driver: RawDriver,
@@ -1124,6 +1220,7 @@ impl RawSpec {
             || self.ding
             || self.deliver.is_some()
             || self.session_driver.is_some()
+            || self.delivery_readiness.is_some()
             || self.driver.claude.is_some()
             || self.driver.codex.is_some()
             // pi predates this predicate gaining driver awareness and was silently skipped too:
@@ -1171,26 +1268,43 @@ impl RawSpec {
             .transpose()?;
         let driver = self.driver.lower(&identity)?;
         let has_driver = driver.is_some();
+        let mut delivery_readiness = self.delivery_readiness;
+        if let Some(readiness) = delivery_readiness.as_mut() {
+            readiness.validate()?;
+        }
         anyhow::ensure!(
             !(self.ding && delivery.is_some()),
             "agent '{identity}' declares both `ding` and `deliver`; choose one transport"
         );
         anyhow::ensure!(
-            !(self.ding && has_driver),
-            "agent '{identity}' declares both `ding` and a typed driver; choose one session owner"
+            !(self.ding && (has_driver || session_driver.is_some() || delivery_readiness.is_some())),
+            "agent '{identity}' declares managed native delivery together with `ding`; generic Ding is only for opaque non-harness PTYs"
         );
-        if session_driver.is_some() {
-            anyhow::ensure!(
-                !self.ding,
-                "agent '{identity}' declares both `session-driver` and `ding`; choose one session owner"
-            );
+        anyhow::ensure!(
+            !(session_driver.is_some() && has_driver),
+            "agent '{identity}' declares both `session-driver` and a typed driver; choose one session owner"
+        );
+        let effective_session_driver =
+            session_driver.or_else(|| driver.as_ref().map(Driver::session_driver));
+        if let (Some(delivery), Some(effective)) = (delivery, effective_session_driver) {
             anyhow::ensure!(
-                delivery.is_none(),
-                "agent '{identity}' declares both `session-driver` and `deliver`; choose one session owner"
+                delivery.session_driver() == effective,
+                "agent '{identity}' delivery transport '{}' requires session-driver '{}', not '{}'",
+                delivery.as_str(),
+                delivery.session_driver().as_str(),
+                effective.as_str()
             );
+        }
+        if let (
+            Some(DeliveryReadiness::Anonymous { harness, .. }),
+            Some(effective),
+        ) = (delivery_readiness.as_ref(), effective_session_driver)
+        {
             anyhow::ensure!(
-                !has_driver,
-                "agent '{identity}' declares both `session-driver` and a typed driver; choose one session owner"
+                *harness == effective,
+                "agent '{identity}' anonymous delivery-readiness harness '{}' does not match effective session-driver '{}'",
+                harness.as_str(),
+                effective.as_str()
             );
         }
         validate_launch(
@@ -1345,6 +1459,7 @@ impl RawSpec {
             delivery,
             session_driver,
             driver,
+            delivery_readiness,
             resources,
             streams,
             tasks,
diff --git a/crates/agent-spec/tests/discovery.rs b/crates/agent-spec/tests/discovery.rs
index 36735be2..3cc8c235 100644
--- a/crates/agent-spec/tests/discovery.rs
+++ b/crates/agent-spec/tests/discovery.rs
@@ -13,8 +13,8 @@ use agent_spec::spec::{
     TaskLifecycle,
 };
 use agent_spec::{
-    AgentDesiredState, AgentSpec, JobType, Resource, SessionDriver, Task, discover, discover_file,
-    discover_strict,
+    AgentDesiredState, AgentSpec, DeliveryReadiness, JobType, Resource, SessionDriver, Task,
+    discover, discover_file, discover_strict,
 };
 
 #[test]
@@ -514,6 +514,83 @@ fn session_driver_is_closed_ownership_for_an_opaque_launch() {
     }
 }
 
+#[test]
+fn delivery_readiness_is_tagged_normalized_and_separate_from_activity() {
+    let tmp = tempfile::tempdir().unwrap();
+    write(
+        tmp.path(),
+        "agents/h/codex/agent.kdl",
+        r#"agent "codex" {
+  host "h"
+  argv "axe" "agent" "launch"
+  session-driver "codex"
+  deliver "app-server"
+  delivery-readiness "credential"
+}"#,
+    );
+    write(
+        tmp.path(),
+        "agents/h/omp/agent.kdl",
+        r#"agent "omp" {
+  host "h"
+  argv "axe" "agent" "launch"
+  session-driver "omp"
+  delivery-readiness "anonymous" "zeta" "alpha" "zeta" harness="omp"
+}"#,
+    );
+
+    let found = discover(tmp.path());
+    assert!(found.errors.is_empty(), "{:?}", found.errors);
+    assert_eq!(
+        find(&found.specs, "codex").delivery_readiness,
+        Some(DeliveryReadiness::Credential { account_id: None })
+    );
+    assert_eq!(
+        find(&found.specs, "omp").delivery_readiness,
+        Some(DeliveryReadiness::Anonymous {
+            harness: SessionDriver::Omp,
+            models: vec!["alpha".into(), "zeta".into()],
+        })
+    );
+}
+
+#[test]
+fn delivery_readiness_rejects_managed_ding_and_mismatched_native_ownership() {
+    for (name, declaration, expected) in [
+        (
+            "ding",
+            r#"agent "worker" { argv "axe"; ding; delivery-readiness "credential" }"#,
+            "generic Ding is only for opaque non-harness PTYs",
+        ),
+        (
+            "transport",
+            r#"agent "worker" { argv "axe"; session-driver "claude"; deliver "app-server"; delivery-readiness "credential" }"#,
+            "requires session-driver 'codex', not 'claude'",
+        ),
+        (
+            "anonymous",
+            r#"agent "worker" { argv "axe"; session-driver "omp"; delivery-readiness "anonymous" "model" harness="codex" }"#,
+            "does not match effective session-driver",
+        ),
+    ] {
+        let tmp = tempfile::tempdir().unwrap();
+        write(
+            tmp.path(),
+            &format!("agents/h/{name}/agent.kdl"),
+            declaration,
+        );
+        let found = discover(tmp.path());
+        assert!(found.specs.is_empty(), "{name}: {:?}", found.specs);
+        assert_eq!(found.errors.len(), 1, "{name}: {:?}", found.errors);
+        assert!(
+            found.errors[0].message.contains(expected),
+            "{name}: expected {expected:?}, got {:?}",
+            found.errors[0]
+        );
+    }
+}
+
+
 #[test]
 fn session_driver_rejects_unknown_duplicate_malformed_and_conflicting_declarations() {
     for (name, declaration, expected) in [
@@ -550,12 +627,12 @@ fn session_driver_rejects_unknown_duplicate_malformed_and_conflicting_declaratio
         (
             "ding",
             r#"agent "worker" { argv "axe"; session-driver "claude"; ding }"#,
-            "declares both `session-driver` and `ding`",
+            "generic Ding is only for opaque non-harness PTYs",
         ),
         (
             "deliver",
-            r#"agent "worker" { argv "axe"; session-driver "claude"; deliver "mcp" }"#,
-            "declares both `session-driver` and `deliver`",
+            r#"agent "worker" { argv "axe"; session-driver "claude"; deliver "app-server" }"#,
+            "requires session-driver 'codex', not 'claude'",
         ),
         (
             "driver",
@@ -633,7 +710,7 @@ fn typed_driver_blocks_reject_legacy_ding() {
         assert!(
             found.errors[0]
                 .message
-                .contains("declares both `ding` and a typed driver"),
+                .contains("generic Ding is only for opaque non-harness PTYs"),
             "{name}: {:?}",
             found.errors[0]
         );
diff --git a/docs/vrs/05-harness-state/spec.md b/docs/vrs/05-harness-state/spec.md
index 8d9ddee3..110fe4cb 100644
--- a/docs/vrs/05-harness-state/spec.md
+++ b/docs/vrs/05-harness-state/spec.md
@@ -183,10 +183,11 @@ comes first. The cross-check is a narrowing of the ungraceful-death window
 (provably dead sessions: pidfile present, process gone), not its closure —
 OHS-T04/OHS-R07 say exactly this, and no death tombstone is attempted: the
 kill that removes the registry entry leaves nothing behind to prove death
-with, and fabricating evidence is the one thing this design never does. And
-hosts running a codex-cli version outside the exact
-`SUPPORTED_CODEX_CLI_VERSIONS` allowlist produce no Codex observed state at all:
-the provider launch is refused before the control channel starts.
+with, and fabricating evidence is the one thing this design never does. A Codex
+binary whose delivery-critical schema projection does not match an admitted
+fingerprint produces no Codex observed state at all: provider launch is refused
+before the control channel starts. Fingerprint admission and live behavioral
+evidence remain separate.
 
 ## Codex producer (OHS-R05)
 
@@ -208,6 +209,7 @@ complement of steerable, a delivery predicate (decision 0001's boundary).
 | `Held { WaitingOnUserInput }` | `active` | `human` | `question` | `waitingOnUserInput` |
 | `Held { NotLoaded }` | *withhold* | — | — | thread not loaded proves nothing about work |
 | `Held { SystemError }` | *withhold* | — | — | see #264's catch-all defect |
+| `Held { UnknownStatus }` | *withhold* | — | — | an unrecognized future status is not a terminal `systemError` and cannot authorize delivery |
 
 `inputBuffer` is `unknown` from this producer: the control stream does not see
 the composer. The projection test must be behavioral — a table that would pass
diff --git a/docs/vrs/requirements.md b/docs/vrs/requirements.md
index 3cad1266..d44b08b8 100644
--- a/docs/vrs/requirements.md
+++ b/docs/vrs/requirements.md
@@ -26,6 +26,10 @@ accepted.
 
 - **T01 Explicit limits:** A documented unsupported case is preferable to a
   hidden distributed guarantee.
+- **T02 Native ownership over heuristic reach:** A managed harness that cannot
+  declare a matching native session driver and non-secret delivery readiness is
+  rejected rather than routed through generic terminal injection. Ding remains
+  available only for opaque non-harness PTYs.
 
 ## Requirements
 
@@ -332,3 +336,30 @@ accepted.
   identity recheck immediately before disruption. Renaming remains
   retire-old/add-new. Every behavior remains complete with an ordinary catalog
   folder and without CAS, captured generations, or replacement journals.
+- **R35 Authoritative admitted graph:** `st2 catalog graph --json` is the sole
+  catalog topology authority. For each uniquely admitted agent it publishes
+  the effective native session driver plus direct parent, root, depth, and
+  nearest-parent-first ancestor facts. Duplicate identity, missing parent,
+  supervisor cycle, bounded-depth overflow, and a per-host root count other
+  than exactly one are errors; affected topology facts are null and the graph
+  is incomplete. Consumers do not reimplement those generic graph rules.
+- **R36 Explicit native delivery readiness:** Managed Claude, Codex, pi,
+  OpenCode, OMP sessions declare their matching native session driver and one
+  tagged delivery-readiness value. Credential readiness may name a non-secret
+  account identifier or leave selection to the driver. Anonymous OMP readiness
+  names OMP and a non-empty normalized model set. Readiness is declaration
+  state, never inferred from activity, argv, process names, or credentials.
+- **R37 Retirement settles the inbox:** Every reconciliation of a retired
+  local agent first tears down every live owned task. Only after every teardown
+  attempt for that agent succeeds does the pass archive its canonical inbox
+  messages; any teardown failure leaves the entire inbox in place and retries
+  teardown plus settlement on the next reconciliation. The archive filename is
+  the durable receipt: repeated settlement and a sync-restored duplicate remove
+  the inbox copy without overwriting the archived bytes. Suspension retains the
+  inbox and does not settle it.
+- **R38 Codex schema admission:** Codex app-server launch is gated by an
+  admitted fingerprint of the delivery-critical schema projection. Admission
+  proves each emitted or consumed method discriminator is linked to its exact
+  payload arm and recursively covers the referenced definitions. Behavioral
+  turn, resume, and receipt evidence is reviewed separately; a schema match
+  alone does not claim it.
diff --git a/docs/vrs/spec.md b/docs/vrs/spec.md
index 2f609fbb..0f9cddba 100644
--- a/docs/vrs/spec.md
+++ b/docs/vrs/spec.md
@@ -735,6 +735,61 @@ call site. Eval run steps and agent log dumps stream child output straight to
 their catalog log files without buffering it. Rationale and rejected
 alternatives: [decision 0007](.decisions/0007-child-output-capture-is-bounded-and-tail-preserving.md).
 
+## Catalog graph and native delivery admission (R35–R38)
+
+Managed harness ownership is explicit. A declaration uses its typed driver or
+one `session-driver "claude|codex|pi|opencode|omp"` and pairs it with exactly
+one non-secret readiness declaration:
+
+```kdl
+delivery-readiness "credential"
+delivery-readiness "credential" account-id="tokengate/shared"
+delivery-readiness "anonymous" "model-a" "model-b" harness="omp"
+```
+
+Credential omission delegates account choice to the native driver. Anonymous
+readiness has at least one model; lowering sorts and deduplicates the model set,
+and `harness` must equal the effective native driver. A legacy `deliver` value
+must match that same explicit driver. No command-basename inference is
+admitted. A managed driver, readiness, or native delivery transport cannot
+coexist with Ding; Ding remains only for opaque non-harness PTYs.
+
+`st2 catalog graph --json` schema `st2.catalog-graph.v2` publishes
+`effectiveSessionDriver` and `deliveryReadiness` separately from `runtime`. It
+also publishes admitted topology:
+
+```json
+{
+  "parentId": "host.parent",
+  "rootId": "host.root",
+  "depth": 2,
+  "ancestorIds": ["host.parent", "host.root"]
+}
+```
+
+A root has null `parentId`, its own `rootId`, depth zero, and an empty ancestor
+array. Duplicate identity, missing or ambiguous parent, cycle, depth beyond 64,
+or a host with other than one root is an error. Every affected topology field
+is null and the graph envelope has `complete: false`; downstream consumers use
+these admitted facts rather than walking supervisor edges themselves.
+
+Retired reconciliation first attempts every live task teardown for the agent.
+Only when all of those attempts succeed does it settle the declaration's whole
+inbox; one failure leaves every inbox file untouched and the next pass retries
+teardown plus settlement. With no live tasks, settlement proceeds immediately.
+Each canonical inbox filename is linked into `resources/archive` and then
+removed from the inbox. An existing archive file wins byte-for-byte, so replay
+and a sync-restored duplicate converge without overwriting the receipt.
+Suspended reconciliation never performs this settlement.
+
+Before starting a Codex provider, st2 asks that binary to generate its
+app-server JSON schemas and fingerprints only the delivery-critical projection:
+every client request and server notification arm st2 uses, the exact
+method-to-`params` reference for each arm, recursively referenced definitions,
+and response definitions st2 reads. Only reviewed fingerprints are admitted.
+Live turn, resume, and durable-receipt evidence remains a separate behavioral
+check and is not inferred from the fingerprint.
+
 ## Message lifecycle
 
 ```text
diff --git a/src/catalog_graph.rs b/src/catalog_graph.rs
index 601a3888..350b9a05 100644
--- a/src/catalog_graph.rs
+++ b/src/catalog_graph.rs
@@ -9,7 +9,7 @@ use agent_spec::spec::AgentSpec;
 use anyhow::{Context, Result};
 use serde::Serialize;
 
-pub const CATALOG_GRAPH_SCHEMA: &str = "st2.catalog-graph.v1";
+pub const CATALOG_GRAPH_SCHEMA: &str = "st2.catalog-graph.v2";
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -45,7 +45,12 @@ pub struct GraphAgent {
     pub persona: Option<String>,
     pub workspace: Option<String>,
     pub resolved_workspace: Option<PathBuf>,
-    pub session_driver: Option<String>,
+    pub effective_session_driver: Option<String>,
+    pub delivery_readiness: Option<agent_spec::DeliveryReadiness>,
+    pub parent_id: Option<String>,
+    pub root_id: Option<String>,
+    pub depth: Option<usize>,
+    pub ancestor_ids: Option<Vec<String>>,
     pub desired_state: String,
     pub desired_state_reason: Option<String>,
     pub source: GraphSource,
@@ -134,7 +139,16 @@ pub fn snapshot(root: &Path, this_host: &str) -> Result<CatalogGraph> {
     let mut agents = found
         .specs
         .iter()
-        .map(|spec| graph_agent(&root, this_host, &found.declarations, spec, &mut runtime_by_path))
+        .map(|spec| {
+            graph_agent(
+                &root,
+                this_host,
+                &found.specs,
+                &found.declarations,
+                spec,
+                &mut runtime_by_path,
+            )
+        })
         .collect::<Vec<_>>();
     agents.sort_by(|left, right| left.id.cmp(&right.id).then(left.source.path.cmp(&right.source.path)));
 
@@ -183,6 +197,7 @@ pub fn snapshot(root: &Path, this_host: &str) -> Result<CatalogGraph> {
 fn graph_agent(
     root: &Path,
     this_host: &str,
+    specs: &[AgentSpec],
     declarations: &[DiscoveredDeclaration],
     spec: &AgentSpec,
     runtime_by_path: &mut BTreeMap<PathBuf, Vec<crate::agents::AgentRow>>,
@@ -204,10 +219,10 @@ fn graph_agent(
         )
         .ok()
     });
-    let session_driver = spec
-        .session_driver
-        .map(|driver| driver.as_str().to_owned())
-        .or_else(|| spec.driver.as_ref().map(|driver| driver.name().to_owned()));
+    let effective_session_driver = spec
+        .effective_session_driver()
+        .map(|driver| driver.as_str().to_owned());
+    let topology = admitted_topology(specs, spec, this_host);
 
     GraphAgent {
         id,
@@ -219,7 +234,12 @@ fn graph_agent(
         persona: spec.role.clone(),
         workspace: spec.workspace.clone(),
         resolved_workspace,
-        session_driver,
+        effective_session_driver,
+        delivery_readiness: spec.delivery_readiness.clone(),
+        parent_id: topology.as_ref().and_then(|facts| facts.parent_id.clone()),
+        root_id: topology.as_ref().map(|facts| facts.root_id.clone()),
+        depth: topology.as_ref().map(|facts| facts.depth),
+        ancestor_ids: topology.map(|facts| facts.ancestor_ids),
         desired_state: spec.desired_state.as_str().to_owned(),
         desired_state_reason: spec.desired_state.reason().map(str::to_owned),
         source: GraphSource {
@@ -254,6 +274,53 @@ fn graph_agent(
     }
 }
 
+struct AdmittedTopology {
+    parent_id: Option<String>,
+    root_id: String,
+    depth: usize,
+    ancestor_ids: Vec<String>,
+}
+
+fn admitted_topology(
+    specs: &[AgentSpec],
+    spec: &AgentSpec,
+    this_host: &str,
+) -> Option<AdmittedTopology> {
+    let id = spec.bus_id(this_host);
+    if specs
+        .iter()
+        .filter(|candidate| candidate.bus_id(this_host) == id)
+        .count()
+        != 1
+    {
+        return None;
+    }
+    let host = spec.resolved_host(this_host);
+    if specs
+        .iter()
+        .filter(|candidate| {
+            candidate.resolved_host(this_host) == host && candidate.supervisor.is_none()
+        })
+        .count()
+        != 1
+    {
+        return None;
+    }
+    let chain = crate::supervisor_chain::chain(specs, spec, this_host).ok()?;
+    let ancestor_ids = chain
+        .iter()
+        .skip(1)
+        .map(|ancestor| ancestor.bus_id(this_host))
+        .collect::<Vec<_>>();
+    let root_id = chain.last()?.bus_id(this_host);
+    Some(AdmittedTopology {
+        parent_id: ancestor_ids.first().cloned(),
+        root_id,
+        depth: ancestor_ids.len(),
+        ancestor_ids,
+    })
+}
+
 fn match_declared<'a>(
     declared: &'a [Declared],
     spec: &AgentSpec,
diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs
index 6d7f6c28..d992c93e 100644
--- a/src/codex_app_server.rs
+++ b/src/codex_app_server.rs
@@ -9,7 +9,7 @@
 //! inbox head and submits typed input only when that state proves an idle or one exact regular
 //! active turn.
 
-use std::collections::{BTreeMap, VecDeque};
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::fs::{self, File, OpenOptions};
 use std::io::{Read as _, Write};
 use std::net::Shutdown;
@@ -33,29 +33,40 @@ use tungstenite::{Message as WebSocketMessage, WebSocket};
 
 use crate::{ding, harness_context, harness_state, message, run, status};
 
-/// Every admitted version has a delivery-critical schema comparison and live remote-TUI evidence.
-/// A later version stays rejected until both checks are repeated; semantic-version proximity is
-/// not compatibility evidence for this experimental provider surface.
+/// A Codex binary is admitted by the exact delivery-critical schema projection st2 consumes, not
+/// by its release number. Behavioral evidence remains a separate gate: a matching schema does not
+/// claim that a model turn, resume, or durable receipt was exercised for an arbitrary build.
 ///
-/// `codex-cli 0.147.0` is admitted on a completed schema comparison against 0.146.0 and on live
-/// evidence that reached a submitted `turn/start` against the real binary. Its live evidence stops
-/// short of a completed model turn: the account was over its usage limit when the check ran. The
-/// `turn/start` response body, `turn/started`, `turn/completed`, the typed `item/completed`
-/// receipt, `turn/steer`, and the `thread/resume` subscription path are therefore unproven on this
-/// version. See #267.
-///
-/// `codex-cli 0.151.0` is admitted after comparing its delivery-critical source and schemas with
-/// 0.147.0. A live native run passed the exact version gate, initialized control, bound a fresh
-/// thread, submitted `turn/start`, and completed the model turn. Restarting the driver then proved
-/// `thread/resume` subscription and reconciled the attempted delivery to `Accepted` from the
-/// matching `clientId` in typed thread history. The live `item/completed` notification did not
-/// independently advance that receipt before restart, so only the durable resume receipt path is
-/// claimed here.
-pub const SUPPORTED_CODEX_CLI_VERSIONS: &[&str] = &[
-    "codex-cli 0.145.0",
-    "codex-cli 0.146.0",
-    "codex-cli 0.147.0",
-    "codex-cli 0.151.0",
+/// This fingerprint is the canonical projection generated by codex-cli 0.151.0. It covers every
+/// outbound method arm st2 emits, the inbound notification arms that drive delivery state, their
+/// recursively referenced payload definitions, and delivery-critical response definitions.
+const ADMITTED_CODEX_PROTOCOL_FINGERPRINTS: &[&str] =
+    &["40d024cc88b67c9d325edd944c87a460dc2fc1094a1b74bba7ccd2347aefe9b6"];
+const CODEX_PROTOCOL_SCHEMA: &str = "codex_app_server_protocol.v2.schemas.json";
+const CLIENT_NOTIFICATION_SCHEMA: &str = "ClientNotification.json";
+const CRITICAL_CLIENT_REQUESTS: &[(&str, &str)] = &[
+    ("initialize", "InitializeParams"),
+    ("hooks/list", "HooksListParams"),
+    ("thread/loaded/list", "ThreadLoadedListParams"),
+    ("thread/resume", "ThreadResumeParams"),
+    ("turn/start", "TurnStartParams"),
+    ("turn/steer", "TurnSteerParams"),
+];
+const CRITICAL_SERVER_NOTIFICATIONS: &[(&str, &str)] = &[
+    ("thread/started", "ThreadStartedNotification"),
+    ("thread/status/changed", "ThreadStatusChangedNotification"),
+    ("turn/started", "TurnStartedNotification"),
+    ("turn/completed", "TurnCompletedNotification"),
+    ("item/started", "ItemStartedNotification"),
+    ("item/completed", "ItemCompletedNotification"),
+    ("thread/compacted", "ContextCompactedNotification"),
+];
+const CRITICAL_RESPONSE_DEFINITIONS: &[&str] = &[
+    "HooksListResponse",
+    "ThreadLoadedListResponse",
+    "ThreadResumeResponse",
+    "TurnStartResponse",
+    "TurnSteerResponse",
 ];
 const RUNTIME_SCHEMA: &str = "st2.codex-runtime.v1";
 const BINDING_SCHEMA: &str = "st2.codex-thread-binding.v1";
@@ -210,6 +221,7 @@ pub enum CodexHoldReason {
     Compaction,
     NotLoaded,
     SystemError,
+    UnknownStatus,
     WaitingOnApproval,
     WaitingOnUserInput,
 }
@@ -270,7 +282,9 @@ impl CodexObservedState {
                 CodexHoldReason::ConflictingTurn => Some(
                     observation(Activity::Active, BlockedOn::None).with_reason("conflictingTurn"),
                 ),
-                CodexHoldReason::NotLoaded | CodexHoldReason::SystemError => None,
+                CodexHoldReason::NotLoaded
+                | CodexHoldReason::SystemError
+                | CodexHoldReason::UnknownStatus => None,
             },
         }
     }
@@ -1342,7 +1356,7 @@ impl CodexControlState {
                 turn_id: None,
             },
             _ => CodexObservedState::Held {
-                reason: CodexHoldReason::SystemError,
+                reason: CodexHoldReason::UnknownStatus,
                 turn_id: None,
             },
         };
@@ -1397,7 +1411,8 @@ impl CodexControlState {
                     | CodexHoldReason::ConflictingTurn
                     | CodexHoldReason::WaitingOnApproval
                     | CodexHoldReason::WaitingOnUserInput
-                    | CodexHoldReason::NotLoaded,
+                    | CodexHoldReason::NotLoaded
+                    | CodexHoldReason::UnknownStatus,
                 ..
             } => self.observed.clone(),
             CodexObservedState::Held {
@@ -1515,7 +1530,7 @@ pub fn run_controlled(
         !codex_argv.is_empty(),
         "Codex controlled launch argv is empty"
     );
-    ensure_supported_version(&codex_argv[0])?;
+    ensure_supported_protocol(&codex_argv[0])?;
     let delivery = CodexDeliveryConfig::resolve(catalog_root, &identity)?;
 
     let state_dir = state_dir(catalog_root, &identity);
@@ -2936,28 +2951,158 @@ fn completed_tui(status: ExitStatus) -> Result<()> {
     Ok(())
 }
 
-fn ensure_supported_version(codex: &str) -> Result<()> {
+fn ensure_supported_protocol(codex: &str) -> Result<()> {
+    let schema_dir = tempfile::tempdir().context("creating Codex schema admission directory")?;
     let output = Command::new(codex)
-        .arg("--version")
+        .args([
+            "app-server",
+            "generate-json-schema",
+            "--experimental",
+            "--out",
+        ])
+        .arg(schema_dir.path())
         .output()
-        .with_context(|| format!("reading Codex version from {codex}"))?;
+        .with_context(|| format!("generating Codex app-server schemas with {codex}"))?;
     anyhow::ensure!(
         output.status.success(),
-        "{codex} --version failed: {}",
+        "{codex} app-server schema generation failed: {}",
         String::from_utf8_lossy(&output.stderr).trim()
     );
-    let actual = String::from_utf8(output.stdout)
-        .context("Codex version output is not UTF-8")?
-        .trim()
-        .to_string();
+    let fingerprint = delivery_critical_schema_fingerprint(schema_dir.path())?;
     anyhow::ensure!(
-        SUPPORTED_CODEX_CLI_VERSIONS.contains(&actual.as_str()),
-        "unsupported Codex app-server protocol version '{actual}' (expected one of: {})",
-        SUPPORTED_CODEX_CLI_VERSIONS.join(", ")
+        ADMITTED_CODEX_PROTOCOL_FINGERPRINTS.contains(&fingerprint.as_str()),
+        "unsupported Codex app-server delivery schema fingerprint '{fingerprint}'"
     );
     Ok(())
 }
 
+fn delivery_critical_schema_fingerprint(schema_dir: &Path) -> Result<String> {
+    let aggregate = read_schema_json(&schema_dir.join(CODEX_PROTOCOL_SCHEMA))?;
+    let client_notification = read_schema_json(&schema_dir.join(CLIENT_NOTIFICATION_SCHEMA))?;
+    let projection = delivery_critical_schema_projection(&aggregate, &client_notification)?;
+    let canonical =
+        serde_json::to_vec(&projection).context("serializing Codex schema projection")?;
+    Ok(format!("{:x}", Sha256::digest(canonical)))
+}
+
+fn read_schema_json(path: &Path) -> Result<Value> {
+    serde_json::from_slice(
+        &fs::read(path).with_context(|| format!("reading Codex schema {}", path.display()))?,
+    )
+    .with_context(|| format!("parsing Codex schema {}", path.display()))
+}
+
+fn delivery_critical_schema_projection(
+    aggregate: &Value,
+    client_notification: &Value,
+) -> Result<Value> {
+    let definitions = aggregate
+        .get("definitions")
+        .and_then(Value::as_object)
+        .context("Codex aggregate schema has no definitions object")?;
+    let mut refs = BTreeSet::new();
+    let mut unions = BTreeMap::<String, Value>::new();
+    for (union, required) in [
+        ("ClientRequest", CRITICAL_CLIENT_REQUESTS),
+        ("ServerNotification", CRITICAL_SERVER_NOTIFICATIONS),
+    ] {
+        let schema = definitions
+            .get(union)
+            .with_context(|| format!("Codex aggregate schema has no {union} definition"))?;
+        let mut arms = BTreeMap::<String, Value>::new();
+        for (method, payload) in required {
+            let arm = required_method_arm(schema, method, Some(payload))?;
+            collect_definition_refs(&arm, &mut refs);
+            arms.insert((*method).to_owned(), arm);
+        }
+        unions.insert(union.to_owned(), serde_json::to_value(arms)?);
+    }
+
+    let initialized = required_method_arm(client_notification, "initialized", None)?;
+    for definition in CRITICAL_RESPONSE_DEFINITIONS {
+        refs.insert((*definition).to_owned());
+    }
+    let mut projected_definitions = BTreeMap::<String, Value>::new();
+    while let Some(name) = refs
+        .iter()
+        .find(|name| !projected_definitions.contains_key(*name))
+        .cloned()
+    {
+        let definition = definitions
+            .get(&name)
+            .with_context(|| format!("Codex schema reference '#/definitions/{name}' is missing"))?
+            .clone();
+        collect_definition_refs(&definition, &mut refs);
+        projected_definitions.insert(name, definition);
+    }
+
+    Ok(json!({
+        "arms": unions,
+        "clientNotification": initialized,
+        "definitions": projected_definitions,
+    }))
+}
+
+fn required_method_arm(
+    union: &Value,
+    method: &str,
+    expected_payload: Option<&str>,
+) -> Result<Value> {
+    let arms = union
+        .get("oneOf")
+        .and_then(Value::as_array)
+        .context("Codex protocol union has no oneOf arms")?;
+    let matches = arms
+        .iter()
+        .filter(|arm| {
+            arm.pointer("/properties/method/enum")
+                .and_then(Value::as_array)
+                .is_some_and(|values| {
+                    values.len() == 1 && values[0].as_str() == Some(method)
+                })
+        })
+        .collect::<Vec<_>>();
+    anyhow::ensure!(
+        matches.len() == 1,
+        "Codex protocol method '{method}' must have exactly one schema arm, found {}",
+        matches.len()
+    );
+    let arm = matches[0];
+    if let Some(payload) = expected_payload {
+        let expected = format!("#/definitions/{payload}");
+        let actual = arm
+            .pointer("/properties/params/$ref")
+            .and_then(Value::as_str);
+        anyhow::ensure!(
+            actual == Some(expected.as_str()),
+            "Codex protocol method '{method}' must reference payload '{expected}', found {}",
+            actual.unwrap_or("<missing>")
+        );
+    }
+    Ok(arm.clone())
+}
+
+fn collect_definition_refs(value: &Value, refs: &mut BTreeSet<String>) {
+    match value {
+        Value::Object(object) => {
+            if let Some(reference) = object.get("$ref").and_then(Value::as_str)
+                && let Some(name) = reference.strip_prefix("#/definitions/")
+            {
+                refs.insert(name.to_owned());
+            }
+            for child in object.values() {
+                collect_definition_refs(child, refs);
+            }
+        }
+        Value::Array(values) => {
+            for child in values {
+                collect_definition_refs(child, refs);
+            }
+        }
+        _ => {}
+    }
+}
+
 pub fn state_dir(catalog_root: &Path, identity: &str) -> PathBuf {
     let base = std::env::var_os("XDG_STATE_HOME")
         .map(PathBuf::from)
@@ -3366,51 +3511,99 @@ mod tests {
         crate::host_lock::process_alive(pid)
     }
 
-    #[test]
-    fn protocol_version_gate_accepts_only_the_exact_allowlist() {
-        assert_eq!(
-            SUPPORTED_CODEX_CLI_VERSIONS,
-            &[
-                "codex-cli 0.145.0",
-                "codex-cli 0.146.0",
-                "codex-cli 0.147.0",
-                "codex-cli 0.151.0",
-            ]
-        );
-        let tmp = tempfile::tempdir().unwrap();
-        let write_version = |name: &str, version: &str| {
-            let path = tmp.path().join(name);
-            fs::write(&path, format!("#!/bin/sh\nprintf '%s\\n' '{version}'\n")).unwrap();
-            fs::set_permissions(&path, fs::Permissions::from_mode(0o755)).unwrap();
-            path
+    fn protocol_schema_fixture() -> (Value, Value) {
+        let arm = |method: &str, payload: &str| {
+            json!({
+                "type": "object",
+                "properties": {
+                    "method": { "enum": [method] },
+                    "params": { "$ref": format!("#/definitions/{payload}") }
+                },
+                "required": ["method", "params"]
+            })
         };
-        for (index, version) in SUPPORTED_CODEX_CLI_VERSIONS.iter().enumerate() {
-            ensure_supported_version(
-                write_version(&format!("codex-admitted-{index}"), version)
-                    .to_str()
-                    .unwrap(),
-            )
-            .unwrap();
-        }
-        // An unadmitted release stays rejected until both policy checks are repeated for it, and
-        // an alpha cannot carry live evidence at all.
-        for (index, unadmitted) in ["codex-cli 0.144.0", "codex-cli 0.148.0-alpha.21"]
-            .into_iter()
-            .enumerate()
-        {
-            let error = ensure_supported_version(
-                write_version(&format!("codex-unadmitted-{index}"), unadmitted)
-                    .to_str()
-                    .unwrap(),
-            )
-            .unwrap_err();
-            assert!(error.to_string().contains(unadmitted));
-            assert!(
-                error
-                    .to_string()
-                    .contains(&SUPPORTED_CODEX_CLI_VERSIONS.join(", "))
+        let mut definitions = serde_json::Map::new();
+        for (union, required) in [
+            ("ClientRequest", CRITICAL_CLIENT_REQUESTS),
+            ("ServerNotification", CRITICAL_SERVER_NOTIFICATIONS),
+        ] {
+            definitions.insert(
+                union.to_owned(),
+                json!({
+                    "oneOf": required
+                        .iter()
+                        .map(|(method, payload)| arm(method, payload))
+                        .collect::<Vec<_>>()
+                }),
             );
+            for (_, payload) in required {
+                definitions.insert((*payload).to_owned(), json!({ "type": "object" }));
+            }
+        }
+        for response in CRITICAL_RESPONSE_DEFINITIONS {
+            definitions.insert((*response).to_owned(), json!({ "type": "object" }));
         }
+        (
+            json!({ "definitions": definitions }),
+            json!({
+                "oneOf": [{
+                    "type": "object",
+                    "properties": { "method": { "enum": ["initialized"] } },
+                    "required": ["method"]
+                }]
+            }),
+        )
+    }
+
+    #[test]
+    fn protocol_admission_proves_each_method_payload_linkage() {
+        let (mut aggregate, client_notification) = protocol_schema_fixture();
+        delivery_critical_schema_projection(&aggregate, &client_notification).unwrap();
+        let start = aggregate
+            .pointer_mut("/definitions/ClientRequest/oneOf")
+            .and_then(Value::as_array_mut)
+            .unwrap()
+            .iter_mut()
+            .find(|arm| {
+                arm.pointer("/properties/method/enum/0")
+                    .and_then(Value::as_str)
+                    == Some("turn/start")
+            })
+            .unwrap();
+        start["properties"]["params"]["$ref"] =
+            Value::String("#/definitions/TurnSteerParams".to_owned());
+
+        let error =
+            delivery_critical_schema_projection(&aggregate, &client_notification).unwrap_err();
+        assert!(error.to_string().contains("turn/start"));
+        assert!(error.to_string().contains("TurnStartParams"));
+    }
+
+    #[test]
+    #[ignore = "requires an installed Codex binary whose schema has completed admission review"]
+    fn installed_codex_protocol_fingerprint_is_admitted() {
+        ensure_supported_protocol("codex").unwrap();
+    }
+
+    #[test]
+    fn unknown_thread_status_remains_a_hold_not_a_terminal_system_error() {
+        let mut state = subscribed_state(CodexObservedState::Idle);
+        state.observe_thread_status("futureStatus", None);
+        assert_eq!(
+            state.observed(),
+            &CodexObservedState::Held {
+                reason: CodexHoldReason::UnknownStatus,
+                turn_id: None,
+            }
+        );
+        state.observe_turn_completed("turn-future");
+        assert_eq!(
+            state.observed(),
+            &CodexObservedState::Held {
+                reason: CodexHoldReason::UnknownStatus,
+                turn_id: None,
+            }
+        );
     }
 
     #[test]
diff --git a/src/driver.rs b/src/driver.rs
index f3b7239c..4f54cba8 100644
--- a/src/driver.rs
+++ b/src/driver.rs
@@ -397,6 +397,7 @@ mod tests {
             delivery: None,
             session_driver: None,
             driver: Some(driver),
+            delivery_readiness: None,
             resources: Vec::new(),
             streams: Vec::new(),
             tasks: Vec::new(),
diff --git a/src/eval_run.rs b/src/eval_run.rs
index e3b43459..8384e2cf 100644
--- a/src/eval_run.rs
+++ b/src/eval_run.rs
@@ -130,6 +130,7 @@ pub fn spec_to_agent_specs(agents: &[SpecAgent], host: &str, root: &Path) -> Vec
                 delivery: None,
                 session_driver: None,
                 driver: None,
+                delivery_readiness: None,
                 resources: Vec::new(),
                 streams: Vec::new(),
                 tasks,
@@ -1707,6 +1708,7 @@ mod tests {
             r#"agent "worker" {
   identity "worker"
   host "evalhost"
+  supervisor "evalhost.sup"
   workspace "$CATALOG/worker"
   argv "sh" "-c" "sleep 60"
 }
@@ -1749,6 +1751,7 @@ mod tests {
   host "evalhost"
   workspace "$CATALOG/worker"
   claude { prompt "Start the assigned work." }
+  delivery-readiness "credential"
 }
 "#,
         );
@@ -1918,7 +1921,7 @@ agent "worker" { identity "worker"; host "evalhost"; argv "true" }
                 )],
             ),
             (
-                "dangling-supervisor",
+                "supervisor-missing",
                 vec![(
                     "agents/evalhost/worker/agent.kdl",
                     r#"agent "worker" {
@@ -1969,6 +1972,7 @@ agent "worker" { identity "worker"; host "evalhost"; argv "true" }
                         r#"agent "two" {
   identity "two"
   host "evalhost"
+  supervisor "evalhost.one"
   pty "agent" { id "two-main"; command "sleep 60" }
   exec "poison" { id "shared"; command "true" }
 }"#,
diff --git a/src/message.rs b/src/message.rs
index 094fa757..30fde258 100644
--- a/src/message.rs
+++ b/src/message.rs
@@ -2202,6 +2202,19 @@ pub fn archive_msg(inbox_dir: &Path, archive_dir: &Path, filename: &str) -> anyh
     }
 }
 
+/// Settle every canonical message still present in an agent's inbox.
+///
+/// Retirement invokes this on every reconciliation pass. The archive receipt remains authoritative,
+/// so replay after an interrupted pass or a sync-restored inbox duplicate is idempotent.
+pub fn archive_inbox(agent_dir: &Path) -> anyhow::Result<()> {
+    let inbox = inbox_dir(agent_dir);
+    let archive = archive_dir(agent_dir);
+    for message in list_inbox(&inbox)? {
+        archive_msg(&inbox, &archive, &message.filename)?;
+    }
+    Ok(())
+}
+
 fn remove_inbox_duplicate(source: &Path, filename: &str) -> anyhow::Result<()> {
     match fs::remove_file(source) {
         Ok(()) => Ok(()),
diff --git a/src/reconcile.rs b/src/reconcile.rs
index 3a885867..8fe3c6b6 100644
--- a/src/reconcile.rs
+++ b/src/reconcile.rs
@@ -663,6 +663,8 @@ pub struct ReconcilePlan<'a> {
     pub launch: Vec<Launch<'a>>,
     /// This host, retired, with live sessions → kill them.
     pub teardown: Vec<Teardown<'a>>,
+    /// This host, retired → archive every inbox message, even when no session remains.
+    pub settle_retirement: Vec<&'a AgentSpec>,
     /// This host, active service, every declared task already present (live, or dead+`keep` frozen).
     pub adopt: Vec<&'a AgentSpec>,
     /// host != this machine → skipped; another machine's st2 owns it.
@@ -722,6 +724,9 @@ pub fn reconcile_selected<'a>(
     validate_task_identities(specs, this_host)?;
     let (owner, task, runtime) = resolve_task(specs, selector, this_host)?;
     let mut plan = ReconcilePlan::default();
+    if owner.desired_state.is_retired() {
+        plan.settle_retirement.push(owner);
+    }
     let actual = sessions.iter().find(|s| s.pty_id == runtime);
     if !owner.desired_state.is_running() {
         if let Some(s) = actual {
@@ -848,6 +853,9 @@ pub fn reconcile<'a>(
         let bus_id = spec.bus_id(this_host);
 
         if !spec.desired_state.is_running() {
+            if spec.desired_state.is_retired() {
+                plan.settle_retirement.push(spec);
+            }
             let mut teardown_ids = Vec::new();
             for t in &spec.tasks {
                 let id = resolve_task_id(&bus_id, &t.name, t.id.as_deref());
diff --git a/src/run.rs b/src/run.rs
index 8822912b..f1704311 100644
--- a/src/run.rs
+++ b/src/run.rs
@@ -1372,6 +1372,7 @@ fn execute_with_presentation_cursor(
     report: &mut UpReport,
     on_canonical_live: &mut dyn FnMut(&agent_spec::spec::AgentSpec),
 ) {
+
     // The corpses tied to a launch target (dead, non-keep, active ptys) are reaped inside the launch
     // loop so a parked flapper keeps its evidence. Everything else in `gc` (e.g. a retired agent's
     // dead sessions) is reaped here.
@@ -1499,13 +1500,33 @@ fn execute_with_presentation_cursor(
     // the same safe direction.
     cap.end_pass(Instant::now(), &plan.live);
 
+    let mut failed_retirement_teardowns = HashSet::new();
     for td in &plan.teardown {
+        let mut failed = false;
         for id in &td.pty_ids {
             match runner.kill(id) {
                 Ok(()) => report.torn_down.push(id.clone()),
-                Err(e) => report.errors.push(format!("kill {id}: {e}")),
+                Err(e) => {
+                    failed = true;
+                    report.errors.push(format!("kill {id}: {e}"));
+                }
             }
         }
+        if failed {
+            failed_retirement_teardowns.insert(td.spec.path.clone());
+        }
+    }
+    for spec in &plan.settle_retirement {
+        if failed_retirement_teardowns.contains(&spec.path) {
+            continue;
+        }
+        let agent_dir = spec.path.parent().unwrap_or_else(|| Path::new("."));
+        if let Err(error) = crate::message::archive_inbox(agent_dir) {
+            report.errors.push(format!(
+                "archive retired inbox for {}: {error:#}",
+                spec.identity
+            ));
+        }
     }
 
     // Presentation never delays lifecycle convergence. Drift repair is bounded to eight sequential
@@ -3217,6 +3238,7 @@ mod tests {
             delivery: None,
             session_driver: None,
             driver: None,
+            delivery_readiness: None,
             resources: vec![],
             streams: Vec::new(),
             tasks: vec![Task {
@@ -3273,6 +3295,7 @@ mod tests {
             delivery: None,
             session_driver: None,
             driver: None,
+            delivery_readiness: None,
             resources: vec![],
             streams: Vec::new(),
             tasks: vec![Task {
@@ -3838,6 +3861,7 @@ mod tests {
             delivery: None,
             session_driver: None,
             driver: None,
+            delivery_readiness: None,
             resources: vec![],
             streams: Vec::new(),
             tasks: vec![],
diff --git a/src/validate.rs b/src/validate.rs
index 192b2145..9f7d5b1a 100644
--- a/src/validate.rs
+++ b/src/validate.rs
@@ -235,22 +235,10 @@ pub(crate) fn validate_discovered(
     }
 
     // 4. Resolved pass: cross-spec + field checks over each agent.
-    let identities: HashSet<&str> = d.specs.iter().map(|s| s.identity.as_str()).collect();
     let mut seen: HashMap<String, PathBuf> = HashMap::new();
     // Placeholder host for bus-id collision: catalogs carry explicit host, and an empty host still
     // makes two unset-host same-identity specs collide (which is the real bug).
     let collision_host = "";
-    let addresses: HashSet<String> = d
-        .specs
-        .iter()
-        .flat_map(|s| {
-            let mut values = vec![s.identity.clone()];
-            if s.host.is_some() {
-                values.push(s.bus_id(collision_host));
-            }
-            values
-        })
-        .collect();
 
     for s in &d.specs {
         let rp = rel(root, &s.path);
@@ -295,6 +283,64 @@ pub(crate) fn validate_discovered(
             issues.push(Issue::error(code, rp.clone(), ag.clone(), error.clone()));
         }
 
+        let effective_session_driver = s.effective_session_driver();
+        if let Some(delivery) = s.delivery {
+            match effective_session_driver {
+                None => issues.push(Issue::error(
+                    "native-driver-missing",
+                    rp.clone(),
+                    ag.clone(),
+                    format!(
+                        "delivery transport '{}' requires an explicit matching native session driver",
+                        delivery.as_str()
+                    ),
+                )),
+                Some(driver) if driver != delivery.session_driver() => issues.push(Issue::error(
+                    "native-driver-mismatch",
+                    rp.clone(),
+                    ag.clone(),
+                    format!(
+                        "delivery transport '{}' requires native session driver '{}', found '{}'",
+                        delivery.as_str(),
+                        delivery.session_driver().as_str(),
+                        driver.as_str()
+                    ),
+                )),
+                Some(_) => {}
+            }
+        }
+        match (&s.delivery_readiness, effective_session_driver) {
+            (Some(_), None) => issues.push(Issue::error(
+                "native-driver-missing",
+                rp.clone(),
+                ag.clone(),
+                "delivery-readiness requires an explicit native session driver".to_string(),
+            )),
+            (None, Some(driver)) => issues.push(Issue::error(
+                "delivery-readiness-missing",
+                rp.clone(),
+                ag.clone(),
+                format!(
+                    "native session driver '{}' requires an explicit non-secret delivery-readiness declaration",
+                    driver.as_str()
+                ),
+            )),
+            (
+                Some(agent_spec::DeliveryReadiness::Anonymous { harness, .. }),
+                Some(driver),
+            ) if *harness != driver => issues.push(Issue::error(
+                "native-driver-mismatch",
+                rp.clone(),
+                ag.clone(),
+                format!(
+                    "anonymous delivery-readiness harness '{}' does not match native session driver '{}'",
+                    harness.as_str(),
+                    driver.as_str()
+                ),
+            )),
+            _ => {}
+        }
+
         // An explicit identity+host pair is authoritative regardless of folder names. When either
         // field is omitted, path defaults remain part of placement and mismatches stay advisory.
         let explicit_placement = s.host.as_ref().is_some_and(|host| {
@@ -357,18 +403,34 @@ pub(crate) fn validate_discovered(
             }
         }
 
-        // Runtime routing accepts either a bare identity or a fully-qualified <host>.<identity>.
-        // Validation must index the same address set or it rejects declarations the bus can route.
-        if let Some(sup) = &s.supervisor
-            && !identities.contains(sup.as_str())
-            && !addresses.contains(sup)
+        if let Err(error) =
+            crate::supervisor_chain::chain(&d.specs, s, this_host.unwrap_or_default())
         {
-            issues.push(Issue::warn(
-                "dangling-supervisor",
-                rp.clone(),
-                ag.clone(),
-                format!("supervisor '{sup}' is not an agent in this catalog"),
-            ));
+            let (code, message) = match error {
+                crate::supervisor_chain::SupervisorChainError::MissingSupervisor => (
+                    "supervisor-missing",
+                    format!(
+                        "supervisor chain from '{}' references a missing or ambiguous parent",
+                        s.bus_id(this_host.unwrap_or_default())
+                    ),
+                ),
+                crate::supervisor_chain::SupervisorChainError::Cycle => (
+                    "supervisor-cycle",
+                    format!(
+                        "supervisor chain from '{}' contains a cycle",
+                        s.bus_id(this_host.unwrap_or_default())
+                    ),
+                ),
+                crate::supervisor_chain::SupervisorChainError::DepthLimit => (
+                    "supervisor-depth",
+                    format!(
+                        "supervisor chain from '{}' exceeds the maximum depth of {}",
+                        s.bus_id(this_host.unwrap_or_default()),
+                        crate::supervisor_chain::SUPERVISOR_CHAIN_LIMIT
+                    ),
+                ),
+            };
+            issues.push(Issue::error(code, rp.clone(), ag.clone(), message));
         }
 
         // Overlay lint: render's persona overlay `@import`s must resolve (WARN — render concern).
@@ -385,6 +447,30 @@ pub(crate) fn validate_discovered(
         }
     }
 
+    let mut root_counts: HashMap<String, usize> = HashMap::new();
+    for spec in &d.specs {
+        let host = spec.resolved_host(this_host.unwrap_or_default()).to_owned();
+        root_counts.entry(host).or_default();
+        if spec.supervisor.is_none() {
+            *root_counts
+                .get_mut(spec.resolved_host(this_host.unwrap_or_default()))
+                .expect("root count entry was just inserted") += 1;
+        }
+    }
+    for (host, count) in root_counts {
+        if count != 1 {
+            issues.push(Issue::error(
+                "root-count",
+                ".".to_string(),
+                None,
+                format!(
+                    "host '{}' must declare exactly one root agent; found {count}",
+                    if host.is_empty() { "<default>" } else { &host }
+                ),
+            ));
+        }
+    }
+
     if let Some(host) = this_host {
         for conflict in crate::materialize::render_ownership_conflicts(root, &d.specs, host) {
             issues.push(Issue::error(
diff --git a/tests/catalog_graph.rs b/tests/catalog_graph.rs
index d1cfe351..bb1db7bc 100644
--- a/tests/catalog_graph.rs
+++ b/tests/catalog_graph.rs
@@ -62,7 +62,7 @@ fn graph_preserves_valid_rows_broken_sources_conflicts_and_incompleteness() {
     let output = st2(root, &["catalog", "graph", "--host", "h", "--json"], None);
     assert_eq!(output.status.code(), Some(1));
     let graph = json(&output);
-    assert_eq!(graph["schema"], "st2.catalog-graph.v1");
+    assert_eq!(graph["schema"], "st2.catalog-graph.v2");
     assert_eq!(graph["complete"], false);
     assert_eq!(graph["roots"]["ptyRoot"], root.join("pty").display().to_string());
 
@@ -77,7 +77,11 @@ fn graph_preserves_valid_rows_broken_sources_conflicts_and_incompleteness() {
     assert_eq!(lead["persona"], "worker");
     assert_eq!(lead["workspace"], "./.workspace");
     assert_eq!(lead["resolvedWorkspace"], root.join("agents/h/lead/.workspace").display().to_string());
-    assert_eq!(lead["sessionDriver"], "claude");
+    assert_eq!(lead["effectiveSessionDriver"], "claude");
+    assert!(lead["parentId"].is_null());
+    assert!(lead["rootId"].is_null());
+    assert!(lead["depth"].is_null());
+    assert!(lead["ancestorIds"].is_null());
     assert_eq!(lead["desiredState"], "suspended");
     assert_eq!(lead["desiredStateReason"], "Waiting for capacity");
     assert_eq!(lead["source"]["identityProvenance"], "declaration");
@@ -106,6 +110,144 @@ fn graph_preserves_valid_rows_broken_sources_conflicts_and_incompleteness() {
     assert!(issue_codes.contains(&"unknown-task-kind"));
 }
 
+#[test]
+fn graph_exposes_admitted_topology_and_delivery_readiness_facts() {
+    let catalog = tempfile::tempdir().unwrap();
+    let root = catalog.path();
+    write(
+        root,
+        "agents/h/root/agent.kdl",
+        r#"agent "root" { host "h"; command "true" }"#,
+    );
+    write(
+        root,
+        "agents/h/worker/agent.kdl",
+        r#"agent "worker" {
+  host "h"
+  supervisor "h.root"
+  argv "axe" "agent" "launch"
+  session-driver "omp"
+  delivery-readiness "anonymous" "zeta" "alpha" "zeta" harness="omp"
+}"#,
+    );
+
+    let output = st2(root, &["catalog", "graph", "--host", "h", "--json"], None);
+    assert!(
+        output.status.success(),
+        "{}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let graph = json(&output);
+    assert_eq!(graph["complete"], true, "{graph:#}");
+    let rows = graph["agents"].as_array().unwrap();
+    let root_row = rows.iter().find(|row| row["id"] == "h.root").unwrap();
+    assert!(root_row["parentId"].is_null());
+    assert_eq!(root_row["rootId"], "h.root");
+    assert_eq!(root_row["depth"], 0);
+    assert_eq!(root_row["ancestorIds"], serde_json::json!([]));
+
+    let worker = rows.iter().find(|row| row["id"] == "h.worker").unwrap();
+    assert_eq!(worker["effectiveSessionDriver"], "omp");
+    assert_eq!(worker["parentId"], "h.root");
+    assert_eq!(worker["rootId"], "h.root");
+    assert_eq!(worker["depth"], 1);
+    assert_eq!(worker["ancestorIds"], serde_json::json!(["h.root"]));
+    assert_eq!(
+        worker["deliveryReadiness"],
+        serde_json::json!({
+            "kind": "anonymous",
+            "harness": "omp",
+            "models": ["alpha", "zeta"]
+        })
+    );
+}
+
+
+#[test]
+fn graph_rejects_missing_cycle_depth_and_per_host_root_count() {
+    let issue_codes = |root: &Path| {
+        let output = st2(root, &["catalog", "graph", "--host", "h", "--json"], None);
+        assert_eq!(output.status.code(), Some(1));
+        let graph = json(&output);
+        assert_eq!(graph["complete"], false);
+        graph["issues"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|issue| issue["code"].as_str().unwrap().to_owned())
+            .collect::<Vec<_>>()
+    };
+
+    let missing = tempfile::tempdir().unwrap();
+    write(
+        missing.path(),
+        "agents/h/root/agent.kdl",
+        r#"agent "root" { host "h"; command "true" }"#,
+    );
+    write(
+        missing.path(),
+        "agents/h/worker/agent.kdl",
+        r#"agent "worker" { host "h"; supervisor "h.absent"; command "true" }"#,
+    );
+    assert!(issue_codes(missing.path()).contains(&"supervisor-missing".to_owned()));
+
+    let cycle = tempfile::tempdir().unwrap();
+    write(
+        cycle.path(),
+        "agents/h/one/agent.kdl",
+        r#"agent "one" { host "h"; supervisor "h.two"; command "true" }"#,
+    );
+    write(
+        cycle.path(),
+        "agents/h/two/agent.kdl",
+        r#"agent "two" { host "h"; supervisor "h.one"; command "true" }"#,
+    );
+    let cycle_codes = issue_codes(cycle.path());
+    assert!(cycle_codes.contains(&"supervisor-cycle".to_owned()));
+    assert!(cycle_codes.contains(&"root-count".to_owned()));
+
+    let deep = tempfile::tempdir().unwrap();
+    for index in 0..=64 {
+        let supervisor = if index == 0 {
+            String::new()
+        } else {
+            format!(" supervisor \"h.node{}\";", index - 1)
+        };
+        write(
+            deep.path(),
+            &format!("agents/h/node{index}/agent.kdl"),
+            &format!(
+                "agent \"node{index}\" {{ host \"h\";{supervisor} command \"true\" }}\n"
+            ),
+        );
+    }
+    assert!(issue_codes(deep.path()).contains(&"supervisor-depth".to_owned()));
+
+    let roots = tempfile::tempdir().unwrap();
+    for identity in ["one", "two"] {
+        write(
+            roots.path(),
+            &format!("agents/h/{identity}/agent.kdl"),
+            &format!("agent \"{identity}\" {{ host \"h\"; command \"true\" }}\n"),
+        );
+    }
+    let roots_output = st2(
+        roots.path(),
+        &["catalog", "graph", "--host", "h", "--json"],
+        None,
+    );
+    assert_eq!(roots_output.status.code(), Some(1));
+    let roots_graph = json(&roots_output);
+    assert!(roots_graph["issues"].as_array().unwrap().iter().any(|issue| {
+        issue["code"] == "root-count"
+    }));
+    for row in roots_graph["agents"].as_array().unwrap() {
+        assert!(row["parentId"].is_null());
+        assert!(row["rootId"].is_null());
+        assert!(row["depth"].is_null());
+        assert!(row["ancestorIds"].is_null());
+    }
+}
 #[test]
 fn candidate_overlay_reports_conflict_on_stdout_and_never_publishes() {
     let catalog = tempfile::tempdir().unwrap();
diff --git a/tests/reconcile.rs b/tests/reconcile.rs
index 935bb383..a3814231 100644
--- a/tests/reconcile.rs
+++ b/tests/reconcile.rs
@@ -436,6 +436,7 @@ fn spec(
         delivery: None,
         session_driver: None,
         driver: None,
+        delivery_readiness: None,
         resources: Vec::new(),
         streams: Vec::new(),
         tasks,
diff --git a/tests/run.rs b/tests/run.rs
index 5303bdc6..69feedb9 100644
--- a/tests/run.rs
+++ b/tests/run.rs
@@ -240,6 +240,7 @@ fn task_spec(identity: &str, host: Option<&str>, id: &str) -> AgentSpec {
         delivery: None,
         session_driver: None,
         driver: None,
+        delivery_readiness: None,
         resources: vec![],
         streams: Vec::new(),
         tasks: vec![Task {
@@ -481,6 +482,7 @@ struct FakeRunner {
     fail_list: bool,
     fail_spawn: Option<String>,
     fail_reap: Option<String>,
+    fail_kill: Option<String>,
     spawned: RefCell<Vec<String>>,
     spawned_targets: RefCell<Vec<TaskTarget>>,
     spawn_dirs: RefCell<Vec<(String, String)>>,
@@ -489,6 +491,7 @@ struct FakeRunner {
     removed: RefCell<Vec<String>>,
     patched: RefCell<Vec<String>>,
     ops: RefCell<Vec<String>>,
+    inbox_expected_during_kill: Option<std::path::PathBuf>,
 }
 
 impl Runner for FakeRunner {
@@ -515,6 +518,15 @@ impl Runner for FakeRunner {
     }
     fn kill(&self, pty_id: &str) -> anyhow::Result<()> {
         self.ops.borrow_mut().push(format!("kill:{pty_id}"));
+        if let Some(inbox) = &self.inbox_expected_during_kill {
+            assert!(
+                inbox.exists(),
+                "retirement must stop live work before archiving the inbox"
+            );
+        }
+        if self.fail_kill.as_deref() == Some(pty_id) {
+            anyhow::bail!("simulated kill failure");
+        }
         self.killed.borrow_mut().push(pty_id.to_string());
         Ok(())
     }
@@ -1062,6 +1074,79 @@ command = "st2 ding hetz.demo"
     assert!(report.launched.is_empty());
 }
 
+#[test]
+fn retired_agent_idempotently_archives_every_inbox_message() {
+    let tmp = tempfile::tempdir().unwrap();
+    let agent_dir = tmp.path().join("agents/hetz/demo");
+    write(
+        tmp.path(),
+        "agents/hetz/demo/agent.kdl",
+        r#"agent "demo" {
+  host "hetz"
+  desired-state "retired" reason="Work complete"
+  command "true"
+}"#,
+    );
+    let filename = "1784649988123-abc23z.md";
+    let inbox = message::inbox_dir(&agent_dir);
+    let archive = message::archive_dir(&agent_dir);
+    fs::create_dir_all(&inbox).unwrap();
+    fs::write(inbox.join(filename), "first receipt").unwrap();
+    let runner = FakeRunner {
+        sessions: vec![live("hetz.demo")],
+        inbox_expected_during_kill: Some(inbox.join(filename)),
+        ..Default::default()
+    };
+
+    let first = up_once(tmp.path(), "hetz", &runner).unwrap();
+    assert!(first.errors.is_empty(), "{:?}", first.errors);
+    assert_eq!(
+        fs::read_to_string(archive.join(filename)).unwrap(),
+        "first receipt"
+    );
+    assert!(!inbox.join(filename).exists());
+
+    fs::write(inbox.join(filename), "restored duplicate").unwrap();
+    let failing_runner = FakeRunner {
+        sessions: vec![live("hetz.demo")],
+        fail_kill: Some("hetz.demo".into()),
+        inbox_expected_during_kill: Some(inbox.join(filename)),
+        ..Default::default()
+    };
+    let second = up_once(tmp.path(), "hetz", &failing_runner).unwrap();
+    assert!(
+        second
+            .errors
+            .iter()
+            .any(|error| error.contains("kill hetz.demo")),
+        "{:?}",
+        second.errors
+    );
+    assert_eq!(
+        fs::read_to_string(archive.join(filename)).unwrap(),
+        "first receipt",
+        "the durable archive receipt must never be overwritten"
+    );
+    assert_eq!(
+        fs::read_to_string(inbox.join(filename)).unwrap(),
+        "restored duplicate",
+        "failed teardown must leave the inbox available to the still-live agent"
+    );
+
+    let retry_runner = FakeRunner {
+        sessions: vec![live("hetz.demo")],
+        inbox_expected_during_kill: Some(inbox.join(filename)),
+        ..Default::default()
+    };
+    let retry = up_once(tmp.path(), "hetz", &retry_runner).unwrap();
+    assert!(retry.errors.is_empty(), "{:?}", retry.errors);
+    assert!(!inbox.join(filename).exists());
+    assert_eq!(
+        fs::read_to_string(archive.join(filename)).unwrap(),
+        "first receipt"
+    );
+}
+
 #[test]
 fn retired_compact_agent_stops_agent_and_derived_ding() {
     let tmp = tempfile::tempdir().unwrap();
diff --git a/tests/validate.rs b/tests/validate.rs
index 0f981d47..e2eb8797 100644
--- a/tests/validate.rs
+++ b/tests/validate.rs
@@ -49,13 +49,14 @@ fn compact_agent_catalog_is_clean() {
     assert_eq!(r.warnings(), 0, "unexpected warnings: {:?}", r.issues);
 }
 #[test]
-fn opaque_session_driver_launch_is_clean_without_ding() {
+fn native_session_driver_is_clean_with_explicit_readiness() {
     let c = catalog(&[(
         "h/worker/agent.kdl",
         r#"agent "worker" {
   host "h"
   argv "axe" "agent" "launch"
   session-driver "claude"
+  delivery-readiness "credential" account-id="tokengate/shared"
 }"#,
     )]);
 
@@ -64,6 +65,30 @@ fn opaque_session_driver_launch_is_clean_without_ding() {
     assert_eq!(report.agents, 1);
 }
 
+#[test]
+fn native_delivery_requires_explicit_matching_driver_and_readiness() {
+    for (name, declaration, code) in [
+        (
+            "legacy-deliver",
+            r#"agent "legacy-deliver" { host "h"; command "codex"; deliver "app-server" }"#,
+            "native-driver-missing",
+        ),
+        (
+            "missing-readiness",
+            r#"agent "missing-readiness" { host "h"; argv "axe"; session-driver "codex" }"#,
+            "delivery-readiness-missing",
+        ),
+    ] {
+        let c = catalog(&[(&format!("h/{name}/agent.kdl"), declaration)]);
+        let report = validate(c.path());
+        assert!(
+            has(&report, code, Severity::Error),
+            "{name}: {:?}",
+            report.issues
+        );
+    }
+}
+
 #[test]
 fn explicit_and_typed_session_drivers_reject_ding() {
     for (name, body) in [