feat(ding): add harness-neutral guarded adapter seam

INVARIANTS.md

@@ -11,6 +11,7 @@ materialization, messaging, DING, or presence must preserve them.
 | **Bounded restart diagnostics** | Relaunching an exec task preserves the just-finished log as one prior generation while bounding retained diagnostics to current plus prior. Final retirement removes the PID and both logs. | `tests/exec_backend.rs::exec_restart_reap_keeps_bounded_diagnostics_and_final_remove_cleans_them`; `tests/run.rs::up_once_finally_removes_dead_retired_tasks_without_restarting_them` |
 | **Exactly-once-safe native bus** | Messages use stable `<unix-ms>-<rand6>.md` files. An archive filename is a durable receipt that shadows and cleans restored inbox replicas and makes repeated archive cleanup idempotent. | `src/message.rs::filename_grammar`; `src/message.rs::archive_receipt_suppresses_and_idempotently_cleans_a_restored_inbox_copy`; `tests/message.rs` |
 | **Fail-closed observed native DING** | Each unread message becomes one normalized `[DING]` frame. Fresh delivery records ownership, then preserves the one combined bracketed-paste, 0.5 second delay, and Return transaction. PTY and Return success are transport only: `Delivered` additionally requires adapter classification of the expected notice text in a submitted-prompt or queued-message pattern while the lowest live composer is empty or an accepted idle placeholder. Every other receipt retains staged ownership. Retry never re-pastes and may send one bare Return only after two adjacent `RetainedSafe` observations; human, modal, active, changed, timed-out, and unknown retry states receive no input. Ownership prevents duplicate paste across command failures, receipt ambiguity, archive races, and restart adoption. Startup backlog otherwise becomes one generic recovery DING; new arrivals remain FIFO; `busy` delivers immediately; only fresh `dnd` defers. | `src/ding/mod.rs::poke_text_normalizes_and_bounds_untrusted_fields`; `src/ding/mod.rs::malicious_controls_cannot_escape_the_single_paste_frame`; `src/ding/mod.rs::pty_delivery_uses_face607_delay_order_and_seconds`; `src/ding/mod.rs::maintained_composer_classifiers_require_exact_idle_state`; `src/ding/mod.rs::successful_transport_with_retained_or_unproven_pixels_is_not_delivered`; `src/ding/mod.rs::ambiguous_transport_receipt_and_retry_errors_retain_staged_ownership`; `src/ding/mod.rs::adapter_recognized_notice_with_an_empty_live_composer_is_a_positive_receipt`; `src/ding/mod.rs::staged_retry_submits_only_retained_safe_and_requires_a_receipt`; `src/ding/mod.rs::staged_ownership_survives_archive_and_never_repastes`; `src/ding/mod.rs::pty_commands_have_a_real_outer_timeout`; `src/ding/mod.rs::session_watch_has_startup_grace_debounce_and_live_reset`; `src/ding/mod.rs::new_arrivals_is_fifo_and_archive_receipts_prevent_reding`; `src/ding/mod.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry`; `src/ding/mod.rs::startup_recovery_notice_retries_in_memory`; `src/ding/mod.rs::startup_backlog_gets_one_generic_recovery_then_new_arrivals_poke` |
+| **Opt-in harness-neutral DING fence** | Adapter presence selects a structured-argv rich path without changing bare DING. Core accepts only receipt-fresh `idle` plus `empty` input-buffer evidence whose session, incarnation, generation, and sequence match PTY's live activity snapshot. Exact attempt ownership is durable before the existing DING bytes cross one atomic generation/revision guard, so restart or ambiguous transport cannot replay them; only a proven zero-byte conflict clears the attempt. Every non-idle, stale, malformed, mismatched, or conflicting case retains FIFO. A successful write becomes PTY-owned without claiming harness acceptance. Separately, exact currently unread filenames injected by a provider-rendered turn-boundary hook become durable hook-owned work with no PTY input until archive. | `crates/agent-spec/tests/discovery.rs::compact_ding_adapter_lowers_structured_argv_without_a_shell`; `crates/agent-spec/tests/discovery.rs::compact_ding_adapter_rejects_ambiguous_or_incomplete_shapes`; `src/ding/adapter.rs::only_fresh_idle_and_empty_authorizes`; `src/ding/adapter.rs::every_non_idle_or_nonempty_input_state_holds`; `src/ding/adapter.rs::malformed_identity_sequence_and_tuple_changes_fail_closed`; `src/ding/guarded_pty.rs::reads_exact_activity_snapshot_and_sends_generation_revision_guard`; `src/ding/guarded_pty.rs::guard_conflict_is_typed_and_never_reinterpreted_as_success`; `src/ding/hook_control.rs::ingress_accepts_only_exact_currently_unread_filenames_and_survives_reload`; `src/ding/hook_control.rs::pty_ownership_is_durable_until_conflict_clear_or_archive`; `src/ding/mod.rs::rich_delivery_requires_matching_fresh_tuple_and_becomes_pty_owned_once`; `src/ding/mod.rs::rich_tuple_mismatch_and_guard_conflict_send_zero_or_one_guarded_packet`; `src/ding/mod.rs::exact_hook_ownership_removes_only_named_unread_work_without_pty_input`; `src/ding/mod.rs::durable_pty_ownership_blocks_replay_after_sidecar_restart`; `tests/ding_control.rs::hook_owned_cli_records_only_exact_unread_filenames`; `tests/exec_backend.rs::generated_rich_ding_uses_task_only_roots_without_shell_expansion` |
 | **Mutation-only filesystem wakeups** | Supervisor and DING filesystem watchers ignore read/open access events and wake early only for create, modify, rename, or remove events. Their own catalog and inbox reads therefore cannot bypass the bounded timer cadence or form a Linux inotify CPU loop. | `src/watch.rs::only_mutations_wake_watch_loops`; `src/watch.rs::linux_reads_are_silent_but_real_mutations_wake`; `src/ding/mod.rs::idle_ding_does_not_spin_on_its_own_inbox_reads`; `src/run.rs::idle_supervisor_does_not_spin_on_its_own_catalog_reads` |
 | **Bounded DING PTY probe churn** | An unsafe or active composer retains its FIFO notice but deferred delivery retries use a bounded backoff, so each inbox poll cannot spawn another short-lived PTY probe. | `src/ding/mod.rs::deferred_delivery_backoff_bounds_short_lived_pty_attempts` |
 | **Agent-declared presence discipline** | The shipped bus contract requires agents to declare `busy` before executing work, use `available` only while yielding or ready, and reserve `dnd` for an explicit hold. Both native harnesses materialize that contract. Busy remains observable but does not suppress DING; fresh `dnd` is the only delivery gate. | `tests/compile_agent.rs::compile_agent_generates_claude_then_materializes_verbatim_persona`; `tests/compile_agent.rs::compile_agent_generates_codex_then_materializes_composed_agents_md`; `src/ding/mod.rs::pending_delivery_ignores_busy_but_respects_fresh_dnd_archive_and_retry` |

README.md

@@ -161,9 +161,45 @@ opaque to st2; catalog readers use the public `agent-spec` crate to inspect the
 
 `argv` launches its first value directly with the remaining values as arguments. It resolves a bare
 program such as `codex` through the task environment's `PATH`, preserves argument boundaries, and
-does not introduce a shell. Use `command #"..."#` instead when the task intentionally needs shell
-syntax such as pipelines, redirects, or variable expansion; `command` continues to run under
-`sh -c`. A runnable task must declare exactly one of `argv` or `command`.
+does not introduce a shell. `$VAR` tokens are expanded only after st2 has resolved the complete
+managed task environment, so st2-owned `CATALOG`/`ST_ROOT` and declared `env` values are available
+to both PTY and exec argv. Use `command #"..."#` instead when the task intentionally needs shell
+syntax such as pipelines or redirects; `command` continues to run under `sh -c`. A runnable task
+must declare exactly one of `argv` or `command`.
+
+An experimental rich DING path can be selected by giving compact `ding` one generic activity
+adapter. The adapter is also structured argv: core introduces no shell, provider selector, arguments, or
+adapter-specific environment. Its tokens resolve against the generated DING task's final managed
+environment, including the agent's declared `env`.
+
+```kdl
+env { ADAPTER_ROOT "/opt/agent-adapters" }
+ding {
+  adapter {
+    argv "$ADAPTER_ROOT/bin/activity" "--format" "jsonl"
+  }
+}
+```
+
+The adapter must publish the matching harness-neutral PTY activity lease and emit the generic
+JSONL input-buffer/freshness evidence described in
+[`docs/vrs/01-ding/spec.md`](docs/vrs/01-ding/spec.md). Only a fresh exact `idle` + `empty` tuple can
+reach PTY's generation/revision-guarded write. Every mismatch, stale event, adapter failure, or
+nonempty/unknown input buffer holds the durable FIFO and writes zero bytes. Omitted/bare `ding`
+preserves the existing delivery path exactly; configured rich DING never falls back to it.
+st2 records exact PTY attempt ownership before the guarded packet, so a transport error or sidecar
+restart cannot replay possibly-written bytes; only a proven zero-byte conflict clears that attempt.
+
+A provider-rendered turn-boundary hook that already injected exact unread files into the next
+context records that fact separately:
+
+```sh
+st2 ding-control --identity <host.identity> hook-owned \
+  --message <exact-unread-filename.md>
+```
+
+The control ingress does not install hooks, parse provider JSON, trigger a model call, or send PTY
+input. It verifies exact unread filenames and keeps ownership durable until archive.
 
 ### Scheduled work is coming soon, not implemented
 
@@ -357,7 +393,7 @@ st2 service uninstall
 
 ```text
 ls, up, down, validate, doctor
-message, ding, agents, status, context, resource
+message, ding, ding-control, agents, status, context, resource
 env, pty, shell, pretrust
 hooks, service, eval
 compile-agent (experimental)

crates/agent-spec/src/kdl_format.rs

@@ -91,7 +91,16 @@ fn agent_node_to_raw(node: &KdlNode) -> anyhow::Result<RawSpec> {
             }
             "command" => raw.command = arg_string(child),
             "argv" => raw.argv = Some(argv(child)?),
-            "ding" => raw.ding = true,
+            "ding" => {
+                if raw.ding {
+                    anyhow::bail!(
+                        "agent '{}' declares compact `ding` more than once",
+                        raw.identity.as_deref().unwrap_or("<unnamed>")
+                    );
+                }
+                raw.ding = true;
+                raw.ding_adapter_argv = ding_adapter_argv(child)?;
+            }
             "env" => {}
             "pty" => {
                 if let Some(name) = arg_string(child) {
@@ -116,6 +125,58 @@ fn agent_node_to_raw(node: &KdlNode) -> anyhow::Result<RawSpec> {
     Ok(raw)
 }
 
+fn ding_adapter_argv(node: &KdlNode) -> anyhow::Result<Option<Vec<String>>> {
+    if !node.entries().is_empty() {
+        anyhow::bail!("compact `ding` accepts no arguments or properties");
+    }
+    let Some(children) = node.children() else {
+        return Ok(None);
+    };
+    let mut adapter = None;
+    for child in children.nodes() {
+        if child.name().value() != "adapter" {
+            anyhow::bail!(
+                "compact `ding` has unsupported child `{}`",
+                child.name().value()
+            );
+        }
+        if adapter.is_some() {
+            anyhow::bail!("compact `ding` declares `adapter` more than once");
+        }
+        if !child.entries().is_empty() {
+            anyhow::bail!("compact `ding` `adapter` accepts no arguments or properties");
+        }
+        let Some(adapter_children) = child.children() else {
+            anyhow::bail!("compact `ding` `adapter` requires non-empty `argv`");
+        };
+        let mut parsed_argv = None;
+        for adapter_child in adapter_children.nodes() {
+            if adapter_child.name().value() != "argv" {
+                anyhow::bail!(
+                    "compact `ding` `adapter` has unsupported child `{}`",
+                    adapter_child.name().value()
+                );
+            }
+            if adapter_child.children().is_some() {
+                anyhow::bail!("compact `ding` adapter `argv` cannot have children");
+            }
+            if parsed_argv.is_some() {
+                anyhow::bail!("compact `ding` adapter declares `argv` more than once");
+            }
+            let values = argv(adapter_child)?;
+            if values.is_empty() || values[0].is_empty() {
+                anyhow::bail!("compact `ding` `adapter` requires non-empty `argv`");
+            }
+            parsed_argv = Some(values);
+        }
+        adapter = Some(
+            parsed_argv
+                .ok_or_else(|| anyhow::anyhow!("compact `ding` `adapter` requires `argv`"))?,
+        );
+    }
+    Ok(adapter)
+}
+
 fn resource_node_to_raw(node: &KdlNode) -> anyhow::Result<(String, RawResource)> {
     if node.children().is_some() {
         anyhow::bail!("resource binding cannot have children");

crates/agent-spec/src/spec.rs

@@ -289,6 +289,10 @@ pub(crate) struct RawSpec {
     /// Compact catalog form: include the built-in `st2 ding` sidecar.
     #[serde(default)]
     pub ding: bool,
+    /// Canonical KDL only: an explicit executable plus arguments for the optional generic DING
+    /// activity adapter. Presence selects the fail-closed rich delivery path.
+    #[serde(skip)]
+    pub ding_adapter_argv: Option<Vec<String>>,
     /// Compact catalog form: reconciliation policy for the generated agent PTY.
     pub lifecycle: Option<String>,
     /// `pty "<name>" {}` / `[pty.<name>]` — interactive tasks.
@@ -642,13 +646,36 @@ impl RawSpec {
             });
         }
         if self.ding {
+            let (command, argv) = match self.ding_adapter_argv {
+                None => (
+                    Some(format!("st2 ding --identity {bus_id} --root $ST_ROOT")),
+                    None,
+                ),
+                Some(adapter) => {
+                    let mut argv = vec![
+                        "st2".to_string(),
+                        "ding".to_string(),
+                        "--identity".to_string(),
+                        bus_id.clone(),
+                        "--root".to_string(),
+                        "$ST_ROOT".to_string(),
+                        "--adapter".to_string(),
+                        adapter[0].clone(),
+                    ];
+                    for argument in &adapter[1..] {
+                        argv.push("--adapter-arg".to_string());
+                        argv.push(argument.clone());
+                    }
+                    (None, Some(argv))
+                }
+            };
             tasks.push(Task {
                 kind: TaskKind::Exec,
                 derived: true,
                 name: "ding".to_string(),
                 id: Some(format!("{bus_id}.ding")),
-                command: Some(format!("st2 ding --identity {bus_id} --root $ST_ROOT")),
-                argv: None,
+                command,
+                argv,
                 cwd: None,
                 tags: BTreeMap::new(),
                 env: self.env,

crates/agent-spec/tests/discovery.rs

@@ -174,6 +174,97 @@ agent "cos" {
     );
 }
 
+#[test]
+fn compact_ding_adapter_lowers_structured_argv_without_a_shell() {
+    let tmp = tempfile::tempdir().unwrap();
+    write(
+        tmp.path(),
+        "agents/host/worker/agent.kdl",
+        r#"
+agent "worker" {
+  host "host"
+  command "agent"
+  ding {
+    adapter {
+      argv "$ADAPTER_ROOT/bin/activity" "--format" "jsonl"
+    }
+  }
+}
+"#,
+    );
+
+    let found = discover(tmp.path());
+    assert!(
+        found.errors.is_empty(),
+        "unexpected errors: {:?}",
+        found.errors
+    );
+    let ding = found.specs[0]
+        .tasks
+        .iter()
+        .find(|task| task.name == "ding")
+        .unwrap();
+    assert_eq!(ding.command, None);
+    assert_eq!(
+        argv(ding),
+        vec![
+            "st2",
+            "ding",
+            "--identity",
+            "host.worker",
+            "--root",
+            "$ST_ROOT",
+            "--adapter",
+            "$ADAPTER_ROOT/bin/activity",
+            "--adapter-arg",
+            "--format",
+            "--adapter-arg",
+            "jsonl",
+        ]
+    );
+}
+
+#[test]
+fn compact_ding_adapter_rejects_ambiguous_or_incomplete_shapes() {
+    for (name, ding, expected) in [
+        (
+            "missing-argv",
+            "ding { adapter }",
+            "requires non-empty `argv`",
+        ),
+        (
+            "empty-argv",
+            "ding { adapter { argv } }",
+            "requires non-empty `argv`",
+        ),
+        (
+            "unknown-child",
+            "ding { delivery \"rich\" }",
+            "unsupported child `delivery`",
+        ),
+        (
+            "shell-command",
+            "ding { adapter { command \"activity\" } }",
+            "unsupported child `command`",
+        ),
+    ] {
+        let tmp = tempfile::tempdir().unwrap();
+        write(
+            tmp.path(),
+            &format!("agents/host/{name}/agent.kdl"),
+            &format!("agent \"{name}\" {{ host \"host\"; command \"agent\"; {ding} }}"),
+        );
+        let found = discover(tmp.path());
+        assert_eq!(found.specs.len(), 0, "{name}");
+        assert_eq!(found.errors.len(), 1, "{name}");
+        assert!(
+            found.errors[0].message.contains(expected),
+            "{name}: {}",
+            found.errors[0].message
+        );
+    }
+}
+
 #[test]
 fn compact_adopt_only_lifecycle_lowers_to_the_generated_agent_task() {
     let tmp = tempfile::tempdir().unwrap();