feat(request): add idempotent service principals

agent-session-id: dev3.dotfiles-cos-misc-issue-1329-upstream
agent-tool: Codex
agent-tool-version: 0.145.0
agent-model: gpt-5.6-sol
agent-runtime-profile: /nix/store/qlk5xbdfmj5nn1q145j0fx73f2pmdjj7-coding-agent-runtime-profile/share/coding-agents/profile.json
agent-skills-manifest: /nix/store/2km8dbiyv3wc484l1hd8n2lh65121qvq-agent-skills-corpus/share/agent-skills/manifest.json
tooling-profile: dotfiles@unknown-dirty

Author: schickling-assistant

INVARIANTS.md

@@ -11,6 +11,7 @@ materialization, messaging, DING, or presence must preserve them.
 | **Clean exec teardown** | Killing an exec task reaps its whole process group. | `tests/exec_backend.rs::exec_kill_reaps_the_whole_process_group_not_just_the_leader` |
 | **Bounded restart diagnostics** | Relaunching an exec task preserves the just-finished log as one prior generation while bounding retained diagnostics to current plus prior. Final retirement removes the PID and both logs. | `tests/exec_backend.rs::exec_restart_reap_keeps_bounded_diagnostics_and_final_remove_cleans_them`; `tests/run.rs::up_once_finally_removes_dead_retired_tasks_without_restarting_them` |
 | **Exactly-once-safe native bus** | Messages use stable `<unix-ms>-<rand6>.md` files. An archive filename is a durable receipt that shadows and cleans restored inbox replicas and makes repeated archive cleanup idempotent. | `src/message.rs::filename_grammar`; `src/message.rs::archive_receipt_suppresses_and_idempotently_cleans_a_restored_inbox_copy`; `tests/message.rs` |
+| **Idempotent service requests** | A declared non-agent service principal publishes one exact JSON request per caller-supplied idempotency key to a canonical Agent Spec inbox. Concurrent or crash-replayed publication reuses the reserved filename; conflicting key reuse fails. The typed reply routes to the principal's canonical inbox without an Agent Spec identity or orphan mailbox. | `tests/request_cli.rs::stable_request_key_atomically_deduplicates_one_canonical_agent_message`; `tests/request_cli.rs::concurrent_replays_publish_exactly_one_request`; `tests/request_cli.rs::typed_reply_routes_to_the_principal_and_status_is_a_tagged_json_union`; `tests/request_cli.rs::request_api_rejects_agent_impersonation_and_unknown_flat_principals` |
 | **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` |
 | **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` |

README.md

@@ -324,6 +324,40 @@ intentional catalog-less fallback used by isolated folder evals. In a catalog-ba
 `st2 message ls` rejects an absent identity; recovery inspection of a deliberately orphaned flat
 box must be explicit with `st2 message ls <identity> --orphan` (and optionally `--archive`).
 
+Services use a separate declared request surface; they do not borrow an Agent
+Spec identity. Declare the endpoint without creating a task:
+
+```kdl
+// <catalog>/principals/dev3/hypermerge/principal.kdl
+principal "hypermerge" host="dev3"
+```
+
+Then publish once, let the addressed agent reply from its normal inbox, and
+observe the typed result. Bodies are JSON and `--tag` is a repeatable
+`key=value` map:
+
+```sh
+st2 request send dev3.repair-agent \
+  --as dev3.hypermerge \
+  --idempotency-key 'escalate:repo#7:abc' \
+  --tag kind=hypermerge.escalation \
+  -m '{"candidate":"abc"}' --json
+
+st2 request read <request-filename> --json
+
+st2 request reply <request-filename> \
+  --tag outcome=needs-human \
+  -m '{"outcome":"needs-human"}' --json
+
+st2 request status --as dev3.hypermerge \
+  --idempotency-key 'escalate:repo#7:abc' --json
+```
+
+The key atomically reserves one message filename and exact envelope before
+publication. Exact retries return that filename with `deduplicated: true`; a
+different request under the same key fails. Replies route only to the declared
+principal's canonical `resources/inbox`, never a flat orphan mailbox.
+
 Adopters should cut directly to the native layout. Before launching a migrated identity, install and
 verify hooks, validate and materialize its hand-authored declaration, stop any predecessor transport,
 and decide how any unread predecessor backlog will be archived or forwarded. Never run predecessor

docs/vrs/spec.md

@@ -60,6 +60,33 @@ materialization, frozen routing after declaration removal, singleton
 completion, custom task-ID supervision/logging/teardown, and the no-opt-in
 legacy control in `tests/eval_run_e2e.rs`.
 
+## Service-principal request transport
+
+A non-agent service that needs bounded judgment work may declare only its bus
+endpoint at
+`principals/<host>/<identity>/principal.kdl`:
+
+```kdl
+principal "hypermerge" host="dev3"
+```
+
+The declaration creates no task, presence, persona, or Agent Spec authority.
+Its content must exactly match its canonical path. `st2 request send` accepts
+only such a principal as the caller and only a discovered Agent Spec as the
+recipient, so a service neither impersonates an agent nor depends on the flat
+orphan-recovery layout.
+
+The caller supplies an idempotency key, a JSON body, and typed string tags.
+Before the native message is published, st2 atomically reserves one random
+canonical message filename and the exact request envelope under the
+principal's `resources/request-state/`. Replays finish that same publication;
+reuse of the key with different caller, recipient, body, or tags fails. An
+agent's `st2 request reply` similarly publishes at most one typed reply to the
+principal's canonical inbox. `st2 request status --json` returns the tagged
+union `pending | replied`, suitable for a durable workflow to observe between
+its own durable waits. st2 provides no wait loop or timer and does not turn the
+request into agent lifecycle authority.
+
 ## Resource bindings (R20-R21)
 
 An agent may directly declare zero or more generic Resource bindings:

src/lib.rs

@@ -21,6 +21,7 @@ pub mod materialize;
 pub mod message;
 pub mod pretrust;
 pub mod reconcile;
+pub mod request;
 pub mod resource;
 pub mod run;
 pub mod service;

src/main.rs

@@ -69,6 +69,9 @@ enum Command {
     /// The stable wire format is a `<unix-ms>-<rand6>.md` Markdown file.
     #[command(subcommand)]
     Message(MessageCmd),
+    /// Idempotent JSON request/reply transport for declared non-agent service principals.
+    #[command(subcommand)]
+    Request(RequestCmd),
     /// An agent's working-state context for lossless restart: read/write/append.
     #[command(subcommand)]
     Context(ContextCmd),
@@ -531,6 +534,62 @@ enum MessageCmd {
     },
 }
 
+#[derive(Subcommand)]
+enum RequestCmd {
+    /// Publish one idempotent JSON request from a declared service principal to an agent.
+    Send {
+        /// Recipient agent: a bus id (`<host>.<identity>`) or a local bare identity.
+        to: String,
+        #[arg(long = "idempotency-key")]
+        idempotency_key: String,
+        /// Typed request tag as `key=value` (repeatable).
+        #[arg(long = "tag")]
+        tags: Vec<String>,
+        /// JSON body. Read from stdin when omitted.
+        #[arg(short = 'm', long = "message")]
+        body: Option<String>,
+        /// Emit the machine receipt as JSON.
+        #[arg(long)]
+        json: bool,
+        #[command(flatten)]
+        ctx: MsgCtx,
+    },
+    /// Decode one typed request from an agent's inbox.
+    Read {
+        request_filename: String,
+        /// Emit the request envelope as JSON.
+        #[arg(long)]
+        json: bool,
+        #[command(flatten)]
+        ctx: MsgCtx,
+    },
+    /// Reply once to a typed request in an agent's inbox.
+    Reply {
+        request_filename: String,
+        /// Typed reply tag as `key=value` (repeatable).
+        #[arg(long = "tag")]
+        tags: Vec<String>,
+        /// JSON body. Read from stdin when omitted.
+        #[arg(short = 'm', long = "message")]
+        body: Option<String>,
+        /// Emit the machine receipt as JSON.
+        #[arg(long)]
+        json: bool,
+        #[command(flatten)]
+        ctx: MsgCtx,
+    },
+    /// Observe the typed reply for one previously published request.
+    Status {
+        #[arg(long = "idempotency-key")]
+        idempotency_key: String,
+        /// Emit the tagged status union as JSON.
+        #[arg(long)]
+        json: bool,
+        #[command(flatten)]
+        ctx: MsgCtx,
+    },
+}
+
 fn main() -> Result<()> {
     let Cli {
         catalog_path,
@@ -562,6 +621,7 @@ fn main() -> Result<()> {
             up(&root, host, once, materialize_only, interval, agent, task)
         }
         Command::Message(cmd) => message_cmd(cmd),
+        Command::Request(cmd) => request_cmd(cmd),
         Command::Context(cmd) => context_cmd(cmd),
         Command::Resource(cmd) => resource_cmd(cmd),
         Command::Service(cmd) => service_cmd(cmd),
@@ -1506,6 +1566,132 @@ fn message_cmd(cmd: MessageCmd) -> Result<()> {
     }
 }
 
+fn request_cmd(cmd: RequestCmd) -> Result<()> {
+    match cmd {
+        RequestCmd::Send {
+            to,
+            idempotency_key,
+            tags,
+            body,
+            json,
+            ctx,
+        } => {
+            let (root, host) = resolve_ctx(&ctx)?;
+            let principal = acting_id(&ctx)?;
+            let body = parse_json_body(body)?;
+            let receipt = st2::request::publish(
+                &root,
+                &host,
+                &principal,
+                &to,
+                &idempotency_key,
+                parse_typed_tags(tags)?,
+                body,
+            )?;
+            print_publish_receipt(&receipt, json)
+        }
+        RequestCmd::Reply {
+            request_filename,
+            tags,
+            body,
+            json,
+            ctx,
+        } => {
+            let (root, host) = resolve_ctx(&ctx)?;
+            let agent = acting_id(&ctx)?;
+            let body = parse_json_body(body)?;
+            let receipt = st2::request::reply(
+                &root,
+                &host,
+                &agent,
+                &request_filename,
+                parse_typed_tags(tags)?,
+                body,
+            )?;
+            print_publish_receipt(&receipt, json)
+        }
+        RequestCmd::Read {
+            request_filename,
+            json,
+            ctx,
+        } => {
+            let (root, host) = resolve_ctx(&ctx)?;
+            let agent = acting_id(&ctx)?;
+            let request = st2::request::read(&root, &host, &agent, &request_filename)?;
+            if json {
+                println!("{}", serde_json::to_string(&request)?);
+            } else {
+                println!(
+                    "request {} from {} ({})",
+                    request.idempotency_key, request.from, request.request_filename
+                );
+            }
+            Ok(())
+        }
+        RequestCmd::Status {
+            idempotency_key,
+            json,
+            ctx,
+        } => {
+            let (root, host) = resolve_ctx(&ctx)?;
+            let principal = acting_id(&ctx)?;
+            let status = st2::request::status(
+                &root,
+                &host,
+                &principal,
+                &idempotency_key,
+            )?;
+            if json {
+                println!("{}", serde_json::to_string(&status)?);
+            } else {
+                match status {
+                    st2::request::RequestStatus::Pending {
+                        idempotency_key,
+                        request_filename,
+                    } => println!("pending {idempotency_key} ({request_filename})"),
+                    st2::request::RequestStatus::Replied {
+                        idempotency_key,
+                        request_filename,
+                        from,
+                        ..
+                    } => println!("replied {idempotency_key} ({request_filename}) from {from}"),
+                }
+            }
+            Ok(())
+        }
+    }
+}
+
+fn parse_json_body(body: Option<String>) -> Result<serde_json::Value> {
+    let body = body_or_stdin(body)?;
+    serde_json::from_str(&body).context("request body must be valid JSON")
+}
+
+fn parse_typed_tags(tags: Vec<String>) -> Result<std::collections::BTreeMap<String, String>> {
+    let mut parsed = std::collections::BTreeMap::new();
+    for tag in tags {
+        let (key, value) = tag
+            .split_once('=')
+            .with_context(|| format!("typed tag must be `key=value`, got `{tag}`"))?;
+        if key.is_empty() || value.is_empty() {
+            anyhow::bail!("typed tag must have a non-empty key and value: `{tag}`");
+        }
+        if parsed.insert(key.to_string(), value.to_string()).is_some() {
+            anyhow::bail!("duplicate typed tag key `{key}`");
+        }
+    }
+    Ok(parsed)
+}
+
+fn print_publish_receipt(receipt: &st2::request::PublishReceipt, json: bool) -> Result<()> {
+    if json {
+        println!("{}", serde_json::to_string(receipt)?);
+    } else {
+        println!("{}", receipt.filename);
+    }
+    Ok(())
+}
+
 /// `st2 message ls --json` row (stable st2 wire contract).
 #[derive(serde::Serialize)]
 struct LsItemJson<'a> {

src/message.rs

@@ -222,6 +222,46 @@ pub fn send_to_inbox(
     )
 }
 
+/// Atomically materialize one already-rendered canonical message filename.
+///
+/// Repeating the same filename and bytes is success; the same filename with different bytes is an
+/// error. Request publication uses this after durably reserving its random filename, so a replay
+/// finishes an interrupted send without allocating a second bus message.
+pub fn materialize_message_once(
+    inbox_dir: &Path,
+    filename: &str,
+    contents: &str,
+) -> anyhow::Result<bool> {
+    if !is_message_filename(filename) {
+        anyhow::bail!("invalid canonical message filename `{filename}`");
+    }
+    fs::create_dir_all(inbox_dir)?;
+    let destination = inbox_dir.join(filename);
+    if destination.is_file() {
+        let existing = fs::read_to_string(&destination)?;
+        if existing == contents {
+            return Ok(false);
+        }
+        anyhow::bail!("message filename collision with different bytes: {filename}");
+    }
+    let temporary = inbox_dir.join(tmp_name());
+    fs::write(&temporary, contents)?;
+    let result = match fs::hard_link(&temporary, &destination) {
+        Ok(()) => Ok(true),
+        Err(_) if destination.is_file() => {
+            let existing = fs::read_to_string(&destination)?;
+            if existing == contents {
+                Ok(false)
+            } else {
+                anyhow::bail!("message filename collision with different bytes: {filename}")
+            }
+        }
+        Err(error) => Err(error.into()),
+    };
+    let _ = fs::remove_file(temporary);
+    result
+}
+
 static TMP_COUNTER: AtomicU64 = AtomicU64::new(0);
 
 fn tmp_name() -> String {
@@ -648,6 +688,22 @@ mod tests {
         assert_eq!(fs::read(archive.join(&filename)).unwrap(), receipt);
     }
 
+    #[test]
+    fn reserved_message_materialization_is_idempotent_but_never_clobbers() {
+        let tmp = tempfile::tempdir().unwrap();
+        let inbox = tmp.path().join("inbox");
+        let filename = "1784649988123-abc23z.md";
+        let contents = render_message("service", Some("request"), None, &[], "{}\n");
+
+        assert!(materialize_message_once(&inbox, filename, &contents).unwrap());
+        assert!(!materialize_message_once(&inbox, filename, &contents).unwrap());
+        let error = materialize_message_once(&inbox, filename, "different")
+            .unwrap_err()
+            .to_string();
+        assert!(error.contains("collision with different bytes"));
+        assert_eq!(fs::read_to_string(inbox.join(filename)).unwrap(), contents);
+    }
+
     #[test]
     fn resolve_inbox_falls_back_to_the_flat_bus_when_catalog_less() {
         let tmp = tempfile::tempdir().unwrap();