From 796c392ce3ddba89eaba53438f0a8cb3b73a254e Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Fri, 31 Jul 2026 11:22:53 +0200 Subject: [PATCH 1/2] feat(ding): add harness-neutral guarded adapter seam --- INVARIANTS.md | 1 + README.md | 35 +- crates/agent-spec/src/kdl_format.rs | 63 ++- crates/agent-spec/src/spec.rs | 31 +- crates/agent-spec/tests/discovery.rs | 91 ++++ docs/vrs/01-ding/requirements.md | 45 +- docs/vrs/01-ding/spec.md | 90 ++++ src/ding/adapter.rs | 464 +++++++++++++++++++ src/ding/guarded_pty.rs | 313 +++++++++++++ src/ding/hook_control.rs | 295 ++++++++++++ src/ding/mod.rs | 641 ++++++++++++++++++++++++++- src/main.rs | 88 +++- tests/ding_control.rs | 72 +++ 13 files changed, 2211 insertions(+), 18 deletions(-) create mode 100644 src/ding/adapter.rs create mode 100644 src/ding/guarded_pty.rs create mode 100644 src/ding/hook_control.rs create mode 100644 tests/ding_control.rs diff --git a/INVARIANTS.md b/INVARIANTS.md index c16a608b..624a384d 100644 --- a/INVARIANTS.md +++ b/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 `-.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` | | **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` | diff --git a/README.md b/README.md index aa2c63d1..2082f2e8 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,39 @@ does not introduce a shell. Use `command #"..."#` instead when the task intentio 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`. +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 +environment. + +```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 hook-owned \ + --message +``` + +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 st2 does not parse or run scheduled entries today. The intended direction starts with a declarative @@ -357,7 +390,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) diff --git a/crates/agent-spec/src/kdl_format.rs b/crates/agent-spec/src/kdl_format.rs index a14a3979..baaa1159 100644 --- a/crates/agent-spec/src/kdl_format.rs +++ b/crates/agent-spec/src/kdl_format.rs @@ -91,7 +91,16 @@ fn agent_node_to_raw(node: &KdlNode) -> anyhow::Result { } "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("") + ); + } + 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 { Ok(raw) } +fn ding_adapter_argv(node: &KdlNode) -> anyhow::Result>> { + 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"); diff --git a/crates/agent-spec/src/spec.rs b/crates/agent-spec/src/spec.rs index f47d2d16..33ac4efd 100644 --- a/crates/agent-spec/src/spec.rs +++ b/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>, /// Compact catalog form: reconciliation policy for the generated agent PTY. pub lifecycle: Option, /// `pty "" {}` / `[pty.]` — 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, diff --git a/crates/agent-spec/tests/discovery.rs b/crates/agent-spec/tests/discovery.rs index e7113abd..f6eae45c 100644 --- a/crates/agent-spec/tests/discovery.rs +++ b/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(); diff --git a/docs/vrs/01-ding/requirements.md b/docs/vrs/01-ding/requirements.md index 9915a19e..10c14abd 100644 --- a/docs/vrs/01-ding/requirements.md +++ b/docs/vrs/01-ding/requirements.md @@ -20,10 +20,9 @@ is in [`spec.md`](./spec.md). ## Assumptions -- **DING-A01 Rendered screens only:** The only available evidence about a - composer's state is a rendered terminal screen. No maintained harness exposes - an evented idle signal, so every precondition below is a measured heuristic - over text. `DQ2` in [`../spec.md`](../spec.md) tracks closing that gap. +- **DING-A01 Legacy rendered-screen default:** Bare `ding` uses the current + rendered-screen path. The optional experimental adapter path is selected only + by an explicit adapter declaration and does not change that default. - **DING-A02 Cooperative human:** The human sharing a pane is not adversarial. A screen that deliberately imitates another harness's composer is a correctness concern, not a security boundary, consistent with `A02`. @@ -95,6 +94,44 @@ is in [`spec.md`](./spec.md). that evidence exists, a transport attempt retains staged ownership and retries by inspection without re-pasting. +### Optional rich delivery must remain generic and fail closed + +- **DING-R11 Structured adapter launch:** An adapter declaration is one + explicit executable plus argument vector. Core introduces no shell, provider + selector, inferred arguments, or environment. Adapter argv participates in + the DING sidecar launch fingerprint. +- **DING-R12 Exact generic activity tuple:** Rich PTY input requires one + receipt-anchored, unexpired `idle` event with `inputBuffer=empty`. Its exact + session, incarnation, PTY generation, and strictly increasing sequence must + match the live PTY activity snapshot. +- **DING-R13 Independent atomic fence:** A qualifying adapter event is only + authority to attempt delivery. st2 snapshots PTY's generation and monotonic + I/O revision, then asks PTY to compare both immediately before one bounded + write. A mismatch writes zero bytes and retains FIFO ownership. +- **DING-R14 No rich fallback:** Configured rich DING never invokes legacy + aggressive or renderer-classified input after missing, stale, malformed, or + conflicting evidence. `active`, child-command-running, `unknown`, nonempty or + unknown input buffer, adapter EOF/error, tuple change, and PTY conflict all + hold. +- **DING-R15 Typed ownership receipts:** Rich decisions distinguish legacy + fallback, hold, staged error, hook ownership, guard conflict, and PTY + ownership. A successful guarded write is staged PTY ownership, not a harness + acceptance receipt; it is not sent again while unread. Exact attempt + ownership is durable before the guarded packet, survives sidecar restart and + ambiguous transport errors, and is cleared only by archive or a proven + zero-byte conflict. Exact hook ownership independently suppresses PTY work. +- **DING-R16 Separate turn-boundary ingress:** A provider-rendered lifecycle + hook may record exact filenames only after it injected those currently unread + files into an already-occurring next context. st2 verifies generic filenames + and unread state, persists hook ownership across sidecar restart, sends no PTY + input for owned files, and clears ownership on archive. +- **DING-R17 No provider or interruption grammar:** Core does not install + provider hooks, parse provider hook payloads, identify a provider from command + spelling, trigger a model call, or define a mid-turn interruption mechanism. +- **DING-R18 Compatibility:** Omitted or bare `ding` lowers to the existing + sidecar command byte-for-byte. Adapter presence selects the only rich policy; + there is no redundant delivery enum. + ## Evidence Each guarantee above is pinned by a named test in diff --git a/docs/vrs/01-ding/spec.md b/docs/vrs/01-ding/spec.md index 025d3234..672b5033 100644 --- a/docs/vrs/01-ding/spec.md +++ b/docs/vrs/01-ding/spec.md @@ -10,6 +10,10 @@ specified in [`01-claude/spec.md`](./01-claude/spec.md) and Active. A map to the implementation and its evidence, not a replacement for the tests. +Bare `ding` is the active compatibility path described first below. The +adapter-selected path later in this document is experimental and depends on +PTY's generic activity and guarded-send protocols. + ## Composer states One inspection of a rendered screen, evaluated against one exact expected @@ -136,6 +140,92 @@ indefinitely occupied composer cannot spawn a terminal probe per inbox poll Declared `busy` never suppresses delivery; only fresh `dnd` defers it (`DING-R09`). +## Experimental adapter-selected delivery + +The sole rich policy is selected by adapter presence; no provider or delivery +enum is built into core: + +```kdl +ding { + adapter { + argv "$ADAPTER_ROOT/bin/activity" "--format" "jsonl" + } +} +``` + +The argv values follow normal task launch semantics. The first value is the +executable, remaining values are exact arguments, and declared task/agent +environment is inherited. st2 neither runs a shell nor adds arguments or +environment. Bare `ding` continues to lower to +`st2 ding --identity --root $ST_ROOT` exactly. + +The long-running adapter owns provider-native interpretation and publishes +PTY's harness-neutral activity lease. Its bounded stdout is newline-delimited +JSON. Core accepts only this strict v1 event: + +```json +{"v":1,"kind":"activity","session":"host.identity","incarnation":"opaque-epoch","generation":"opaque-pty-generation","sequence":7,"state":"idle","inputBuffer":"empty","validForMs":250,"reason":"opaque"} +``` + +- Required common fields are `v`, `kind`, `session`, `incarnation`, + `generation`, and a nonzero strictly increasing `sequence`. +- `state` is `idle`, `active`, `child`, or `unknown`. +- `inputBuffer` is `empty`, `nonempty`, or `unknown`. This is deliberately not + named composer state; adapters translate their own UI facts. +- `validForMs` is anchored when st2 receives the line and capped at two seconds. + A newer event invalidates the prior lease. +- Unknown fields, versions, or kinds; malformed/oversized/non-UTF-8 lines; + identity or sequence mismatch; adapter EOF/error; and tuple changes fail + closed. Opaque `reason` is not interpreted. + +Only fresh `idle` + `empty` proceeds. st2 reads one PTY STATUS packet and +requires exact equality across: + +```text +event.session = PTY session +event.incarnation = activity.producerEpoch +event.generation = STATUS generation = activity.generation +event.sequence = activity.sequence +event.state = activity.state = idle +``` + +It then records exact durable attempt ownership and sends the existing +normalized bracketed-paste DING notice plus Return as opaque bytes in PTY's +generation/I/O-revision guarded packet. PTY compares both tokens and writes once +in the same event-loop turn. A successful guard creates PTY ownership, not a +positive harness receipt, and the notice is not written again while unread, +including after a sidecar restart. A proven conflict writes zero guarded bytes +and clears attempt ownership. A transport error is ambiguous after packet write, +so ownership remains fail-closed. Both outcomes invalidate the lease and require +a newer qualifying event before any unowned work can proceed. + +The PTY contract is defined by the stacked experimental +[activity PR](https://github.com/compoundingtech/pty/pull/131) and +[guarded-send PR](https://github.com/compoundingtech/pty/pull/133). Rich DING +must remain opt-in until that substrate and external acceptance are available. + +## Turn-boundary hook ownership + +Hook delivery is separate from the activity adapter. A rendered provider hook +first injects exact unread filenames into an already-occurring next context, +then calls: + +```text +st2 ding-control --identity hook-owned \ + --message [--message ...] +``` + +The ingress resolves the durable inbox, rejects invalid, duplicate, archived, +or missing filenames, and atomically records ownership beside the inbox. The +sidecar rechecks receipts on every wake/restart. Hook-owned filenames are +removed from rich PTY work while unread; archive removes their durable +ownership. The command does not read provider hook JSON, install hooks, trigger +a model call, press Return, or provide mid-turn interruption. + +Rich logs emit typed JSON receipts for `held`, staged `error`, `hook-owned`, +`guard-conflict`, and `pty-owned`. The legacy path emits one `fallback` receipt +only to make its unchanged selection explicit. + ## Known limits - Idle proof depends on footer chrome that a harness may render differently diff --git a/src/ding/adapter.rs b/src/ding/adapter.rs new file mode 100644 index 00000000..34a754a0 --- /dev/null +++ b/src/ding/adapter.rs @@ -0,0 +1,464 @@ +use std::io::{BufRead as _, BufReader, Read as _}; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc::{Receiver, Sender, channel}; +use std::thread; +use std::time::{Duration, Instant}; + +use serde::Deserialize; + +const MAX_EVENT_BYTES: usize = 64 * 1024; +const MAX_LEASE: Duration = Duration::from_secs(2); + +#[derive(Debug)] +pub(super) enum AdapterSignal { + Line(String), + Error(String), + Eof, +} + +/// One explicitly configured, long-running generic activity adapter. +/// +/// Its argv is launched directly. Core introduces no shell, arguments, or environment. Provider-native +/// interpretation stays in that external process; core consumes only its bounded JSONL stdout. +pub(super) struct ExternalAdapter { + child: Child, + signals: Receiver, +} + +impl ExternalAdapter { + pub(super) fn spawn(argv: &[String], wake: Sender<()>) -> anyhow::Result { + let (program, arguments) = argv + .split_first() + .ok_or_else(|| anyhow::anyhow!("DING adapter argv cannot be empty"))?; + let mut child = Command::new(program) + .args(arguments) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(|error| anyhow::anyhow!("starting DING adapter `{program}`: {error}"))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| anyhow::anyhow!("DING adapter stdout was not piped"))?; + let (tx, signals) = channel(); + thread::spawn(move || { + let mut reader = BufReader::new(stdout); + loop { + let mut bytes = Vec::new(); + match (&mut reader) + .take((MAX_EVENT_BYTES + 1) as u64) + .read_until(b'\n', &mut bytes) + { + Ok(0) => { + let _ = tx.send(AdapterSignal::Eof); + let _ = wake.send(()); + break; + } + Ok(_) if bytes.len() > MAX_EVENT_BYTES => { + let _ = tx.send(AdapterSignal::Error(format!( + "adapter event exceeds {MAX_EVENT_BYTES} bytes" + ))); + let _ = wake.send(()); + break; + } + Ok(_) => { + while matches!(bytes.last(), Some(b'\n' | b'\r')) { + bytes.pop(); + } + match String::from_utf8(bytes) { + Ok(line) if !line.is_empty() => { + let _ = tx.send(AdapterSignal::Line(line)); + } + Ok(_) => { + let _ = tx.send(AdapterSignal::Error( + "adapter emitted an empty event".to_string(), + )); + } + Err(error) => { + let _ = tx.send(AdapterSignal::Error(format!( + "adapter event is not UTF-8: {error}" + ))); + } + } + let _ = wake.send(()); + } + Err(error) => { + let _ = tx.send(AdapterSignal::Error(format!( + "reading adapter stdout: {error}" + ))); + let _ = wake.send(()); + break; + } + } + } + }); + Ok(Self { child, signals }) + } + + pub(super) fn try_recv(&self) -> Option { + self.signals.try_recv().ok() + } +} + +impl Drop for ExternalAdapter { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "kebab-case")] +pub(super) enum ActivityState { + Idle, + Active, + Child, + Unknown, +} + +impl ActivityState { + pub(super) fn pty_name(self) -> &'static str { + match self { + Self::Idle => "idle", + Self::Active => "active", + Self::Child => "child_command", + Self::Unknown => "unknown", + } + } +} + +#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub(super) enum InputBufferState { + Empty, + Nonempty, + Unknown, +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] +struct ActivityEvent { + v: u8, + kind: String, + session: String, + incarnation: String, + generation: String, + sequence: u64, + state: ActivityState, + #[serde(rename = "inputBuffer")] + input_buffer: InputBufferState, + #[serde(rename = "validForMs")] + valid_for_ms: u64, + #[serde(default)] + reason: Option, +} + +#[derive(Debug, Clone)] +pub(super) struct IdleLease { + pub(super) session: String, + pub(super) incarnation: String, + pub(super) generation: String, + pub(super) sequence: u64, + pub(super) state: ActivityState, + expires_at: Instant, +} + +impl IdleLease { + pub(super) fn is_fresh(&self, now: Instant) -> bool { + now < self.expires_at + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum HoldReason { + AdapterUnavailable, + AdapterError, + TupleChanged, + Active, + Child, + Unknown, + InputNonempty, + InputUnknown, + Stale, +} + +impl HoldReason { + pub(super) fn as_str(self) -> &'static str { + match self { + Self::AdapterUnavailable => "adapter-unavailable", + Self::AdapterError => "adapter-error", + Self::TupleChanged => "adapter-tuple-changed", + Self::Active => "activity-active", + Self::Child => "activity-child", + Self::Unknown => "activity-unknown", + Self::InputNonempty => "input-buffer-nonempty", + Self::InputUnknown => "input-buffer-unknown", + Self::Stale => "activity-stale", + } + } +} + +/// Current fail-closed adapter evidence. The first event anchors an incarnation/generation tuple. +/// A tuple change is observed but cannot authorize delivery until a later event confirms it. +#[derive(Debug)] +pub(super) struct AdapterState { + session: String, + last_sequence: Option, + tuple: Option<(String, String)>, + last_state: Option, + last_input_buffer: Option, + lease: Option, + faulted: bool, + tuple_changed: bool, +} + +impl AdapterState { + pub(super) fn new(session: impl Into) -> Self { + Self { + session: session.into(), + last_sequence: None, + tuple: None, + last_state: None, + last_input_buffer: None, + lease: None, + faulted: false, + tuple_changed: false, + } + } + + pub(super) fn apply_line(&mut self, line: &str, received_at: Instant) -> anyhow::Result<()> { + let event: ActivityEvent = match serde_json::from_str(line) { + Ok(event) => event, + Err(error) => { + return self.fail(format!("invalid adapter JSONL event: {error}")); + } + }; + self.apply(event, received_at) + } + + fn apply(&mut self, event: ActivityEvent, received_at: Instant) -> anyhow::Result<()> { + if event.v != 1 { + return self.fail(format!("unsupported adapter protocol version {}", event.v)); + } + if event.kind != "activity" { + return self.fail(format!("unsupported adapter event kind `{}`", event.kind)); + } + if event.session != self.session { + return self.fail(format!( + "adapter session mismatch: expected `{}`, got `{}`", + self.session, event.session + )); + } + if event.incarnation.is_empty() || event.generation.is_empty() { + return self.fail("adapter incarnation and generation must be non-empty".to_string()); + } + let next_tuple = (event.incarnation.clone(), event.generation.clone()); + let tuple_changed = self + .tuple + .as_ref() + .is_some_and(|current| current != &next_tuple); + if event.sequence == 0 + || (!tuple_changed + && self + .last_sequence + .is_some_and(|previous| event.sequence <= previous)) + { + return self.fail(format!( + "adapter sequence {} is not strictly newer than {:?}", + event.sequence, self.last_sequence + )); + } + if event.valid_for_ms == 0 { + return self.fail("adapter validForMs must be greater than zero".to_string()); + } + + let _opaque_reason = event.reason.as_deref(); + self.last_sequence = Some(event.sequence); + self.faulted = false; + self.last_state = Some(event.state); + self.last_input_buffer = Some(event.input_buffer); + if tuple_changed { + self.tuple = Some(next_tuple); + self.lease = None; + self.tuple_changed = true; + return Ok(()); + } + self.tuple = Some(next_tuple); + self.tuple_changed = false; + self.lease = None; + if event.state == ActivityState::Idle && event.input_buffer == InputBufferState::Empty { + let duration = Duration::from_millis(event.valid_for_ms).min(MAX_LEASE); + self.lease = Some(IdleLease { + session: event.session, + incarnation: event.incarnation, + generation: event.generation, + sequence: event.sequence, + state: event.state, + expires_at: received_at + duration, + }); + } + Ok(()) + } + + fn fail(&mut self, message: String) -> anyhow::Result<()> { + self.invalidate(); + anyhow::bail!(message) + } + + pub(super) fn invalidate(&mut self) { + self.lease = None; + self.faulted = true; + } + + pub(super) fn authority(&self, now: Instant) -> Result<&IdleLease, HoldReason> { + if self.faulted { + return Err(HoldReason::AdapterError); + } + if self.tuple_changed { + return Err(HoldReason::TupleChanged); + } + if let Some(lease) = self.lease.as_ref() { + return lease + .is_fresh(now) + .then_some(lease) + .ok_or(HoldReason::Stale); + } + match (self.last_state, self.last_input_buffer) { + (None, _) => Err(HoldReason::AdapterUnavailable), + (Some(ActivityState::Active), _) => Err(HoldReason::Active), + (Some(ActivityState::Child), _) => Err(HoldReason::Child), + (Some(ActivityState::Unknown), _) => Err(HoldReason::Unknown), + (Some(ActivityState::Idle), Some(InputBufferState::Nonempty)) => { + Err(HoldReason::InputNonempty) + } + (Some(ActivityState::Idle), Some(InputBufferState::Unknown)) => { + Err(HoldReason::InputUnknown) + } + (Some(ActivityState::Idle), Some(InputBufferState::Empty)) => Err(HoldReason::Stale), + (Some(ActivityState::Idle), None) => Err(HoldReason::InputUnknown), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc::channel; + + fn event(sequence: u64, state: &str, input_buffer: &str, valid_for_ms: u64) -> String { + format!( + r#"{{"v":1,"kind":"activity","session":"host.agent","incarnation":"epoch-a","generation":"generation-a","sequence":{sequence},"state":"{state}","inputBuffer":"{input_buffer}","validForMs":{valid_for_ms},"reason":"opaque"}}"# + ) + } + + #[test] + fn only_fresh_idle_and_empty_authorizes() { + let now = Instant::now(); + let mut state = AdapterState::new("host.agent"); + state + .apply_line(&event(1, "idle", "empty", 250), now) + .unwrap(); + let lease = state.authority(now).unwrap(); + assert_eq!(lease.incarnation, "epoch-a"); + assert_eq!(lease.generation, "generation-a"); + assert_eq!(lease.sequence, 1); + assert!(matches!( + state.authority(now + Duration::from_millis(250)), + Err(HoldReason::Stale) + )); + } + + #[test] + fn every_non_idle_or_nonempty_input_state_holds() { + for (sequence, activity, input, expected) in [ + (1, "active", "empty", HoldReason::Active), + (2, "child", "empty", HoldReason::Child), + (3, "unknown", "empty", HoldReason::Unknown), + (4, "idle", "nonempty", HoldReason::InputNonempty), + (5, "idle", "unknown", HoldReason::InputUnknown), + ] { + let mut state = AdapterState::new("host.agent"); + state + .apply_line(&event(sequence, activity, input, 250), Instant::now()) + .unwrap(); + assert!(matches!(state.authority(Instant::now()), Err(reason) if reason == expected)); + } + } + + #[test] + fn malformed_identity_sequence_and_tuple_changes_fail_closed() { + let now = Instant::now(); + let mut state = AdapterState::new("host.agent"); + state + .apply_line(&event(1, "idle", "empty", 250), now) + .unwrap(); + assert!( + state + .apply_line(&event(1, "idle", "empty", 250), now) + .is_err() + ); + assert!(matches!( + state.authority(now), + Err(HoldReason::AdapterError) + )); + + let wrong = event(2, "idle", "empty", 250).replace("host.agent", "other.agent"); + assert!(state.apply_line(&wrong, now).is_err()); + + state + .apply_line(&event(3, "idle", "empty", 250), now) + .unwrap(); + let changed = event(1, "idle", "empty", 250) + .replace("epoch-a", "epoch-b") + .replace("generation-a", "generation-b"); + state.apply_line(&changed, now).unwrap(); + assert!(matches!( + state.authority(now), + Err(HoldReason::TupleChanged) + )); + let confirmed = event(2, "idle", "empty", 250) + .replace("epoch-a", "epoch-b") + .replace("generation-a", "generation-b"); + state.apply_line(&confirmed, now).unwrap(); + assert!(state.authority(now).is_ok()); + } + + #[test] + fn unknown_fields_and_event_kinds_are_rejected() { + let mut state = AdapterState::new("host.agent"); + let extra = event(1, "idle", "empty", 250).replace( + r#","reason":"opaque"}"#, + r#","reason":"opaque","provider":"forbidden"}"#, + ); + assert!(state.apply_line(&extra, Instant::now()).is_err()); + let other = event(2, "idle", "empty", 250).replace("activity", "turn-boundary"); + assert!(state.apply_line(&other, Instant::now()).is_err()); + } + + #[test] + fn external_adapter_uses_exact_direct_argv_and_wakes_on_jsonl() { + let payload = event(1, "idle", "empty", 250); + let (wake_tx, wake_rx) = channel(); + let adapter = ExternalAdapter::spawn( + &[ + "sh".to_string(), + "-c".to_string(), + "printf '%s\\n' \"$1\"".to_string(), + "sh".to_string(), + payload.clone(), + ], + wake_tx, + ) + .unwrap(); + wake_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + let signal = loop { + if let Some(signal) = adapter.try_recv() { + break signal; + } + thread::yield_now(); + }; + assert!(matches!(signal, AdapterSignal::Line(line) if line == payload)); + assert!(ExternalAdapter::spawn(&[], channel::<()>().0).is_err()); + } +} diff --git a/src/ding/guarded_pty.rs b/src/ding/guarded_pty.rs new file mode 100644 index 00000000..2abd8948 --- /dev/null +++ b/src/ding/guarded_pty.rs @@ -0,0 +1,313 @@ +use std::io::{Read as _, Write as _}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +const STATUS_PACKET: u8 = 7; +const GUARDED_DATA_PACKET: u8 = 9; +const MAX_RESPONSE_BYTES: usize = 1024 * 1024; +const SOCKET_TIMEOUT: Duration = Duration::from_secs(2); + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub(super) struct PtyActivity { + pub(super) state: String, + pub(super) generation: String, + #[serde(rename = "producerEpoch")] + pub(super) producer_epoch: Option, + pub(super) sequence: u64, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub(super) struct PtySnapshot { + pub(super) name: String, + pub(super) generation: String, + #[serde(rename = "ioRevision")] + pub(super) io_revision: u64, + pub(super) activity: PtyActivity, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) enum GuardedSend { + Sent { + revision: u64, + }, + Conflict { + actual_generation: String, + actual_revision: u64, + error: String, + }, +} + +pub(super) trait GuardedPty { + fn snapshot(&self) -> anyhow::Result; + fn compare_and_send( + &self, + generation: &str, + revision: u64, + bytes: &str, + ) -> anyhow::Result; + fn session_alive(&self) -> bool; +} + +pub(super) struct SocketPty { + session: String, + root: PathBuf, +} + +impl SocketPty { + pub(super) fn new(session: impl Into) -> anyhow::Result { + let session = session.into(); + validate_session(&session)?; + Ok(Self { + session, + root: super::pty_session_dir(), + }) + } + + #[cfg(test)] + fn with_root(session: impl Into, root: PathBuf) -> Self { + Self { + session: session.into(), + root, + } + } + + fn request(&self, packet_type: u8, payload: &[u8]) -> anyhow::Result> { + let path = self.root.join(format!("{}.sock", self.session)); + let mut stream = UnixStream::connect(&path).map_err(|error| { + anyhow::anyhow!("connecting to PTY socket {}: {error}", path.display()) + })?; + stream.set_read_timeout(Some(SOCKET_TIMEOUT))?; + stream.set_write_timeout(Some(SOCKET_TIMEOUT))?; + write_packet(&mut stream, packet_type, payload)?; + let (response_type, response) = read_packet(&mut stream)?; + if response_type != packet_type { + anyhow::bail!("PTY returned packet type {response_type}, expected {packet_type}"); + } + Ok(response) + } +} + +impl GuardedPty for SocketPty { + fn snapshot(&self) -> anyhow::Result { + let response = self.request(STATUS_PACKET, &[])?; + let snapshot: PtySnapshot = serde_json::from_slice(&response) + .map_err(|error| anyhow::anyhow!("parsing PTY STATUS response: {error}"))?; + if snapshot.name != self.session { + anyhow::bail!( + "PTY STATUS session mismatch: expected `{}`, got `{}`", + self.session, + snapshot.name + ); + } + Ok(snapshot) + } + + fn compare_and_send( + &self, + generation: &str, + revision: u64, + bytes: &str, + ) -> anyhow::Result { + #[derive(Serialize)] + struct Request<'a> { + generation: &'a str, + #[serde(rename = "ioRevision")] + io_revision: u64, + data: &'a str, + } + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct Response { + ok: bool, + generation: String, + #[serde(rename = "ioRevision")] + io_revision: u64, + error: Option, + } + + let payload = serde_json::to_vec(&Request { + generation, + io_revision: revision, + data: bytes, + })?; + let response: Response = + serde_json::from_slice(&self.request(GUARDED_DATA_PACKET, &payload)?) + .map_err(|error| anyhow::anyhow!("parsing PTY guarded-send response: {error}"))?; + if response.ok { + if response.generation != generation || response.io_revision <= revision { + anyhow::bail!("PTY guarded-send success returned an invalid generation/revision"); + } + Ok(GuardedSend::Sent { + revision: response.io_revision, + }) + } else { + Ok(GuardedSend::Conflict { + actual_generation: response.generation, + actual_revision: response.io_revision, + error: response + .error + .unwrap_or_else(|| "guard rejected".to_string()), + }) + } + } + + fn session_alive(&self) -> bool { + super::session_alive(&self.session) + } +} + +fn validate_session(session: &str) -> anyhow::Result<()> { + if session.is_empty() + || session.len() > 255 + || !session + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')) + { + anyhow::bail!( + "invalid PTY session `{session}`: expected 1-255 letters, digits, dots, dashes, or underscores" + ); + } + Ok(()) +} + +fn write_packet(stream: &mut UnixStream, packet_type: u8, payload: &[u8]) -> anyhow::Result<()> { + let length = u32::try_from(payload.len()) + .map_err(|_| anyhow::anyhow!("PTY packet payload is too large"))?; + let mut header = [0_u8; 5]; + header[0] = packet_type; + header[1..].copy_from_slice(&length.to_be_bytes()); + stream.write_all(&header)?; + stream.write_all(payload)?; + Ok(()) +} + +fn read_packet(stream: &mut UnixStream) -> anyhow::Result<(u8, Vec)> { + let mut header = [0_u8; 5]; + stream.read_exact(&mut header)?; + let length = u32::from_be_bytes(header[1..].try_into().expect("four-byte length")) as usize; + if length > MAX_RESPONSE_BYTES { + anyhow::bail!("PTY response exceeds {MAX_RESPONSE_BYTES} bytes"); + } + let mut payload = vec![0_u8; length]; + stream.read_exact(&mut payload)?; + Ok((header[0], payload)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::net::UnixListener; + use std::thread; + + fn serve_once( + listener: &UnixListener, + expected_type: u8, + response: serde_json::Value, + ) -> Vec { + let (mut stream, _) = listener.accept().unwrap(); + let (packet_type, payload) = read_packet(&mut stream).unwrap(); + assert_eq!(packet_type, expected_type); + write_packet( + &mut stream, + expected_type, + &serde_json::to_vec(&response).unwrap(), + ) + .unwrap(); + payload + } + + #[test] + fn reads_exact_activity_snapshot_and_sends_generation_revision_guard() { + let tmp = tempfile::tempdir().unwrap(); + let socket = tmp.path().join("host.agent.sock"); + let listener = UnixListener::bind(socket).unwrap(); + let root = tmp.path().to_path_buf(); + let server = thread::spawn(move || { + assert!( + serve_once( + &listener, + STATUS_PACKET, + serde_json::json!({ + "name": "host.agent", + "generation": "generation-a", + "ioRevision": 41, + "activity": { + "state": "idle", + "generation": "generation-a", + "producerEpoch": "epoch-a", + "sequence": 7 + } + }), + ) + .is_empty() + ); + let payload = serve_once( + &listener, + GUARDED_DATA_PACKET, + serde_json::json!({ + "ok": true, + "generation": "generation-a", + "ioRevision": 42 + }), + ); + let request: serde_json::Value = serde_json::from_slice(&payload).unwrap(); + assert_eq!( + request, + serde_json::json!({ + "generation": "generation-a", + "ioRevision": 41, + "data": "notice" + }) + ); + }); + + let pty = SocketPty::with_root("host.agent", root); + let snapshot = pty.snapshot().unwrap(); + assert_eq!(snapshot.activity.producer_epoch.as_deref(), Some("epoch-a")); + assert_eq!(snapshot.activity.sequence, 7); + assert_eq!( + pty.compare_and_send("generation-a", 41, "notice").unwrap(), + GuardedSend::Sent { revision: 42 } + ); + server.join().unwrap(); + } + + #[test] + fn guard_conflict_is_typed_and_never_reinterpreted_as_success() { + let tmp = tempfile::tempdir().unwrap(); + let socket = tmp.path().join("host.agent.sock"); + let listener = UnixListener::bind(socket).unwrap(); + let root = tmp.path().to_path_buf(); + let server = thread::spawn(move || { + let _ = serve_once( + &listener, + GUARDED_DATA_PACKET, + serde_json::json!({ + "ok": false, + "generation": "generation-b", + "ioRevision": 9, + "error": "daemon generation mismatch" + }), + ); + }); + let pty = SocketPty::with_root("host.agent", root); + assert_eq!( + pty.compare_and_send("generation-a", 8, "notice").unwrap(), + GuardedSend::Conflict { + actual_generation: "generation-b".to_string(), + actual_revision: 9, + error: "daemon generation mismatch".to_string(), + } + ); + server.join().unwrap(); + } + + #[test] + fn session_names_cannot_escape_the_pty_root() { + assert!(SocketPty::new("../other").is_err()); + assert!(SocketPty::new("host.agent").is_ok()); + } +} diff --git a/src/ding/hook_control.rs b/src/ding/hook_control.rs new file mode 100644 index 00000000..a1e32889 --- /dev/null +++ b/src/ding/hook_control.rs @@ -0,0 +1,295 @@ +use std::collections::HashSet; +use std::fs; +use std::io::Write as _; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +const HOOK_RECEIPT_VERSION: u8 = 1; +const HOOK_RECEIPT_KIND: &str = "hook-owned"; + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct HookOwnedReceipt { + v: u8, + kind: String, + messages: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PtyOwnedReceipt { + v: u8, + kind: String, + messages: Vec, + generation: String, + revision: u64, + sequence: u64, +} + +pub(super) fn control_dir(inbox_dir: &Path) -> PathBuf { + inbox_dir.parent().unwrap_or(inbox_dir).join("ding-control") +} + +/// Record that a provider-native hook already injected exact unread filenames into an +/// already-occurring next context. The caller owns provider translation; this ingress validates +/// only generic durable-inbox facts. +pub fn record_hook_owned(inbox_dir: &Path, messages: &[String]) -> anyhow::Result { + if messages.is_empty() { + anyhow::bail!("hook-owned receipt requires at least one --message"); + } + let mut unique = HashSet::new(); + for filename in messages { + if !crate::message::is_message_filename(filename) { + anyhow::bail!("hook-owned receipt has invalid message filename `{filename}`"); + } + if !unique.insert(filename.as_str()) { + anyhow::bail!("hook-owned receipt repeats message filename `{filename}`"); + } + } + let unread: HashSet = crate::message::list_inbox(inbox_dir)? + .into_iter() + .map(|message| message.filename) + .collect(); + if let Some(missing) = messages.iter().find(|filename| !unread.contains(*filename)) { + anyhow::bail!("hook-owned message `{missing}` is not currently unread"); + } + + persist_receipt( + inbox_dir, + "hook-owned", + &HookOwnedReceipt { + v: HOOK_RECEIPT_VERSION, + kind: HOOK_RECEIPT_KIND.to_string(), + messages: messages.to_vec(), + }, + ) +} + +pub(super) fn record_pty_owned( + inbox_dir: &Path, + messages: &HashSet, + generation: &str, + revision: u64, + sequence: u64, +) -> anyhow::Result { + if messages.is_empty() { + anyhow::bail!("PTY ownership requires at least one unread message"); + } + let unread = unread_filenames(inbox_dir)?; + if let Some(missing) = messages.iter().find(|filename| !unread.contains(*filename)) { + anyhow::bail!("PTY-owned message `{missing}` is not currently unread"); + } + let mut messages = messages.iter().cloned().collect::>(); + messages.sort(); + persist_receipt( + inbox_dir, + "pty-owned", + &PtyOwnedReceipt { + v: HOOK_RECEIPT_VERSION, + kind: "pty-owned".to_string(), + messages, + generation: generation.to_string(), + revision, + sequence, + }, + ) +} + +fn persist_receipt( + inbox_dir: &Path, + name: &str, + receipt: &impl Serialize, +) -> anyhow::Result { + let directory = control_dir(inbox_dir); + fs::create_dir_all(&directory)?; + let mut temporary = tempfile::NamedTempFile::new_in(&directory)?; + serde_json::to_writer(&mut temporary, receipt)?; + temporary.write_all(b"\n")?; + temporary.as_file().sync_all()?; + let target = directory.join(format!( + "{}.{name}.json", + crate::message::new_filename() + .strip_suffix(".md") + .expect("canonical message filename") + )); + temporary + .persist_noclobber(&target) + .map_err(|error| anyhow::anyhow!("persisting {name} receipt: {}", error.error))?; + Ok(target) +} + +pub(super) fn clear_pty_owned(path: &Path) -> anyhow::Result<()> { + fs::remove_file(path) + .map_err(|error| anyhow::anyhow!("clearing PTY ownership {}: {error}", path.display())) +} + +/// Record fields remain generic and exact; this helper validates only the hook-owned envelope. +fn validate_hook_receipt(receipt: &HookOwnedReceipt, path: &Path) -> anyhow::Result<()> { + if receipt.v != HOOK_RECEIPT_VERSION || receipt.kind != HOOK_RECEIPT_KIND { + anyhow::bail!("unsupported hook-owned receipt {}", path.display()); + } + Ok(()) +} + +/// Record fields remain generic and exact; this helper validates only the PTY-owned envelope. +fn validate_pty_receipt(receipt: &PtyOwnedReceipt, path: &Path) -> anyhow::Result<()> { + if receipt.v != HOOK_RECEIPT_VERSION + || receipt.kind != "pty-owned" + || receipt.generation.is_empty() + || receipt.sequence == 0 + { + anyhow::bail!("unsupported PTY-owned receipt {}", path.display()); + } + Ok(()) +} + +fn retain_current_messages( + path: &Path, + messages: &[String], + unread: &HashSet, + owned: &mut HashSet, +) -> anyhow::Result<()> { + let mut still_unread = false; + for filename in messages { + if !crate::message::is_message_filename(filename) { + anyhow::bail!( + "DING ownership receipt {} has invalid filename `{filename}`", + path.display() + ); + } + if unread.contains(filename) { + still_unread = true; + owned.insert(filename.clone()); + } + } + if !still_unread { + fs::remove_file(path)?; + } + Ok(()) +} + +fn receipt_paths(inbox_dir: &Path, suffix: &str) -> Vec { + let directory = control_dir(inbox_dir); + let Ok(entries) = fs::read_dir(&directory) else { + return Vec::new(); + }; + let mut paths = entries + .filter_map(Result::ok) + .map(|entry| entry.path()) + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.ends_with(suffix)) + }) + .collect::>(); + paths.sort(); + paths +} + +fn unread_filenames(inbox_dir: &Path) -> anyhow::Result> { + Ok(crate::message::list_inbox(inbox_dir)? + .into_iter() + .map(|message| message.filename) + .collect()) +} + +/// Load durable hook ownership, rechecking every filename against the current unread inbox. +/// Receipts remain until all named messages are archived, so sidecar restarts cannot reding work +/// that a hook already injected. +pub(super) fn load_hook_owned(inbox_dir: &Path) -> anyhow::Result> { + let unread = unread_filenames(inbox_dir)?; + let mut owned = HashSet::new(); + for path in receipt_paths(inbox_dir, ".hook-owned.json") { + let receipt: HookOwnedReceipt = serde_json::from_slice(&fs::read(&path)?) + .map_err(|error| anyhow::anyhow!("parsing {}: {error}", path.display()))?; + validate_hook_receipt(&receipt, &path)?; + retain_current_messages(&path, &receipt.messages, &unread, &mut owned)?; + } + Ok(owned) +} + +pub(super) fn load_pty_owned(inbox_dir: &Path) -> anyhow::Result> { + let unread = unread_filenames(inbox_dir)?; + let mut owned = HashSet::new(); + for path in receipt_paths(inbox_dir, ".pty-owned.json") { + let receipt: PtyOwnedReceipt = serde_json::from_slice(&fs::read(&path)?) + .map_err(|error| anyhow::anyhow!("parsing {}: {error}", path.display()))?; + validate_pty_receipt(&receipt, &path)?; + retain_current_messages(&path, &receipt.messages, &unread, &mut owned)?; + } + Ok(owned) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::message::{archive_dir, archive_msg, inbox_dir, send_to_inbox}; + + #[test] + fn ingress_accepts_only_exact_currently_unread_filenames_and_survives_reload() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = inbox_dir(tmp.path()); + let archive = archive_dir(tmp.path()); + let filename = send_to_inbox(&inbox, "sender", Some("subject"), None, &[], "body").unwrap(); + let receipt = record_hook_owned(&inbox, std::slice::from_ref(&filename)).unwrap(); + assert!(receipt.is_file()); + assert_eq!( + load_hook_owned(&inbox).unwrap(), + HashSet::from([filename.clone()]) + ); + assert!(record_hook_owned(&inbox, &["not-a-message".to_string()]).is_err()); + assert!(record_hook_owned(&inbox, &["1785000000000-abc123.md".to_string()]).is_err()); + + archive_msg(&inbox, &archive, &filename).unwrap(); + assert!(load_hook_owned(&inbox).unwrap().is_empty()); + assert!(!receipt.exists()); + } + + #[test] + fn ingress_rejects_duplicates() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = inbox_dir(tmp.path()); + let filename = send_to_inbox(&inbox, "sender", None, None, &[], "body").unwrap(); + assert!( + record_hook_owned(&inbox, &[filename.clone(), filename]) + .unwrap_err() + .to_string() + .contains("repeats") + ); + } + + #[test] + fn pty_ownership_is_durable_until_conflict_clear_or_archive() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = inbox_dir(tmp.path()); + let archive = archive_dir(tmp.path()); + let filename = send_to_inbox(&inbox, "sender", None, None, &[], "body").unwrap(); + let receipt = record_pty_owned( + &inbox, + &HashSet::from([filename.clone()]), + "generation-a", + 41, + 7, + ) + .unwrap(); + assert_eq!( + load_pty_owned(&inbox).unwrap(), + HashSet::from([filename.clone()]) + ); + clear_pty_owned(&receipt).unwrap(); + assert!(load_pty_owned(&inbox).unwrap().is_empty()); + + let receipt = record_pty_owned( + &inbox, + &HashSet::from([filename.clone()]), + "generation-a", + 42, + 8, + ) + .unwrap(); + archive_msg(&inbox, &archive, &filename).unwrap(); + assert!(load_pty_owned(&inbox).unwrap().is_empty()); + assert!(!receipt.exists()); + } +} diff --git a/src/ding/mod.rs b/src/ding/mod.rs index 1777ea49..a72ee6cc 100644 --- a/src/ding/mod.rs +++ b/src/ding/mod.rs @@ -23,13 +23,19 @@ use std::sync::mpsc::{Receiver, channel}; use std::thread; use std::time::{Duration, Instant}; +mod adapter; mod composer; +mod guarded_pty; mod harness; +mod hook_control; use crate::message::{self, Message}; use crate::status; +use adapter::{AdapterSignal, AdapterState, ExternalAdapter, HoldReason, IdleLease}; use composer::{ComposerState, classify_composer, classify_receipt}; +use guarded_pty::{GuardedPty, GuardedSend, PtySnapshot, SocketPty}; use harness::ReceiptState; +pub use hook_control::record_hook_owned; const BRACKETED_PASTE_START: &str = "\x1b[200~"; const BRACKETED_PASTE_END: &str = "\x1b[201~"; @@ -155,6 +161,52 @@ pub enum PokeOutcome { Deferred, } +/// Observable decisions from the opt-in rich path. These are transport/ownership receipts, not +/// claims that a harness accepted a prompt. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +#[serde(tag = "result", rename_all = "kebab-case")] +enum DeliveryReceipt { + Fallback { + reason: &'static str, + }, + Held { + reason: String, + }, + Error { + stage: &'static str, + error: String, + }, + HookOwned { + messages: Vec, + }, + GuardConflict { + generation: String, + revision: u64, + error: String, + }, + PtyOwned { + generation: String, + revision: u64, + sequence: u64, + }, +} + +#[derive(Default)] +struct ReceiptLog { + last: Option, +} + +impl ReceiptLog { + fn emit(&mut self, receipt: DeliveryReceipt) { + let serialized = + serde_json::to_string(&receipt).expect("DING delivery receipts are serializable"); + if self.last.as_deref() != Some(&serialized) { + eprintln!("st2 ding receipt: {serialized}"); + self.last = Some(serialized); + } + } +} + /// How DING delivers a poke and checks liveness, abstracted so the watch loop is testable without a /// real `pty`. pub trait Poker { @@ -939,6 +991,356 @@ fn flush_pending( } } +#[derive(Debug)] +struct RichNotice { + filenames: HashSet, + text: String, + pty_owned: bool, +} + +impl RichNotice { + fn recovery(messages: &[Message]) -> Self { + Self { + filenames: messages + .iter() + .map(|message| message.filename.clone()) + .collect(), + text: RECOVERY_POKE.to_string(), + pty_owned: false, + } + } + + fn message(message: Message) -> Self { + Self { + filenames: HashSet::from([message.filename.clone()]), + text: poke_text(&message), + pty_owned: false, + } + } +} + +fn refresh_rich_pending( + inbox_dir: &Path, + seen: &mut HashSet, + pending: &mut VecDeque, +) -> anyhow::Result> { + pending.extend( + new_arrivals(inbox_dir, seen) + .into_iter() + .map(RichNotice::message), + ); + let unread: HashSet = message::list_inbox(inbox_dir)? + .into_iter() + .map(|message| message.filename) + .collect(); + let hook_owned = hook_control::load_hook_owned(inbox_dir)?; + let pty_owned = hook_control::load_pty_owned(inbox_dir)?; + let mut applied_hook_ownership = HashSet::new(); + for notice in pending.iter_mut() { + notice.filenames.retain(|filename| { + if !unread.contains(filename) { + return false; + } + if hook_owned.contains(filename) { + applied_hook_ownership.insert(filename.clone()); + return false; + } + true + }); + if notice + .filenames + .iter() + .any(|filename| pty_owned.contains(filename)) + { + notice.pty_owned = true; + } + } + pending.retain(|notice| !notice.filenames.is_empty()); + let mut applied = applied_hook_ownership.into_iter().collect::>(); + applied.sort(); + Ok(applied) +} + +fn guarded_delivery_bytes(text: &str) -> String { + // This is the existing DING terminal transaction carried as one opaque guarded write: + // normalized bracketed paste followed by Return. Core assigns no provider meaning to it. + format!("{}\r", bracketed_paste(text)) +} + +fn snapshot_matches_lease(snapshot: &PtySnapshot, lease: &IdleLease) -> bool { + snapshot.name == lease.session + && snapshot.generation == lease.generation + && snapshot.activity.generation == lease.generation + && snapshot.activity.producer_epoch.as_deref() == Some(lease.incarnation.as_str()) + && snapshot.activity.sequence == lease.sequence + && snapshot.activity.state == lease.state.pty_name() +} + +fn attempt_rich_delivery( + inbox_dir: &Path, + pending: &mut VecDeque, + state: &mut AdapterState, + pty: &dyn GuardedPty, + last_attempted: &mut Option<(String, String, u64)>, + receipts: &mut ReceiptLog, +) { + let Some(notice) = pending.front_mut() else { + return; + }; + if notice.pty_owned { + receipts.emit(DeliveryReceipt::Held { + reason: "pty-owned-awaiting-archive-or-hook".to_string(), + }); + return; + } + + let lease = match state.authority(Instant::now()) { + Ok(lease) => lease.clone(), + Err(reason) => { + receipts.emit(DeliveryReceipt::Held { + reason: reason.as_str().to_string(), + }); + return; + } + }; + let attempt_key = ( + lease.incarnation.clone(), + lease.generation.clone(), + lease.sequence, + ); + if last_attempted.as_ref() == Some(&attempt_key) { + receipts.emit(DeliveryReceipt::Held { + reason: "activity-tuple-already-attempted".to_string(), + }); + return; + } + *last_attempted = Some(attempt_key); + + let snapshot = match pty.snapshot() { + Ok(snapshot) => snapshot, + Err(error) => { + state.invalidate(); + receipts.emit(DeliveryReceipt::Error { + stage: "pty-status", + error: format!("PTY STATUS failed: {error}"), + }); + return; + } + }; + if !snapshot_matches_lease(&snapshot, &lease) { + state.invalidate(); + receipts.emit(DeliveryReceipt::Held { + reason: "pty-activity-tuple-mismatch".to_string(), + }); + return; + } + if !lease.is_fresh(Instant::now()) { + state.invalidate(); + receipts.emit(DeliveryReceipt::Held { + reason: HoldReason::Stale.as_str().to_string(), + }); + return; + } + + let ownership = match hook_control::record_pty_owned( + inbox_dir, + ¬ice.filenames, + &snapshot.generation, + snapshot.io_revision, + lease.sequence, + ) { + Ok(path) => path, + Err(error) => { + state.invalidate(); + receipts.emit(DeliveryReceipt::Error { + stage: "ownership-record", + error: format!("recording PTY ownership failed: {error}"), + }); + return; + } + }; + let bytes = guarded_delivery_bytes(¬ice.text); + match pty.compare_and_send(&snapshot.generation, snapshot.io_revision, &bytes) { + Ok(GuardedSend::Sent { revision }) => { + notice.pty_owned = true; + receipts.emit(DeliveryReceipt::PtyOwned { + generation: snapshot.generation, + revision, + sequence: lease.sequence, + }); + } + Ok(GuardedSend::Conflict { + actual_generation, + actual_revision, + error, + }) => { + state.invalidate(); + if let Err(clear_error) = hook_control::clear_pty_owned(&ownership) { + notice.pty_owned = true; + receipts.emit(DeliveryReceipt::Error { + stage: "ownership-clear", + error: format!( + "guard conflict was zero-byte but ownership clear failed: {clear_error}" + ), + }); + return; + } + receipts.emit(DeliveryReceipt::GuardConflict { + generation: actual_generation, + revision: actual_revision, + error, + }); + } + Err(error) => { + state.invalidate(); + // The socket error is ambiguous after the packet write. Preserve the pre-recorded + // ownership and never retry these bytes without an archive or exact hook receipt. + notice.pty_owned = true; + receipts.emit(DeliveryReceipt::Error { + stage: "pty-guarded-send", + error: format!("PTY guarded send failed: {error}"), + }); + } + } +} + +fn run_rich_ding( + inbox_dir: &Path, + status_path: Option<&Path>, + session: &str, + adapter_argv: &[String], + pty: &dyn GuardedPty, + config: &DingConfig, + stop: &AtomicBool, +) -> anyhow::Result<()> { + let (wake_tx, wake_rx) = channel::<()>(); + // Rich delivery also consumes the sibling `ding-control` receipt folder. + let watch_at = inbox_dir.parent().unwrap_or(inbox_dir); + let _watcher = crate::watch::watch_recursive_mutations(watch_at, wake_tx.clone()); + let adapter = ExternalAdapter::spawn(adapter_argv, wake_tx)?; + let mut adapter_state = AdapterState::new(session); + let mut receipts = ReceiptLog::default(); + + let backlog = message::list_inbox(inbox_dir).unwrap_or_default(); + let mut seen: HashSet = backlog + .iter() + .map(|message| message.filename.clone()) + .collect(); + let mut pending = VecDeque::new(); + if !backlog.is_empty() { + pending.push_back(RichNotice::recovery(&backlog)); + } + eprintln!( + "st2 ding: rich adapter ready — found {} existing unread message(s){}; watching for generic activity.", + backlog.len(), + if backlog.is_empty() { + "" + } else { + " and queued one recovery notice" + } + ); + + let mut watch = SessionWatch::default(); + let mut logged_waiting = false; + let mut last_refresh: Option = None; + let mut last_attempted = None; + + loop { + if stop.load(Ordering::SeqCst) { + break; + } + + while let Some(signal) = adapter.try_recv() { + match signal { + AdapterSignal::Line(line) => { + if let Err(error) = adapter_state.apply_line(&line, Instant::now()) { + receipts.emit(DeliveryReceipt::Error { + stage: "adapter-event", + error: error.to_string(), + }); + } + } + AdapterSignal::Error(error) => { + adapter_state.invalidate(); + receipts.emit(DeliveryReceipt::Error { + stage: "adapter-io", + error, + }); + } + AdapterSignal::Eof => { + adapter_state.invalidate(); + receipts.emit(DeliveryReceipt::Error { + stage: "adapter-exit", + error: "activity adapter exited".to_string(), + }); + } + } + } + + let alive = pty.session_alive(); + if watch.step(alive) == WatchStep::Gone { + eprintln!("st2 ding: target pty session is gone — exiting."); + break; + } + if alive { + if let Some(path) = status_path + && last_refresh.is_none_or(|instant| instant.elapsed() >= config.status_refresh) + { + let _ = status::refresh(path); + last_refresh = Some(Instant::now()); + } + + let control_ready = match refresh_rich_pending(inbox_dir, &mut seen, &mut pending) { + Ok(hook_owned) => { + if !hook_owned.is_empty() { + receipts.emit(DeliveryReceipt::HookOwned { + messages: hook_owned, + }); + } + true + } + Err(error) => { + receipts.emit(DeliveryReceipt::Error { + stage: "hook-control", + error: format!("hook control failed: {error}"), + }); + false + } + }; + if control_ready { + if delivery_suppressed(status_path) { + receipts.emit(DeliveryReceipt::Held { + reason: "presence-dnd".to_string(), + }); + } else { + attempt_rich_delivery( + inbox_dir, + &mut pending, + &mut adapter_state, + pty, + &mut last_attempted, + &mut receipts, + ); + } + } + } else if !watch.seen_alive && !logged_waiting { + eprintln!( + "st2 ding: target pty session not yet registered; waiting before enabling exit-when-gone." + ); + logged_waiting = true; + } + + match wake_rx.recv_timeout(config.poll) { + Ok(()) => drain(&wake_rx), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + thread::sleep(config.poll); + } + } + } + Ok(()) +} + /// Set by SIGINT/SIGTERM so `st2 ding` exits cleanly when st2 tears the sidecar down. static STOP: AtomicBool = AtomicBool::new(false); @@ -959,17 +1361,38 @@ pub fn serve( inbox_dir: &Path, status_path: &Path, session: &str, + adapter_argv: Option<&[String]>, config: &DingConfig, ) -> anyhow::Result<()> { - probe_pty_on_path()?; install_signal_handler(); - run_ding( - inbox_dir, - Some(status_path), - &PtyPoker::new(session), - config, - &STOP, - ) + match adapter_argv { + None => { + probe_pty_on_path()?; + let mut receipts = ReceiptLog::default(); + receipts.emit(DeliveryReceipt::Fallback { + reason: "adapter-not-configured", + }); + run_ding( + inbox_dir, + Some(status_path), + &PtyPoker::new(session), + config, + &STOP, + ) + } + Some(argv) => { + let pty = SocketPty::new(session)?; + run_rich_ding( + inbox_dir, + Some(status_path), + session, + argv, + &pty, + config, + &STOP, + ) + } + } } fn drain(rx: &Receiver<()>) { @@ -1070,6 +1493,208 @@ mod tests { } } + struct FakeGuardedPty { + snapshot: PtySnapshot, + result: Mutex, + sends: Mutex>, + } + + impl FakeGuardedPty { + fn matching(result: GuardedSend) -> Self { + Self { + snapshot: PtySnapshot { + name: "host.agent".to_string(), + generation: "generation-a".to_string(), + io_revision: 41, + activity: guarded_pty::PtyActivity { + state: "idle".to_string(), + generation: "generation-a".to_string(), + producer_epoch: Some("epoch-a".to_string()), + sequence: 7, + }, + }, + result: Mutex::new(result), + sends: Mutex::new(Vec::new()), + } + } + } + + impl GuardedPty for FakeGuardedPty { + fn snapshot(&self) -> anyhow::Result { + Ok(self.snapshot.clone()) + } + + fn compare_and_send( + &self, + generation: &str, + revision: u64, + bytes: &str, + ) -> anyhow::Result { + self.sends + .lock() + .unwrap() + .push((generation.to_string(), revision, bytes.to_string())); + Ok(self.result.lock().unwrap().clone()) + } + + fn session_alive(&self) -> bool { + true + } + } + + fn rich_activity(sequence: u64, state: &str, input_buffer: &str, valid_for_ms: u64) -> String { + format!( + r#"{{"v":1,"kind":"activity","session":"host.agent","incarnation":"epoch-a","generation":"generation-a","sequence":{sequence},"state":"{state}","inputBuffer":"{input_buffer}","validForMs":{valid_for_ms}}}"# + ) + } + + #[test] + fn rich_delivery_requires_matching_fresh_tuple_and_becomes_pty_owned_once() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = tmp.path().join("inbox"); + let pty = FakeGuardedPty::matching(GuardedSend::Sent { revision: 42 }); + let mut state = AdapterState::new("host.agent"); + state + .apply_line(&rich_activity(7, "idle", "empty", 500), Instant::now()) + .unwrap(); + let filename = send_to_inbox(&inbox, "sender", Some("subject"), None, &[], "body").unwrap(); + let mut pending = VecDeque::from([RichNotice::message( + message::list_inbox(&inbox).unwrap().remove(0), + )]); + let mut sequence = None; + let mut receipts = ReceiptLog::default(); + attempt_rich_delivery( + &inbox, + &mut pending, + &mut state, + &pty, + &mut sequence, + &mut receipts, + ); + assert!(pending[0].pty_owned); + assert_eq!(pty.sends.lock().unwrap().len(), 1); + let sent = pty.sends.lock().unwrap()[0].clone(); + assert_eq!(sent.0, "generation-a"); + assert_eq!(sent.1, 41); + assert_eq!(sent.2, guarded_delivery_bytes(&pending[0].text)); + assert!(sent.2.starts_with(BRACKETED_PASTE_START)); + assert!(sent.2.ends_with("\x1b[201~\r")); + assert_eq!( + hook_control::load_pty_owned(&inbox).unwrap(), + HashSet::from([filename]) + ); + + attempt_rich_delivery( + &inbox, + &mut pending, + &mut state, + &pty, + &mut sequence, + &mut receipts, + ); + assert_eq!(pty.sends.lock().unwrap().len(), 1); + } + + #[test] + fn rich_tuple_mismatch_and_guard_conflict_send_zero_or_one_guarded_packet() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = tmp.path().join("inbox"); + let mut mismatch_pty = FakeGuardedPty::matching(GuardedSend::Sent { revision: 42 }); + mismatch_pty.snapshot.activity.sequence = 8; + let mut state = AdapterState::new("host.agent"); + state + .apply_line(&rich_activity(7, "idle", "empty", 500), Instant::now()) + .unwrap(); + send_to_inbox(&inbox, "sender", Some("subject"), None, &[], "body").unwrap(); + let mut pending = VecDeque::from([RichNotice::message( + message::list_inbox(&inbox).unwrap().remove(0), + )]); + let mut sequence = None; + let mut receipts = ReceiptLog::default(); + attempt_rich_delivery( + &inbox, + &mut pending, + &mut state, + &mismatch_pty, + &mut sequence, + &mut receipts, + ); + assert!(mismatch_pty.sends.lock().unwrap().is_empty()); + assert!(!pending[0].pty_owned); + + let conflict_pty = FakeGuardedPty::matching(GuardedSend::Conflict { + actual_generation: "generation-a".to_string(), + actual_revision: 42, + error: "I/O revision mismatch".to_string(), + }); + let mut state = AdapterState::new("host.agent"); + state + .apply_line(&rich_activity(7, "idle", "empty", 500), Instant::now()) + .unwrap(); + let mut sequence = None; + attempt_rich_delivery( + &inbox, + &mut pending, + &mut state, + &conflict_pty, + &mut sequence, + &mut receipts, + ); + assert_eq!(conflict_pty.sends.lock().unwrap().len(), 1); + assert!(!pending[0].pty_owned); + assert!(hook_control::load_pty_owned(&inbox).unwrap().is_empty()); + assert!(matches!( + state.authority(Instant::now()), + Err(HoldReason::AdapterError) + )); + } + + #[test] + fn exact_hook_ownership_removes_only_named_unread_work_without_pty_input() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = inbox_dir(tmp.path()); + let first = send_to_inbox(&inbox, "sender", Some("one"), None, &[], "first").unwrap(); + let second = send_to_inbox(&inbox, "sender", Some("two"), None, &[], "second").unwrap(); + let messages = message::list_inbox(&inbox).unwrap(); + let mut pending = VecDeque::from([RichNotice::recovery(&messages)]); + let mut seen = messages + .iter() + .map(|message| message.filename.clone()) + .collect(); + record_hook_owned(&inbox, std::slice::from_ref(&first)).unwrap(); + + assert_eq!( + refresh_rich_pending(&inbox, &mut seen, &mut pending).unwrap(), + vec![first] + ); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].filenames, HashSet::from([second])); + } + + #[test] + fn durable_pty_ownership_blocks_replay_after_sidecar_restart() { + let tmp = tempfile::tempdir().unwrap(); + let inbox = inbox_dir(tmp.path()); + let filename = send_to_inbox(&inbox, "sender", Some("work"), None, &[], "body").unwrap(); + let messages = message::list_inbox(&inbox).unwrap(); + hook_control::record_pty_owned(&inbox, &HashSet::from([filename]), "generation-a", 41, 7) + .unwrap(); + + // A fresh in-memory queue simulates a sidecar restart. Durable ownership is reloaded before + // any activity lease can authorize another guarded write. + let mut pending = VecDeque::from([RichNotice::recovery(&messages)]); + let mut seen = messages + .iter() + .map(|message| message.filename.clone()) + .collect(); + assert!( + refresh_rich_pending(&inbox, &mut seen, &mut pending) + .unwrap() + .is_empty() + ); + assert!(pending[0].pty_owned); + } + #[test] fn poke_id_extracts_rand6() { assert_eq!(poke_id("1785070000000-abc123.md"), "abc123"); diff --git a/src/main.rs b/src/main.rs index 907bceb6..4aba88d6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -108,6 +108,29 @@ enum Command { /// Poll/liveness cadence in milliseconds (folder changes poke immediately regardless). #[arg(long, default_value_t = 1000)] interval: u64, + /// Opt into fail-closed rich DING with this long-running generic activity-adapter + /// executable. Omission preserves the existing aggressive behavior exactly. + #[arg(long)] + adapter: Option, + /// One argument passed directly to `--adapter` (repeatable; no shell parsing). + #[arg(long, requires = "adapter", allow_hyphen_values = true)] + adapter_arg: Vec, + }, + /// EXPERIMENTAL generic local ingress for provider-rendered DING hooks. This records only that + /// an already-running lifecycle hook injected exact unread filenames into the next context. + #[command(name = "ding-control")] + DingControl { + #[command(subcommand)] + operation: DingControlCmd, + /// Whose unread inbox the hook injected. Defaults to `$ST_AGENT`. + #[arg(long)] + identity: Option, + /// Catalog root. Defaults to `$CATALOG`. + #[arg(long, conflicts_with = "catalog_path")] + root: Option, + /// Host used to resolve `.` bus ids. + #[arg(long)] + host: Option, }, /// Get or set an agent's presence status. No `--set` prints the status; no identity means yours /// (`$ST_AGENT`). Settable: offline | available | busy | away | dnd (`unknown` is derived). @@ -339,6 +362,15 @@ enum HooksCmd { VerifyOwn, } +#[derive(Subcommand)] +enum DingControlCmd { + /// Record exact currently unread filenames already injected by a lifecycle hook. + HookOwned { + #[arg(long = "message", required = true)] + messages: Vec, + }, +} + #[derive(Subcommand)] enum ResourceCmd { /// Link a resource (a URL you produced or reference) into your resource list. @@ -562,7 +594,23 @@ fn main() -> Result<()> { root, host, interval, - } => ding_cmd(session, identity, root, host, interval), + adapter, + adapter_arg, + } => ding_cmd( + session, + identity, + root, + host, + interval, + adapter, + adapter_arg, + ), + Command::DingControl { + operation, + identity, + root, + host, + } => ding_control_cmd(operation, identity, root, host), Command::Status { identity, set, ctx } => status_cmd(identity, set, ctx), Command::Agents { catalog, @@ -1193,6 +1241,8 @@ fn ding_cmd( root: Option, host: Option, interval: u64, + adapter: Option, + adapter_args: Vec, ) -> Result<()> { let ctx = MsgCtx { root, @@ -1220,7 +1270,41 @@ fn ding_cmd( poll: Duration::from_millis(interval), ..Default::default() }; - ding::serve(&inbox, &status_path, &session, &config) + let adapter_argv = adapter.map(|program| { + std::iter::once(program) + .chain(adapter_args) + .collect::>() + }); + ding::serve( + &inbox, + &status_path, + &session, + adapter_argv.as_deref(), + &config, + ) +} + +fn ding_control_cmd( + operation: DingControlCmd, + identity: Option, + root: Option, + host: Option, +) -> Result<()> { + let ctx = MsgCtx { + root, + as_id: identity, + host, + }; + let (catalog_root, this_host) = resolve_ctx(&ctx)?; + let id = acting_id(&ctx)?; + let inbox = message::resolve_inbox(&catalog_root, &id, &this_host); + match operation { + DingControlCmd::HookOwned { messages } => { + let receipt = ding::record_hook_owned(&inbox, &messages)?; + println!("{}", receipt.display()); + Ok(()) + } + } } /// Resolve the catalog root and local host from a message subcommand's shared context. diff --git a/tests/ding_control.rs b/tests/ding_control.rs new file mode 100644 index 00000000..43492daf --- /dev/null +++ b/tests/ding_control.rs @@ -0,0 +1,72 @@ +use std::fs; +use std::path::Path; +use std::process::Command; + +use st2::message::{inbox_dir, send_to_inbox}; + +fn st2() -> Command { + Command::new(env!("CARGO_BIN_EXE_st2")) +} + +fn write(root: &Path, relative: &str, contents: &str) { + let path = root.join(relative); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, contents).unwrap(); +} + +#[test] +fn hook_owned_cli_records_only_exact_unread_filenames() { + let tmp = tempfile::tempdir().unwrap(); + let agent_dir = tmp.path().join("agents/host/worker"); + write( + tmp.path(), + "agents/host/worker/agent.kdl", + r#"agent "worker" { host "host"; command "agent"; ding }"#, + ); + let inbox = inbox_dir(&agent_dir); + let filename = send_to_inbox(&inbox, "sender", Some("work"), None, &[], "body").unwrap(); + + let output = st2() + .arg("--catalog") + .arg(tmp.path()) + .args([ + "ding-control", + "--identity", + "host.worker", + "--host", + "host", + "hook-owned", + "--message", + &filename, + ]) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let receipt = Path::new(std::str::from_utf8(&output.stdout).unwrap().trim()); + assert!(receipt.is_file()); + let value: serde_json::Value = serde_json::from_slice(&fs::read(receipt).unwrap()).unwrap(); + assert_eq!(value["kind"], "hook-owned"); + assert_eq!(value["messages"], serde_json::json!([filename])); + + let rejected = st2() + .arg("--catalog") + .arg(tmp.path()) + .args([ + "ding-control", + "--identity", + "host.worker", + "--host", + "host", + "hook-owned", + "--message", + "1785000000000-abc123.md", + ]) + .output() + .unwrap(); + assert!(!rejected.status.success()); + assert!(String::from_utf8_lossy(&rejected.stderr).contains("not currently unread")); +} From d7500b0fcad8bb268da9da96c0226d9caddbe305 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Fri, 31 Jul 2026 11:46:49 +0200 Subject: [PATCH 2/2] fix(ding): expand rich argv from task environment --- INVARIANTS.md | 2 +- README.md | 11 ++- docs/vrs/01-ding/requirements.md | 7 +- docs/vrs/01-ding/spec.md | 9 +- src/exec_backend.rs | 33 +------ src/expand.rs | 6 +- src/run.rs | 104 ++++++++++++++-------- tests/exec_backend.rs | 142 ++++++++++++++++++++++++++++++- 8 files changed, 236 insertions(+), 78 deletions(-) diff --git a/INVARIANTS.md b/INVARIANTS.md index 624a384d..cbb410db 100644 --- a/INVARIANTS.md +++ b/INVARIANTS.md @@ -11,7 +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 `-.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` | +| **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` | diff --git a/README.md b/README.md index 2082f2e8..4609cf45 100644 --- a/README.md +++ b/README.md @@ -161,13 +161,16 @@ 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 -environment. +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" } diff --git a/docs/vrs/01-ding/requirements.md b/docs/vrs/01-ding/requirements.md index 10c14abd..954c2d6b 100644 --- a/docs/vrs/01-ding/requirements.md +++ b/docs/vrs/01-ding/requirements.md @@ -97,9 +97,10 @@ is in [`spec.md`](./spec.md). ### Optional rich delivery must remain generic and fail closed - **DING-R11 Structured adapter launch:** An adapter declaration is one - explicit executable plus argument vector. Core introduces no shell, provider - selector, inferred arguments, or environment. Adapter argv participates in - the DING sidecar launch fingerprint. + executable plus argument vector, expanded against the complete resolved DING + task environment before direct launch. Core introduces no shell, provider + selector, inferred arguments, or adapter-specific environment. Adapter argv + participates in the DING sidecar launch fingerprint. - **DING-R12 Exact generic activity tuple:** Rich PTY input requires one receipt-anchored, unexpired `idle` event with `inputBuffer=empty`. Its exact session, incarnation, PTY generation, and strictly increasing sequence must diff --git a/docs/vrs/01-ding/spec.md b/docs/vrs/01-ding/spec.md index 672b5033..25dbc927 100644 --- a/docs/vrs/01-ding/spec.md +++ b/docs/vrs/01-ding/spec.md @@ -154,9 +154,12 @@ ding { ``` The argv values follow normal task launch semantics. The first value is the -executable, remaining values are exact arguments, and declared task/agent -environment is inherited. st2 neither runs a shell nor adds arguments or -environment. Bare `ding` continues to lower to +executable and remaining values are exact arguments. st2 first resolves the +generated sidecar's complete managed environment, including `CATALOG`, +`ST_ROOT`, and declared task/agent `env`, and then expands each argv token +against that final map without changing argument boundaries. It neither runs a +shell nor adds adapter arguments or adapter-specific environment. Bare `ding` +continues to lower to `st2 ding --identity --root $ST_ROOT` exactly. The long-running adapter owns provider-native interpretation and publishes diff --git a/src/exec_backend.rs b/src/exec_backend.rs index 97eb580d..01ee0b3a 100644 --- a/src/exec_backend.rs +++ b/src/exec_backend.rs @@ -22,7 +22,7 @@ use std::process::Stdio; use crate::host_lock::process_alive; use crate::reconcile::{Session, TaskLaunch, TaskTarget}; -use crate::run::resolve_task_cwd; +use crate::run::{expand_task_argv, managed_task_env, resolve_task_cwd}; /// Supervises `exec` tasks as terminal-free processes. pub struct ExecBackend { @@ -74,6 +74,7 @@ impl ExecBackend { // kill cannot take it; elsewhere a plain pass-through detached by the `setsid` below. Env, cwd, // and stdio set here reach the task in both modes. let unit = crate::isolate::scope_unit(&target.pty_id); + let managed_env = managed_task_env(target, &self.catalog_root, None); let (program, args): (OsString, Vec) = match &target.launch { TaskLaunch::Shell(command) => ( OsString::from("sh"), @@ -81,12 +82,7 @@ impl ExecBackend { ), TaskLaunch::Argv(argv) => { debug_assert!(!argv.is_empty()); - let mut expanded = argv - .iter() - .map(|arg| { - OsString::from(crate::expand::expand_catalog(arg, &self.catalog_root)) - }) - .collect::>(); + let mut expanded = expand_task_argv(argv, &managed_env); let program = expanded.remove(0); (program, expanded) } @@ -97,28 +93,7 @@ impl ExecBackend { .stdin(Stdio::null()) .stdout(log.try_clone()?) .stderr(log) - .env("CATALOG", &self.catalog_root) - .env("ST_ROOT", &self.catalog_root) - .env( - "PTY_ROOT", - crate::run::effective_pty_root(&self.catalog_root), - ); - if let Ok(path) = crate::hooks::hooks_root() { - cmd.env("ST_HOOKS", path); - } - for (k, v) in &target.env { - // PTY_ROOT resolves to the EFFECTIVE root (an exported ambient one wins over the rendered - // `$CATALOG/pty`) so a ding — an exec task that reads PTY_ROOT to find the pty it pokes — - // targets the same partition st2's pty ops use. - if k == "PTY_ROOT" { - cmd.env( - "PTY_ROOT", - crate::run::effective_pty_root(&self.catalog_root), - ); - } else { - cmd.env(k, crate::expand::expand_catalog(v, &self.catalog_root)); - } - } + .envs(&managed_env); // New session: no controlling terminal (R09) and decoupled from st2's process group, so it // survives st2 exit and `kill(-pid)` teardown reaps the whole group. In Scope mode this runs // on the outer `systemd-run`, which exec-chains into the workload, so the workload is the diff --git a/src/expand.rs b/src/expand.rs index acb137f5..6db67eda 100644 --- a/src/expand.rs +++ b/src/expand.rs @@ -1,8 +1,8 @@ //! Spawn-time environment-variable expansion. //! -//! st2 spawns via `pty run` with values passed through the child environment. Those values, `cwd`, -//! and tags do not pass through a shell, so st2 expands them explicitly. The command itself runs -//! under `sh -c` and expands there. +//! st2 spawns with values, `cwd`, tags, and structured argv that do not pass through a shell, so st2 +//! expands them explicitly. Direct argv is expanded after the complete managed task environment is +//! resolved. The command form itself runs under `sh -c` and expands there. //! //! Supported forms: `$VAR`, `${VAR}`, and `$$` → literal `$`. An **unset** variable is left as its //! literal token (`$VAR`) rather than blanked — a spawn-time path with an undefined var is a diff --git a/src/run.rs b/src/run.rs index 6caa7e4a..3daff656 100644 --- a/src/run.rs +++ b/src/run.rs @@ -12,7 +12,7 @@ use std::cell::RefCell; use std::collections::{BTreeMap, HashMap, HashSet}; -use std::ffi::OsString; +use std::ffi::{OsStr, OsString}; use std::io::{Read as _, Seek as _}; use std::os::unix::process::CommandExt as _; use std::path::{Path, PathBuf}; @@ -101,6 +101,66 @@ pub(crate) fn resolve_task_cwd( } } +/// Resolve the st2-owned launch environment before expanding any structured argv token. +/// +/// Declared values retain their existing `$CATALOG` plus ambient expansion semantics. The returned +/// map is then authoritative for direct argv expansion, so task-only values such as `ST_ROOT` and a +/// declared adapter root do not depend on the supervisor's ambient environment. +pub(crate) fn managed_task_env( + target: &TaskTarget, + catalog_root: &Path, + default_term: Option<&str>, +) -> BTreeMap { + let mut env = BTreeMap::from([ + ( + OsString::from("CATALOG"), + catalog_root.as_os_str().to_os_string(), + ), + ( + OsString::from("ST_ROOT"), + catalog_root.as_os_str().to_os_string(), + ), + ( + OsString::from("PTY_ROOT"), + effective_pty_root(catalog_root).into_os_string(), + ), + ]); + if let Some(term) = default_term { + env.insert(OsString::from("TERM"), OsString::from(term)); + } + if let Ok(path) = crate::hooks::hooks_root() { + env.insert(OsString::from("ST_HOOKS"), path.into_os_string()); + } + for (key, value) in &target.env { + let value = if key == "PTY_ROOT" { + effective_pty_root(catalog_root).into_os_string() + } else { + OsString::from(crate::expand::expand_catalog(value, catalog_root)) + }; + env.insert(OsString::from(key), value); + } + env +} + +/// Expand one structured argv against the complete resolved task environment while preserving +/// every argument boundary and introducing no shell. Managed/declared values outrank ambient ones; +/// the ambient fallback preserves the existing direct-argv contract for undeclared variables. +pub(crate) fn expand_task_argv( + argv: &[String], + task_env: &BTreeMap, +) -> Vec { + argv.iter() + .map(|argument| { + OsString::from(crate::expand::expand_vars(argument, |key| { + task_env + .get(OsStr::new(key)) + .map(|value| value.to_string_lossy().into_owned()) + .or_else(|| std::env::var(key).ok()) + })) + }) + .collect() +} + /// The set of task operations st2 needs. Abstracted so execution is testable against a fake. pub trait Runner { /// ACTUAL state: every task session the runner can see (unioned across backends). @@ -189,41 +249,16 @@ impl PtyCli { /// initial `pty run` process and persisted through repeatable `--env KEY=VALUE` arguments, so a /// manual `pty restart` recreates the task without snapshotting unrelated ambient OS variables. fn managed_task_env(&self, target: &TaskTarget) -> BTreeMap { - let mut env = BTreeMap::from([ - ( - OsString::from("CATALOG"), - self.catalog_root.as_os_str().to_os_string(), - ), - ( - OsString::from("ST_ROOT"), - self.catalog_root.as_os_str().to_os_string(), - ), - ( - OsString::from("PTY_ROOT"), - effective_pty_root(&self.catalog_root).into_os_string(), - ), - (OsString::from("TERM"), OsString::from("xterm-256color")), - ]); - if let Ok(path) = crate::hooks::hooks_root() { - env.insert(OsString::from("ST_HOOKS"), path.into_os_string()); - } - for (key, value) in &target.env { - let value = if key == "PTY_ROOT" { - effective_pty_root(&self.catalog_root).into_os_string() - } else { - OsString::from(self.expand(value)) - }; - env.insert(OsString::from(key), value); - } - env + managed_task_env(target, &self.catalog_root, Some("xterm-256color")) } /// Build (but do not run) the `pty run` invocation for `target`. Split out so the exact argv + /// env can be unit-tested without spawning anything. /// /// `$VAR`s are expanded here for everything that does NOT pass through a shell — env values, tag - /// values, `cwd`, and direct argv — because `pty` passes them through verbatim. Shell source is - /// left unexpanded: `sh -c` expands it at spawn from the same env (which includes `$CATALOG`). + /// values, `cwd`, and direct argv — because `pty` passes them through verbatim. Direct argv uses + /// the complete resolved managed task environment. Shell source is left unexpanded: `sh -c` + /// expands it at spawn from the same env (which includes `$CATALOG`). fn build_run_command(&self, target: &TaskTarget, spec_dir: &Path) -> Command { let cwd = self.resolve_cwd(target, spec_dir); let mut cmd = Command::new(&self.bin); @@ -270,7 +305,7 @@ impl PtyCli { // Direct mode preserves argument boundaries and introduces no shell process. TaskLaunch::Argv(argv) => { debug_assert!(!argv.is_empty()); - cmd.args(argv.iter().map(|arg| self.expand(arg))); + cmd.args(expand_task_argv(argv, &managed_env)); } } cmd @@ -1748,14 +1783,15 @@ mod tests { let cli = PtyCli::new(PathBuf::from("/my/catalog")); let mut t = target("hetz.demo.agent", "unused"); t.launch = TaskLaunch::Argv(vec![ - "axe".into(), + "$TOOL_ROOT/axe".into(), "agent".into(), "exec".into(), "--".into(), "claude".into(), "--resume".into(), - "$CATALOG/session id".into(), + "$ST_ROOT/session id".into(), ]); + t.env.insert("TOOL_ROOT".into(), "/opt/task tools".into()); let cmd = cli.build_run_command(&t, Path::new("/cat/hetz/demo")); let args = cmd .get_args() @@ -1766,7 +1802,7 @@ mod tests { assert_eq!( &args[sep + 1..], [ - "axe", + "/opt/task tools/axe", "agent", "exec", "--", diff --git a/tests/exec_backend.rs b/tests/exec_backend.rs index ef2d13b4..225e29c7 100644 --- a/tests/exec_backend.rs +++ b/tests/exec_backend.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; use std::fs; -#[cfg(target_os = "macos")] +use std::os::unix::fs::{PermissionsExt as _, symlink}; use std::process::Command; use std::thread::sleep; use std::time::Duration; @@ -168,6 +168,146 @@ fn exec_launches_direct_argv_with_literal_boundaries_and_catalog_expansion() { backend.remove(id).unwrap(); } +#[test] +fn generated_rich_ding_uses_task_only_roots_without_shell_expansion() { + let output = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "generated_rich_ding_task_env_helper", + "--nocapture", + ]) + .env("ST2_RICH_DING_TASK_ENV_HELPER", "1") + .env_remove("ST_ROOT") + .env_remove("ADAPTER_ROOT") + .output() + .unwrap(); + assert!( + output.status.success(), + "isolated helper failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +#[test] +fn generated_rich_ding_task_env_helper() { + if std::env::var_os("ST2_RICH_DING_TASK_ENV_HELPER").is_none() { + return; + } + assert!(std::env::var_os("ST_ROOT").is_none()); + assert!(std::env::var_os("ADAPTER_ROOT").is_none()); + + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let state = tmp.path().join("state"); + let bin = tmp.path().join("bin"); + let adapter_root = tmp.path().join("adapters"); + let record = tmp.path().join("adapter-record.txt"); + fs::create_dir_all(catalog.join("agents/testhost/worker")).unwrap(); + fs::create_dir_all(&bin).unwrap(); + fs::create_dir_all(&adapter_root).unwrap(); + symlink(env!("CARGO_BIN_EXE_st2"), bin.join("st2")).unwrap(); + + let adapter = adapter_root.join("activity"); + fs::write( + &adapter, + r#"#!/bin/sh +{ + printf 'root=%s\n' "$ST_ROOT" + for argument in "$@"; do + printf 'arg=%s\n' "$argument" + done +} > "$ADAPTER_RECORD" +exec /bin/sleep 30 +"#, + ) + .unwrap(); + let mut permissions = fs::metadata(&adapter).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(&adapter, permissions).unwrap(); + + let task_path = format!( + "{}:{}", + bin.display(), + std::env::var("PATH").expect("test helper PATH") + ); + let declaration = format!( + r#" +agent "worker" {{ + host "testhost" + command "/bin/true" + env {{ + PATH "{}" + ADAPTER_ROOT "{}" + ADAPTER_RECORD "{}" + }} + ding {{ + adapter {{ + argv "$ADAPTER_ROOT/activity" "--format" "jsonl" "space arg" "; touch forbidden" + }} + }} +}} +"#, + task_path, + adapter_root.display(), + record.display() + ); + fs::write( + catalog.join("agents/testhost/worker/agent.kdl"), + declaration, + ) + .unwrap(); + + let found = st2::discover(&catalog); + assert!( + found.errors.is_empty(), + "generated declaration did not parse: {:?}", + found.errors + ); + let plan = st2::reconcile::reconcile(&found.specs, &[], "testhost"); + let target = plan + .launch + .iter() + .flat_map(|launch| &launch.tasks) + .find(|target| target.name == "ding") + .expect("generated rich DING task") + .clone(); + assert!(matches!(target.launch, TaskLaunch::Argv(_))); + + let backend = ExecBackend::new(state, catalog.clone()); + let spec_dir = found.specs[0].path.parent().unwrap(); + backend.spawn(&target, spec_dir).unwrap(); + assert!( + wait_until(|| record.is_file()), + "generated sidecar did not start its task-env adapter; log: {:?}", + fs::read_to_string(catalog.join("logs").join(format!("{}.log", target.pty_id))).ok() + ); + assert_eq!( + fs::read_to_string(&record).unwrap(), + format!( + "root={}\narg=--format\narg=jsonl\narg=space arg\narg=; touch forbidden\n", + catalog.display() + ) + ); + assert!(!spec_dir.join("forbidden").exists()); + + let log = catalog.join("logs").join(format!("{}.log", target.pty_id)); + let expected_inbox = catalog.join("agents/testhost/worker/resources/inbox"); + assert!( + wait_until(|| fs::read_to_string(&log) + .unwrap_or_default() + .contains(&expected_inbox.display().to_string())), + "generated `--root $ST_ROOT` did not resolve the catalog inbox: {:?}", + fs::read_to_string(&log).ok() + ); + + backend.kill(&target.pty_id).unwrap(); + assert!(wait_until(|| backend.list().unwrap().iter().any( + |session| session.pty_id == target.pty_id && !session.alive + ))); + backend.remove(&target.pty_id).unwrap(); +} + /// Auto-log observability: a detached exec's stdout AND stderr must be captured to a discoverable /// `/logs/