1use super::*;
4
5pub fn start_mobkit_runtime(
6 config: MobKitConfig,
7 agent_events: Vec<EventEnvelope<UnifiedEvent>>,
8 timeout: Duration,
9) -> Result<MobkitRuntimeHandle, MobkitRuntimeError> {
10 start_mobkit_runtime_with_options(config, agent_events, timeout, RuntimeOptions::default())
11}
12
13pub fn start_mobkit_runtime_with_options(
14 config: MobKitConfig,
15 agent_events: Vec<EventEnvelope<UnifiedEvent>>,
16 timeout: Duration,
17 options: RuntimeOptions,
18) -> Result<MobkitRuntimeHandle, MobkitRuntimeError> {
19 let delivery_runtime_epoch_ms = current_time_ms();
20 let mut lifecycle_events = Vec::new();
21 let mut seq = 0_u64;
22 lifecycle_events.push(LifecycleEvent {
23 seq,
24 stage: LifecycleStage::MobStarted,
25 });
26 seq += 1;
27
28 let mut supervisor_transitions = Vec::new();
29 let mut module_events = Vec::new();
30 let mut loaded_modules = BTreeSet::new();
31 let mut live_children = BTreeMap::new();
32
33 for module_id in &config.discovery.modules {
34 let module = config
35 .modules
36 .iter()
37 .find(|module| &module.id == module_id)
38 .ok_or_else(|| {
39 MobkitRuntimeError::Config(ConfigResolutionError::ModuleNotConfigured(
40 module_id.clone(),
41 ))
42 })?;
43
44 let pre_spawn = config
45 .pre_spawn
46 .iter()
47 .find(|data| data.module_id == *module_id);
48
49 let mut start_result = supervise_module_start(module, pre_spawn, timeout, &options);
50 supervisor_transitions.append(&mut start_result.transitions);
51 if let Some(error) = start_result.terminal_error.as_ref() {
52 let timestamp_ms = current_time_ms();
53 module_events.push(EventEnvelope {
54 event_id: format!("evt-supervisor-warning-{}-{timestamp_ms}", module.id),
55 source: "module".to_string(),
56 timestamp_ms,
57 event: UnifiedEvent::Module(ModuleEvent {
58 module: module.id.clone(),
59 event_type: "supervisor.warning".to_string(),
60 payload: serde_json::json!({
61 "error": format!("{error:?}")
62 }),
63 }),
64 });
65 }
66 if let Some(event) = start_result.event {
67 loaded_modules.insert(module_id.clone());
68 if let Some(child) = start_result.child {
69 live_children.insert(module_id.clone(), child);
70 }
71 module_events.push(event);
72 }
73 }
74
75 lifecycle_events.push(LifecycleEvent {
76 seq,
77 stage: LifecycleStage::ModulesStarted,
78 });
79 seq += 1;
80
81 let merged_events = merge_unified_events(module_events, agent_events);
82 lifecycle_events.push(LifecycleEvent {
83 seq,
84 stage: LifecycleStage::MergedStreamStarted,
85 });
86
87 let memory_backend = match options.memory_backend.as_ref() {
88 Some(MemoryBackendConfig::Elephant(config)) => Some(
89 ElephantMemoryStoreAdapter::from_config(config)
90 .map_err(MobkitRuntimeError::MemoryBackend)?,
91 ),
92 None => None,
93 };
94 let persisted_memory = match memory_backend.as_ref() {
95 Some(backend) => backend
96 .read_state()
97 .map_err(MobkitRuntimeError::MemoryBackend)?,
98 None => PersistedMemoryState::default(),
99 };
100 let mut memory_assertions = persisted_memory
101 .assertions
102 .into_iter()
103 .filter_map(|assertion| {
104 let entity = MobkitRuntimeHandle::canonical_memory_token(&assertion.entity)?;
105 let topic = MobkitRuntimeHandle::canonical_memory_token(&assertion.topic)?;
106 let store = MobkitRuntimeHandle::canonical_memory_store(&assertion.store)?;
107 let fact = assertion.fact.trim();
108 if fact.is_empty() {
109 return None;
110 }
111 Some(MemoryAssertion {
112 assertion_id: assertion.assertion_id,
113 entity,
114 topic,
115 store,
116 fact: fact.to_string(),
117 metadata: assertion.metadata,
118 indexed_at_ms: assertion.indexed_at_ms,
119 })
120 })
121 .collect::<Vec<_>>();
122 while memory_assertions.len() > MEMORY_ASSERTIONS_MAX_RETAINED {
123 memory_assertions.remove(0);
124 }
125 let mut memory_conflicts = BTreeMap::new();
126 for signal in persisted_memory.conflicts {
127 let Some(entity) = MobkitRuntimeHandle::canonical_memory_token(&signal.entity) else {
128 continue;
129 };
130 let Some(topic) = MobkitRuntimeHandle::canonical_memory_token(&signal.topic) else {
131 continue;
132 };
133 let Some(store) = MobkitRuntimeHandle::canonical_memory_store(&signal.store) else {
134 continue;
135 };
136 let normalized_signal = MemoryConflictSignal {
137 entity: entity.clone(),
138 topic: topic.clone(),
139 store: store.clone(),
140 reason: signal
141 .reason
142 .as_deref()
143 .map(str::trim)
144 .filter(|value| !value.is_empty())
145 .map(ToString::to_string),
146 updated_at_ms: signal.updated_at_ms,
147 };
148 memory_conflicts.insert(
149 MemoryConflictKey {
150 entity,
151 topic,
152 store,
153 },
154 normalized_signal,
155 );
156 }
157 let memory_sequence = memory_assertions
158 .iter()
159 .filter_map(|assertion| parse_memory_assertion_sequence(&assertion.assertion_id))
160 .max()
161 .map(|last_sequence| last_sequence.saturating_add(1))
162 .unwrap_or(memory_assertions.len() as u64);
163
164 Ok(MobkitRuntimeHandle {
165 config,
166 runtime_options: options,
167 loaded_modules,
168 live_children,
169 lifecycle_events,
170 supervisor_report: SupervisorReport {
171 transitions: supervisor_transitions,
172 },
173 merged_events,
174 scheduling_claims: BTreeSet::new(),
175 scheduling_claim_ticks: BTreeMap::new(),
176 scheduling_last_due_ticks: BTreeMap::new(),
177 scheduling_dispatch_sequence: 0,
178 routing_sequence: 0,
179 routing_resolutions: BTreeMap::new(),
180 routing_resolution_order: Vec::new(),
181 runtime_routes: BTreeMap::new(),
182 delivery_sequence: 0,
183 delivery_runtime_epoch_ms,
184 delivery_now_floor_ms: 0,
185 delivery_clock_ms: 0,
186 delivery_history: Vec::new(),
187 delivery_idempotency: BTreeMap::new(),
188 delivery_idempotency_by_delivery: BTreeMap::new(),
189 delivery_rate_window_counts: BTreeMap::new(),
190 gating_sequence: 0,
191 gating_pending: BTreeMap::new(),
192 gating_pending_order: Vec::new(),
193 gating_audit: Vec::new(),
194 memory_sequence,
195 memory_assertions,
196 memory_conflicts,
197 memory_backend,
198 running: true,
199 })
200}
201
202fn parse_memory_assertion_sequence(assertion_id: &str) -> Option<u64> {
203 assertion_id
204 .strip_prefix("memory-assert-")
205 .and_then(|suffix| suffix.parse::<u64>().ok())
206}