action handler: SSH executor target (ExecuteCommand-over-SSH) + recovered B2 docs

.claude/board/EPIPHANIES.md

@@ -7,6 +7,46 @@
 
 ---
 
+## 2026-06-24 — E-ACTIONHANDLER-TRANSPORT — the daemon is transport-agnostic because HIRO is multi-wire; and the OGIT Auth type unifies "who connects" with "who the gate authorizes"
+
+**Status:** FINDING (`[G]` for the core + WebSocket edge; `[H]` for the Kafka edge).
+
+Two design facts surfaced building B2-transport (the live action daemon, in
+rs-graph-llm `graph-flow-action-ogar::daemon`):
+
+1. **HIRO distributes actions over more than one wire** — a handler-facing
+   WebSocket (`action-ws`) AND an internal Kafka bus that legacy handlers consume
+   directly (operator note, 2026-06-24). The wire differs; the dispatch doesn't.
+   So the daemon is factored as: `Daemon::react` (the transport-agnostic core —
+   one inbound `action-ws` frame → outbound frames, running the gate + executor,
+   pure/no-I/O) + a `Transport` trait (`recv`/`send`, the swappable edge) +
+   `Daemon::serve` (the loop, generic over `Transport`). The WebSocket edge
+   (`WsTransport`) and a future Kafka edge (`rdkafka`) share `serve` verbatim —
+   the gated dispatch is written once, the wire is a thin shell. This is the
+   action-arm analogue of the codec stack's "one algebra, many carriers": one
+   dispatch, many transports.
+
+2. **The OGIT Auth type unifies the two identities that must be the same.** A
+   handler's connection presents a credential; the gate authorizes an actor.
+   These MUST be the same principal — and OGIT's `NTO/Auth/Configuration` (the
+   `auth_store` class, OGAR `0x0B01`) already unifies them: it is keyed by
+   `accountId` and maps `sub` → actor (`0x0104`), org/tenant → scope. So the
+   daemon's `Auth` type is shaped after it: one value carries the `token` the
+   transport presents (the `token-$TOKEN` subprotocol) AND the `account` the gate
+   authorizes as (`accountId` → actor). `Daemon::new` takes `&Auth` and derives
+   the gate actor from `auth.account`; `WsTransport::connect` takes `&Auth` and
+   presents `auth.token`. The identity that connects IS the identity the RBAC
+   grant is checked against — structurally, not by convention. (A future
+   producer-side `auth_from_ogit(entity)` lift would populate `Auth` from a real
+   `NTO/Auth/Configuration` node, the same way `assemble_action_handler` lifts the
+   handler contract.)
+
+Proven by `ws_roundtrip_against_a_mock_server` (engine `submitAction` → ack → real
+command → result over a live socket) + 10 pure-core tests. Scorecard: B2-transport
+WebSocket edge SHIPPED; Kafka edge reserved (`D-ACTIONHANDLER-TRANSPORT`).
+
+---
+
 ## 2026-06-24 — E-ACTIONHANDLER-B2LIFT — the producer stays parser-free even when lifting a JSON REST response: it defines the `Deserialize` DTOs + the pure lift, the runtime does the `from_str`
 
 **Status:** FINDING (`[G]` for capabilities; `[H]` for the applicabilities envelope).

crates/ogar-action-handler/src/lib.rs

@@ -141,6 +141,118 @@ impl CapabilityExecutor for NativeCommandExecutor {
     }
 }
 
+/// Reference executor for the **SSH** target: runs an `ExecuteCommand`
+/// capability's `command` on a remote host via the system `ssh` binary. This is
+/// arago's canonical ActionHandler shape (`ExecuteCommand`-over-SSH) — the
+/// [`NativeCommandExecutor`], remote.
+///
+/// Dependency-free by design: it shells out to `ssh` (like the native one shells
+/// out to `sh`), so it carries no SSH-library / C dependency and fits the same
+/// sync [`CapabilityExecutor`] seam. Non-interactive by construction
+/// (`BatchMode=yes` — never prompt for a password/passphrase; a connection that
+/// would prompt fails fast). Returns the same `output` / `stderr` / `exitcode`
+/// shape as the native executor — an `ssh` connection failure surfaces as a
+/// non-zero `exitcode` (255), the same way the native one reports a failed
+/// command, *not* an executor error.
+///
+/// **Trust model:** identical to the native executor — the gate
+/// (`commit_via<ClassRbac>`) runs upstream; this executor assumes its caller
+/// already authorized the action and runs the command verbatim on the target.
+#[derive(Debug, Clone)]
+pub struct SshExecutor {
+    target: String,
+    identity: Option<String>,
+    port: Option<u16>,
+}
+
+impl SshExecutor {
+    /// The single capability this executor implements (same verb as native —
+    /// it is remote execution of the same `ExecuteCommand`).
+    pub const CAPABILITY: &'static str = "ExecuteCommand";
+
+    /// An SSH executor for `target` (`user@host` or `host`).
+    pub fn new(target: impl Into<String>) -> Self {
+        Self {
+            target: target.into(),
+            identity: None,
+            port: None,
+        }
+    }
+
+    /// Use a specific identity file (`ssh -i`).
+    #[must_use]
+    pub fn with_identity(mut self, identity: impl Into<String>) -> Self {
+        self.identity = Some(identity.into());
+        self
+    }
+
+    /// Connect on a non-default port (`ssh -p`).
+    #[must_use]
+    pub fn with_port(mut self, port: u16) -> Self {
+        self.port = Some(port);
+        self
+    }
+
+    /// Build the `ssh` argument vector for `command` — the pure, testable half
+    /// (no spawn). Always non-interactive (`BatchMode=yes`); `--` terminates ssh
+    /// options so the remote command is never re-parsed as a flag.
+    fn build_args(&self, command: &str) -> Vec<String> {
+        let mut args = vec![
+            "-o".to_owned(),
+            "BatchMode=yes".to_owned(),
+            "-o".to_owned(),
+            "StrictHostKeyChecking=accept-new".to_owned(),
+        ];
+        if let Some(port) = self.port {
+            args.push("-p".to_owned());
+            args.push(port.to_string());
+        }
+        if let Some(identity) = &self.identity {
+            args.push("-i".to_owned());
+            args.push(identity.clone());
+        }
+        args.push(self.target.clone());
+        args.push("--".to_owned());
+        args.push(command.to_owned());
+        args
+    }
+}
+
+impl CapabilityExecutor for SshExecutor {
+    fn execute(
+        &self,
+        capability: &str,
+        bound: &[(String, String)],
+    ) -> Result<Vec<(String, String)>, String> {
+        if capability != Self::CAPABILITY {
+            return Err(format!(
+                "ssh executor implements only `{}`, not `{capability}`",
+                Self::CAPABILITY
+            ));
+        }
+        let command = bound
+            .iter()
+            .find(|(k, _)| k == "command")
+            .map(|(_, v)| v.as_str())
+            .ok_or_else(|| "missing `command` parameter".to_owned())?;
+
+        let output = Command::new("ssh")
+            .args(self.build_args(command))
+            .output()
+            .map_err(|e| format!("failed to spawn ssh to `{}`: {e}", self.target))?;
+
+        let trim = |b: &[u8]| String::from_utf8_lossy(b).trim_end().to_owned();
+        Ok(vec![
+            ("output".to_owned(), trim(&output.stdout)),
+            ("stderr".to_owned(), trim(&output.stderr)),
+            (
+                "exitcode".to_owned(),
+                output.status.code().unwrap_or(-1).to_string(),
+            ),
+        ])
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -294,4 +406,43 @@ mod tests {
             result.result
         );
     }
+
+    #[test]
+    fn ssh_builds_a_non_interactive_argv() {
+        // The pure half: the ssh argv is well-formed and non-interactive.
+        let args = SshExecutor::new("ops@node-1").build_args("uptime");
+        // BatchMode=yes is mandatory (never prompt).
+        assert!(args.windows(2).any(|w| w == ["-o", "BatchMode=yes"]));
+        // target, then `--`, then the remote command (never re-parsed as a flag).
+        let dashdash = args.iter().position(|a| a == "--").unwrap();
+        assert_eq!(args[dashdash - 1], "ops@node-1");
+        assert_eq!(args[dashdash + 1], "uptime");
+    }
+
+    #[test]
+    fn ssh_identity_and_port_are_threaded() {
+        let args = SshExecutor::new("host")
+            .with_identity("/keys/id_ed25519")
+            .with_port(2222)
+            .build_args("ls");
+        assert!(args.windows(2).any(|w| w == ["-p", "2222"]));
+        assert!(args.windows(2).any(|w| w == ["-i", "/keys/id_ed25519"]));
+    }
+
+    #[test]
+    fn ssh_missing_command_is_an_error_before_spawn() {
+        // No `command` → error without ever invoking ssh (no host needed).
+        let err = SshExecutor::new("host")
+            .execute("ExecuteCommand", &[])
+            .unwrap_err();
+        assert!(err.contains("command"), "got: {err}");
+    }
+
+    #[test]
+    fn ssh_unknown_capability_is_rejected() {
+        let err = SshExecutor::new("host")
+            .execute("RunScript", &[("command".to_owned(), "x".to_owned())])
+            .unwrap_err();
+        assert!(err.contains("ExecuteCommand"), "got: {err}");
+    }
 }

docs/ARAGO-ACTIONHANDLER-PARITY.md

@@ -190,8 +190,18 @@ downstream. The remaining bricks:
   **native target is SHIPPED**: `ogar-action-handler::NativeCommandExecutor` runs
   `ExecuteCommand` via a local POSIX shell and returns `output`/`stderr`/`exitcode`
   — proven end-to-end by `full_dispatch_runs_a_real_command` ("OGAR running it
-  here," native). SSH / REST / WinRM targets follow the same trait; rs-graph-llm's
-  `graph-flow-action` provides the production executors (and runs `commit_via`).
+  here," native). The **REST target is SHIPPED too**: rs-graph-llm
+  `graph-flow-action-ogar::rest::RestExecutor` (`feature = "rest"`, pure-Rust
+  `ureq`) POSTs the bound params to an HTTP endpoint and returns the response as
+  `resultParameters` — the arago HTTP-callout shape — and runs only behind the
+  gate (`rest_executor_runs_only_behind_the_gate`). The **SSH target** is coded
+  too: `ogar-action-handler::SshExecutor` shells out to the system `ssh`
+  (dep-free, non-interactive `BatchMode=yes`) — arago's canonical
+  `ExecuteCommand`-over-SSH, the native executor made remote; its argv
+  construction + pre-spawn guards are tested, end-to-end exec needs a live host
+  (no sshd in CI). WinRM is the one executor target left. The native + SSH
+  (Command-based, dep-free) executors live in OGAR `ogar-action-handler`; the
+  network ones (REST, library-based) live in rs-graph-llm.
 - **B1-uplink — the hard gate before the executor (SHIPPED).** rs-graph-llm's
   `graph-flow-action-ogar` crate is the seam: `GatedOgarHandler` wraps an OGAR
   `CapabilityExecutor` as a `graph-flow-action::ActionHandler`, so the executor's
@@ -201,10 +211,21 @@ downstream. The remaining bricks:
   unauthorized actor (`Denied`) or a MUL `Block` (`Escalated`) never reaches the
   OGAR executor. Three tests pin it; `NativeCommandExecutor` runs the real command
   only on the authorized path. OGAR owns the executor; rs-graph-llm owns the gate.
-- **B2-transport — the live WebSocket loop.** Wrap `handle_submit` in a
-  `tokio-tungstenite` client (connect with the `token-$TOKEN` subprotocol, JSON-
-  codec the six `action_ws` message types, drive the dispatch, retry-on-no-ack).
-  All the message types, connection path, and auth are now pinned (§2a).
+- **B2-transport — the live daemon (SHIPPED, WebSocket edge).** Built in
+  rs-graph-llm's `graph-flow-action-ogar::daemon` as a **transport-agnostic** core:
+  `Daemon::react` turns one inbound `action-ws` JSON frame into the outbound frames
+  it warrants (`acknowledged` + `sendActionResult`, or `negativeAcknowledged`),
+  running the hard gate (`run_gated`) + the executor in between — pure, no I/O. A
+  `Transport` trait is the swappable edge (`recv`/`send`); `Daemon::serve` is the
+  loop. The **`WsTransport`** WebSocket edge (`feature = "ws"`) connects with the
+  `token-$TOKEN` subprotocol and is proven by a mock-server roundtrip
+  (`ws_roundtrip_against_a_mock_server`: engine `submitAction` → ack → run → result
+  over a real socket). The connection identity is an `Auth` type shaped after OGIT
+  `NTO/Auth/Configuration` (`auth_store` `0x0B01`) — the same principal the
+  transport authenticates as (`accountId`) is the actor the gate authorizes.
+  **HIRO also distributes actions over Kafka**; that edge (`rdkafka` over the same
+  `Transport` trait) is reserved — the core is ready, it needs the topic/record
+  shape pinned.
 - **B2-lift — the instance config lift (SHIPPED for capabilities).** Parse a
   deployed handler's REST registration → the concrete signatures the *schema*
   half cannot supply. `GET /capabilities` is **shipped**: `registration::{RegisteredCapability,
@@ -243,8 +264,11 @@ transport over them.
 | **Reactive dispatch + B1 seam** | ✅ `[G]` SHIPPED | `action_ws::handle_submit` + the `CapabilityExecutor` trait (validate→ack→bind→execute→result; tested with a mock) |
 | **Executor — native target (B1)** | ✅ `[G]` SHIPPED | `ogar-action-handler::NativeCommandExecutor` runs `ExecuteCommand` for real; `full_dispatch_runs_a_real_command` |
 | **Hard gate before executor (B1-uplink)** | ✅ `[G]` SHIPPED | rs-graph-llm `graph-flow-action-ogar::GatedOgarHandler` — `commit_via` (RBAC ∧ guard ∧ MUL) lands before `handle`; `take_result()` is `None` iff the gate refused (3 tests) |
-| **Executor — SSH/REST/WinRM (B1)** | ⛔ `[H]` | further `CapabilityExecutor` impls (rs-graph-llm `graph-flow-action`) |
-| **Live WebSocket transport (B2-transport)** | ⛔ `[H]` | wrap `handle_submit` in a `tokio-tungstenite` loop + JSON codec (all shapes pinned, §2a) |
+| **Executor — REST target (B1)** | ✅ `[G]` SHIPPED | rs-graph-llm `graph-flow-action-ogar::rest::RestExecutor` (`feature = "rest"`, ureq) POSTs bound params → resultParameters; runs only behind the gate (`rest_executor_runs_only_behind_the_gate`) |
+| **Executor — SSH target (B1)** | 🟡 `[G]` code / `[H]` live | `ogar-action-handler::SshExecutor` shells out to system `ssh` (non-interactive `BatchMode=yes`, same `output`/`stderr`/`exitcode` shape as native) — arago's canonical `ExecuteCommand`-over-SSH, dep-free. argv construction + pre-spawn guards tested; end-to-end needs a live host (no sshd in CI) |
+| **Executor — WinRM (B1)** | ⛔ `[H]` | a further `CapabilityExecutor` impl (Windows remote exec) |
+| **Live transport — daemon core + WebSocket (B2-transport)** | ✅ `[G]` SHIPPED | rs-graph-llm `graph-flow-action-ogar::daemon`: transport-agnostic `Daemon::react`/`serve` + `Transport` trait + `WsTransport` (action-ws), gate-driving; mock-server roundtrip. `Auth` ← OGIT `NTO/Auth/Configuration` |
+| **Live transport — Kafka edge (B2-transport)** | ⛔ `[H]` | `rdkafka` over the same `Transport` trait (action topic → result topic); core ready, needs the topic/record shape pinned |
 | **Instance config lift — capabilities (B2-lift)** | ✅ `[G]` SHIPPED | `registration::lift_registration` + `ogar-action-handler::parse_capabilities`: real `GET /capabilities` JSON → `ConcreteCapability` (`ActionParam[]`); `rest_registration_lifts_binds_and_runs` (JSON → lift → bind → run) |
 | **Instance config lift — applicabilities (B2-lift)** | ✅ `[G]` SHIPPED | `registration::lift_applicabilities` + `ogar-action-handler::parse_applicabilities`: real `GET /applicabilities` JSON → per-handler `StateGuard` sets; `rest_applicabilities_lift_to_per_handler_guards`. Residual: inner filter-list field name is alias-flexible pending a live response |
 
@@ -259,13 +283,21 @@ MUL-blocked action never executes (proven structurally — `take_result()` is
 `None`). The **whole instance lift is shipped too** — real `GET /capabilities`
 and `GET /applicabilities` JSON bodies lift to concrete `ActionParam[]` (runs
 end-to-end, `rest_registration_lifts_binds_and_runs`) and per-handler `StateGuard`
-sets (`rest_applicabilities_lift_to_per_handler_guards`). What's left for
-a **live** drop-in replacement of arago's Python daemon: **B2-transport** (the
-WebSocket loop — all shapes/auth pinned) and the **non-native executor targets**
-(SSH/REST). Each is transport/runner glue over existing types — **no missing IR,
-no missing protocol mapping**. That is the honest state: OGAR *is* an ActionHandler
-that runs commands here, reads its own registration, and gates every action; a thin
-WebSocket transport away from connecting to a live HIRO engine.
+sets (`rest_applicabilities_lift_to_per_handler_guards`). And the **live daemon
+runs over a real socket** — `graph-flow-action-ogar::daemon` drives the gated
+dispatch through a `Transport` trait, with the `action-ws` WebSocket edge proven
+by a mock-server roundtrip. Three executor targets run gated — **native** (local
+command), **SSH** (remote command, arago's canonical shape — coded, live-host test
+pending) and **REST** (HTTP callout). The one thing left for a **live** drop-in
+replacement of arago's Python daemon that needs a real input: the **Kafka
+transport edge** (HIRO's internal bus — `rdkafka` over the same `Transport` trait,
+needs the topic/record shape pinned + a broker to test). WinRM is a further
+executor for completeness. Everything is a single edge/runner impl over existing
+types — **no missing IR, no missing protocol mapping**. That is the honest state:
+OGAR *is* an ActionHandler that reads its own registration, gates every action,
+runs commands locally / over SSH / as HTTP callouts, and speaks `action-ws` over a
+live socket; a Kafka consumer away from being arago's Python daemon, on a HIRO
+deployment that distributes over Kafka.
 
 ---
 
@@ -299,6 +331,13 @@ is replaceable; the parity claim is certified, not argued.
 - `rs-graph-llm/graph-flow-action-ogar` — the **uplink**: OGAR's
   `CapabilityExecutor` behind the hard gate (`GatedOgarHandler` / `run_gated`);
   `commit_via` lands before any execution.
+- `rs-graph-llm/graph-flow-action-ogar/src/daemon.rs` — **B2-transport**: the
+  transport-agnostic `Daemon` (`react`/`serve`) + the `Transport` trait +
+  `WsTransport` (action-ws WebSocket edge) + the OGIT-`Auth`-derived identity.
+- `rs-graph-llm/graph-flow-action-ogar/src/rest.rs` — the **REST executor**
+  (`RestExecutor`, `feature = "rest"`): the arago HTTP-callout target, gated.
+- `crates/ogar-action-handler/src/lib.rs` — the **native** (`NativeCommandExecutor`)
+  + **SSH** (`SshExecutor`, shells out to `ssh`) executor targets, dep-free.
 - arago: `github.com/arago/ActionHandlers`,
   `arago/python-hiro-stonebranch-actionhandler`, HIRO 7 Action API tutorial.
 - **HIRO 7 Action API machine-readable specs (the authoritative harvest, §2a):**