meerkat_mobkit/
runtime.rs

1//! Runtime subsystem types — routing, delivery, gating, memory, scheduling, and session persistence.
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::fs::{self, OpenOptions};
5use std::io::Write;
6use std::io::{BufRead, BufReader};
7use std::net::{TcpStream, ToSocketAddrs};
8use std::path::{Path, PathBuf};
9use std::process::{Child, Command, Stdio};
10use std::sync::mpsc;
11use std::time::Duration;
12use std::time::{SystemTime, UNIX_EPOCH};
13
14use chrono::{Datelike, Offset, Timelike, Utc};
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17
18use crate::auth::{
19    JwtClaimsValidationConfig, build_jwt_verification_key, inspect_jwt_header, parse_jwks_json,
20    parse_oidc_discovery_json, select_jwk_for_token, validate_jwt_with_verification_key,
21};
22use crate::baseline::{
23    BaselineVerificationError, BaselineVerificationReport, verify_meerkat_baseline_symbols,
24};
25use crate::decisions::{
26    AuthPolicy, AuthProvider, BigQueryNaming, ConsoleAccessRequest, ConsolePolicy,
27    DecisionPolicyError, ReleaseMetadata, RuntimeOpsPolicy, enforce_console_route_access,
28    load_trusted_mobkit_modules_from_toml, parse_release_metadata_json, validate_bigquery_naming,
29    validate_release_metadata, validate_runtime_ops_policy,
30};
31use crate::process::{ProcessBoundaryError, run_process_json_line};
32use crate::protocol::parse_unified_event_line;
33use crate::rpc::{RpcCapabilities, RpcCapabilitiesError, parse_rpc_capabilities};
34use crate::types::{
35    EventEnvelope, MobKitConfig, ModuleConfig, ModuleEvent, PreSpawnData, RestartPolicy,
36    UnifiedEvent,
37};
38
39mod bootstrap;
40mod console_ingress;
41pub mod cross_mob_control;
42pub mod cross_mob_remote;
43mod delivery;
44mod event_transport;
45mod gating;
46mod memory;
47pub mod metadata;
48mod module_boundary;
49mod routing;
50mod rpc;
51mod scheduling;
52mod session_store;
53mod supervisor;
54
55pub use bootstrap::{start_mobkit_runtime, start_mobkit_runtime_with_options};
56pub(crate) use console_ingress::resolve_authorized_console_auth_from_token;
57pub use console_ingress::{
58    ConsoleAgentLiveSnapshot, ConsoleLiveSnapshot, ConsoleMember, ConsoleModelCapabilities,
59    ConsoleRestJsonRequest, ConsoleRestJsonResponse, extract_bearer_token_from_header,
60    handle_console_rest_json_route, handle_console_rest_json_route_with_snapshot,
61    handle_console_rest_json_route_with_snapshot_and_access, validate_console_token,
62};
63pub use event_transport::normalize_event_line;
64pub use metadata::{
65    InMemoryMetadataStore, LabelRpcResult, MetadataScope, MetadataStoreError,
66    PersistentMetadataStore, RuntimeMetadataTable, SqliteMetadataStore, dispatch_labels_delete,
67    dispatch_labels_get, dispatch_labels_set, labels_to_json_value, parse_labels_param,
68    parse_run_id_param,
69};
70pub use routing::WILDCARD_ROUTE;
71pub use routing::route_module_call;
72pub use rpc::{
73    route_module_call_rpc_json, route_module_call_rpc_subprocess,
74    run_rpc_capabilities_boundary_once,
75};
76pub use scheduling::evaluate_schedules_at_tick;
77pub use session_store::{
78    BigQueryGcConfig, BigQuerySessionStoreAdapter, BigQuerySessionStoreError, GcErrorCallback,
79    JsonFileSessionStore, JsonFileSessionStoreError, JsonStoreLockRecord, SessionPersistenceRow,
80    SessionStoreContract, SessionStoreKind, materialize_latest_session_rows,
81    materialize_live_session_rows, run_periodic_gc, run_periodic_gc_with_error_callback,
82    session_store_contracts,
83};
84pub use supervisor::{run_discovered_module_once, run_module_boundary_once};
85
86pub(crate) use scheduling::validate_schedules;
87
88use event_transport::{insert_event_sorted, merge_unified_events};
89use supervisor::supervise_module_start;
90
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub enum NormalizationError {
93    InvalidJson,
94    InvalidSchema,
95    MissingField(&'static str),
96    InvalidFieldType(&'static str),
97    SourceMismatch { expected: &'static str, got: String },
98}
99
100impl std::fmt::Display for NormalizationError {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        match self {
103            Self::InvalidJson => write!(f, "invalid JSON"),
104            Self::InvalidSchema => write!(f, "invalid schema"),
105            Self::MissingField(field) => write!(f, "missing field: {field}"),
106            Self::InvalidFieldType(field) => write!(f, "invalid field type: {field}"),
107            Self::SourceMismatch { expected, got } => {
108                write!(f, "source mismatch: expected {expected}, got {got}")
109            }
110        }
111    }
112}
113
114impl std::error::Error for NormalizationError {}
115
116#[derive(Debug, Clone, PartialEq, Eq)]
117pub enum RuntimeBoundaryError {
118    Process(ProcessBoundaryError),
119    Normalize(NormalizationError),
120    Mcp(McpBoundaryError),
121}
122
123impl std::fmt::Display for RuntimeBoundaryError {
124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        match self {
126            Self::Process(err) => write!(f, "process boundary: {err}"),
127            Self::Normalize(err) => write!(f, "normalization: {err}"),
128            Self::Mcp(err) => write!(f, "MCP boundary: {err}"),
129        }
130    }
131}
132
133impl std::error::Error for RuntimeBoundaryError {
134    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
135        match self {
136            Self::Process(err) => Some(err),
137            Self::Normalize(err) => Some(err),
138            Self::Mcp(err) => Some(err),
139        }
140    }
141}
142
143#[derive(Debug, Clone, PartialEq, Eq)]
144pub enum McpBoundaryError {
145    RuntimeUnavailable(String),
146    McpRequired {
147        module_id: String,
148        flow: String,
149    },
150    Timeout {
151        module_id: String,
152        operation: String,
153        timeout_ms: u64,
154    },
155    ConnectionFailed {
156        module_id: String,
157        reason: String,
158    },
159    ToolListFailed {
160        module_id: String,
161        reason: String,
162    },
163    ToolNotFound {
164        module_id: String,
165        tool: String,
166        available_tools: Vec<String>,
167    },
168    ToolCallFailed {
169        module_id: String,
170        tool: String,
171        reason: String,
172    },
173    CloseFailed {
174        module_id: String,
175        reason: String,
176    },
177    OperationFailedWithCloseFailure {
178        primary: Box<McpBoundaryError>,
179        close: Box<McpBoundaryError>,
180    },
181    InvalidToolPayload {
182        module_id: String,
183        tool: String,
184        reason: String,
185    },
186    InvalidJsonResponse {
187        module_id: String,
188        tool: String,
189        response: String,
190    },
191}
192
193impl std::fmt::Display for McpBoundaryError {
194    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
195        match self {
196            Self::RuntimeUnavailable(msg) => write!(f, "runtime unavailable: {msg}"),
197            Self::McpRequired { module_id, flow } => {
198                write!(f, "MCP required for module {module_id} flow {flow}")
199            }
200            Self::Timeout {
201                module_id,
202                operation,
203                timeout_ms,
204            } => {
205                write!(
206                    f,
207                    "timeout for module {module_id} operation {operation} after {timeout_ms}ms"
208                )
209            }
210            Self::ConnectionFailed { module_id, reason } => {
211                write!(f, "connection failed for module {module_id}: {reason}")
212            }
213            Self::ToolListFailed { module_id, reason } => {
214                write!(f, "tool list failed for module {module_id}: {reason}")
215            }
216            Self::ToolNotFound {
217                module_id,
218                tool,
219                available_tools,
220            } => {
221                write!(
222                    f,
223                    "tool {tool} not found for module {module_id} (available: {})",
224                    available_tools.join(", ")
225                )
226            }
227            Self::ToolCallFailed {
228                module_id,
229                tool,
230                reason,
231            } => {
232                write!(
233                    f,
234                    "tool call {tool} failed for module {module_id}: {reason}"
235                )
236            }
237            Self::CloseFailed { module_id, reason } => {
238                write!(f, "close failed for module {module_id}: {reason}")
239            }
240            Self::OperationFailedWithCloseFailure { primary, close } => {
241                write!(f, "operation failed: {primary}; close also failed: {close}")
242            }
243            Self::InvalidToolPayload {
244                module_id,
245                tool,
246                reason,
247            } => {
248                write!(
249                    f,
250                    "invalid tool payload for module {module_id} tool {tool}: {reason}"
251                )
252            }
253            Self::InvalidJsonResponse {
254                module_id,
255                tool,
256                response,
257            } => {
258                write!(
259                    f,
260                    "invalid JSON response for module {module_id} tool {tool}: {response}"
261                )
262            }
263        }
264    }
265}
266
267impl std::error::Error for McpBoundaryError {}
268
269#[derive(Debug, Clone, PartialEq, Eq)]
270pub enum ConfigResolutionError {
271    ModuleNotConfigured(String),
272    ModuleNotDiscovered(String),
273}
274
275impl std::fmt::Display for ConfigResolutionError {
276    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
277        match self {
278            Self::ModuleNotConfigured(id) => write!(f, "module not configured: {id}"),
279            Self::ModuleNotDiscovered(id) => write!(f, "module not discovered: {id}"),
280        }
281    }
282}
283
284impl std::error::Error for ConfigResolutionError {}
285
286#[derive(Debug, Clone, PartialEq, Eq)]
287pub enum RuntimeFromConfigError {
288    Config(ConfigResolutionError),
289    Runtime(RuntimeBoundaryError),
290}
291
292impl std::fmt::Display for RuntimeFromConfigError {
293    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
294        match self {
295            Self::Config(err) => write!(f, "config resolution: {err}"),
296            Self::Runtime(err) => write!(f, "runtime boundary: {err}"),
297        }
298    }
299}
300
301impl std::error::Error for RuntimeFromConfigError {
302    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
303        match self {
304            Self::Config(err) => Some(err),
305            Self::Runtime(err) => Some(err),
306        }
307    }
308}
309
310#[derive(Debug, Clone, PartialEq, Eq)]
311pub enum RpcRuntimeError {
312    Process(ProcessBoundaryError),
313    Capabilities(RpcCapabilitiesError),
314}
315
316impl std::fmt::Display for RpcRuntimeError {
317    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
318        match self {
319            Self::Process(err) => write!(f, "process boundary: {err}"),
320            Self::Capabilities(err) => write!(f, "capabilities: {err}"),
321        }
322    }
323}
324
325impl std::error::Error for RpcRuntimeError {
326    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
327        match self {
328            Self::Process(err) => Some(err),
329            Self::Capabilities(err) => Some(err),
330        }
331    }
332}
333
334#[derive(Debug, Clone, PartialEq, Eq)]
335pub enum BaselineRuntimeError {
336    Process(ProcessBoundaryError),
337    InvalidRepoPathJson,
338    MissingRepoRoot,
339    InvalidRepoRoot,
340    Baseline(BaselineVerificationError),
341}
342
343impl std::fmt::Display for BaselineRuntimeError {
344    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
345        match self {
346            Self::Process(err) => write!(f, "process boundary: {err}"),
347            Self::InvalidRepoPathJson => write!(f, "invalid repo path JSON"),
348            Self::MissingRepoRoot => write!(f, "missing repo root"),
349            Self::InvalidRepoRoot => write!(f, "invalid repo root"),
350            Self::Baseline(err) => write!(f, "baseline verification: {err}"),
351        }
352    }
353}
354
355impl std::error::Error for BaselineRuntimeError {
356    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
357        match self {
358            Self::Process(err) => Some(err),
359            Self::Baseline(err) => Some(err),
360            _ => None,
361        }
362    }
363}
364
365#[derive(Debug, Clone, PartialEq, Eq)]
366pub enum MobkitRuntimeError {
367    Config(ConfigResolutionError),
368    MemoryBackend(ElephantMemoryStoreError),
369}
370
371impl std::fmt::Display for MobkitRuntimeError {
372    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
373        match self {
374            Self::Config(err) => write!(f, "config resolution: {err}"),
375            Self::MemoryBackend(err) => write!(f, "memory backend: {err}"),
376        }
377    }
378}
379
380impl std::error::Error for MobkitRuntimeError {
381    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
382        match self {
383            Self::Config(err) => Some(err),
384            Self::MemoryBackend(err) => Some(err),
385        }
386    }
387}
388
389#[derive(Debug, Clone, PartialEq, Eq)]
390pub enum DecisionRuntimeError {
391    Policy(DecisionPolicyError),
392}
393
394impl std::fmt::Display for DecisionRuntimeError {
395    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
396        match self {
397            Self::Policy(err) => write!(f, "decision policy: {err}"),
398        }
399    }
400}
401
402impl std::error::Error for DecisionRuntimeError {
403    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
404        match self {
405            Self::Policy(err) => Some(err),
406        }
407    }
408}
409
410#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
411pub struct RuntimeDecisionInputs {
412    pub bigquery: BigQueryNaming,
413    pub trusted_mobkit_toml: String,
414    pub auth: AuthPolicy,
415    pub trusted_oidc: TrustedOidcRuntimeConfig,
416    pub console: ConsolePolicy,
417    pub ops: RuntimeOpsPolicy,
418    pub release_metadata_json: String,
419}
420
421#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
422pub struct RuntimeDecisionState {
423    pub bigquery: BigQueryNaming,
424    pub modules: Vec<ModuleConfig>,
425    pub auth: AuthPolicy,
426    pub trusted_oidc: TrustedOidcRuntimeConfig,
427    pub console: ConsolePolicy,
428    pub ops: RuntimeOpsPolicy,
429    pub release_metadata: ReleaseMetadata,
430}
431
432#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
433pub struct TrustedOidcRuntimeConfig {
434    pub discovery_json: String,
435    pub jwks_json: String,
436    pub audience: String,
437}
438
439#[derive(Debug, Clone, PartialEq, Eq)]
440pub enum ElephantMemoryStoreError {
441    InvalidConfig(String),
442    Io(String),
443    Serialize(String),
444    InvalidStoreData(String),
445    ExternalCallFailed(String),
446}
447
448impl std::fmt::Display for ElephantMemoryStoreError {
449    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
450        match self {
451            Self::InvalidConfig(msg) => write!(f, "invalid config: {msg}"),
452            Self::Io(msg) => write!(f, "I/O error: {msg}"),
453            Self::Serialize(msg) => write!(f, "serialization error: {msg}"),
454            Self::InvalidStoreData(msg) => write!(f, "invalid store data: {msg}"),
455            Self::ExternalCallFailed(msg) => write!(f, "external call failed: {msg}"),
456        }
457    }
458}
459
460impl std::error::Error for ElephantMemoryStoreError {}
461
462#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
463pub struct ElephantMemoryBackendConfig {
464    pub endpoint: String,
465    pub state_path: String,
466}
467
468#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
469#[serde(tag = "kind", rename_all = "snake_case")]
470pub enum MemoryBackendConfig {
471    Elephant(ElephantMemoryBackendConfig),
472}
473
474#[derive(Debug, Clone, PartialEq, Eq)]
475struct ElephantMemoryStoreAdapter {
476    endpoint: String,
477    state_path: PathBuf,
478}
479
480#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
481pub struct RuntimeOptions {
482    pub on_failure_retry_budget: u32,
483    pub always_restart_budget: u32,
484    #[serde(default)]
485    pub supervisor_restart_backoff_ms: u64,
486    #[serde(default)]
487    pub supervisor_test_force_terminate_failure: bool,
488    #[serde(default)]
489    pub memory_backend: Option<MemoryBackendConfig>,
490    #[serde(default = "default_implicit_delegate_idle_retire_secs")]
491    pub implicit_delegate_idle_retire_secs: Option<u64>,
492    #[serde(default = "default_implicit_delegate_idle_sweep_interval_ms")]
493    pub implicit_delegate_idle_sweep_interval_ms: u64,
494}
495
496fn default_implicit_delegate_idle_retire_secs() -> Option<u64> {
497    Some(300)
498}
499
500fn default_implicit_delegate_idle_sweep_interval_ms() -> u64 {
501    10_000
502}
503
504impl Default for RuntimeOptions {
505    fn default() -> Self {
506        Self {
507            on_failure_retry_budget: 1,
508            always_restart_budget: 1,
509            supervisor_restart_backoff_ms: 0,
510            supervisor_test_force_terminate_failure: false,
511            memory_backend: None,
512            implicit_delegate_idle_retire_secs: default_implicit_delegate_idle_retire_secs(),
513            implicit_delegate_idle_sweep_interval_ms:
514                default_implicit_delegate_idle_sweep_interval_ms(),
515        }
516    }
517}
518
519#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
520pub enum LifecycleStage {
521    MobStarted,
522    ModulesStarted,
523    MergedStreamStarted,
524    ShutdownRequested,
525    ShutdownComplete,
526}
527
528#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
529pub struct LifecycleEvent {
530    pub seq: u64,
531    pub stage: LifecycleStage,
532}
533
534#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
535pub enum ModuleHealthState {
536    Starting,
537    Healthy,
538    Failed,
539    Restarting,
540    Stopped,
541}
542
543#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
544pub struct ModuleHealthTransition {
545    pub module_id: String,
546    pub from: Option<ModuleHealthState>,
547    pub to: ModuleHealthState,
548    pub attempt: u32,
549}
550
551#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
552pub struct SupervisorReport {
553    pub transitions: Vec<ModuleHealthTransition>,
554}
555
556#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
557pub struct RuntimeShutdownReport {
558    pub terminated_modules: Vec<String>,
559    pub orphan_processes: u32,
560}
561
562#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
563pub struct ScheduleDefinition {
564    pub schedule_id: String,
565    pub interval: String,
566    pub timezone: String,
567    pub enabled: bool,
568    #[serde(default)]
569    pub jitter_ms: u64,
570    #[serde(default)]
571    pub catch_up: bool,
572}
573
574#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
575pub struct ScheduleTrigger {
576    pub schedule_id: String,
577    pub interval: String,
578    pub timezone: String,
579    pub due_tick_ms: u64,
580}
581
582#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
583pub struct ScheduleEvaluation {
584    pub tick_ms: u64,
585    pub due_triggers: Vec<ScheduleTrigger>,
586}
587
588#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
589pub struct SchedulingSupervisorSignal {
590    pub module_id: String,
591    pub latest_state: ModuleHealthState,
592    pub latest_attempt: u32,
593    pub restart_observed: bool,
594}
595
596#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
597pub struct ScheduleDispatch {
598    pub claim_key: String,
599    pub schedule_id: String,
600    pub interval: String,
601    pub timezone: String,
602    pub due_tick_ms: u64,
603    pub tick_ms: u64,
604    pub event_id: String,
605    pub supervisor_signal: Option<SchedulingSupervisorSignal>,
606    #[serde(default)]
607    pub runtime_injection: Option<ScheduleRuntimeInjection>,
608    #[serde(default)]
609    pub runtime_injection_error: Option<String>,
610}
611
612#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
613pub struct ScheduleRuntimeInjection {
614    pub member_id: String,
615    pub message: String,
616    pub injection_event_id: String,
617}
618
619#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
620pub struct ScheduleDispatchReport {
621    pub tick_ms: u64,
622    pub due_count: usize,
623    pub dispatched: Vec<ScheduleDispatch>,
624    pub skipped_claims: Vec<String>,
625}
626
627#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
628pub struct RoutingResolveRequest {
629    pub recipient: String,
630    #[serde(default)]
631    pub channel: Option<String>,
632    #[serde(default)]
633    pub retry_max: Option<u32>,
634    #[serde(default)]
635    pub backoff_ms: Option<u64>,
636    #[serde(default)]
637    pub rate_limit_per_minute: Option<u32>,
638}
639
640#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
641pub struct RuntimeRoute {
642    pub route_key: String,
643    pub recipient: String,
644    #[serde(default)]
645    pub channel: Option<String>,
646    pub sink: String,
647    pub target_module: String,
648    #[serde(default)]
649    pub retry_max: Option<u32>,
650    #[serde(default)]
651    pub backoff_ms: Option<u64>,
652    #[serde(default)]
653    pub rate_limit_per_minute: Option<u32>,
654}
655
656#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
657pub struct RoutingResolution {
658    pub route_id: String,
659    pub recipient: String,
660    pub channel: String,
661    pub sink: String,
662    pub target_module: String,
663    pub retry_max: u32,
664    pub backoff_ms: u64,
665    pub rate_limit_per_minute: u32,
666}
667
668#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
669pub struct DeliverySendRequest {
670    pub resolution: RoutingResolution,
671    pub payload: Value,
672    #[serde(default)]
673    pub idempotency_key: Option<String>,
674}
675
676#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
677pub struct DeliveryAttempt {
678    pub attempt: u32,
679    pub status: String,
680    pub backoff_ms: u64,
681}
682
683#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
684pub struct DeliveryRecord {
685    pub delivery_id: String,
686    pub route_id: String,
687    pub recipient: String,
688    pub sink: String,
689    pub target_module: String,
690    pub payload: Value,
691    pub status: String,
692    pub attempts: Vec<DeliveryAttempt>,
693    pub first_attempt_ms: u64,
694    pub final_attempt_ms: u64,
695    pub idempotency_key: Option<String>,
696    pub sink_adapter: Option<String>,
697}
698
699#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
700pub struct DeliveryHistoryRequest {
701    #[serde(default)]
702    pub recipient: Option<String>,
703    #[serde(default)]
704    pub sink: Option<String>,
705    #[serde(default = "default_delivery_history_limit")]
706    pub limit: usize,
707}
708
709#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
710pub struct DeliveryHistoryResponse {
711    pub deliveries: Vec<DeliveryRecord>,
712}
713
714#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
715pub struct MemoryStoreInfo {
716    pub store: String,
717    pub record_count: usize,
718}
719
720#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
721pub struct MemoryIndexRequest {
722    pub entity: String,
723    pub topic: String,
724    #[serde(default)]
725    pub store: Option<String>,
726    #[serde(default)]
727    pub fact: Option<String>,
728    #[serde(default)]
729    pub metadata: Option<Value>,
730    #[serde(default)]
731    pub conflict: Option<bool>,
732    #[serde(default)]
733    pub conflict_reason: Option<String>,
734}
735
736#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
737pub struct MemoryAssertion {
738    pub assertion_id: String,
739    pub entity: String,
740    pub topic: String,
741    pub store: String,
742    pub fact: String,
743    #[serde(default)]
744    pub metadata: Option<Value>,
745    pub indexed_at_ms: u64,
746}
747
748#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
749pub struct MemoryConflictSignal {
750    pub entity: String,
751    pub topic: String,
752    pub store: String,
753    #[serde(default)]
754    pub reason: Option<String>,
755    pub updated_at_ms: u64,
756}
757
758#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
759pub struct MemoryIndexResult {
760    pub entity: String,
761    pub topic: String,
762    pub store: String,
763    #[serde(default)]
764    pub assertion_id: Option<String>,
765    pub conflict_active: bool,
766}
767
768#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
769pub struct MemoryQueryRequest {
770    #[serde(default)]
771    pub entity: Option<String>,
772    #[serde(default)]
773    pub topic: Option<String>,
774    #[serde(default)]
775    pub store: Option<String>,
776}
777
778#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
779pub struct MemoryQueryResult {
780    pub assertions: Vec<MemoryAssertion>,
781    pub conflicts: Vec<MemoryConflictSignal>,
782}
783
784#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
785struct PersistedMemoryState {
786    #[serde(default)]
787    assertions: Vec<MemoryAssertion>,
788    #[serde(default)]
789    conflicts: Vec<MemoryConflictSignal>,
790}
791
792#[derive(Debug, Clone, PartialEq, Eq)]
793pub enum MemoryIndexError {
794    EntityRequired,
795    TopicRequired,
796    UnsupportedStore(String),
797    FactRequiredWhenConflictUnset,
798    BackendPersistFailed(ElephantMemoryStoreError),
799}
800
801impl std::fmt::Display for MemoryIndexError {
802    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
803        match self {
804            Self::EntityRequired => write!(f, "entity is required"),
805            Self::TopicRequired => write!(f, "topic is required"),
806            Self::UnsupportedStore(store) => write!(f, "unsupported store: {store}"),
807            Self::FactRequiredWhenConflictUnset => {
808                write!(f, "fact is required when conflict is unset")
809            }
810            Self::BackendPersistFailed(err) => write!(f, "backend persist failed: {err}"),
811        }
812    }
813}
814
815impl std::error::Error for MemoryIndexError {
816    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
817        match self {
818            Self::BackendPersistFailed(err) => Some(err),
819            _ => None,
820        }
821    }
822}
823
824#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
825#[serde(rename_all = "snake_case")]
826pub enum GatingRiskTier {
827    R0,
828    R1,
829    R2,
830    R3,
831}
832
833#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
834pub struct GatingEvaluateRequest {
835    pub action: String,
836    pub actor_id: String,
837    pub risk_tier: GatingRiskTier,
838    #[serde(default)]
839    pub rationale: Option<String>,
840    #[serde(default)]
841    pub requested_approver: Option<String>,
842    #[serde(default)]
843    pub approval_recipient: Option<String>,
844    #[serde(default)]
845    pub approval_channel: Option<String>,
846    #[serde(default)]
847    pub approval_timeout_ms: Option<u64>,
848    #[serde(default)]
849    pub entity: Option<String>,
850    #[serde(default)]
851    pub topic: Option<String>,
852}
853
854#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
855#[serde(rename_all = "snake_case")]
856pub enum GatingOutcome {
857    Allowed,
858    AllowedWithAudit,
859    PendingApproval,
860    SafeDraft,
861}
862
863#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
864pub struct GatingEvaluateResult {
865    pub action_id: String,
866    pub action: String,
867    pub actor_id: String,
868    pub risk_tier: GatingRiskTier,
869    pub outcome: GatingOutcome,
870    #[serde(default)]
871    pub pending_id: Option<String>,
872    #[serde(default)]
873    pub fallback_reason: Option<String>,
874}
875
876#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
877pub struct GatingPendingEntry {
878    pub pending_id: String,
879    pub action_id: String,
880    pub action: String,
881    pub actor_id: String,
882    pub risk_tier: GatingRiskTier,
883    #[serde(default)]
884    pub requested_approver: Option<String>,
885    #[serde(default)]
886    pub approval_recipient: Option<String>,
887    #[serde(default)]
888    pub approval_channel: Option<String>,
889    #[serde(default)]
890    pub approval_route_id: Option<String>,
891    #[serde(default)]
892    pub approval_delivery_id: Option<String>,
893    pub created_at_ms: u64,
894    pub deadline_at_ms: u64,
895}
896
897#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
898#[serde(rename_all = "snake_case")]
899pub enum GatingDecision {
900    Approve,
901    Reject,
902    Escalate,
903}
904
905#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
906pub struct GatingDecideRequest {
907    pub pending_id: String,
908    pub approver_id: String,
909    pub decision: GatingDecision,
910    #[serde(default)]
911    pub reason: Option<String>,
912}
913
914#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
915pub struct GatingDecisionResult {
916    pub pending_id: String,
917    pub action_id: String,
918    pub approver_id: String,
919    pub decision: GatingDecision,
920    pub outcome: GatingOutcome,
921    pub decided_at_ms: u64,
922    #[serde(default)]
923    pub reason: Option<String>,
924    #[serde(default)]
925    pub next_pending_id: Option<String>,
926}
927
928#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
929pub struct GatingAuditEntry {
930    pub audit_id: String,
931    pub timestamp_ms: u64,
932    pub event_type: String,
933    pub action_id: String,
934    #[serde(default)]
935    pub pending_id: Option<String>,
936    pub actor_id: String,
937    pub risk_tier: GatingRiskTier,
938    pub outcome: GatingOutcome,
939    pub detail: Value,
940}
941
942#[derive(Debug, Clone, PartialEq, Eq)]
943pub enum GatingDecideError {
944    UnknownPendingId(String),
945    SelfApprovalForbidden,
946    ApproverMismatch { expected: String, provided: String },
947}
948
949impl std::fmt::Display for GatingDecideError {
950    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
951        match self {
952            Self::UnknownPendingId(id) => write!(f, "unknown pending id: {id}"),
953            Self::SelfApprovalForbidden => write!(f, "self-approval is forbidden"),
954            Self::ApproverMismatch { expected, provided } => {
955                write!(f, "approver mismatch: expected {expected}, got {provided}")
956            }
957        }
958    }
959}
960
961impl std::error::Error for GatingDecideError {}
962
963#[derive(Debug, Clone, PartialEq, Eq)]
964struct DeliveryIdempotencyEntry {
965    delivery_id: String,
966    payload: Value,
967    canonical_resolution: RoutingResolution,
968}
969
970#[derive(Debug, Clone, PartialEq, Eq, Default)]
971struct RouterBoundaryOverrides {
972    channel: Option<String>,
973    sink: Option<String>,
974    target_module: Option<String>,
975    retry_max: Option<u32>,
976    backoff_ms: Option<u64>,
977    rate_limit_per_minute: Option<u32>,
978}
979
980#[derive(Debug, Clone, PartialEq, Eq, Default)]
981struct DeliveryBoundaryOutcome {
982    sink_adapter: Option<String>,
983    force_fail: bool,
984}
985
986#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
987struct DeliveryRateWindowKey {
988    route_id: String,
989    recipient: String,
990    sink: String,
991    window_start_ms: u64,
992}
993
994#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
995struct MemoryConflictKey {
996    entity: String,
997    topic: String,
998    store: String,
999}
1000
1001#[derive(Debug)]
1002pub struct MobkitRuntimeHandle {
1003    config: MobKitConfig,
1004    runtime_options: RuntimeOptions,
1005    loaded_modules: BTreeSet<String>,
1006    live_children: BTreeMap<String, Child>,
1007    pub lifecycle_events: Vec<LifecycleEvent>,
1008    pub supervisor_report: SupervisorReport,
1009    pub merged_events: Vec<EventEnvelope<UnifiedEvent>>,
1010    scheduling_claims: BTreeSet<String>,
1011    scheduling_claim_ticks: BTreeMap<u64, Vec<String>>,
1012    scheduling_last_due_ticks: BTreeMap<String, u64>,
1013    scheduling_dispatch_sequence: u64,
1014    routing_sequence: u64,
1015    routing_resolutions: BTreeMap<String, RoutingResolution>,
1016    routing_resolution_order: Vec<String>,
1017    runtime_routes: BTreeMap<String, RuntimeRoute>,
1018    delivery_sequence: u64,
1019    delivery_runtime_epoch_ms: u64,
1020    delivery_now_floor_ms: u64,
1021    delivery_clock_ms: u64,
1022    delivery_history: Vec<DeliveryRecord>,
1023    delivery_idempotency: BTreeMap<String, DeliveryIdempotencyEntry>,
1024    delivery_idempotency_by_delivery: BTreeMap<String, Vec<String>>,
1025    delivery_rate_window_counts: BTreeMap<DeliveryRateWindowKey, u32>,
1026    gating_sequence: u64,
1027    gating_pending: BTreeMap<String, GatingPendingEntry>,
1028    gating_pending_order: Vec<String>,
1029    gating_audit: Vec<GatingAuditEntry>,
1030    memory_sequence: u64,
1031    memory_assertions: Vec<MemoryAssertion>,
1032    memory_conflicts: BTreeMap<MemoryConflictKey, MemoryConflictSignal>,
1033    memory_backend: Option<ElephantMemoryStoreAdapter>,
1034    running: bool,
1035}
1036
1037impl MobkitRuntimeHandle {
1038    pub fn lifecycle_events(&self) -> &[LifecycleEvent] {
1039        &self.lifecycle_events
1040    }
1041
1042    pub fn supervisor_report(&self) -> &SupervisorReport {
1043        &self.supervisor_report
1044    }
1045
1046    #[doc(hidden)]
1047    pub fn inject_test_events(&mut self, events: Vec<EventEnvelope<UnifiedEvent>>) {
1048        for event in events {
1049            insert_event_sorted(&mut self.merged_events, event);
1050        }
1051    }
1052
1053    fn next_sequence(counter: &mut u64) -> u64 {
1054        let seq = *counter;
1055        *counter = counter.saturating_add(1);
1056        seq
1057    }
1058}
1059
1060#[derive(Debug, Clone, PartialEq, Eq)]
1061pub enum ScheduleValidationError {
1062    EmptyScheduleId,
1063    DuplicateScheduleId(String),
1064    InvalidTickMs(u64),
1065    InvalidInterval {
1066        schedule_id: String,
1067        interval: String,
1068    },
1069    InvalidTimezone {
1070        schedule_id: String,
1071        timezone: String,
1072    },
1073    /// The total cron lookback work for this request exceeded
1074    /// `CRON_LOOKBACK_BUDGET_PER_REQUEST`. Surfaced (rather than silently
1075    /// stalling) when a request packs enough sparse crons to blow past the
1076    /// per-request iteration ceiling — a soft-DoS guard.
1077    LookbackBudgetExceeded {
1078        schedule_id: String,
1079    },
1080}
1081
1082impl std::fmt::Display for ScheduleValidationError {
1083    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1084        match self {
1085            Self::EmptyScheduleId => write!(f, "empty schedule id"),
1086            Self::DuplicateScheduleId(id) => write!(f, "duplicate schedule id: {id}"),
1087            Self::InvalidTickMs(ms) => write!(f, "invalid tick ms: {ms}"),
1088            Self::InvalidInterval {
1089                schedule_id,
1090                interval,
1091            } => {
1092                write!(f, "invalid interval for schedule {schedule_id}: {interval}")
1093            }
1094            Self::InvalidTimezone {
1095                schedule_id,
1096                timezone,
1097            } => {
1098                write!(f, "invalid timezone for schedule {schedule_id}: {timezone}")
1099            }
1100            Self::LookbackBudgetExceeded { schedule_id } => {
1101                write!(
1102                    f,
1103                    "cron lookback budget exceeded while resolving schedule {schedule_id}"
1104                )
1105            }
1106        }
1107    }
1108}
1109
1110impl std::error::Error for ScheduleValidationError {}
1111
1112#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1113pub struct ModuleRouteRequest {
1114    pub module_id: String,
1115    pub method: String,
1116    pub params: Value,
1117}
1118
1119#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1120pub struct ModuleRouteResponse {
1121    pub module_id: String,
1122    pub method: String,
1123    pub payload: Value,
1124}
1125
1126#[derive(Debug, Clone, PartialEq, Eq)]
1127pub enum ModuleRouteError {
1128    UnloadedModule(String),
1129    ModuleRuntime(RuntimeBoundaryError),
1130    UnexpectedRouteResponse,
1131}
1132
1133impl std::fmt::Display for ModuleRouteError {
1134    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1135        match self {
1136            Self::UnloadedModule(id) => write!(f, "unloaded module: {id}"),
1137            Self::ModuleRuntime(err) => write!(f, "module runtime: {err}"),
1138            Self::UnexpectedRouteResponse => write!(f, "unexpected route response"),
1139        }
1140    }
1141}
1142
1143impl std::error::Error for ModuleRouteError {
1144    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1145        match self {
1146            Self::ModuleRuntime(err) => Some(err),
1147            _ => None,
1148        }
1149    }
1150}
1151
1152#[derive(Debug, Clone, PartialEq, Eq)]
1153pub enum RoutingResolveError {
1154    RouterModuleNotLoaded,
1155    DeliveryModuleNotLoaded,
1156    EmptyRecipient,
1157    InvalidChannel,
1158    InvalidRateLimitPerMinute,
1159    RetryMaxExceedsCap { provided: u32, cap: u32 },
1160    RouterBoundary(RuntimeBoundaryError),
1161}
1162
1163impl std::fmt::Display for RoutingResolveError {
1164    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1165        match self {
1166            Self::RouterModuleNotLoaded => write!(f, "router module not loaded"),
1167            Self::DeliveryModuleNotLoaded => write!(f, "delivery module not loaded"),
1168            Self::EmptyRecipient => write!(f, "empty recipient"),
1169            Self::InvalidChannel => write!(f, "invalid channel"),
1170            Self::InvalidRateLimitPerMinute => write!(f, "invalid rate limit per minute"),
1171            Self::RetryMaxExceedsCap { provided, cap } => {
1172                write!(f, "retry max {provided} exceeds cap {cap}")
1173            }
1174            Self::RouterBoundary(err) => write!(f, "router boundary: {err}"),
1175        }
1176    }
1177}
1178
1179impl std::error::Error for RoutingResolveError {
1180    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1181        match self {
1182            Self::RouterBoundary(err) => Some(err),
1183            _ => None,
1184        }
1185    }
1186}
1187
1188#[derive(Debug, Clone, PartialEq, Eq)]
1189pub enum DeliverySendError {
1190    DeliveryModuleNotLoaded,
1191    InvalidRouteTarget(String),
1192    InvalidRouteId,
1193    UnknownRouteId(String),
1194    ForgedResolution,
1195    InvalidRecipient,
1196    InvalidSink,
1197    InvalidIdempotencyKey,
1198    IdempotencyPayloadMismatch,
1199    RateLimited {
1200        sink: String,
1201        window_start_ms: u64,
1202        limit: u32,
1203    },
1204    DeliveryBoundary(RuntimeBoundaryError),
1205}
1206
1207impl std::fmt::Display for DeliverySendError {
1208    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1209        match self {
1210            Self::DeliveryModuleNotLoaded => write!(f, "delivery module not loaded"),
1211            Self::InvalidRouteTarget(target) => write!(f, "invalid route target: {target}"),
1212            Self::InvalidRouteId => write!(f, "invalid route id"),
1213            Self::UnknownRouteId(id) => write!(f, "unknown route id: {id}"),
1214            Self::ForgedResolution => write!(f, "forged resolution"),
1215            Self::InvalidRecipient => write!(f, "invalid recipient"),
1216            Self::InvalidSink => write!(f, "invalid sink"),
1217            Self::InvalidIdempotencyKey => write!(f, "invalid idempotency key"),
1218            Self::IdempotencyPayloadMismatch => write!(f, "idempotency payload mismatch"),
1219            Self::RateLimited {
1220                sink,
1221                window_start_ms,
1222                limit,
1223            } => {
1224                write!(
1225                    f,
1226                    "rate limited on sink {sink} (window {window_start_ms}ms, limit {limit})"
1227                )
1228            }
1229            Self::DeliveryBoundary(err) => write!(f, "delivery boundary: {err}"),
1230        }
1231    }
1232}
1233
1234impl std::error::Error for DeliverySendError {
1235    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1236        match self {
1237            Self::DeliveryBoundary(err) => Some(err),
1238            _ => None,
1239        }
1240    }
1241}
1242
1243#[derive(Debug, Clone, PartialEq, Eq)]
1244pub enum RuntimeRouteMutationError {
1245    EmptyRouteKey,
1246    EmptyRecipient,
1247    InvalidChannel,
1248    EmptySink,
1249    EmptyTargetModule,
1250    InvalidRateLimitPerMinute,
1251    RetryMaxExceedsCap { provided: u32, cap: u32 },
1252    RouteNotFound(String),
1253}
1254
1255impl std::fmt::Display for RuntimeRouteMutationError {
1256    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1257        match self {
1258            Self::EmptyRouteKey => write!(f, "empty route key"),
1259            Self::EmptyRecipient => write!(f, "empty recipient"),
1260            Self::InvalidChannel => write!(f, "invalid channel"),
1261            Self::EmptySink => write!(f, "empty sink"),
1262            Self::EmptyTargetModule => write!(f, "empty target module"),
1263            Self::InvalidRateLimitPerMinute => write!(f, "invalid rate limit per minute"),
1264            Self::RetryMaxExceedsCap { provided, cap } => {
1265                write!(f, "retry max {provided} exceeds cap {cap}")
1266            }
1267            Self::RouteNotFound(key) => write!(f, "route not found: {key}"),
1268        }
1269    }
1270}
1271
1272impl std::error::Error for RuntimeRouteMutationError {}
1273
1274#[derive(Debug, Clone, PartialEq, Eq)]
1275pub enum RpcRouteError {
1276    InvalidRequest,
1277    BoundaryProcess(ProcessBoundaryError),
1278    Route(ModuleRouteError),
1279    InvalidResponse,
1280}
1281
1282impl std::fmt::Display for RpcRouteError {
1283    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1284        match self {
1285            Self::InvalidRequest => write!(f, "invalid request"),
1286            Self::BoundaryProcess(err) => write!(f, "boundary process: {err}"),
1287            Self::Route(err) => write!(f, "route: {err}"),
1288            Self::InvalidResponse => write!(f, "invalid response"),
1289        }
1290    }
1291}
1292
1293impl std::error::Error for RpcRouteError {
1294    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1295        match self {
1296            Self::BoundaryProcess(err) => Some(err),
1297            Self::Route(err) => Some(err),
1298            _ => None,
1299        }
1300    }
1301}
1302
1303#[derive(Debug, Clone, PartialEq, Eq)]
1304pub enum RuntimeMutationError {
1305    Config(ConfigResolutionError),
1306    Runtime(RuntimeBoundaryError),
1307}
1308
1309impl std::fmt::Display for RuntimeMutationError {
1310    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1311        match self {
1312            Self::Config(err) => write!(f, "config resolution: {err}"),
1313            Self::Runtime(err) => write!(f, "runtime boundary: {err}"),
1314        }
1315    }
1316}
1317
1318impl std::error::Error for RuntimeMutationError {
1319    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1320        match self {
1321            Self::Config(err) => Some(err),
1322            Self::Runtime(err) => Some(err),
1323        }
1324    }
1325}
1326
1327#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1328#[serde(rename_all = "snake_case")]
1329pub enum SubscribeScope {
1330    Mob,
1331    Agent,
1332    Interaction,
1333}
1334
1335#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1336pub struct SubscribeRequest {
1337    pub scope: SubscribeScope,
1338    pub last_event_id: Option<String>,
1339    pub agent_id: Option<String>,
1340}
1341
1342impl Default for SubscribeRequest {
1343    fn default() -> Self {
1344        Self {
1345            scope: SubscribeScope::Mob,
1346            last_event_id: None,
1347            agent_id: None,
1348        }
1349    }
1350}
1351
1352#[derive(Debug, Clone, PartialEq, Eq)]
1353pub enum SubscribeError {
1354    EmptyCheckpoint,
1355    UnknownCheckpoint(String),
1356    MissingAgentId,
1357    InvalidAgentId,
1358}
1359
1360impl std::fmt::Display for SubscribeError {
1361    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1362        match self {
1363            Self::EmptyCheckpoint => write!(f, "empty checkpoint"),
1364            Self::UnknownCheckpoint(id) => write!(f, "unknown checkpoint: {id}"),
1365            Self::MissingAgentId => write!(f, "missing agent id"),
1366            Self::InvalidAgentId => write!(f, "invalid agent id"),
1367        }
1368    }
1369}
1370
1371impl std::error::Error for SubscribeError {}
1372
1373#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1374pub struct SubscribeKeepAlive {
1375    pub interval_ms: u64,
1376    pub event: String,
1377}
1378
1379#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1380pub struct SubscribeResponse {
1381    pub scope: SubscribeScope,
1382    pub replay_from_event_id: Option<String>,
1383    pub keep_alive: SubscribeKeepAlive,
1384    pub keep_alive_comment: String,
1385    pub event_frames: Vec<String>,
1386    pub events: Vec<EventEnvelope<UnifiedEvent>>,
1387}
1388
1389const SSE_KEEP_ALIVE_INTERVAL_MS: u64 = 15_000;
1390const SSE_KEEP_ALIVE_EVENT_NAME: &str = "keep-alive";
1391const SSE_KEEP_ALIVE_COMMENT_FRAME: &str = ": keep-alive\n\n";
1392const SUBSCRIBE_REPLAY_EVENT_CAP: usize = 3;
1393const SCHEDULING_CLAIM_RETENTION_WINDOW_MS: u64 = 86_400_000;
1394const SCHEDULING_CLAIMS_MAX_RETAINED: usize = 4_096;
1395const SCHEDULING_LAST_DUE_MAX_RETAINED: usize = 4_096;
1396const DELIVERY_HISTORY_LIMIT_DEFAULT: usize = 20;
1397const DELIVERY_HISTORY_LIMIT_MAX: usize = 200;
1398const ROUTING_RESOLUTION_LIMIT_MAX: usize = 512;
1399pub const ROUTING_RETRY_MAX_CAP: u32 = 10;
1400const DELIVERY_RATE_WINDOW_MS: u64 = 60_000;
1401const DELIVERY_RATE_WINDOWS_RETAINED: u64 = 2;
1402const DELIVERY_CLOCK_STEP_MS: u64 = 1_000;
1403const GATING_APPROVAL_TIMEOUT_DEFAULT_MS: u64 = 60_000;
1404/// Lower bound on a caller-supplied R3 approval timeout. Without it a value of
1405/// `0` (or a few ms) makes `deadline_at_ms == created_at_ms`, so the pending
1406/// entry is treated as already-expired by `refresh_gating_timeouts` on the very
1407/// next gating RPC and is replaced with a `timeout_fallback` SafeDraft before
1408/// any approver can act — R3 approval can never complete. 1 second is the floor
1409/// that keeps a `PendingApproval` outcome actually approvable.
1410const GATING_APPROVAL_TIMEOUT_MIN_MS: u64 = 1_000;
1411/// Upper bound on a caller-supplied R3 approval timeout. Without it a huge
1412/// `approval_timeout_ms` makes `created_at_ms.saturating_add(timeout_ms)`
1413/// saturate to `u64::MAX`, so `now_ms >= deadline_at_ms` is never true and the
1414/// pending approval can never time out to a safe draft. 7 days is far beyond
1415/// any real human-approval window.
1416const GATING_APPROVAL_TIMEOUT_MAX_MS: u64 = 7 * 24 * 60 * 60 * 1_000;
1417const GATING_AUDIT_MAX_RETAINED: usize = 512;
1418const GATING_PENDING_MAX_RETAINED: usize = 512;
1419const MEMORY_ASSERTIONS_MAX_RETAINED: usize = 4_096;
1420const MEMORY_SUPPORTED_STORES: [&str; 5] = [
1421    "knowledge_graph",
1422    "vector",
1423    "timeline",
1424    "todo",
1425    "top_of_mind",
1426];
1427const ELEPHANT_HEALTHCHECK_TIMEOUT: Duration = Duration::from_secs(2);
1428// Multi-year bounded lookback so sparse valid cron schedules (for example leap-day)
1429// are not silently skipped when polling cadence is coarse.
1430const CRON_LOOKBACK_MINUTES: u64 = 5_270_400;
1431// Per-request ceiling on TOTAL cron lookback iterations across all schedules in
1432// one evaluate/dispatch call. Each schedule may still walk back the full
1433// `CRON_LOOKBACK_MINUTES` (≈10y) for a legitimately-sparse cron such as
1434// `0 0 29 2 *` (Feb 29, ≈1.5M iterations), but a request packed with many such
1435// crons cannot multiply that cost without bound. The ceiling allows several
1436// worst-case sparse crons per request (well above any real schedule set) while
1437// capping the adversarial `MAX_SCHEDULES_PER_REQUEST * ~1.5M` (~380M) blow-up
1438// that would otherwise stall the tokio worker and starve the module mutex.
1439const CRON_LOOKBACK_BUDGET_PER_REQUEST: u64 = CRON_LOOKBACK_MINUTES.saturating_mul(8);
1440const CONSOLE_EXPERIENCE_ROUTE: &str = "/console/experience";
1441const CONSOLE_MODULES_ROUTE: &str = "/console/modules";
1442const EVENTS_SUBSCRIBE_METHOD: &str = "mobkit/events/subscribe";
1443
1444fn default_delivery_history_limit() -> usize {
1445    DELIVERY_HISTORY_LIMIT_DEFAULT
1446}
1447
1448pub fn run_meerkat_baseline_verification_once(
1449    command: &str,
1450    args: &[String],
1451    env: &[(String, String)],
1452    timeout: Duration,
1453) -> Result<BaselineVerificationReport, BaselineRuntimeError> {
1454    let line = run_process_json_line(command, args, env, timeout)
1455        .map_err(BaselineRuntimeError::Process)?;
1456    let value: Value =
1457        serde_json::from_str(&line).map_err(|_| BaselineRuntimeError::InvalidRepoPathJson)?;
1458    let repo = value
1459        .as_object()
1460        .and_then(|obj| obj.get("repo_root"))
1461        .ok_or(BaselineRuntimeError::MissingRepoRoot)?
1462        .as_str()
1463        .ok_or(BaselineRuntimeError::InvalidRepoRoot)?;
1464    if repo.trim().is_empty() {
1465        return Err(BaselineRuntimeError::InvalidRepoRoot);
1466    }
1467    verify_meerkat_baseline_symbols(Some(std::path::Path::new(repo)))
1468        .map_err(BaselineRuntimeError::Baseline)
1469}
1470
1471pub fn build_runtime_decision_state(
1472    input: RuntimeDecisionInputs,
1473) -> Result<RuntimeDecisionState, DecisionRuntimeError> {
1474    validate_bigquery_naming(&input.bigquery).map_err(DecisionRuntimeError::Policy)?;
1475    let modules = load_trusted_mobkit_modules_from_toml(&input.trusted_mobkit_toml)
1476        .map_err(DecisionRuntimeError::Policy)?;
1477    validate_runtime_ops_policy(&input.ops).map_err(DecisionRuntimeError::Policy)?;
1478    let release_metadata = parse_release_metadata_json(&input.release_metadata_json)
1479        .map_err(DecisionRuntimeError::Policy)?;
1480    validate_release_metadata(&release_metadata).map_err(DecisionRuntimeError::Policy)?;
1481    if input.trusted_oidc.audience.trim().is_empty() {
1482        return Err(DecisionRuntimeError::Policy(
1483            DecisionPolicyError::InvalidTrustedAuthConfig(
1484                "trusted OIDC audience must not be empty".to_string(),
1485            ),
1486        ));
1487    }
1488    parse_oidc_discovery_json(&input.trusted_oidc.discovery_json).map_err(|err| {
1489        DecisionRuntimeError::Policy(DecisionPolicyError::InvalidTrustedAuthConfig(format!(
1490            "invalid trusted OIDC discovery: {err:?}"
1491        )))
1492    })?;
1493    parse_jwks_json(&input.trusted_oidc.jwks_json).map_err(|err| {
1494        DecisionRuntimeError::Policy(DecisionPolicyError::InvalidTrustedAuthConfig(format!(
1495            "invalid trusted JWKS: {err:?}"
1496        )))
1497    })?;
1498
1499    Ok(RuntimeDecisionState {
1500        bigquery: input.bigquery,
1501        modules,
1502        auth: input.auth,
1503        trusted_oidc: input.trusted_oidc,
1504        console: input.console,
1505        ops: input.ops,
1506        release_metadata,
1507    })
1508}
1509
1510fn current_time_ms() -> u64 {
1511    SystemTime::now()
1512        .duration_since(UNIX_EPOCH)
1513        .unwrap_or_default()
1514        .as_millis() as u64
1515}
meerkat_mobkit/runtime.rs

meerkat_mobkit/
runtime.rs
