779 lines
31 KiB
Rust
779 lines
31 KiB
Rust
//! Production management target backed by the agent's bounded runtime stores.
|
|
|
|
#![allow(clippy::missing_errors_doc)]
|
|
|
|
use crate::behavior::{BehaviorIngress, BehaviorMode};
|
|
use crate::control_plane::{
|
|
AuditEventView, CONTROL_PROTOCOL_VERSION, ControlContext, ControlError, ControlErrorCode,
|
|
ControlFuture, ControlPayload, ControlRequest, ControlTarget, ConversationChannelView,
|
|
HealthView, Page, PageRequest, PendingApprovalView, RuntimeView, ScheduledJobView,
|
|
SessionMetadataView,
|
|
};
|
|
use crate::conversation::{ConversationChannel, ConversationKey, ConversationStore};
|
|
use crate::observability::{
|
|
CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, MetricsSnapshot,
|
|
Observability, pseudonymous_identifier,
|
|
};
|
|
use crate::policy::{
|
|
ApprovalId, AuthenticatedPrincipal, MemoryPolicyAudit, PolicyFinalOutcome, PolicyGateway,
|
|
PolicyReasonCode,
|
|
};
|
|
use crate::session::{SessionState, SessionStatus};
|
|
use libremetaverse_types::compat::CancellationToken;
|
|
use std::collections::BTreeMap;
|
|
use std::fmt;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::{Arc, Mutex, MutexGuard};
|
|
use std::time::{Instant, SystemTime, UNIX_EPOCH};
|
|
use tokio::sync::mpsc;
|
|
|
|
const ROAMING_JOB_ID: &str = "default-roaming";
|
|
|
|
/// Commands whose ownership remains with the service lifecycle loop.
|
|
pub enum RuntimeControlCommand {
|
|
Pause,
|
|
Resume,
|
|
ForceReconnect,
|
|
OperatorMessage { message: String },
|
|
GracefulShutdown,
|
|
}
|
|
|
|
impl fmt::Debug for RuntimeControlCommand {
|
|
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
match self {
|
|
Self::OperatorMessage { message } => formatter
|
|
.debug_struct("OperatorMessage")
|
|
.field("message_bytes", &message.len())
|
|
.finish(),
|
|
Self::Pause => formatter.write_str("Pause"),
|
|
Self::Resume => formatter.write_str("Resume"),
|
|
Self::ForceReconnect => formatter.write_str("ForceReconnect"),
|
|
Self::GracefulShutdown => formatter.write_str("GracefulShutdown"),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
struct RuntimeState {
|
|
session: SessionStatus,
|
|
behavior: BehaviorMode,
|
|
service_state: &'static str,
|
|
region_id: Option<String>,
|
|
region_handle: Option<u64>,
|
|
region_name: Option<String>,
|
|
position: Option<[f64; 3]>,
|
|
}
|
|
|
|
impl Default for RuntimeState {
|
|
fn default() -> Self {
|
|
Self {
|
|
session: SessionStatus::default(),
|
|
behavior: BehaviorMode::Offline,
|
|
service_state: "starting",
|
|
region_id: None,
|
|
region_handle: None,
|
|
region_name: None,
|
|
position: None,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Real control target shared by integrated and split transports.
|
|
pub struct AgentControlTarget {
|
|
started: Instant,
|
|
state: Mutex<RuntimeState>,
|
|
conversations: Arc<ConversationStore>,
|
|
policy: Arc<PolicyGateway>,
|
|
audit: Arc<MemoryPolicyAudit>,
|
|
observability: Mutex<Option<Arc<Observability>>>,
|
|
build: Mutex<Option<Arc<dyn crate::build::BuildControl>>>,
|
|
vision: Mutex<Option<Arc<dyn crate::vision::VisionControl>>>,
|
|
behavior: BehaviorIngress,
|
|
commands: mpsc::Sender<RuntimeControlCommand>,
|
|
command_capacity: usize,
|
|
jobs: Mutex<BTreeMap<String, bool>>,
|
|
next_operation: AtomicU64,
|
|
}
|
|
|
|
impl fmt::Debug for AgentControlTarget {
|
|
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
formatter
|
|
.debug_struct("AgentControlTarget")
|
|
.field("state", &lock(&self.state))
|
|
.field("command_capacity", &self.command_capacity)
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl AgentControlTarget {
|
|
pub fn new(
|
|
conversations: Arc<ConversationStore>,
|
|
policy: Arc<PolicyGateway>,
|
|
audit: Arc<MemoryPolicyAudit>,
|
|
behavior: BehaviorIngress,
|
|
command_capacity: usize,
|
|
) -> Result<(Arc<Self>, mpsc::Receiver<RuntimeControlCommand>), ControlError> {
|
|
if command_capacity == 0 || command_capacity > 8_192 {
|
|
return Err(control_error(
|
|
ControlErrorCode::InvalidRequest,
|
|
"invalid runtime control queue capacity",
|
|
false,
|
|
));
|
|
}
|
|
let (commands, receiver) = mpsc::channel(command_capacity);
|
|
let mut jobs = BTreeMap::new();
|
|
jobs.insert(ROAMING_JOB_ID.to_owned(), false);
|
|
Ok((
|
|
Arc::new(Self {
|
|
started: Instant::now(),
|
|
state: Mutex::new(RuntimeState::default()),
|
|
conversations,
|
|
policy,
|
|
audit,
|
|
observability: Mutex::new(None),
|
|
build: Mutex::new(None),
|
|
vision: Mutex::new(None),
|
|
behavior,
|
|
commands,
|
|
command_capacity,
|
|
jobs: Mutex::new(jobs),
|
|
next_operation: AtomicU64::new(1),
|
|
}),
|
|
receiver,
|
|
))
|
|
}
|
|
|
|
/// Attaches the unified recorder used by the live service and control API.
|
|
pub fn attach_observability(&self, observability: Arc<Observability>) {
|
|
*lock(&self.observability) = Some(observability);
|
|
}
|
|
|
|
/// Attaches transactional build cancellation to the operator action API.
|
|
pub fn attach_build_control(&self, build: Arc<dyn crate::build::BuildControl>) {
|
|
*lock(&self.build) = Some(build);
|
|
}
|
|
|
|
/// Attaches synthetic viewport cancellation and metadata to operator control.
|
|
pub fn attach_vision_control(&self, vision: Arc<dyn crate::vision::VisionControl>) {
|
|
*lock(&self.vision) = Some(vision);
|
|
}
|
|
|
|
pub fn update_session(&self, session: SessionStatus) {
|
|
let mut state = lock(&self.state);
|
|
state.session = session;
|
|
state.service_state = if session.state == SessionState::ShuttingDown {
|
|
"stopping"
|
|
} else {
|
|
"running"
|
|
};
|
|
}
|
|
|
|
pub fn update_behavior(&self, behavior: BehaviorMode) {
|
|
lock(&self.state).behavior = behavior;
|
|
}
|
|
|
|
pub fn update_region(
|
|
&self,
|
|
region_id: Option<String>,
|
|
region_handle: Option<u64>,
|
|
region_name: Option<String>,
|
|
position: Option<[f64; 3]>,
|
|
) {
|
|
let mut state = lock(&self.state);
|
|
state.region_id = region_id.filter(|value| value.len() <= 64);
|
|
state.region_handle = region_handle.filter(|value| *value != 0);
|
|
state.region_name = region_name.filter(|value| value.len() <= 256);
|
|
state.position =
|
|
position.filter(|value| value.iter().all(|component| component.is_finite()));
|
|
}
|
|
|
|
pub fn mark_stopping(&self) {
|
|
lock(&self.state).service_state = "stopping";
|
|
}
|
|
|
|
fn enqueue(
|
|
&self,
|
|
operation: &'static str,
|
|
command: RuntimeControlCommand,
|
|
) -> Result<ControlPayload, ControlError> {
|
|
self.commands
|
|
.try_send(command)
|
|
.map_err(|error| match error {
|
|
mpsc::error::TrySendError::Full(_) => control_error(
|
|
ControlErrorCode::Backpressure,
|
|
"runtime control queue is full",
|
|
true,
|
|
),
|
|
mpsc::error::TrySendError::Closed(_) => control_error(
|
|
ControlErrorCode::TransportClosed,
|
|
"runtime control queue is closed",
|
|
true,
|
|
),
|
|
})?;
|
|
Ok(ControlPayload::Accepted {
|
|
operation_id: format!(
|
|
"{operation}-{}",
|
|
self.next_operation.fetch_add(1, Ordering::Relaxed)
|
|
),
|
|
})
|
|
}
|
|
|
|
#[allow(clippy::too_many_lines)] // Exhaustive protocol-to-runtime mapping stays auditable.
|
|
fn execute_now(
|
|
&self,
|
|
context: &ControlContext,
|
|
request: ControlRequest,
|
|
) -> Result<ControlPayload, ControlError> {
|
|
match request {
|
|
ControlRequest::Health => {
|
|
let state = lock(&self.state);
|
|
Ok(ControlPayload::Health(HealthView {
|
|
service_state: state.service_state.to_owned(),
|
|
ready: state.session.agent_ready,
|
|
uptime_seconds: self.started.elapsed().as_secs(),
|
|
protocol_version: CONTROL_PROTOCOL_VERSION,
|
|
}))
|
|
}
|
|
ControlRequest::Metrics => {
|
|
let mut metrics = lock(&self.observability)
|
|
.as_ref()
|
|
.map_or_else(MetricsSnapshot::default, |observer| {
|
|
observer.metrics().snapshot()
|
|
});
|
|
let state = lock(&self.state);
|
|
metrics.ready = state.session.agent_ready;
|
|
metrics.active_sessions =
|
|
u64::try_from(self.conversations.list_metadata().len()).unwrap_or(u64::MAX);
|
|
metrics.queue_depth = u64::try_from(
|
|
self.command_capacity
|
|
.saturating_sub(self.commands.capacity()),
|
|
)
|
|
.unwrap_or(u64::MAX);
|
|
metrics.queue_capacity = u64::try_from(self.command_capacity).unwrap_or(u64::MAX);
|
|
Ok(ControlPayload::Metrics(metrics))
|
|
}
|
|
ControlRequest::Runtime => {
|
|
let state = lock(&self.state).clone();
|
|
let usage = self.policy.global_budget_usage();
|
|
let build = lock(&self.build);
|
|
let vision = lock(&self.vision);
|
|
Ok(ControlPayload::Runtime(RuntimeView {
|
|
grid_state: state.session.state.as_str().to_owned(),
|
|
generation: state.session.generation,
|
|
transport_connected: state.session.transport_connected,
|
|
agent_ready: state.session.agent_ready,
|
|
region_id: state.region_id,
|
|
region_handle: state.region_handle,
|
|
region_name: state.region_name,
|
|
position: state.position,
|
|
behavior_mode: behavior_name(state.behavior).to_owned(),
|
|
control_queue_used: self
|
|
.command_capacity
|
|
.saturating_sub(self.commands.capacity()),
|
|
control_queue_capacity: self.command_capacity,
|
|
budget_tool_calls_used: usage.tool_calls,
|
|
budget_movement_millimeters_used: usage.movement_millimeters,
|
|
active_build_transaction: build.as_ref().and_then(|build| build.active_build()),
|
|
build_progress: build.as_ref().and_then(|build| build.build_progress()),
|
|
build_orphan_ids: build
|
|
.as_ref()
|
|
.map_or_else(Vec::new, |build| build.build_orphans()),
|
|
active_visual_capture: vision
|
|
.as_ref()
|
|
.and_then(|vision| vision.active_capture()),
|
|
visual_progress: vision.as_ref().and_then(|vision| vision.capture_progress()),
|
|
visual_image_sha256: vision
|
|
.as_ref()
|
|
.and_then(|vision| vision.last_image_sha256()),
|
|
}))
|
|
}
|
|
ControlRequest::ListSessions { page } => {
|
|
let values = self
|
|
.conversations
|
|
.list_metadata()
|
|
.into_iter()
|
|
.map(|metadata| SessionMetadataView {
|
|
session_id: metadata.session_id.as_str().to_owned(),
|
|
avatar_id: metadata.avatar_id.to_string(),
|
|
channel: match metadata.channel {
|
|
ConversationChannel::PublicChat => "public_chat",
|
|
ConversationChannel::DirectIm => "direct_im",
|
|
}
|
|
.to_owned(),
|
|
created_unix_millis: metadata.created_unix_millis,
|
|
last_active_unix_millis: metadata.last_active_unix_millis,
|
|
turns: metadata.turns,
|
|
bytes: metadata.bytes,
|
|
})
|
|
.collect();
|
|
Ok(ControlPayload::Sessions(page_values(&page, values)))
|
|
}
|
|
ControlRequest::ListScheduledJobs { page } => {
|
|
let values = lock(&self.jobs)
|
|
.iter()
|
|
.map(|(job_id, enabled)| ScheduledJobView {
|
|
job_id: job_id.clone(),
|
|
kind: "bounded_roaming".to_owned(),
|
|
enabled: *enabled,
|
|
next_run_unix_millis: None,
|
|
})
|
|
.collect();
|
|
Ok(ControlPayload::ScheduledJobs(page_values(&page, values)))
|
|
}
|
|
ControlRequest::ListPendingApprovals { page } => {
|
|
let values = self
|
|
.policy
|
|
.pending_approvals(unix_seconds())
|
|
.into_iter()
|
|
.map(|approval| PendingApprovalView {
|
|
approval_id: approval.id.get(),
|
|
tool: approval.tool,
|
|
principal: approval.principal,
|
|
expires_unix_seconds: approval.expires_at,
|
|
movement_millimeters: approval.cost.movement_millimeters,
|
|
inventory_operations: approval.cost.inventory_operations,
|
|
build_prims: approval.cost.build_prims,
|
|
})
|
|
.collect();
|
|
Ok(ControlPayload::PendingApprovals(page_values(&page, values)))
|
|
}
|
|
ControlRequest::ListAuditEvents { page } => {
|
|
let values = self
|
|
.audit
|
|
.snapshot()
|
|
.into_iter()
|
|
.enumerate()
|
|
.map(|(index, record)| AuditEventView {
|
|
sequence: u64::try_from(index).unwrap_or(u64::MAX).saturating_add(1),
|
|
unix_millis: record.recorded_unix_millis,
|
|
principal: pseudonymous_identifier(record.principal.as_str()),
|
|
operation: record.tool.as_str().to_owned(),
|
|
outcome: policy_outcome_name(record.final_outcome).to_owned(),
|
|
authorization_id: record.authorization_id,
|
|
})
|
|
.collect();
|
|
Ok(ControlPayload::AuditEvents(page_values(&page, values)))
|
|
}
|
|
ControlRequest::ListObservabilityEvents { page } => {
|
|
let values = lock(&self.observability)
|
|
.as_ref()
|
|
.map_or_else(Vec::new, |observer| observer.snapshot());
|
|
Ok(ControlPayload::ObservabilityEvents(page_values(
|
|
&page, values,
|
|
)))
|
|
}
|
|
ControlRequest::PauseAutonomy => {
|
|
let response = self.enqueue("pause", RuntimeControlCommand::Pause)?;
|
|
self.behavior.pause();
|
|
let vision = lock(&self.vision).clone();
|
|
if let Some((vision, active)) =
|
|
vision.and_then(|vision| vision.active_capture().map(|active| (vision, active)))
|
|
{
|
|
let _ = vision.cancel_capture(&active);
|
|
}
|
|
Ok(response)
|
|
}
|
|
ControlRequest::ResumeAutonomy => {
|
|
let response = self.enqueue("resume", RuntimeControlCommand::Resume)?;
|
|
self.behavior.resume();
|
|
Ok(response)
|
|
}
|
|
ControlRequest::CancelAction { action_id } => {
|
|
let build_cancelled = lock(&self.build)
|
|
.as_ref()
|
|
.is_some_and(|build| build.cancel_build(&action_id));
|
|
let vision_cancelled = lock(&self.vision)
|
|
.as_ref()
|
|
.is_some_and(|vision| vision.cancel_capture(&action_id));
|
|
let cancelled = self.behavior.cancel_action(&action_id).map_err(|_| {
|
|
control_error(ControlErrorCode::InvalidRequest, "invalid action ID", false)
|
|
})?;
|
|
if !cancelled && !build_cancelled && !vision_cancelled {
|
|
return Err(control_error(
|
|
ControlErrorCode::NotFound,
|
|
"active behavior action not found",
|
|
false,
|
|
));
|
|
}
|
|
Ok(ControlPayload::Completed)
|
|
}
|
|
ControlRequest::DecideApproval {
|
|
approval_id,
|
|
approve,
|
|
} => {
|
|
let id = ApprovalId::from_raw(approval_id).ok_or_else(|| {
|
|
control_error(
|
|
ControlErrorCode::InvalidRequest,
|
|
"invalid approval ID",
|
|
false,
|
|
)
|
|
})?;
|
|
let principal =
|
|
AuthenticatedPrincipal::from_authenticated_control(context.principal.clone())
|
|
.map_err(|_| {
|
|
control_error(
|
|
ControlErrorCode::PermissionDenied,
|
|
"invalid operator principal",
|
|
false,
|
|
)
|
|
})?;
|
|
let reason = if approve {
|
|
self.policy.grant_approval(id, &principal, unix_seconds())
|
|
} else {
|
|
self.policy.deny_approval(id, &principal, unix_seconds())
|
|
}
|
|
.map_err(|_| {
|
|
control_error(
|
|
ControlErrorCode::Internal,
|
|
"approval decision failed",
|
|
false,
|
|
)
|
|
})?;
|
|
match reason {
|
|
PolicyReasonCode::Allowed | PolicyReasonCode::ApprovalNotGranted => {
|
|
Ok(ControlPayload::Completed)
|
|
}
|
|
PolicyReasonCode::ApprovalUnknown => Err(control_error(
|
|
ControlErrorCode::NotFound,
|
|
"approval not found",
|
|
false,
|
|
)),
|
|
PolicyReasonCode::ApprovalExpired | PolicyReasonCode::ApprovalReplayed => {
|
|
Err(control_error(
|
|
ControlErrorCode::Conflict,
|
|
"approval is no longer pending",
|
|
false,
|
|
))
|
|
}
|
|
_ => Err(control_error(
|
|
ControlErrorCode::Conflict,
|
|
"approval decision rejected",
|
|
false,
|
|
)),
|
|
}
|
|
}
|
|
ControlRequest::ForceReconnect => {
|
|
self.enqueue("reconnect", RuntimeControlCommand::ForceReconnect)
|
|
}
|
|
ControlRequest::ExpireConversation { avatar_id, channel } => {
|
|
let avatar_id = libremetaverse_types::UUID::parse(avatar_id).map_err(|_| {
|
|
control_error(ControlErrorCode::InvalidRequest, "invalid avatar ID", false)
|
|
})?;
|
|
let channel = match channel {
|
|
ConversationChannelView::PublicChat => ConversationChannel::PublicChat,
|
|
ConversationChannelView::DirectIm => ConversationChannel::DirectIm,
|
|
};
|
|
let key = ConversationKey::new(avatar_id, channel).map_err(|_| {
|
|
control_error(
|
|
ControlErrorCode::InvalidRequest,
|
|
"invalid conversation key",
|
|
false,
|
|
)
|
|
})?;
|
|
if self.conversations.expire(key) {
|
|
Ok(ControlPayload::Completed)
|
|
} else {
|
|
Err(control_error(
|
|
ControlErrorCode::NotFound,
|
|
"conversation not found",
|
|
false,
|
|
))
|
|
}
|
|
}
|
|
ControlRequest::SetRoamingJob { job_id, enabled } => {
|
|
let mut jobs = lock(&self.jobs);
|
|
let Some(current) = jobs.get_mut(&job_id) else {
|
|
return Err(control_error(
|
|
ControlErrorCode::NotFound,
|
|
"roaming job not found",
|
|
false,
|
|
));
|
|
};
|
|
self.behavior.set_roaming(enabled).map_err(|_| {
|
|
control_error(
|
|
ControlErrorCode::Backpressure,
|
|
"behavior queue unavailable",
|
|
true,
|
|
)
|
|
})?;
|
|
*current = enabled;
|
|
Ok(ControlPayload::Completed)
|
|
}
|
|
ControlRequest::InjectOperatorMessage { message } => self.enqueue(
|
|
"operator-message",
|
|
RuntimeControlCommand::OperatorMessage { message },
|
|
),
|
|
ControlRequest::GracefulShutdown => {
|
|
let response = self.enqueue("shutdown", RuntimeControlCommand::GracefulShutdown)?;
|
|
self.mark_stopping();
|
|
Ok(response)
|
|
}
|
|
ControlRequest::SubscribeEvents { .. } | ControlRequest::CancelRequest { .. } => {
|
|
Err(control_error(
|
|
ControlErrorCode::Internal,
|
|
"transport request reached runtime",
|
|
false,
|
|
))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ControlTarget for AgentControlTarget {
|
|
fn execute(
|
|
&self,
|
|
context: ControlContext,
|
|
request: ControlRequest,
|
|
cancellation: CancellationToken,
|
|
) -> ControlFuture<'_> {
|
|
Box::pin(async move {
|
|
if cancellation.is_cancellation_requested() {
|
|
return Err(control_error(
|
|
ControlErrorCode::Cancelled,
|
|
"control request cancelled",
|
|
false,
|
|
));
|
|
}
|
|
let operation = runtime_request_name(&request);
|
|
let correlation = request_correlation(&request);
|
|
let domain_event = runtime_domain_event(&request).ok().flatten();
|
|
let result = self.execute_now(&context, request);
|
|
if let Some(observability) = lock(&self.observability).as_ref() {
|
|
let (severity, result_code) = match &result {
|
|
Ok(_) => (EventSeverity::Info, "accepted"),
|
|
Err(error) => (
|
|
if error.retryable {
|
|
EventSeverity::Warning
|
|
} else {
|
|
EventSeverity::Error
|
|
},
|
|
control_error_name(error.code),
|
|
),
|
|
};
|
|
let event = EventDraft::new(
|
|
EventFamily::ControlCommand,
|
|
severity,
|
|
"control",
|
|
EventOrigin::Control,
|
|
)
|
|
.and_then(|event| event.correlation(correlation))
|
|
.and_then(|event| event.result_code(result_code))
|
|
.and_then(|event| event.code_field("operation", operation))
|
|
.and_then(|event| {
|
|
event.code_field(
|
|
"role",
|
|
match context.role {
|
|
crate::control_plane::ControlRole::Observer => "observer",
|
|
crate::control_plane::ControlRole::Operator => "operator",
|
|
},
|
|
)
|
|
})
|
|
.and_then(|event| event.redacted("operator_payload"));
|
|
if let Ok(event) = event {
|
|
let _ = observability.record(event);
|
|
}
|
|
if result.is_ok()
|
|
&& let Some(event) = domain_event
|
|
{
|
|
let _ = observability.record(event);
|
|
}
|
|
}
|
|
result
|
|
})
|
|
}
|
|
}
|
|
|
|
fn runtime_domain_event(
|
|
request: &ControlRequest,
|
|
) -> Result<Option<EventDraft>, crate::observability::ObservabilityError> {
|
|
let event = match request {
|
|
ControlRequest::SetRoamingJob { job_id, enabled } => Some(
|
|
EventDraft::new(
|
|
EventFamily::ScheduledJob,
|
|
EventSeverity::Info,
|
|
"scheduler",
|
|
EventOrigin::Control,
|
|
)?
|
|
.correlation(CorrelationIds {
|
|
action_id: Some(job_id.clone()),
|
|
..CorrelationIds::default()
|
|
})?
|
|
.result_code("accepted")?
|
|
.code_field("state", if *enabled { "enabled" } else { "disabled" })?,
|
|
),
|
|
ControlRequest::ExpireConversation { avatar_id, channel } => Some(
|
|
EventDraft::new(
|
|
EventFamily::SessionExpired,
|
|
EventSeverity::Info,
|
|
"conversation",
|
|
EventOrigin::Control,
|
|
)?
|
|
.correlation(CorrelationIds {
|
|
avatar_id: Some(avatar_id.clone()),
|
|
..CorrelationIds::default()
|
|
})?
|
|
.result_code("completed")?
|
|
.reason_code("operator_expired")?
|
|
.code_field(
|
|
"channel",
|
|
match channel {
|
|
ConversationChannelView::PublicChat => "public_chat",
|
|
ConversationChannelView::DirectIm => "direct_im",
|
|
},
|
|
)?,
|
|
),
|
|
ControlRequest::DecideApproval {
|
|
approval_id,
|
|
approve,
|
|
} => Some(
|
|
EventDraft::new(
|
|
EventFamily::ApprovalDecision,
|
|
EventSeverity::Info,
|
|
"policy",
|
|
EventOrigin::Control,
|
|
)?
|
|
.correlation(CorrelationIds {
|
|
action_id: Some(format!("approval-{approval_id}")),
|
|
..CorrelationIds::default()
|
|
})?
|
|
.result_code("completed")?
|
|
.code_field("decision", if *approve { "approved" } else { "denied" })?,
|
|
),
|
|
ControlRequest::GracefulShutdown => Some(
|
|
EventDraft::new(
|
|
EventFamily::Shutdown,
|
|
EventSeverity::Info,
|
|
"service",
|
|
EventOrigin::Control,
|
|
)?
|
|
.result_code("requested")?,
|
|
),
|
|
_ => None,
|
|
};
|
|
Ok(event)
|
|
}
|
|
|
|
fn request_correlation(request: &ControlRequest) -> CorrelationIds {
|
|
let action_id = match request {
|
|
ControlRequest::CancelAction { action_id } => Some(action_id.clone()),
|
|
ControlRequest::DecideApproval { approval_id, .. } => {
|
|
Some(format!("approval-{approval_id}"))
|
|
}
|
|
_ => None,
|
|
};
|
|
CorrelationIds {
|
|
action_id,
|
|
..CorrelationIds::default()
|
|
}
|
|
}
|
|
|
|
const fn runtime_request_name(request: &ControlRequest) -> &'static str {
|
|
match request {
|
|
ControlRequest::Health => "health",
|
|
ControlRequest::Metrics => "metrics",
|
|
ControlRequest::Runtime => "runtime",
|
|
ControlRequest::ListSessions { .. } => "list_sessions",
|
|
ControlRequest::ListScheduledJobs { .. } => "list_scheduled_jobs",
|
|
ControlRequest::ListPendingApprovals { .. } => "list_pending_approvals",
|
|
ControlRequest::ListAuditEvents { .. } => "list_audit_events",
|
|
ControlRequest::ListObservabilityEvents { .. } => "list_observability_events",
|
|
ControlRequest::SubscribeEvents { .. } => "subscribe_events",
|
|
ControlRequest::CancelRequest { .. } => "cancel_request",
|
|
ControlRequest::PauseAutonomy => "pause_autonomy",
|
|
ControlRequest::ResumeAutonomy => "resume_autonomy",
|
|
ControlRequest::CancelAction { .. } => "cancel_action",
|
|
ControlRequest::DecideApproval { approve: true, .. } => "approve_proposal",
|
|
ControlRequest::DecideApproval { approve: false, .. } => "deny_proposal",
|
|
ControlRequest::ForceReconnect => "force_reconnect",
|
|
ControlRequest::ExpireConversation { .. } => "expire_conversation",
|
|
ControlRequest::SetRoamingJob { enabled: true, .. } => "enable_roaming_job",
|
|
ControlRequest::SetRoamingJob { enabled: false, .. } => "disable_roaming_job",
|
|
ControlRequest::InjectOperatorMessage { .. } => "inject_operator_message",
|
|
ControlRequest::GracefulShutdown => "graceful_shutdown",
|
|
}
|
|
}
|
|
|
|
const fn control_error_name(code: ControlErrorCode) -> &'static str {
|
|
match code {
|
|
ControlErrorCode::AuthenticationFailed => "authentication_failed",
|
|
ControlErrorCode::VersionMismatch => "version_mismatch",
|
|
ControlErrorCode::PermissionDenied => "permission_denied",
|
|
ControlErrorCode::InvalidRequest => "invalid_request",
|
|
ControlErrorCode::Replay => "replay",
|
|
ControlErrorCode::NotFound => "not_found",
|
|
ControlErrorCode::Conflict => "conflict",
|
|
ControlErrorCode::Cancelled => "cancelled",
|
|
ControlErrorCode::TimedOut => "timed_out",
|
|
ControlErrorCode::Busy => "busy",
|
|
ControlErrorCode::Backpressure => "backpressure",
|
|
ControlErrorCode::FrameTooLarge => "frame_too_large",
|
|
ControlErrorCode::IdleTimeout => "idle_timeout",
|
|
ControlErrorCode::TransportClosed => "transport_closed",
|
|
ControlErrorCode::Internal => "internal",
|
|
}
|
|
}
|
|
|
|
fn page_values<T>(request: &PageRequest, values: Vec<T>) -> Page<T> {
|
|
let start = usize::try_from(request.cursor.unwrap_or(0))
|
|
.unwrap_or(usize::MAX)
|
|
.min(values.len());
|
|
let end = start
|
|
.saturating_add(usize::from(request.limit))
|
|
.min(values.len());
|
|
let next_cursor = (end < values.len()).then(|| u64::try_from(end).unwrap_or(u64::MAX));
|
|
Page {
|
|
items: values.into_iter().skip(start).take(end - start).collect(),
|
|
next_cursor,
|
|
}
|
|
}
|
|
|
|
const fn behavior_name(mode: BehaviorMode) -> &'static str {
|
|
match mode {
|
|
BehaviorMode::Offline => "offline",
|
|
BehaviorMode::Settling => "settling",
|
|
BehaviorMode::Available => "available",
|
|
BehaviorMode::Engaged => "engaged",
|
|
BehaviorMode::Executing => "executing",
|
|
BehaviorMode::Roaming => "roaming",
|
|
BehaviorMode::Paused => "paused",
|
|
BehaviorMode::Recovering => "recovering",
|
|
}
|
|
}
|
|
|
|
const fn policy_outcome_name(outcome: PolicyFinalOutcome) -> &'static str {
|
|
match outcome {
|
|
PolicyFinalOutcome::Denied => "denied",
|
|
PolicyFinalOutcome::ApprovalRequired => "approval_required",
|
|
PolicyFinalOutcome::ApprovalGranted => "approval_granted",
|
|
PolicyFinalOutcome::Authorized => "authorized",
|
|
PolicyFinalOutcome::Completed => "completed",
|
|
PolicyFinalOutcome::Rejected => "rejected",
|
|
PolicyFinalOutcome::Failed => "failed",
|
|
PolicyFinalOutcome::AmbiguousMutation => "ambiguous_mutation",
|
|
}
|
|
}
|
|
|
|
fn unix_seconds() -> u64 {
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.map_or(0, |duration| duration.as_secs())
|
|
}
|
|
|
|
const fn control_error(
|
|
code: ControlErrorCode,
|
|
message: &'static str,
|
|
retryable: bool,
|
|
) -> ControlError {
|
|
ControlError {
|
|
code,
|
|
message,
|
|
retryable,
|
|
}
|
|
}
|
|
|
|
fn lock<T>(value: &Mutex<T>) -> MutexGuard<'_, T> {
|
|
value
|
|
.lock()
|
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
}
|