Files
MetaCrate/crates/metacrate-grid-agent/src/main.rs
Chili Palmer b56b28043f
Some checks failed
CI / rust-skia (Rust only) (push) Successful in 2m47s
CI / required (push) Failing after 52s
Add grid agent milestone acceptance gate (#135)
2026-08-18 12:43:25 +02:00

1235 lines
48 KiB
Rust

use metacrate_grid_agent::{
AgentService, ConfigLoader, GridEventKind, ObservableEvent, OperatingMode,
};
use std::error::Error;
use std::fmt;
use std::path::PathBuf;
#[cfg(feature = "live-grid")]
use std::sync::Arc;
#[cfg(feature = "live-grid")]
use std::time::{Duration, SystemTime, UNIX_EPOCH};
const MAX_ARGUMENTS: usize = 8;
#[derive(Debug)]
struct CliError(String);
impl fmt::Display for CliError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
impl Error for CliError {}
fn color_disabled() -> bool {
std::env::var_os("NO_COLOR").is_some()
|| std::env::var_os("TERM").is_some_and(|value| value == "dumb")
}
#[derive(Clone, Copy, Default, Eq, PartialEq)]
enum Operation {
#[default]
Serve,
CheckConfig,
RunOnce,
Tui,
TuiClient,
PrintPaths,
Acceptance,
CheckLiveOptIns,
}
#[derive(Default)]
struct Options {
config: Option<PathBuf>,
operation: Operation,
evidence: Option<PathBuf>,
}
fn options() -> Result<Option<Options>, CliError> {
let mut result = Options::default();
let mut arguments = std::env::args_os().skip(1);
let mut count = 0;
while let Some(argument) = arguments.next() {
count += 1;
if count > MAX_ARGUMENTS {
return Err(CliError(format!(
"at most {MAX_ARGUMENTS} command-line arguments are accepted"
)));
}
if argument == "--help" || argument == "-h" {
println!(
"metacrate-grid-agent [--config PATH] [--check-config | --run-once | --tui | --tui-client | --print-paths | --acceptance-evidence PATH | --check-live-opt-ins]\n\
Configuration precedence: defaults < JSON < secret files < environment.\n\
With no --config, the platform default is used when it exists."
);
return Ok(None);
} else if argument == "--version" || argument == "-V" {
println!("metacrate-grid-agent {}", env!("CARGO_PKG_VERSION"));
return Ok(None);
}
if argument == "--check-config" {
set_operation(&mut result, Operation::CheckConfig)?;
} else if argument == "--run-once" {
set_operation(&mut result, Operation::RunOnce)?;
} else if argument == "--tui" {
set_operation(&mut result, Operation::Tui)?;
} else if argument == "--tui-client" {
set_operation(&mut result, Operation::TuiClient)?;
} else if argument == "--print-paths" {
set_operation(&mut result, Operation::PrintPaths)?;
} else if argument == "--acceptance-evidence" {
set_operation(&mut result, Operation::Acceptance)?;
let path = arguments
.next()
.ok_or_else(|| CliError("--acceptance-evidence requires a new path".into()))?;
count += 1;
result.evidence = Some(PathBuf::from(path));
} else if argument == "--check-live-opt-ins" {
set_operation(&mut result, Operation::CheckLiveOptIns)?;
} else if argument == "--config" {
let path = arguments
.next()
.ok_or_else(|| CliError("--config requires a path".into()))?;
count += 1;
result.config = Some(PathBuf::from(path));
} else {
let argument = PathBuf::from(argument);
return Err(CliError(format!(
"unknown argument {}; use --help",
argument.display()
)));
}
}
Ok(Some(result))
}
fn set_operation(options: &mut Options, operation: Operation) -> Result<(), CliError> {
if options.operation != Operation::Serve {
return Err(CliError("operation flags are mutually exclusive".into()));
}
options.operation = operation;
Ok(())
}
#[tokio::main]
#[allow(clippy::too_many_lines)]
async fn main() -> Result<(), Box<dyn Error>> {
let Some(options) = options()? else {
return Ok(());
};
let platform_paths = metacrate_grid_agent::PlatformPaths::discover();
if options.operation == Operation::PrintPaths {
println!("config={}", platform_paths.config_file.display());
println!("data={}", platform_paths.data_directory.display());
return Ok(());
}
if options.operation == Operation::Acceptance {
let evidence = options
.evidence
.ok_or_else(|| CliError("acceptance evidence path is required".into()))?;
let written = metacrate_grid_agent::run_deterministic_acceptance(
evidence,
metacrate_grid_agent::AcceptanceBudgets::default(),
)
.await?;
println!(
"deterministic acceptance passed; evidence={}",
written.display()
);
return Ok(());
}
if options.operation == Operation::CheckLiveOptIns {
let opt_ins =
metacrate_grid_agent::LiveGridOptIns::from_environment(|name| std::env::var(name).ok())
.validate()?;
println!(
"live opt-ins: login={} chat_im={} script={} landmarks={} build={} visual={}",
opt_ins.login,
opt_ins.chat_and_im,
opt_ins.script_delivery,
opt_ins.landmarks_and_roaming,
opt_ins.reversible_build,
opt_ins.visual_capture
);
return Ok(());
}
let mut loader = ConfigLoader::new();
let config_path = options.config.or_else(|| {
platform_paths
.config_file
.is_file()
.then_some(platform_paths.config_file)
});
if let Some(path) = config_path {
loader = loader.with_file(path);
}
let config = loader.load()?;
if options.operation == Operation::CheckConfig {
println!("configuration is valid for {:?} mode", config.mode);
return Ok(());
}
if options.operation == Operation::TuiClient {
let token = config
.control
.operator_token
.as_ref()
.ok_or_else(|| CliError("split TUI requires a configured operator token".into()))?;
let client = metacrate_grid_agent::ReconnectingTcpTransport::new(
config.control.listen,
token.clone(),
config.control.limits,
);
metacrate_grid_agent::tui::run_terminal(std::sync::Arc::new(client), !color_disabled())
.await?;
return Ok(());
}
if config.mode != OperatingMode::OfflineFake {
#[cfg(feature = "live-grid")]
return run_live(
config,
options.operation == Operation::RunOnce,
options.operation == Operation::Tui,
)
.await;
#[cfg(not(feature = "live-grid"))]
return Err(CliError(
"live grid mode requires rebuilding with --features live-grid".into(),
)
.into());
}
if options.operation == Operation::Tui {
return Err(CliError("embedded TUI requires live-grid mode".into()).into());
}
let startup_timeout = config.timeouts.startup;
let mut handle = AgentService::offline(config)?.start()?;
if options.operation == Operation::RunOnce {
tokio::time::timeout(startup_timeout, async {
loop {
match handle.next_event().await {
Some(ObservableEvent::Grid(event))
if event.kind == GridEventKind::BackendReady =>
{
return Ok::<(), CliError>(());
}
Some(_) => {}
None => {
return Err(CliError("service stopped before backend readiness".into()));
}
}
}
})
.await
.map_err(|_| CliError("timed out waiting for offline backend readiness".into()))??;
handle.shutdown().await?;
println!("grid agent completed one offline startup/shutdown cycle");
return Ok(());
}
println!("grid agent started in offline/fake mode; press Ctrl-C to stop");
loop {
tokio::select! {
signal = tokio::signal::ctrl_c() => {
signal?;
break;
}
event = handle.next_event() => {
if event.is_none() {
break;
}
}
}
}
handle.shutdown().await?;
println!("grid agent stopped cleanly");
Ok(())
}
#[cfg(feature = "live-grid")]
#[allow(clippy::too_many_lines)]
async fn run_live(
config: metacrate_grid_agent::AgentConfig,
run_once: bool,
tui: bool,
) -> Result<(), Box<dyn Error>> {
use metacrate_grid_agent::{
AgentControlTarget, BehaviorObservation, ControlEventKind, ControlPlane, ControlTarget,
GridSessionBackend, LibremetaverseClientOwner, OperatingMode, RuntimeControlCommand,
SessionControl, SessionObservation, SessionState, SessionSupervisor, TcpControlConfig,
TcpControlServer,
};
let connection = config.grid.clone().ok_or_else(|| {
CliError("validated live configuration did not contain a grid connection".into())
})?;
let owner = LibremetaverseClientOwner::new()?;
let mut live = start_live_interactions(&config, &owner)?;
let backend = match owner.session_backend_with_agent_services(
connection,
live.interaction.ingress(),
live.perception.clone(),
live.behavior.ingress(),
) {
Ok(backend) => backend,
Err(error) => {
live.interaction.shutdown().await?;
live.behavior.shutdown().await?;
return Err(error.into());
}
};
let erased: Arc<dyn GridSessionBackend> = Arc::new(backend);
let mut handle = SessionSupervisor::new(
erased,
config.reconnect,
config.limits.control_queue,
config.limits.observable_queue,
)?
.start();
live.vision.set_generation(handle.status().generation);
if run_once {
let readiness = tokio::time::timeout(config.timeouts.startup, async {
loop {
match handle.next_observation().await {
Some(event @ SessionObservation::Transition { status, .. })
if status.agent_ready =>
{
live.vision.set_generation(status.generation);
record_session_observation(&live.observability, &event);
return Ok::<(), CliError>(());
}
Some(
event @ SessionObservation::Transition {
status:
status @ metacrate_grid_agent::SessionStatus {
state: SessionState::AuthenticationBlocked,
..
},
..
},
) => {
live.vision.set_generation(status.generation);
record_session_observation(&live.observability, &event);
return Err(CliError(
"grid authentication/configuration requires operator action".into(),
));
}
Some(event) => record_session_observation(&live.observability, &event),
None => {
return Err(CliError(
"session supervisor stopped before readiness".into(),
));
}
}
}
})
.await;
let readiness = match readiness {
Ok(result) => result,
Err(_) => Err(CliError("timed out waiting for full grid readiness".into())),
};
if let Err(error) = readiness {
let session_result = handle.shutdown().await;
let interaction_result = live.interaction.shutdown().await;
let behavior_result = live.behavior.shutdown().await;
session_result?;
interaction_result?;
behavior_result?;
return Err(error.into());
}
let session_result = handle.shutdown().await;
let interaction_result = live.interaction.shutdown().await;
let behavior_result = live.behavior.shutdown().await;
session_result?;
interaction_result?;
behavior_result?;
println!("grid agent completed one supervised login/logout cycle");
return Ok(());
}
let (control_target, mut control_commands) = AgentControlTarget::new(
live.conversations.clone(),
live.policy.clone(),
live.audit.clone(),
live.behavior.ingress(),
config.limits.control_queue,
)?;
control_target.attach_observability(live.observability.clone());
control_target.attach_build_control(live.build_control.clone());
control_target.attach_vision_control(live.vision.clone());
control_target.update_session(handle.status());
let erased_target: Arc<dyn ControlTarget> = control_target.clone();
let (control_plane, integrated_client, control_server) = match config.mode {
OperatingMode::Integrated => {
let (plane, client) = ControlPlane::integrated(erased_target, config.control.limits)?;
(plane, Some(client), None)
}
OperatingMode::SplitService => {
let operator = config.control.operator_token.clone().ok_or_else(|| {
CliError("validated split configuration omitted its operator token".into())
})?;
let plane = ControlPlane::new(
erased_target,
operator,
config.control.observer_token.clone(),
config.control.limits,
)?;
let transport = TcpControlConfig {
listen: config.control.listen,
limits: config.control.limits,
};
let server = if let Some(tls) = &config.control.remote_tls {
eprintln!(
"WARNING: remote control is enabled with TLS; protect operator tokens and certificate keys"
);
TcpControlServer::bind_tls(plane.clone(), transport, tls.server_config()?).await?
} else {
TcpControlServer::bind(plane.clone(), transport).await?
};
println!(
"grid agent control plane listening on {}",
server.local_addr()
);
(plane, None, Some(server))
}
OperatingMode::OfflineFake => {
return Err(CliError("offline mode reached the live control plane".into()).into());
}
};
let _tui_task = if tui {
let client = integrated_client.ok_or_else(|| {
CliError("--tui requires integrated mode; use --tui-client for split mode".into())
})?;
Some(tokio::spawn(metacrate_grid_agent::tui::run_terminal(
Arc::new(client),
!color_disabled(),
)))
} else {
None
};
println!("grid agent session supervisor started; press Ctrl-C to stop");
let mut signal_error = None;
let mut control_failure = None;
let mut active_interactions = 0_usize;
loop {
tokio::select! {
signal = tokio::signal::ctrl_c() => {
if let Err(error) = signal {
signal_error = Some(error);
}
break;
}
event = handle.next_observation() => {
let Some(event) = event else { break; };
record_session_observation(&live.observability, &event);
if let SessionObservation::Transition { status, reason, retry_in } = event {
live.vision.set_generation(status.generation);
control_target.update_session(status);
live.landmark_roaming
.update_pause(|pause| pause.degraded = !status.agent_ready);
control_plane.publish(ControlEventKind::StateChanged {
component: "session".into(),
state: status.state.as_str().into(),
});
println!(
"grid session state={} generation={} transport_connected={} agent_ready={} reason={reason:?} retry_in={retry_in:?}",
status.state.as_str(),
status.generation,
status.transport_connected,
status.agent_ready,
);
}
}
event = live.interaction.next_observation() => {
let Some(event) = event else { break; };
match &event {
metacrate_grid_agent::InteractionObservation::InferenceStarted { .. } => {
active_interactions = active_interactions.saturating_add(1);
}
metacrate_grid_agent::InteractionObservation::InferenceFinished { .. } => {
active_interactions = active_interactions.saturating_sub(1);
}
_ => {}
}
live.landmark_roaming.update_pause(|pause| {
pause.conversation = active_interactions != 0;
});
record_interaction_observation(&live.observability, &event);
println!("grid interaction event={event:?}");
}
event = live.perception_observations.recv() => {
let Some(event) = event else { break; };
record_perception_observation(&live.observability, &event);
println!("grid perception event={event:?}");
}
event = live.behavior.next_observation() => {
let Some(event) = event else { break; };
record_behavior_observation(&live.observability, &event);
if let BehaviorObservation::Transition { to, .. } = &event {
control_target.update_behavior(*to);
control_plane.publish(ControlEventKind::StateChanged {
component: "behavior".into(),
state: format!("{to:?}").to_ascii_lowercase(),
});
}
println!("grid behavior event={event:?}");
}
command = control_commands.recv() => {
let Some(command) = command else { break; };
match command {
RuntimeControlCommand::Pause => {
live.landmark_roaming
.update_pause(|pause| pause.operator = true);
if let Err(error) = handle.control(SessionControl::Pause).await {
control_failure = Some(error.to_string());
break;
}
}
RuntimeControlCommand::Resume => {
if let Err(error) = handle.control(SessionControl::Resume).await {
control_failure = Some(error.to_string());
break;
}
live.landmark_roaming
.update_pause(|pause| pause.operator = false);
}
RuntimeControlCommand::ForceReconnect => {
if let Err(error) = handle.control(SessionControl::ForceReconnect).await {
control_failure = Some(error.to_string());
break;
}
}
RuntimeControlCommand::OperatorMessage { message } => {
control_plane.publish(ControlEventKind::StateChanged {
component: "operator_message".into(),
state: format!("accepted_{}_bytes", message.len()),
});
}
RuntimeControlCommand::GracefulShutdown => break,
}
}
}
}
let shutdown_event = metacrate_grid_agent::EventDraft::new(
metacrate_grid_agent::EventFamily::Shutdown,
metacrate_grid_agent::EventSeverity::Info,
"service",
metacrate_grid_agent::EventOrigin::Service,
)?
.result_code("requested")?;
let _ = live.observability.record(shutdown_event);
control_target.mark_stopping();
live.landmark_roaming.shutdown().await;
let session_result = handle.shutdown().await;
let interaction_result = live.interaction.shutdown().await;
let behavior_result = live.behavior.shutdown().await;
let control_result = if let Some(server) = control_server {
server.shutdown().await
} else {
Ok(())
};
session_result?;
interaction_result?;
behavior_result?;
control_result?;
if let Some(error) = signal_error {
return Err(error.into());
}
if let Some(error) = control_failure {
return Err(CliError(format!("control command failed: {error}")).into());
}
println!("grid agent stopped cleanly");
Ok(())
}
#[cfg(feature = "live-grid")]
struct LiveInteractions {
interaction: metacrate_grid_agent::InteractionHandle,
perception: metacrate_grid_agent::PerceptionIngress,
perception_observations:
tokio::sync::mpsc::Receiver<metacrate_grid_agent::PerceptionObservation>,
behavior: metacrate_grid_agent::BehaviorHandle,
conversations: Arc<metacrate_grid_agent::ConversationStore>,
policy: Arc<metacrate_grid_agent::PolicyGateway>,
audit: Arc<metacrate_grid_agent::MemoryPolicyAudit>,
observability: Arc<metacrate_grid_agent::Observability>,
_landmark_intake: metacrate_grid_agent::LibremetaverseLandmarkIntake,
landmark_roaming: metacrate_grid_agent::LandmarkRoamingHandle,
build_control: Arc<dyn metacrate_grid_agent::BuildControl>,
vision:
Arc<metacrate_grid_agent::VisionService<metacrate_grid_agent::LibremetaverseSceneSource>>,
}
#[cfg(feature = "live-grid")]
#[allow(clippy::too_many_lines)]
fn start_live_interactions(
config: &metacrate_grid_agent::AgentConfig,
owner: &metacrate_grid_agent::LibremetaverseClientOwner,
) -> Result<LiveInteractions, Box<dyn Error>> {
use metacrate_grid_agent::{
AuthorizedBackendRouter, AuthorizedToolBackend, BehaviorBackend, BehaviorController,
BuildLimits, BuildService, BuildToolBackend, ConversationStore, InteractionCoordinator,
LandmarkLimits, LandmarkService, LandmarkToolBackend, LibremetaverseBuildGrid,
LibremetaverseLandmarkGrid, LibremetaverseLandmarkIntake, LibremetaverseSceneSource,
LibremetaverseScriptInventory, LlmClient, LlmTransportLimits, MemoryPolicyAudit,
Observability, ObservabilityLimits, PerceptionBackend, PolicyAuditSink, PolicyGateway,
PolicyLimits, PolicyLlmResponder, ScriptDeliveryBackend, ScriptDeliverySettings,
SystemRoamingRandom, ToolLoopLimits, UnifiedPolicyAudit, VisionAugmentedResponder,
VisionLimits, VisionService, behavior_policy_tools, build_policy_tools,
landmark_policy_tools, perception_policy_tools, script_delivery_policy_tool,
};
let transport_limits = LlmTransportLimits {
request_timeout: config.timeouts.request,
total_timeout: config.interaction.model_timeout,
max_prompt_bytes: config.limits.max_body_bytes,
max_response_bytes: config.limits.max_body_bytes,
max_concurrent_requests: config.interaction.max_concurrent_inference,
..LlmTransportLimits::default()
};
let conversation = Arc::new(ConversationStore::from_config(config)?);
let perception = Arc::new(PerceptionBackend::new(
Arc::new(owner.world_snapshot_source()),
Arc::clone(&conversation),
config.limits.observable_queue,
)?);
let perception_ingress = perception.ingress();
let perception_observations = perception
.take_observations()
.ok_or_else(|| CliError("perception observation receiver already claimed".into()))?;
let behavior = BehaviorController::new(
config.behavior.clone(),
Arc::new(owner.embodiment_sink()),
config.limits.control_queue,
config.limits.observable_queue,
config.timeouts.shutdown,
)?
.start();
let behavior_ingress = behavior.ingress();
let client = Arc::new(LlmClient::new(config.llm.clone(), transport_limits)?);
let audit = Arc::new(MemoryPolicyAudit::new(config.limits.observable_queue)?);
let observability = Observability::memory(ObservabilityLimits {
ring_events: config.limits.observable_queue,
subscriber_queue: config.limits.observable_queue.min(512),
journal_queue: config.limits.observable_queue,
..ObservabilityLimits::default()
})?;
let audit_sink: Arc<dyn PolicyAuditSink> = Arc::new(UnifiedPolicyAudit::new(
audit.clone(),
observability.clone(),
));
let mut tools = perception_policy_tools()?;
tools.extend(behavior_policy_tools(&config.behavior)?);
let script_inventory = Arc::new(LibremetaverseScriptInventory::new(owner.client()));
let script_backend: Arc<dyn AuthorizedToolBackend> = Arc::new(ScriptDeliveryBackend::new(
script_inventory,
ScriptDeliverySettings::default(),
)?);
tools.push(script_delivery_policy_tool(
ScriptDeliverySettings::default(),
)?);
let build_service = Arc::new(BuildService::new(
Arc::new(LibremetaverseBuildGrid::new(owner)),
BuildLimits::default(),
)?);
let build_control: Arc<dyn metacrate_grid_agent::BuildControl> = build_service.clone();
let build_backend: Arc<dyn AuthorizedToolBackend> =
Arc::new(BuildToolBackend::new(build_service));
tools.extend(build_policy_tools(BuildLimits::default())?);
let landmark_service = Arc::new(LandmarkService::new(
Arc::new(LibremetaverseLandmarkGrid::new(owner)),
Arc::new(SystemRoamingRandom::default()),
config.authorized_avatar_uuids.clone(),
LandmarkLimits::default(),
Some(config.storage_path.join("landmarks.json")),
)?);
let landmark_backend: Arc<dyn AuthorizedToolBackend> = Arc::new(
LandmarkToolBackend::new(Arc::clone(&landmark_service))
.with_behavior(behavior_ingress.clone()),
);
let landmark_intake = LibremetaverseLandmarkIntake::start(&landmark_service, owner)?;
let landmark_roaming = landmark_service.start_roaming_runner(
metacrate_grid_agent::RoamingPause {
degraded: true,
..metacrate_grid_agent::RoamingPause::default()
},
Some(behavior_ingress.clone()),
);
tools.extend(landmark_policy_tools(LandmarkLimits::default())?);
let routes = tools
.iter()
.map(|tool| tool.definition.name.as_str().to_owned())
.map(|name| {
let backend: Arc<dyn AuthorizedToolBackend> = if name.starts_with("build_object_") {
Arc::clone(&build_backend)
} else if name == metacrate_grid_agent::SCRIPT_DELIVERY_TOOL {
Arc::clone(&script_backend)
} else if name.starts_with("landmark_") {
Arc::clone(&landmark_backend)
} else if name.starts_with("behavior_") {
Arc::new(BehaviorBackend::new(behavior_ingress.clone()))
} else {
perception.clone()
};
(name, backend)
})
.collect::<Vec<_>>();
let gateway = Arc::new(PolicyGateway::new(
config.authorized_avatar_uuids.clone(),
tools,
PolicyLimits::default(),
audit_sink,
)?);
let loop_limits = ToolLoopLimits {
max_tool_calls_per_turn: config.limits.max_tool_calls,
max_tool_calls_per_session: config.limits.max_tool_calls,
max_history_messages: config.conversation.limits.max_turns_per_session,
max_history_bytes: config.conversation.limits.max_session_bytes,
wall_clock_timeout: config.interaction.model_timeout,
..ToolLoopLimits::default()
};
let now = Arc::new(|| {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |duration| duration.as_secs())
});
let routed_backend: Arc<dyn AuthorizedToolBackend> =
Arc::new(AuthorizedBackendRouter::new(routes)?);
let responder = Arc::new(PolicyLlmResponder::new(
client,
gateway.clone(),
routed_backend,
loop_limits,
now,
)?);
let vision_limits = VisionLimits::default();
let vision = Arc::new(VisionService::new(
Arc::new(LibremetaverseSceneSource::new(owner, vision_limits)),
vision_limits,
)?);
let responder = Arc::new(VisionAugmentedResponder::new(vision.clone(), responder));
let sink = Arc::new(owner.interaction_sink());
let pacer: Arc<dyn metacrate_grid_agent::ResponsePacer> = Arc::new(behavior_ingress);
let interaction = InteractionCoordinator::new(
config.interaction.clone(),
libremetaverse_types::UUID::zero(),
config.authorized_avatar_uuids.clone(),
conversation.clone(),
responder,
sink,
config.limits.grid_event_queue,
config.limits.observable_queue,
config.timeouts.shutdown,
)?
.with_response_pacer(pacer)
.start();
Ok(LiveInteractions {
interaction,
perception: perception_ingress,
perception_observations,
behavior,
conversations: conversation,
policy: gateway,
audit,
observability,
_landmark_intake: landmark_intake,
landmark_roaming,
build_control,
vision,
})
}
#[cfg(feature = "live-grid")]
fn record_session_observation(
observer: &metacrate_grid_agent::Observability,
observation: &metacrate_grid_agent::SessionObservation,
) {
use metacrate_grid_agent::{
EventDraft, EventFamily, EventOrigin, EventSeverity, SessionObservation,
};
let SessionObservation::Transition {
status,
reason,
retry_in,
} = observation
else {
return;
};
observer.metrics().set_ready(status.agent_ready);
if matches!(
reason,
metacrate_grid_agent::SessionReason::TransientTransport
| metacrate_grid_agent::SessionReason::Maintenance
| metacrate_grid_agent::SessionReason::Kicked
| metacrate_grid_agent::SessionReason::SimulatorDisconnected
| metacrate_grid_agent::SessionReason::ServerFailure
| metacrate_grid_agent::SessionReason::OperatorForceReconnect
) {
observer.metrics().record_reconnect();
}
let event = EventDraft::new(
EventFamily::LifecycleTransition,
if status.agent_ready {
EventSeverity::Info
} else {
EventSeverity::Warning
},
"session",
EventOrigin::Grid,
)
.and_then(|event| event.reason_code(session_reason_code(*reason)))
.and_then(|event| {
event
.retry_count(status.consecutive_failures)
.code_field("to", status.state.as_str())
})
.and_then(|event| event.field("generation", serde_json::Value::from(status.generation)))
.and_then(|event| event.field("ready", serde_json::Value::from(status.agent_ready)))
.and_then(|event| {
event.field(
"transport_connected",
serde_json::Value::from(status.transport_connected),
)
})
.and_then(|event| {
event.field(
"retry_millis",
serde_json::Value::from(retry_in.map_or(0, |duration| {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
})),
)
});
if let Ok(event) = event {
let _ = observer.record(event);
}
}
#[cfg(feature = "live-grid")]
#[allow(clippy::too_many_lines)]
fn record_interaction_observation(
observer: &metacrate_grid_agent::Observability,
observation: &metacrate_grid_agent::InteractionObservation,
) {
use metacrate_grid_agent::{
CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, InferenceOutcome,
InteractionObservation, MemoryReason, OutcomeCode,
};
let event = match observation {
InteractionObservation::SessionMemory { event } => {
let (family, state) = match event.reason {
MemoryReason::SessionCreated => (EventFamily::SessionCreated, "created"),
MemoryReason::PublicExpired
| MemoryReason::DirectImExpired
| MemoryReason::SessionLimitEviction
| MemoryReason::TotalByteEviction
| MemoryReason::OperatorDeleted
| MemoryReason::OperatorExpired => (EventFamily::SessionExpired, "expired"),
MemoryReason::TurnCompacted
| MemoryReason::ByteCompacted
| MemoryReason::ToolResultCompacted => {
(EventFamily::LifecycleTransition, "compacted")
}
MemoryReason::CorruptSnapshotQuarantined
| MemoryReason::UnsupportedSnapshotQuarantined
| MemoryReason::SnapshotTooLargeQuarantined
| MemoryReason::PermissionsNotVerified => {
(EventFamily::LifecycleTransition, "quarantined")
}
};
EventDraft::new(
family,
EventSeverity::Info,
"conversation",
EventOrigin::Service,
)
.and_then(|draft| {
draft.correlation(CorrelationIds {
avatar_id: event.avatar_id.map(|id| id.to_string()),
session_id: event.session_id.as_ref().map(|id| id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.and_then(|draft| draft.reason_code(memory_reason_code(event.reason)))
.and_then(|draft| draft.code_field("state", state))
}
InteractionObservation::Suppressed {
delivery_id,
reason,
} => EventDraft::new(
EventFamily::InboundMessage,
EventSeverity::Info,
"interaction",
EventOrigin::Grid,
)
.and_then(|event| {
event.correlation(CorrelationIds {
request_id: Some(delivery_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.and_then(|event| event.result_code("suppressed"))
.and_then(|event| event.reason_code(suppression_reason_code(*reason)))
.and_then(|event| event.redacted("message_content")),
InteractionObservation::IntentRouted {
delivery_id,
session_id,
origin,
intent,
} => EventDraft::new(
EventFamily::InboundMessage,
EventSeverity::Info,
"interaction",
EventOrigin::Grid,
)
.and_then(|event| {
event.correlation(CorrelationIds {
session_id: Some(session_id.as_str().to_owned()),
request_id: Some(delivery_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.and_then(|event| event.result_code("routed"))
.and_then(|event| event.code_field("origin", interaction_origin_code(*origin)))
.and_then(|event| event.code_field("intent", interaction_intent_code(*intent)))
.and_then(|event| event.redacted("message_content")),
InteractionObservation::AttentionRequested {
delivery_id,
avatar_id,
} => EventDraft::new(
EventFamily::ModelActionSummary,
EventSeverity::Info,
"behavior",
EventOrigin::Service,
)
.and_then(|event| {
event.correlation(CorrelationIds {
avatar_id: Some(avatar_id.to_string()),
request_id: Some(delivery_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.and_then(|event| event.code_field("summary", "face_speaker")),
InteractionObservation::InferenceStarted {
delivery_id,
session_id,
prompt_messages,
} => EventDraft::new(
EventFamily::InferenceRequest,
EventSeverity::Info,
"llm",
EventOrigin::Service,
)
.and_then(|event| {
event.correlation(CorrelationIds {
session_id: Some(session_id.as_str().to_owned()),
request_id: Some(delivery_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.and_then(|event| event.field("prompt_messages", serde_json::Value::from(*prompt_messages)))
.and_then(|event| event.redacted("prompt_content")),
InteractionObservation::InferenceFinished {
delivery_id,
session_id,
outcome,
duration_millis,
} => {
observer
.metrics()
.record_inference_latency(Duration::from_millis(*duration_millis));
let (severity, result, metric) = match outcome {
InferenceOutcome::Completed => {
(EventSeverity::Info, "completed", OutcomeCode::Completed)
}
InferenceOutcome::Cancelled => {
(EventSeverity::Warning, "cancelled", OutcomeCode::Cancelled)
}
InferenceOutcome::TimedOut => {
(EventSeverity::Warning, "timed_out", OutcomeCode::Failed)
}
InferenceOutcome::Failed => (EventSeverity::Error, "failed", OutcomeCode::Failed),
InferenceOutcome::PolicyRejected => (
EventSeverity::Warning,
"policy_rejected",
OutcomeCode::Denied,
),
};
observer.metrics().record_inference_outcome(metric);
EventDraft::new(
EventFamily::InferenceResult,
severity,
"llm",
EventOrigin::Model,
)
.and_then(|event| {
event.correlation(CorrelationIds {
session_id: Some(session_id.as_str().to_owned()),
request_id: Some(delivery_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.map(|event| event.duration_millis(*duration_millis))
.and_then(|event| event.result_code(result))
.and_then(|event| event.redacted("response_content"))
}
InteractionObservation::Delivery {
delivery_id,
session_id,
channel,
outcome,
delivered_parts,
} => EventDraft::new(
EventFamily::OutboundMessage,
EventSeverity::Info,
"interaction",
EventOrigin::Service,
)
.and_then(|event| {
event.correlation(CorrelationIds {
session_id: Some(session_id.as_str().to_owned()),
request_id: Some(delivery_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.and_then(|event| event.result_code(delivery_outcome_code(*outcome)))
.and_then(|event| event.code_field("channel", interaction_channel_code(*channel)))
.and_then(|event| event.field("delivered_parts", serde_json::Value::from(*delivered_parts)))
.and_then(|event| event.redacted("message_content")),
};
if let Ok(event) = event {
let _ = observer.record(event);
}
}
#[cfg(feature = "live-grid")]
fn record_perception_observation(
observer: &metacrate_grid_agent::Observability,
observation: &metacrate_grid_agent::PerceptionObservation,
) {
use metacrate_grid_agent::{
CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity, PerceptionOutcome,
};
observer
.metrics()
.record_tool_latency(Duration::from_millis(observation.duration_millis));
let result = match observation.outcome {
PerceptionOutcome::Completed => "completed",
PerceptionOutcome::Rejected => "rejected",
};
let event = EventDraft::new(
EventFamily::ToolResult,
EventSeverity::Info,
"perception",
EventOrigin::Service,
)
.and_then(|event| {
event.correlation(CorrelationIds {
action_id: Some(observation.call_id.as_str().to_owned()),
..CorrelationIds::default()
})
})
.map(|event| event.duration_millis(observation.duration_millis))
.and_then(|event| event.result_code(result))
.and_then(|event| event.identifier_field("tool", observation.tool.as_str()))
.and_then(|event| {
event.field(
"result_bytes",
serde_json::Value::from(observation.result_bytes),
)
})
.and_then(|event| event.field("cache_hit", serde_json::Value::from(observation.cache_hit)))
.and_then(|event| event.redacted("tool_result"));
if let Ok(event) = event {
let _ = observer.record(event);
}
}
#[cfg(feature = "live-grid")]
fn record_behavior_observation(
observer: &metacrate_grid_agent::Observability,
observation: &metacrate_grid_agent::BehaviorObservation,
) {
use metacrate_grid_agent::{
BehaviorObservation, CorrelationIds, EventDraft, EventFamily, EventOrigin, EventSeverity,
};
let event = match observation {
BehaviorObservation::Transition { from, to, .. } => EventDraft::new(
EventFamily::BehaviorTransition,
EventSeverity::Info,
"behavior",
EventOrigin::Service,
)
.and_then(|event| event.code_field("from", behavior_mode_code(*from)))
.and_then(|event| event.code_field("to", behavior_mode_code(*to))),
BehaviorObservation::Action {
action_id,
generation,
action,
duration_millis,
outcome,
..
} => EventDraft::new(
EventFamily::ToolResult,
EventSeverity::Info,
"behavior",
EventOrigin::Service,
)
.and_then(|event| {
event.correlation(CorrelationIds {
action_id: Some(action_id.clone()),
..CorrelationIds::default()
})
})
.map(|event| event.duration_millis(*duration_millis))
.and_then(|event| event.result_code(behavior_outcome_code(*outcome)))
.and_then(|event| event.identifier_field("tool", action))
.and_then(|event| {
event.field(
"generation",
serde_json::Value::from(generation.unwrap_or(0)),
)
}),
};
if let Ok(event) = event {
let _ = observer.record(event);
}
}
#[cfg(feature = "live-grid")]
const fn session_reason_code(reason: metacrate_grid_agent::SessionReason) -> &'static str {
use metacrate_grid_agent::SessionReason;
match reason {
SessionReason::Startup => "startup",
SessionReason::LoginSucceeded => "login_succeeded",
SessionReason::AgentReady => "agent_ready",
SessionReason::ReadinessLost => "readiness_lost",
SessionReason::TransientTransport => "transient_transport",
SessionReason::Maintenance => "maintenance",
SessionReason::Kicked => "kicked",
SessionReason::SimulatorDisconnected => "simulator_disconnected",
SessionReason::ServerFailure => "server_failure",
SessionReason::InvalidCredentials => "invalid_credentials",
SessionReason::InvalidConfiguration => "invalid_configuration",
SessionReason::StableSessionReset => "stable_session_reset",
SessionReason::OperatorPause => "operator_pause",
SessionReason::OperatorResume => "operator_resume",
SessionReason::OperatorForceReconnect => "operator_force_reconnect",
SessionReason::OperatorLogout => "operator_logout",
SessionReason::ShutdownRequested => "shutdown_requested",
SessionReason::ShutdownComplete => "shutdown_complete",
}
}
#[cfg(feature = "live-grid")]
const fn memory_reason_code(reason: metacrate_grid_agent::MemoryReason) -> &'static str {
use metacrate_grid_agent::MemoryReason;
match reason {
MemoryReason::SessionCreated => "session_created",
MemoryReason::PublicExpired => "public_expired",
MemoryReason::DirectImExpired => "direct_im_expired",
MemoryReason::TurnCompacted => "turn_compacted",
MemoryReason::ByteCompacted => "byte_compacted",
MemoryReason::ToolResultCompacted => "tool_result_compacted",
MemoryReason::SessionLimitEviction => "session_limit_eviction",
MemoryReason::TotalByteEviction => "total_byte_eviction",
MemoryReason::OperatorDeleted => "operator_deleted",
MemoryReason::OperatorExpired => "operator_expired",
MemoryReason::CorruptSnapshotQuarantined => "corrupt_snapshot_quarantined",
MemoryReason::UnsupportedSnapshotQuarantined => "unsupported_snapshot_quarantined",
MemoryReason::SnapshotTooLargeQuarantined => "snapshot_too_large_quarantined",
MemoryReason::PermissionsNotVerified => "permissions_not_verified",
}
}
#[cfg(feature = "live-grid")]
const fn suppression_reason_code(reason: metacrate_grid_agent::SuppressionReason) -> &'static str {
use metacrate_grid_agent::SuppressionReason;
match reason {
SuppressionReason::SelfEcho => "self_echo",
SuppressionReason::Duplicate => "duplicate",
SuppressionReason::Muted => "muted",
SuppressionReason::UnsupportedSource => "unsupported_source",
SuppressionReason::UnsupportedDialog => "unsupported_dialog",
SuppressionReason::AmbientPublicChat => "ambient_public_chat",
SuppressionReason::SenderQueueFull => "sender_queue_full",
SuppressionReason::SenderLimit => "sender_limit",
SuppressionReason::Disconnected => "disconnected",
SuppressionReason::UnsafeResponse => "unsafe_response",
SuppressionReason::Cancelled => "cancelled",
}
}
#[cfg(feature = "live-grid")]
const fn interaction_origin_code(origin: metacrate_grid_agent::InteractionOrigin) -> &'static str {
use metacrate_grid_agent::InteractionOrigin;
match origin {
InteractionOrigin::Public => "public",
InteractionOrigin::UnprivilegedIm => "unprivileged_im",
InteractionOrigin::AuthorizedIm => "authorized_im",
}
}
#[cfg(feature = "live-grid")]
const fn interaction_intent_code(intent: metacrate_grid_agent::InteractionIntent) -> &'static str {
use metacrate_grid_agent::InteractionIntent;
match intent {
InteractionIntent::Informational => "informational",
InteractionIntent::PublicCommandDenied => "public_command_denied",
InteractionIntent::PolicyGatedCommand => "policy_gated_command",
InteractionIntent::PolicyGatedLslRequest => "policy_gated_lsl_request",
}
}
#[cfg(feature = "live-grid")]
const fn delivery_outcome_code(outcome: metacrate_grid_agent::DeliveryOutcome) -> &'static str {
use metacrate_grid_agent::DeliveryOutcome;
match outcome {
DeliveryOutcome::Succeeded => "succeeded",
DeliveryOutcome::Failed => "failed",
DeliveryOutcome::TimedOut => "timed_out",
DeliveryOutcome::PolicyDenied => "policy_denied",
}
}
#[cfg(feature = "live-grid")]
const fn interaction_channel_code(
channel: metacrate_grid_agent::InteractionChannel,
) -> &'static str {
use metacrate_grid_agent::InteractionChannel;
match channel {
InteractionChannel::PublicChat => "public_chat",
InteractionChannel::DirectIm => "direct_im",
}
}
#[cfg(feature = "live-grid")]
const fn behavior_mode_code(mode: metacrate_grid_agent::BehaviorMode) -> &'static str {
use metacrate_grid_agent::BehaviorMode;
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",
}
}
#[cfg(feature = "live-grid")]
const fn behavior_outcome_code(outcome: metacrate_grid_agent::BehaviorOutcome) -> &'static str {
use metacrate_grid_agent::BehaviorOutcome;
match outcome {
BehaviorOutcome::Completed => "completed",
BehaviorOutcome::Cancelled => "cancelled",
BehaviorOutcome::TimedOut => "timed_out",
BehaviorOutcome::Stuck => "stuck",
BehaviorOutcome::Rejected => "rejected",
BehaviorOutcome::Preempted => "preempted",
}
}