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

@@ -13,6 +13,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` |
 | **Derived companion lifecycle** | A generated DING starts only with an eligible canonical agent, is suppressed or stopped while that target is held, suspended, unavailable, retired, or terminally parked, and remains coupled without changing explicit sibling task behavior. Healthy compact startup still launches both tasks in one pass, and targeted reconciliation holds a missing generated DING rather than broadening to its agent. | `tests/run.rs::fresh_compact_agent_launches_with_its_derived_ding`; `tests/run.rs::absent_adopt_only_compact_agent_holds_its_derived_ding`; `tests/run.rs::held_adopt_only_compact_agent_stops_its_live_derived_ding`; `tests/run.rs::failed_compact_agent_restart_stops_its_live_derived_ding`; `tests/run.rs::failed_compact_agent_reap_stops_its_live_derived_ding`; `tests/run.rs::parked_compact_agent_stops_its_live_derived_ding`; `tests/run.rs::parked_compact_agent_does_not_relaunch_its_exited_derived_ding`; `tests/run.rs::retired_compact_agent_stops_agent_and_derived_ding`; `tests/run.rs::suspend_and_resume_cover_derived_ding_sibling_continuity_and_inbox_retention`; `tests/run.rs::selected_missing_derived_ding_is_held_without_broadening_to_its_agent`; `tests/run.rs::up_once_collects_spawn_errors_without_aborting` |
 | **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. Retry never re-pastes and may send one bare Return only after two adjacent `RetainedSafe` observations. A maintained adapter's positive `NotRetained` observation releases only an already archived staged head; unread, blocked, timed-out, errored, unknown, and unrecognized states retain ownership and later FIFO work remains blocked. Ownership prevents duplicate paste across command failures, receipt ambiguity, archive races, and restart adoption without letting a vanished archived head block FIFO indefinitely. 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_retry_keeps_unproven_and_retained_blocked_owned`; `src/ding/mod.rs::staged_ownership_survives_archive_and_never_repastes`; `src/ding/mod.rs::archived_not_retained_releases_fifo_without_repasting_owned_notice`; `src/ding/mod.rs::unread_not_retained_keeps_fifo_ownership_without_repasting`; `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

@@ -416,6 +416,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

@@ -151,6 +151,32 @@ not duplicated in tags. Clearing removes only the owned native value or tag,
 and unrelated PTY metadata is preserved. Repeating the same projection is a
 no-op. Failure is reported and retried by the ordinary loop, never converted
 into launch, teardown, garbage collection, replacement, or flapping authority.
+## 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)
 

src/lib.rs

@@ -24,6 +24,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),
@@ -677,6 +680,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,
@@ -708,6 +767,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),
@@ -2017,6 +2077,132 @@ fn send_resolved_message(
     }
 }
 
+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

@@ -224,6 +224,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 {
@@ -1161,6 +1201,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();