Add supervised operator command service boundary

epiphany-core/src/bin/epiphany-operator-command.rs

@@ -0,0 +1,177 @@
+use anyhow::{Context, Result, anyhow, bail};
+use epiphany_core::{
+    OperatorCapability, OperatorCommandPolicy, OperatorCommandServiceConfig,
+    OperatorCommandServiceReadiness, open_host_identity_at, read_operator_command_trust_anchor,
+    serve_operator_command_rudp,
+};
+use std::collections::BTreeMap;
+use std::net::{SocketAddr, UdpSocket};
+use std::path::PathBuf;
+
+fn main() -> Result<()> {
+    let args = Args::parse(std::env::args().skip(1))?;
+    let signer = open_host_identity_at(&args.executor_identity_store)?;
+    let config = OperatorCommandServiceConfig {
+        command_store: args.command_store.clone(),
+        local_verse_store: args.local_verse_store.clone(),
+        resident_self_store: args.resident_self_store.clone(),
+        policy: OperatorCommandPolicy {
+            runtime_id: args.runtime_id.clone(),
+            discord_guild_id: args.guild_id.clone(),
+            allowed_channel_ids: args.channel_ids.clone(),
+            actor_capabilities: args.actor_capabilities.clone(),
+            max_ttl_seconds: args.max_ttl_seconds,
+        },
+        trusted_bifrost_identity: read_operator_command_trust_anchor(&args.bifrost_trust_anchor)?,
+    };
+    if args.status_only {
+        print_readiness(&args, &signer, "configured")?;
+        return Ok(());
+    }
+    let socket = UdpSocket::bind(args.bind).with_context(|| {
+        format!(
+            "binding Epiphany operator command RUDP service at {}",
+            args.bind
+        )
+    })?;
+    print_readiness(&args, &signer, "serving")?;
+    serve_operator_command_rudp(socket, &config, &signer)
+}
+
+fn print_readiness(
+    args: &Args,
+    signer: &epiphany_core::HostIdentitySigner,
+    status: &str,
+) -> Result<()> {
+    println!(
+        "{}",
+        serde_json::to_string(&OperatorCommandServiceReadiness {
+            schema_version: "epiphany.operator_command.service_readiness.v0".into(),
+            status: status.into(),
+            runtime_id: args.runtime_id.clone(),
+            bind: args.bind.to_string(),
+            executor_identity_id: signer.entry().identity_id.clone(),
+            private_state_exposed: false,
+        })?
+    );
+    Ok(())
+}
+
+struct Args {
+    bind: SocketAddr,
+    runtime_id: String,
+    bifrost_trust_anchor: PathBuf,
+    executor_identity_store: PathBuf,
+    command_store: PathBuf,
+    local_verse_store: PathBuf,
+    resident_self_store: PathBuf,
+    guild_id: String,
+    channel_ids: Vec<String>,
+    actor_capabilities: BTreeMap<String, Vec<OperatorCapability>>,
+    max_ttl_seconds: i64,
+    status_only: bool,
+}
+
+impl Args {
+    fn parse(values: impl Iterator<Item = String>) -> Result<Self> {
+        let mut values = values.peekable();
+        let mut fields = BTreeMap::<String, String>::new();
+        let mut channels = Vec::new();
+        let mut actors = BTreeMap::new();
+        let mut status_only = false;
+        while let Some(flag) = values.next() {
+            match flag.as_str() {
+                "--status" => status_only = true,
+                "--channel-id" => channels.push(values.next().context("missing --channel-id")?),
+                "--actor-capabilities" => {
+                    let value = values.next().context("missing --actor-capabilities")?;
+                    let (actor, capabilities) = value.split_once('=').context(
+                        "--actor-capabilities must be actor=status,sleep,wake,directive",
+                    )?;
+                    if actor.trim().is_empty() || actors.contains_key(actor) {
+                        bail!("operator actor capability binding is empty or duplicated");
+                    }
+                    let parsed = capabilities
+                        .split(',')
+                        .map(parse_capability)
+                        .collect::<Result<Vec<_>>>()?;
+                    if parsed.is_empty() {
+                        bail!("operator actor requires at least one capability");
+                    }
+                    actors.insert(actor.to_string(), parsed);
+                }
+                "--bind"
+                | "--runtime-id"
+                | "--bifrost-trust-anchor"
+                | "--executor-identity-store"
+                | "--command-store"
+                | "--local-verse-store"
+                | "--resident-self-store"
+                | "--guild-id"
+                | "--max-ttl-seconds" => {
+                    let value = values
+                        .next()
+                        .ok_or_else(|| anyhow!("missing value for {flag}"))?;
+                    if fields.insert(flag.clone(), value).is_some() {
+                        bail!("duplicate operator service argument {flag}");
+                    }
+                }
+                _ => bail!("unknown operator service argument {flag}"),
+            }
+        }
+        let required = |name: &str| {
+            fields
+                .get(name)
+                .cloned()
+                .ok_or_else(|| anyhow!("missing {name}"))
+        };
+        if channels.is_empty() || actors.is_empty() {
+            bail!("operator service requires channels and actor capabilities");
+        }
+        let max_ttl_seconds = fields
+            .get("--max-ttl-seconds")
+            .map(String::as_str)
+            .unwrap_or("60")
+            .parse()?;
+        if !(1..=60).contains(&max_ttl_seconds) {
+            bail!("operator service max TTL must be 1..=60 seconds");
+        }
+        Ok(Self {
+            bind: required("--bind")?.parse()?,
+            runtime_id: required("--runtime-id")?,
+            bifrost_trust_anchor: required("--bifrost-trust-anchor")?.into(),
+            executor_identity_store: required("--executor-identity-store")?.into(),
+            command_store: required("--command-store")?.into(),
+            local_verse_store: required("--local-verse-store")?.into(),
+            resident_self_store: required("--resident-self-store")?.into(),
+            guild_id: required("--guild-id")?,
+            channel_ids: channels,
+            actor_capabilities: actors,
+            max_ttl_seconds,
+            status_only,
+        })
+    }
+}
+
+fn parse_capability(value: &str) -> Result<OperatorCapability> {
+    match value {
+        "status" => Ok(OperatorCapability::Status),
+        "sleep" => Ok(OperatorCapability::Sleep),
+        "wake" => Ok(OperatorCapability::Wake),
+        "directive" => Ok(OperatorCapability::Directive),
+        _ => bail!("unknown operator capability {value}"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parser_has_no_argv_or_store_selector_escape_hatch() {
+        assert!(Args::parse(["--argv".into(), "whoami".into()].into_iter()).is_err());
+        assert!(
+            Args::parse(["--actor-capabilities".into(), "actor=shell".into()].into_iter()).is_err()
+        );
+    }
+}

epiphany-core/src/lib.rs

@@ -22,6 +22,7 @@ mod imagination_consideration;
 mod memory_graph;
 mod mind_gateway;
 mod operator_command;
+mod operator_command_service;
 mod organ_dependencies;
 mod packaged_release;
 mod persona_feedback_admission;
@@ -823,6 +824,15 @@ pub use operator_command::{
     operator_command_admission_signing_payload, operator_command_admission_signing_purpose,
     operator_command_packet_sha256,
 };
+pub use operator_command_service::{
+    EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_SCHEMA_VERSION,
+    EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_TYPE, EpiphanyOperatorCommandResultReceipt,
+    EpiphanyOperatorCommandWireResult, OperatorCommandServiceConfig,
+    OperatorCommandServiceReadiness, execute_operator_command_admission,
+    operator_command_result_receipt_signing_payload,
+    operator_command_result_receipt_signing_purpose, read_operator_command_trust_anchor,
+    serve_operator_command_rudp,
+};
 pub use organ_dependencies::EPIPHANY_LAUNCH_ORGAN_CONTRACT_SCHEMA_VERSION;
 pub use organ_dependencies::EPIPHANY_ORGAN_DEPENDENCY_SCHEMA_VERSION;
 pub use organ_dependencies::EPIPHANY_STANDING_ORGANS;

epiphany-core/src/operator_command.rs

@@ -33,7 +33,7 @@ pub enum OperatorCapability {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(tag = "kind", rename_all = "snake_case")]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
 pub enum OperatorCommand {
     Status,
     Sleep { reason: String },
@@ -53,6 +53,7 @@ impl OperatorCommand {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct OperatorCommandPacket {
     pub command_id: String,
     pub nonce: String,
@@ -68,7 +69,9 @@ pub struct OperatorCommandPacket {
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct BifrostOperatorCommandAdmission {
+    pub schema_name: String,
     pub schema_version: String,
     pub admission_id: String,
     pub packet: OperatorCommandPacket,
@@ -228,6 +231,7 @@ fn validate_admission(
     anchor: &HostIdentityTrustAnchorEntry,
     policy: &OperatorCommandPolicy,
     now: chrono::DateTime<chrono::Utc>,
+    enforce_time: bool,
 ) -> Result<OperatorCapability> {
     let packet = &admission.packet;
     let issued =
@@ -242,7 +246,8 @@ fn validate_admission(
         }
         _ => true,
     };
-    if admission.schema_version != BIFROST_OPERATOR_COMMAND_ADMISSION_SCHEMA_VERSION
+    if admission.schema_name != BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE
+        || admission.schema_version != BIFROST_OPERATOR_COMMAND_ADMISSION_SCHEMA_VERSION
         || admission.admission_id.trim().is_empty()
         || packet.command_id.trim().is_empty()
         || packet.nonce.trim().is_empty()
@@ -260,8 +265,8 @@ fn validate_admission(
         || !policy
             .allowed_channel_ids
             .contains(&packet.discord_channel_id)
-        || issued > now
-        || expires < now
+        || (enforce_time && issued > now)
+        || (enforce_time && expires < now)
         || expires <= issued
         || (expires - issued).num_seconds() > policy.max_ttl_seconds
         || policy.max_ttl_seconds <= 0
@@ -299,7 +304,8 @@ pub fn admit_and_execute_bifrost_operator_command(
     now: &str,
 ) -> Result<OperatorCommandResult> {
     let now_dt = chrono::DateTime::parse_from_rfc3339(now)?.with_timezone(&chrono::Utc);
-    let capability = validate_admission(admission, trusted_bifrost_identity, policy, now_dt)?;
+    let capability =
+        validate_admission(admission, trusted_bifrost_identity, policy, now_dt, false)?;
     let packet = &admission.packet;
     let mut admitted = LocalAdmittedOperatorCommand {
         schema_version: LOCAL_OPERATOR_COMMAND_ADMISSION_SCHEMA_VERSION.into(),
@@ -340,6 +346,7 @@ pub fn admit_and_execute_bifrost_operator_command(
     if let Some(existing) = cache.get::<OperatorCommandResult>(&result_id)? {
         return Ok(existing);
     }
+    validate_admission(admission, trusted_bifrost_identity, policy, now_dt, true)?;
     if cache
         .get::<LocalAdmittedOperatorCommand>(&packet.command_id)?
         .as_ref()
@@ -672,6 +679,7 @@ mod tests {
             command,
         };
         let mut admission = BifrostOperatorCommandAdmission {
+            schema_name: BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE.into(),
             schema_version: BIFROST_OPERATOR_COMMAND_ADMISSION_SCHEMA_VERSION.into(),
             admission_id: format!("admission-{id}"),
             packet_sha256: operator_command_packet_sha256(&packet)?,

epiphany-core/src/operator_command_service.rs

@@ -0,0 +1,436 @@
+use anyhow::{Context, Result, anyhow, bail};
+use cultnet_rs::{
+    CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding,
+    CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract,
+    decode_cultnet_message_from_slice, encode_cultnet_message_to_vec,
+};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::net::UdpSocket;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use crate::{
+    BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE, BifrostOperatorCommandAdmission, HostIdentitySigner,
+    HostIdentityTrustAnchorEntry, OperatorCommandPolicy, OperatorCommandResult,
+    admit_and_execute_bifrost_operator_command,
+};
+
+pub const EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_TYPE: &str =
+    "epiphany.operator_command.sealed_result";
+pub const EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_SCHEMA_VERSION: &str =
+    "epiphany.operator_command.sealed_result.v0";
+const RESULT_SIGNING_PURPOSE: &str = "epiphany.operator-command.sealed-result.v0";
+const RUDP_CONNECTION_ID: u32 = 0xe91f_0001;
+
+#[derive(Clone, Debug)]
+pub struct OperatorCommandServiceConfig {
+    pub command_store: PathBuf,
+    pub local_verse_store: PathBuf,
+    pub resident_self_store: PathBuf,
+    pub policy: OperatorCommandPolicy,
+    pub trusted_bifrost_identity: HostIdentityTrustAnchorEntry,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct EpiphanyOperatorCommandResultReceipt {
+    pub schema_version: String,
+    pub result: EpiphanyOperatorCommandWireResult,
+    pub result_payload_sha256: String,
+    pub command_id: String,
+    pub packet_sha256: String,
+    pub target_runtime_id: String,
+    pub completed_at: String,
+    pub provider_identity_id: String,
+    pub private_state_exposed: bool,
+    pub executor_signature: Vec<u8>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct EpiphanyOperatorCommandWireResult {
+    pub schema_version: String,
+    pub result_id: String,
+    pub command_id: String,
+    pub packet_sha256: String,
+    pub target_runtime_id: String,
+    pub disposition: String,
+    pub consequence_kind: String,
+    pub consequence_ref: String,
+    pub completed_at: String,
+    pub private_state_exposed: bool,
+    pub operator_status: String,
+    pub state_status: String,
+    pub coordinator_action: String,
+    pub brake_status: String,
+    pub detail: String,
+}
+
+impl From<&OperatorCommandResult> for EpiphanyOperatorCommandWireResult {
+    fn from(value: &OperatorCommandResult) -> Self {
+        Self {
+            schema_version: value.schema_version.clone(),
+            result_id: value.result_id.clone(),
+            command_id: value.command_id.clone(),
+            packet_sha256: value.packet_sha256.clone(),
+            target_runtime_id: value.target_runtime_id.clone(),
+            disposition: value.disposition.clone(),
+            consequence_kind: value.consequence_kind.clone(),
+            consequence_ref: value.consequence_ref.clone(),
+            completed_at: value.completed_at.clone(),
+            private_state_exposed: value.private_state_exposed,
+            operator_status: value.operator_status.clone(),
+            state_status: value.state_status.clone(),
+            coordinator_action: value.coordinator_action.clone(),
+            brake_status: value.brake_status.clone(),
+            detail: value.detail.clone(),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct OperatorCommandServiceReadiness {
+    pub schema_version: String,
+    pub status: String,
+    pub runtime_id: String,
+    pub bind: String,
+    pub executor_identity_id: String,
+    pub private_state_exposed: bool,
+}
+
+pub fn operator_command_result_receipt_signing_purpose() -> &'static str {
+    RESULT_SIGNING_PURPOSE
+}
+
+pub fn operator_command_result_receipt_signing_payload(
+    receipt: &EpiphanyOperatorCommandResultReceipt,
+) -> Result<Vec<u8>> {
+    rmp_serde::to_vec(&(
+        receipt.schema_version.as_str(),
+        &receipt.result,
+        receipt.result_payload_sha256.as_str(),
+        receipt.command_id.as_str(),
+        receipt.packet_sha256.as_str(),
+        receipt.target_runtime_id.as_str(),
+        receipt.completed_at.as_str(),
+        receipt.provider_identity_id.as_str(),
+        receipt.private_state_exposed,
+    ))
+    .map_err(Into::into)
+}
+
+pub fn execute_operator_command_admission(
+    config: &OperatorCommandServiceConfig,
+    signer: &HostIdentitySigner,
+    admission_payload: &[u8],
+    now: &str,
+) -> Result<EpiphanyOperatorCommandResultReceipt> {
+    let admission: BifrostOperatorCommandAdmission = rmp_serde::from_slice(admission_payload)
+        .context("operator service rejected malformed or non-strict Bifrost admission")?;
+    let result = admit_and_execute_bifrost_operator_command(
+        &config.command_store,
+        &config.local_verse_store,
+        &config.resident_self_store,
+        &admission,
+        &config.trusted_bifrost_identity,
+        &config.policy,
+        now,
+    )?;
+    let wire_result = EpiphanyOperatorCommandWireResult::from(&result);
+    let result_payload = rmp_serde::to_vec_named(&wire_result)?;
+    let mut receipt = EpiphanyOperatorCommandResultReceipt {
+        schema_version: EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_SCHEMA_VERSION.into(),
+        result_payload_sha256: format!("sha256-{:x}", Sha256::digest(&result_payload)),
+        command_id: result.command_id.clone(),
+        packet_sha256: result.packet_sha256.clone(),
+        target_runtime_id: result.target_runtime_id.clone(),
+        completed_at: result.completed_at.clone(),
+        provider_identity_id: signer.entry().identity_id.clone(),
+        private_state_exposed: false,
+        executor_signature: Vec::new(),
+        result: wire_result,
+    };
+    receipt.executor_signature = signer
+        .sign(
+            RESULT_SIGNING_PURPOSE,
+            &operator_command_result_receipt_signing_payload(&receipt)?,
+        )?
+        .signature;
+    Ok(receipt)
+}
+
+pub fn serve_operator_command_rudp(
+    socket: UdpSocket,
+    config: &OperatorCommandServiceConfig,
+    signer: &HostIdentitySigner,
+) -> Result<()> {
+    socket.set_read_timeout(Some(Duration::from_millis(100)))?;
+    let mut transport =
+        CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::server(
+            &config.policy.runtime_id,
+            socket,
+            RUDP_CONNECTION_ID,
+        ))?;
+    loop {
+        if let Some(frame) = transport.receive_once()? {
+            let response = process_wire_frame(config, signer, &frame.payload);
+            if let Ok(message) = response {
+                transport.send(
+                    "schema",
+                    encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?,
+                )?;
+            }
+            // Malformed, foreign, unsigned, expired-new, and capability-invalid
+            // admissions receive no signed application reply. A receipt would
+            // falsely turn unauthenticated material into an Epiphany statement.
+            // Authenticated owner precondition failures are instead returned by
+            // the core as sealed `refused` results.
+        }
+        transport.poll_resends()?;
+    }
+}
+
+fn process_wire_frame(
+    config: &OperatorCommandServiceConfig,
+    signer: &HostIdentitySigner,
+    payload: &[u8],
+) -> Result<CultNetMessage> {
+    process_wire_frame_at(config, signer, payload, &chrono::Utc::now().to_rfc3339())
+}
+
+fn process_wire_frame_at(
+    config: &OperatorCommandServiceConfig,
+    signer: &HostIdentitySigner,
+    payload: &[u8],
+    now: &str,
+) -> Result<CultNetMessage> {
+    let message = decode_cultnet_message_from_slice(payload, CultNetWireContract::CultNetSchemaV0)
+        .context("operator service rejected non-CultNet payload")?;
+    let CultNetMessage::DocumentPutRaw { document, .. } = message else {
+        bail!("operator service accepts only one raw typed admission document");
+    };
+    if document.schema_id != BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE
+        || document.payload_encoding != CultNetRawPayloadEncoding::Messagepack
+        || document.source_role.as_deref() != Some("bifrost-operator-admission")
+    {
+        bail!("operator service rejected foreign wire document authority");
+    }
+    let admission: BifrostOperatorCommandAdmission =
+        rmp_serde::from_slice(&document.payload).context("operator wire admission is malformed")?;
+    if document.record_key != admission.admission_id
+        || document.source_runtime_id.as_deref()
+            != Some(admission.source_observer_runtime_id.as_str())
+    {
+        bail!("operator wire envelope substituted admission identity");
+    }
+    let receipt = execute_operator_command_admission(config, signer, &document.payload, now)?;
+    Ok(CultNetMessage::DocumentPutRaw {
+        message_id: format!("operator-result-{}", receipt.command_id),
+        document: CultNetRawDocumentRecord {
+            schema_id: EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_TYPE.into(),
+            record_key: receipt.command_id.clone(),
+            stored_at: receipt.completed_at.clone(),
+            payload_encoding: CultNetRawPayloadEncoding::Messagepack,
+            payload: rmp_serde::to_vec_named(&receipt)?,
+            source_runtime_id: Some(config.policy.runtime_id.clone()),
+            source_agent_id: Some(receipt.provider_identity_id.clone()),
+            source_role: Some("epiphany-operator-command-executor".into()),
+            tags: Some(vec!["cultnet.transport.rudp.v0".into()]),
+        },
+    })
+}
+
+pub fn read_operator_command_trust_anchor(path: &Path) -> Result<HostIdentityTrustAnchorEntry> {
+    rmp_serde::from_slice(&std::fs::read(path)?)
+        .map_err(|error| anyhow!("operator Bifrost trust anchor is malformed: {error}"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{OperatorCapability, OperatorCommand, OperatorCommandPacket};
+    use std::collections::BTreeMap;
+
+    fn fixture(
+        root: &Path,
+    ) -> Result<(
+        OperatorCommandServiceConfig,
+        HostIdentitySigner,
+        HostIdentitySigner,
+    )> {
+        let bifrost = crate::enroll_host_identity_at(&root.join("bifrost.cc"))?;
+        let anchor = crate::export_host_identity_trust_anchor(&bifrost, &root.join("anchor.cc"))?;
+        let executor = crate::enroll_host_identity_at(&root.join("executor.cc"))?;
+        Ok((
+            OperatorCommandServiceConfig {
+                command_store: root.join("commands.cc"),
+                local_verse_store: root.join("verse.cc"),
+                resident_self_store: root.join("resident.cc"),
+                policy: OperatorCommandPolicy {
+                    runtime_id: "epiphany-yggdrasil".into(),
+                    discord_guild_id: "guild".into(),
+                    allowed_channel_ids: vec!["ops".into()],
+                    actor_capabilities: BTreeMap::from([(
+                        "actor".into(),
+                        vec![OperatorCapability::Sleep],
+                    )]),
+                    max_ttl_seconds: 60,
+                },
+                trusted_bifrost_identity: anchor,
+            },
+            executor,
+            bifrost,
+        ))
+    }
+
+    fn admission(signer: &HostIdentitySigner) -> Result<BifrostOperatorCommandAdmission> {
+        let packet = OperatorCommandPacket {
+            command_id: "command-1".into(),
+            nonce: "nonce-1".into(),
+            source_event_id: "event-1".into(),
+            source_actor_id: "actor".into(),
+            discord_guild_id: "guild".into(),
+            discord_channel_id: "ops".into(),
+            discord_message_id: "message-1".into(),
+            target_runtime_id: "epiphany-yggdrasil".into(),
+            issued_at: "2026-07-19T12:00:00Z".into(),
+            expires_at: "2026-07-19T12:01:00Z".into(),
+            command: OperatorCommand::Sleep {
+                reason: "Supervised sleep.".into(),
+            },
+        };
+        let mut value = BifrostOperatorCommandAdmission {
+            schema_name: BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE.into(),
+            schema_version: crate::BIFROST_OPERATOR_COMMAND_ADMISSION_SCHEMA_VERSION.into(),
+            admission_id: "admission-1".into(),
+            packet_sha256: crate::operator_command_packet_sha256(&packet)?,
+            packet,
+            source_observer_id: "voidbot".into(),
+            source_observer_runtime_id: "voidbot-yggdrasil".into(),
+            provider: "bifrost".into(),
+            bifrost_admission_receipt_id: "receipt-1".into(),
+            authority: "exact_operator_command_only".into(),
+            provider_identity_id: signer.entry().identity_id.clone(),
+            provider_signature: Vec::new(),
+        };
+        value.provider_signature = signer
+            .sign(
+                crate::operator_command_admission_signing_purpose(),
+                &crate::operator_command_admission_signing_payload(&value)?,
+            )?
+            .signature;
+        Ok(value)
+    }
+
+    #[test]
+    fn seals_exact_result_and_replays_same_receipt() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let (config, executor, bifrost) = fixture(temp.path())?;
+        let payload = rmp_serde::to_vec_named(&admission(&bifrost)?)?;
+        let first = execute_operator_command_admission(
+            &config,
+            &executor,
+            &payload,
+            "2026-07-19T12:00:01Z",
+        )?;
+        let reopened_executor = crate::open_host_identity_at(&temp.path().join("executor.cc"))?;
+        let replay = execute_operator_command_admission(
+            &config,
+            &reopened_executor,
+            &payload,
+            "2026-07-19T12:02:00Z",
+        )?;
+        assert_eq!(first, replay);
+        assert!(!first.private_state_exposed);
+        crate::verify_host_identity_signature(
+            executor.entry(),
+            RESULT_SIGNING_PURPOSE,
+            &operator_command_result_receipt_signing_payload(&first)?,
+            &crate::HostIdentitySignature {
+                identity_id: first.provider_identity_id.clone(),
+                signature: first.executor_signature.clone(),
+            },
+        )?;
+        Ok(())
+    }
+
+    #[test]
+    fn strict_wire_rejects_tamper_before_command_store_write() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let (config, executor, bifrost) = fixture(temp.path())?;
+        let mut value = serde_json::to_value(admission(&bifrost)?)?;
+        value
+            .as_object_mut()
+            .unwrap()
+            .insert("argv".into(), serde_json::json!(["whoami"]));
+        let payload = rmp_serde::to_vec_named(&value)?;
+        assert!(
+            execute_operator_command_admission(
+                &config,
+                &executor,
+                &payload,
+                "2026-07-19T12:00:01Z"
+            )
+            .is_err()
+        );
+        assert!(!config.command_store.exists());
+        Ok(())
+    }
+
+    #[test]
+    fn cultnet_wire_accepts_only_exact_typed_admission_and_returns_sealed_receipt() -> Result<()> {
+        let temp = tempfile::tempdir()?;
+        let (config, executor, bifrost) = fixture(temp.path())?;
+        let admission = admission(&bifrost)?;
+        let request = CultNetMessage::DocumentPutRaw {
+            message_id: "request-1".into(),
+            document: CultNetRawDocumentRecord {
+                schema_id: BIFROST_OPERATOR_COMMAND_DELIVERY_TYPE.into(),
+                record_key: admission.admission_id.clone(),
+                stored_at: admission.packet.issued_at.clone(),
+                payload_encoding: CultNetRawPayloadEncoding::Messagepack,
+                payload: rmp_serde::to_vec_named(&admission)?,
+                source_runtime_id: Some(admission.source_observer_runtime_id.clone()),
+                source_agent_id: Some(admission.provider_identity_id.clone()),
+                source_role: Some("bifrost-operator-admission".into()),
+                tags: Some(vec!["cultnet.transport.rudp.v0".into()]),
+            },
+        };
+        let encoded =
+            encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)?;
+        let response = process_wire_frame_at(&config, &executor, &encoded, "2026-07-19T12:00:01Z")?;
+        let CultNetMessage::DocumentPutRaw { document, .. } = response else {
+            panic!("non-document response")
+        };
+        assert_eq!(
+            document.schema_id,
+            EPIPHANY_OPERATOR_COMMAND_RESULT_RECEIPT_TYPE
+        );
+        let receipt: EpiphanyOperatorCommandResultReceipt =
+            rmp_serde::from_slice(&document.payload)?;
+        assert_eq!(receipt.provider_identity_id, executor.entry().identity_id);
+        assert_eq!(receipt.command_id, admission.packet.command_id);
+        assert_eq!(receipt.packet_sha256, admission.packet_sha256);
+        assert!(!receipt.private_state_exposed);
+
+        let mut alien = request;
+        let CultNetMessage::DocumentPutRaw { document, .. } = &mut alien else {
+            unreachable!()
+        };
+        document.source_role = Some("persona-feedback".into());
+        let before = std::fs::read(&config.command_store)?;
+        assert!(
+            process_wire_frame_at(
+                &config,
+                &executor,
+                &encode_cultnet_message_to_vec(&alien, CultNetWireContract::CultNetSchemaV0)?,
+                "2026-07-19T12:00:02Z"
+            )
+            .is_err()
+        );
+        assert_eq!(std::fs::read(&config.command_store)?, before);
+        Ok(())
+    }
+}

epiphany-core/src/packaged_release.rs

@@ -97,6 +97,7 @@ pub fn required_packaged_release_binaries(target_triple: &str) -> Vec<(&'static
         ("repository-body", file_name("epiphany-repository-body")),
         ("host-identity", file_name("epiphany-host-identity")),
         ("swarm", file_name("epiphany-swarm")),
+        ("operator-command", file_name("epiphany-operator-command")),
         ("heartbeat", file_name("epiphany-heartbeat-store")),
         (
             "persona-feedback-ingress",
@@ -367,6 +368,7 @@ fn required_release_build_target(role: &str) -> Result<(&'static str, &'static s
         "repository-body" => Ok(("epiphany-core", "epiphany-repository-body")),
         "host-identity" => Ok(("epiphany-core", "epiphany-host-identity")),
         "swarm" => Ok(("epiphany-core", "epiphany-swarm")),
+        "operator-command" => Ok(("epiphany-core", "epiphany-operator-command")),
         "heartbeat" => Ok(("epiphany-core", "epiphany-heartbeat-store")),
         "persona-feedback-ingress" => Ok(("epiphany-core", "epiphany-persona-feedback-ingress")),
         "coordinator" => Ok(("epiphany-core", "epiphany-mvp-coordinator")),
@@ -826,6 +828,7 @@ mod tests {
             .collect::<BTreeSet<_>>();
         for role in [
             "swarm",
+            "operator-command",
             "heartbeat",
             "coordinator",
             "model-runtime",
@@ -837,6 +840,10 @@ mod tests {
             required_release_build_target("swarm").unwrap(),
             ("epiphany-core", "epiphany-swarm")
         );
+        assert_eq!(
+            required_release_build_target("operator-command").unwrap(),
+            ("epiphany-core", "epiphany-operator-command")
+        );
         assert_eq!(
             required_release_build_target("heartbeat").unwrap(),
             ("epiphany-core", "epiphany-heartbeat-store")

notes/epiphany-discord-operator-nerve.md

@@ -2,16 +2,26 @@
 
 ## Objective
 
-Yggdrasil must continue operating while Starfire is packed. The operator must
-be able to inspect and direct Epiphany through Discord without turning ordinary
-conversation into command authority or giving VoidBot access to Epiphany's
-private stores.
+This is Epiphany's ordinary organizational interface, not a remote-control
+fixture for one deployment. A resident swarm continuously models its authorized
+domain, derives bounded direction from the admitted map, and lets Imagination
+suggest improvements and possible features without waiting for an instruction.
+The repository Persona gives the organization a shared conversational crossing:
+members can add attributed feedback, challenge the model, and discuss proposals
+without becoming machine operators merely by speaking.
+
+Explicit operator commands form a second, narrower nerve for named consequences
+such as inspection, sleep, wake, direction pressure, and Mind review. Yggdrasil
+continuing while Starfire is packed is the July 22 acceptance case for that
+product contract, not the reason the contract exists. VoidBot must never gain
+access to Epiphany's private stores.
 
 Persona feedback and operator command are different nerves:
 
 - Persona conversation is `feedback_only` at VoidBot/Bifrost and
   `resident-pressure-only` in Epiphany. It may provoke attention and an
-  Imagination proposal. It cannot adopt Mind state or grant consequence.
+  Imagination proposal, but it is not required to ignite either Modeling or
+  autonomous direction. It cannot adopt Mind state or grant consequence.
 - Operator commands are explicit owner-authenticated slash commands. They cross
   Bifrost as signed, expiring, target-bound typed intent and invoke one named
   Epiphany owner primitive. They are never inferred from prose.