1use std::collections::HashSet;
2use std::process::{Command, Stdio};
3
4use anyhow::{Context, Result};
5use serde::{Deserialize, Serialize, de::DeserializeOwned};
6use serde_json::{Value, json};
7
8use super::{
9 ApprovalRecord, ApprovalReplayResult, CheckpointManifest, CheckpointRecord, CompactionRecord,
10 MessageRecord, NewProcess, NewProviderProbe, PairingTokenRecord, PluginInstallRecord,
11 ProcessRecord, ProcessStatus, ProviderProbeRecord, RuntimeStore, SessionRecord, TaskRecord,
12 TaskStatus, TaskTimelineEvent, ToolRunRecord, approve_and_replay, deny_approval,
13 request_daemon_json, restore_checkpoint,
14};
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum RuntimeClientSource {
19 Daemon,
20 Local,
21}
22
23impl RuntimeClientSource {
24 pub fn as_str(self) -> &'static str {
25 match self {
26 RuntimeClientSource::Daemon => "daemon",
27 RuntimeClientSource::Local => "local",
28 }
29 }
30}
31
32#[derive(Debug, Clone)]
33enum RuntimeClientMode {
34 PreferDaemon,
35 DaemonOnly,
36 LocalOnly,
37}
38
39#[derive(Debug, Clone)]
40pub struct RuntimeClient {
41 mode: RuntimeClientMode,
42 auth_token: Option<String>,
43}
44
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct RuntimeRead<T> {
47 pub source: RuntimeClientSource,
48 pub value: T,
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize)]
52pub struct RuntimeHealth {
53 pub ok: bool,
54 pub service: String,
55 pub database: String,
56}
57
58#[derive(Debug, Clone, Serialize, Deserialize)]
59pub struct RuntimeSnapshot {
60 pub ok: bool,
61 pub database: String,
62 pub sessions: Vec<SessionRecord>,
63 pub tasks: Vec<TaskRecord>,
64 pub tool_runs: Vec<ToolRunRecord>,
65 pub processes: Vec<ProcessRecord>,
66 pub approvals: Vec<ApprovalRecord>,
67 pub checkpoints: Vec<CheckpointRecord>,
68 pub compactions: Vec<CompactionRecord>,
69 pub plugins: Vec<PluginInstallRecord>,
70 pub provider_probes: Vec<ProviderProbeRecord>,
71 pub pairings: Vec<PairingTokenRecord>,
72 pub safety: crate::app::SafetyConfig,
73}
74
75#[derive(Debug, Clone, Serialize, Deserialize)]
76pub struct RuntimeDashboardCounts {
77 pub pending_approvals: usize,
78 pub running_tasks: usize,
79 pub waiting_tasks: usize,
80 pub blocked_tasks: usize,
81 pub ready_processes: usize,
82 pub recent_checkpoints: usize,
83 pub installed_plugins: usize,
84 pub archived_approvals: usize,
85 pub archived_checkpoints: usize,
86}
87
88#[derive(Debug, Clone, Serialize, Deserialize)]
89pub struct RuntimeDashboard {
90 pub ok: bool,
91 pub health: RuntimeHealth,
92 pub safety: crate::app::SafetyConfig,
93 pub counts: RuntimeDashboardCounts,
94 pub sessions: Vec<SessionRecord>,
95 pub tasks: Vec<TaskRecord>,
96 pub tool_runs: Vec<ToolRunRecord>,
97 pub processes: Vec<ProcessRecord>,
98 pub approvals: Vec<ApprovalRecord>,
99 pub checkpoints: Vec<CheckpointRecord>,
100 pub compactions: Vec<CompactionRecord>,
101 pub plugins: Vec<PluginInstallRecord>,
102 pub provider_probes: Vec<ProviderProbeRecord>,
103 pub pairings: Vec<PairingTokenRecord>,
104}
105
106#[derive(Debug, Clone, Serialize, Deserialize)]
107pub struct RuntimeDiagnostics {
108 #[serde(flatten)]
109 pub snapshot: RuntimeSnapshot,
110 pub mode: String,
111 pub hygiene: RuntimeDiagnosticsHygiene,
112}
113
114#[derive(Debug, Clone, Serialize, Deserialize)]
115pub struct RuntimeDiagnosticsHygiene {
116 pub preview: Value,
117 pub visible: RuntimeDiagnosticsVisible,
118 pub archived: RuntimeDiagnosticsArchived,
119}
120
121#[derive(Debug, Clone, Serialize, Deserialize)]
122pub struct RuntimeDiagnosticsVisible {
123 pub approvals: usize,
124 pub checkpoints: usize,
125}
126
127#[derive(Debug, Clone, Serialize, Deserialize)]
128pub struct RuntimeDiagnosticsArchived {
129 pub approvals: usize,
130 pub checkpoints: usize,
131}
132
133#[derive(Debug, Clone, Serialize, Deserialize)]
134pub struct RuntimeHygienePreview {
135 pub ok: bool,
136 pub reason: String,
137 pub approvals: Vec<ApprovalRecord>,
138 pub checkpoints: Vec<CheckpointRecord>,
139 pub counts: RuntimeHygieneCounts,
140}
141
142#[derive(Debug, Clone, Serialize, Deserialize)]
143pub struct RuntimeHygieneCounts {
144 pub approvals: usize,
145 pub checkpoints: usize,
146 pub total: usize,
147}
148
149#[derive(Debug, Clone, Serialize, Deserialize)]
150pub struct RuntimeHygieneArchive {
151 pub ok: bool,
152 pub reason: String,
153 pub archived: RuntimeHygieneCounts,
154 pub matched: RuntimeHygieneCounts,
155}
156
157#[derive(Debug, Clone, Serialize, Deserialize)]
158pub struct RuntimeTaskDetail {
159 pub ok: bool,
160 pub task: TaskRecord,
161 pub events: Vec<TaskTimelineEvent>,
162 pub session: Option<SessionRecord>,
163 pub messages: Vec<MessageRecord>,
164 pub approvals: Vec<ApprovalRecord>,
165 pub checkpoints: Vec<CheckpointRecord>,
166 pub processes: Vec<ProcessRecord>,
167 pub tool_runs: Vec<ToolRunRecord>,
168 pub compactions: Vec<CompactionRecord>,
169}
170
171#[derive(Debug, Clone, Serialize, Deserialize)]
172pub struct RuntimeApprovalDetail {
173 pub ok: bool,
174 pub approval: ApprovalRecord,
175 pub task: Option<TaskRecord>,
176 pub checkpoint: Option<CheckpointRecord>,
177 pub pending_action: Value,
178 pub args: Value,
179 pub changed_files: Value,
180 pub affected_paths: Vec<String>,
181}
182
183#[derive(Debug, Clone, Serialize, Deserialize)]
184pub struct RuntimeCheckpointDetail {
185 pub ok: bool,
186 pub checkpoint: CheckpointRecord,
187 pub task: Option<TaskRecord>,
188 pub approval: Option<ApprovalRecord>,
189 pub pending_action: Value,
190 pub changed_files: Value,
191 pub affected_paths: Vec<String>,
192}
193
194#[derive(Debug, Clone, Serialize, Deserialize)]
195pub struct RuntimeProcessLog {
196 pub ok: bool,
197 pub content: String,
198}
199
200#[derive(Debug, Clone, Serialize, Deserialize)]
201pub struct RuntimeProcessOpen {
202 pub ok: bool,
203 pub target: String,
204}
205
206#[derive(Debug, Clone, Serialize, Deserialize)]
207pub struct RuntimePorts {
208 pub ok: bool,
209 pub ports: String,
210}
211
212#[derive(Debug, Clone, Serialize, Deserialize)]
213pub struct RuntimeApprovalDecision {
214 pub ok: bool,
215 pub approval: Option<ApprovalRecord>,
216 pub replayed: bool,
217 pub summary: String,
218}
219
220#[derive(Debug, Clone, Serialize, Deserialize)]
221pub struct RuntimeCheckpointRestore {
222 pub ok: bool,
223 pub checkpoint: CheckpointManifest,
224}
225
226#[derive(Debug, Clone, Serialize, Deserialize)]
227struct RuntimeItems<T> {
228 ok: bool,
229 items: Vec<T>,
230}
231
232#[derive(Debug, Clone, Serialize, Deserialize)]
233pub struct RuntimeOne<T> {
234 pub ok: bool,
235 pub item: T,
236}
237
238pub struct RuntimeService {
239 store: RuntimeStore,
240}
241
242impl RuntimeClient {
243 pub fn auto() -> Self {
244 Self {
245 mode: RuntimeClientMode::PreferDaemon,
246 auth_token: None,
247 }
248 }
249
250 pub fn daemon() -> Self {
251 Self {
252 mode: RuntimeClientMode::DaemonOnly,
253 auth_token: None,
254 }
255 }
256
257 pub fn daemon_with_token(token: impl Into<String>) -> Self {
258 Self {
259 mode: RuntimeClientMode::DaemonOnly,
260 auth_token: Some(token.into()),
261 }
262 }
263
264 pub fn local() -> Self {
265 Self {
266 mode: RuntimeClientMode::LocalOnly,
267 auth_token: None,
268 }
269 }
270
271 pub fn health(&self) -> Result<RuntimeRead<RuntimeHealth>> {
272 self.read(json!({"command": "health"}), |service| service.health())
273 }
274
275 pub fn snapshot(&self) -> Result<RuntimeRead<RuntimeSnapshot>> {
276 self.read(json!({"command": "runtime_snapshot"}), |service| {
277 service.snapshot()
278 })
279 }
280
281 pub fn dashboard(&self) -> Result<RuntimeRead<RuntimeDashboard>> {
282 self.read(json!({"command": "runtime_dashboard"}), |service| {
283 service.dashboard()
284 })
285 }
286
287 pub fn diagnostics(&self) -> Result<RuntimeRead<RuntimeDiagnostics>> {
288 self.read(json!({"command": "runtime_diagnostics"}), |service| {
289 service.diagnostics()
290 })
291 }
292
293 pub fn hygiene_preview(&self) -> Result<RuntimeRead<RuntimeHygienePreview>> {
294 self.read(json!({"command": "runtime_hygiene_preview"}), |service| {
295 service.hygiene_preview()
296 })
297 }
298
299 pub fn hygiene_archive(&self) -> Result<RuntimeRead<RuntimeHygieneArchive>> {
300 self.read_inner(
301 json!({"command": "runtime_hygiene_archive"}),
302 true,
303 |service| service.hygiene_archive(),
304 )
305 }
306
307 pub fn task_detail(&self, id: &str) -> Result<RuntimeRead<RuntimeTaskDetail>> {
308 self.read(
309 json!({"command": "runtime_task_detail", "id": id}),
310 |service| service.task_detail(id),
311 )
312 }
313
314 pub fn approval_detail(&self, id: &str) -> Result<RuntimeRead<RuntimeApprovalDetail>> {
315 self.read(
316 json!({"command": "runtime_approval_detail", "id": id}),
317 |service| service.approval_detail(id),
318 )
319 }
320
321 pub fn checkpoint_detail(&self, id: &str) -> Result<RuntimeRead<RuntimeCheckpointDetail>> {
322 self.read(
323 json!({"command": "runtime_checkpoint_detail", "id": id}),
324 |service| service.checkpoint_detail(id),
325 )
326 }
327
328 pub fn list_tasks(&self, limit: usize) -> Result<RuntimeRead<Vec<TaskRecord>>> {
329 self.list(
330 json!({"command": "runtime_tasks", "limit": limit}),
331 |service| service.list_tasks(limit),
332 )
333 }
334
335 pub fn list_processes(&self, limit: usize) -> Result<RuntimeRead<Vec<ProcessRecord>>> {
336 self.list(
337 json!({"command": "runtime_processes", "limit": limit}),
338 |service| service.list_processes(limit),
339 )
340 }
341
342 pub fn list_approvals(&self) -> Result<RuntimeRead<Vec<ApprovalRecord>>> {
343 self.list(json!({"command": "runtime_approvals"}), |service| {
344 service.list_approvals()
345 })
346 }
347
348 pub fn list_tool_runs(&self, limit: usize) -> Result<RuntimeRead<Vec<ToolRunRecord>>> {
349 self.list(
350 json!({"command": "runtime_tool_runs", "limit": limit}),
351 |service| service.list_tool_runs(limit),
352 )
353 }
354
355 pub fn list_checkpoints(&self, limit: usize) -> Result<RuntimeRead<Vec<CheckpointRecord>>> {
356 self.list(
357 json!({"command": "runtime_checkpoints", "limit": limit}),
358 |service| service.list_checkpoints(limit),
359 )
360 }
361
362 pub fn list_plugins(&self) -> Result<RuntimeRead<Vec<PluginInstallRecord>>> {
363 self.list(json!({"command": "runtime_plugins"}), |service| {
364 service.list_plugins()
365 })
366 }
367
368 pub fn process_log(&self, id: &str, tail_bytes: Option<u64>) -> Result<RuntimeProcessLog> {
369 self.action(
370 json!({
371 "command": "logs",
372 "id": id,
373 "tail_bytes": tail_bytes,
374 }),
375 |service| service.process_log(id, tail_bytes),
376 )
377 }
378
379 pub fn stop_process(&self, id: &str) -> Result<RuntimeOne<ProcessRecord>> {
380 self.action_authed(json!({"command": "stop_process", "id": id}), |service| {
381 service
382 .stop_process(id)
383 .map(|item| RuntimeOne { ok: true, item })
384 })
385 }
386
387 pub fn restart_process(&self, id: &str) -> Result<RuntimeOne<ProcessRecord>> {
388 self.action_authed(json!({"command": "restart_process", "id": id}), |service| {
389 service
390 .restart_process(id)
391 .map(|item| RuntimeOne { ok: true, item })
392 })
393 }
394
395 pub fn open_process(&self, id: &str) -> Result<RuntimeProcessOpen> {
396 self.action_authed(json!({"command": "open_process", "id": id}), |service| {
397 service.open_process(id)
398 })
399 }
400
401 pub fn ports(&self) -> Result<RuntimePorts> {
402 self.action(json!({"command": "ports"}), |service| service.ports())
403 }
404
405 pub fn approve(&self, id: &str) -> Result<RuntimeApprovalDecision> {
406 self.action_authed(json!({"command": "approve", "id": id}), |_service| {
407 RuntimeService::approval_decision(approve_and_replay(id)?)
408 })
409 }
410
411 pub fn deny(&self, id: &str) -> Result<RuntimeApprovalDecision> {
412 self.action_authed(json!({"command": "deny", "id": id}), |_service| {
413 RuntimeService::approval_decision(deny_approval(id)?)
414 })
415 }
416
417 pub fn restore_checkpoint(&self, id: &str) -> Result<RuntimeCheckpointRestore> {
418 self.action_authed(
419 json!({"command": "restore_checkpoint", "id": id}),
420 |_service| {
421 Ok(RuntimeCheckpointRestore {
422 ok: true,
423 checkpoint: restore_checkpoint(id)?,
424 })
425 },
426 )
427 }
428
429 pub fn set_plugin_enabled(&self, id: &str, enabled: bool) -> Result<()> {
430 self.action_authed::<Value, _>(
431 json!({"command": "set_plugin_enabled", "id": id, "enabled": enabled}),
432 |service| {
433 service.set_plugin_enabled(id, enabled)?;
434 Ok(json!({"ok": true}))
435 },
436 )?;
437 Ok(())
438 }
439
440 pub fn model_info(&self, model: &str) -> Result<Value> {
441 self.action(
442 json!({"command": "model_info", "model": model}),
443 |service| Ok(json!({"ok": true, "model": service.model_info(model)})),
444 )
445 }
446
447 pub fn set_safety_mode(&self, mode: &str) -> Result<Value> {
448 self.action_authed(
449 json!({"command": "set_safety_mode", "mode": mode}),
450 |service| {
451 let safety = service.set_safety_mode(mode)?;
452 Ok(json!({"ok": true, "safety": safety}))
453 },
454 )
455 }
456
457 fn list<T, F>(&self, body: Value, local: F) -> Result<RuntimeRead<Vec<T>>>
458 where
459 T: DeserializeOwned,
460 F: FnOnce(&RuntimeService) -> Result<Vec<T>>,
461 {
462 self.read(body, |service| {
463 local(service).map(|items| RuntimeItems { ok: true, items })
464 })
465 .map(|read| RuntimeRead {
466 source: read.source,
467 value: read.value.items,
468 })
469 }
470
471 fn read<T, F>(&self, body: Value, local: F) -> Result<RuntimeRead<T>>
472 where
473 T: DeserializeOwned,
474 F: FnOnce(&RuntimeService) -> Result<T>,
475 {
476 self.read_inner(body, false, local)
477 }
478
479 fn action<T, F>(&self, body: Value, local: F) -> Result<T>
480 where
481 T: DeserializeOwned,
482 F: FnOnce(&RuntimeService) -> Result<T>,
483 {
484 self.action_inner(body, false, local)
485 }
486
487 fn action_authed<T, F>(&self, body: Value, local: F) -> Result<T>
488 where
489 T: DeserializeOwned,
490 F: FnOnce(&RuntimeService) -> Result<T>,
491 {
492 self.action_inner(body, true, local)
493 }
494
495 fn read_inner<T, F>(&self, body: Value, authed: bool, local: F) -> Result<RuntimeRead<T>>
496 where
497 T: DeserializeOwned,
498 F: FnOnce(&RuntimeService) -> Result<T>,
499 {
500 match self.mode {
501 RuntimeClientMode::LocalOnly => RuntimeService::open_default()
502 .and_then(|service| local(&service))
503 .map(|value| RuntimeRead {
504 source: RuntimeClientSource::Local,
505 value,
506 }),
507 RuntimeClientMode::DaemonOnly => {
508 self.request_daemon(body, authed).map(|value| RuntimeRead {
509 source: RuntimeClientSource::Daemon,
510 value,
511 })
512 },
513 RuntimeClientMode::PreferDaemon => match self.request_daemon(body, authed) {
514 Ok(value) => Ok(RuntimeRead {
515 source: RuntimeClientSource::Daemon,
516 value,
517 }),
518 Err(_) => RuntimeService::open_default()
519 .and_then(|service| local(&service))
520 .map(|value| RuntimeRead {
521 source: RuntimeClientSource::Local,
522 value,
523 }),
524 },
525 }
526 }
527
528 fn action_inner<T, F>(&self, body: Value, authed: bool, local: F) -> Result<T>
529 where
530 T: DeserializeOwned,
531 F: FnOnce(&RuntimeService) -> Result<T>,
532 {
533 match self.mode {
534 RuntimeClientMode::LocalOnly => {
535 RuntimeService::open_default().and_then(|service| local(&service))
536 },
537 RuntimeClientMode::DaemonOnly => self.request_daemon(body, authed),
538 RuntimeClientMode::PreferDaemon => match self.request_daemon(body, authed) {
539 Ok(value) => Ok(value),
540 Err(_) => RuntimeService::open_default().and_then(|service| local(&service)),
541 },
542 }
543 }
544
545 fn request_daemon<T: DeserializeOwned>(&self, mut body: Value, authed: bool) -> Result<T> {
546 if authed && let Some(token) = self.auth_token.as_ref() {
547 body["auth"] = json!({ "token": token });
548 }
549 let value = request_daemon_json(body)?;
550 serde_json::from_value(value).context("daemon response had unexpected shape")
551 }
552}
553
554impl RuntimeService {
555 pub fn open_default() -> Result<Self> {
556 Ok(Self {
557 store: RuntimeStore::open_default()?,
558 })
559 }
560
561 pub fn from_store(store: RuntimeStore) -> Self {
562 Self { store }
563 }
564
565 pub fn health(&self) -> Result<RuntimeHealth> {
566 Ok(RuntimeHealth {
567 ok: true,
568 service: "mermaidd".to_string(),
569 database: self.store.path().display().to_string(),
570 })
571 }
572
573 pub fn snapshot(&self) -> Result<RuntimeSnapshot> {
574 Ok(RuntimeSnapshot {
575 ok: true,
576 database: self.store.path().display().to_string(),
577 sessions: self.store.sessions().list(100)?,
578 tasks: self.store.tasks().list(100)?,
579 tool_runs: self.store.tool_runs().list(100)?,
580 processes: self.store.processes().list(100)?,
581 approvals: self.store.approvals().list_all(100)?,
582 checkpoints: self.store.checkpoints().list_all(100)?,
583 compactions: self.store.compactions().list(100)?,
584 plugins: self.store.plugins().list()?,
585 provider_probes: self.store.provider_probes().list(None, None)?,
586 pairings: self.store.pairing_tokens().list()?,
587 safety: crate::app::load_config().unwrap_or_default().safety,
588 })
589 }
590
591 pub fn dashboard(&self) -> Result<RuntimeDashboard> {
592 let sessions = self.store.sessions().list(100)?;
593 let tasks = self.store.tasks().list(200)?;
594 let tool_runs = self.store.tool_runs().list(100)?;
595 let processes = self.store.processes().list(100)?;
596 let approvals = self.store.approvals().list_pending()?;
597 let checkpoints = self.store.checkpoints().list(100)?;
598 let compactions = self.store.compactions().list(100)?;
599 let plugins = self.store.plugins().list()?;
600 let provider_probes = self.store.provider_probes().list(None, None)?;
601 let pairings = self.store.pairing_tokens().list()?;
602
603 let running_tasks = tasks
604 .iter()
605 .filter(|task| task.status == TaskStatus::Running)
606 .count();
607 let waiting_tasks = tasks
608 .iter()
609 .filter(|task| task.status == TaskStatus::WaitingForApproval)
610 .count();
611 let blocked_tasks = tasks
612 .iter()
613 .filter(|task| matches!(task.status, TaskStatus::Blocked | TaskStatus::Failed))
614 .count();
615 let ready_processes = processes
616 .iter()
617 .filter(|process| process.detected_url.is_some())
618 .count();
619
620 Ok(RuntimeDashboard {
621 ok: true,
622 health: self.health()?,
623 safety: crate::app::load_config().unwrap_or_default().safety,
624 counts: RuntimeDashboardCounts {
625 pending_approvals: approvals.len(),
626 running_tasks,
627 waiting_tasks,
628 blocked_tasks,
629 ready_processes,
630 recent_checkpoints: checkpoints.len(),
631 installed_plugins: plugins.len(),
632 archived_approvals: self.store.approvals().count_archived()?,
633 archived_checkpoints: self.store.checkpoints().count_archived()?,
634 },
635 sessions,
636 tasks,
637 tool_runs,
638 processes,
639 approvals,
640 checkpoints,
641 compactions,
642 plugins,
643 provider_probes,
644 pairings,
645 })
646 }
647
648 pub fn diagnostics(&self) -> Result<RuntimeDiagnostics> {
649 let snapshot = self.snapshot()?;
650 let hygiene = self.hygiene_preview()?;
651 Ok(RuntimeDiagnostics {
652 snapshot,
653 mode: "diagnostics".to_string(),
654 hygiene: RuntimeDiagnosticsHygiene {
655 preview: serde_json::to_value(hygiene)?,
656 visible: RuntimeDiagnosticsVisible {
657 approvals: self.store.approvals().list_pending()?.len(),
658 checkpoints: self.store.checkpoints().list(100)?.len(),
659 },
660 archived: RuntimeDiagnosticsArchived {
661 approvals: self.store.approvals().count_archived()?,
662 checkpoints: self.store.checkpoints().count_archived()?,
663 },
664 },
665 })
666 }
667
668 pub fn hygiene_preview(&self) -> Result<RuntimeHygienePreview> {
669 let preview = self.raw_hygiene_preview()?;
670 Ok(RuntimeHygienePreview {
671 ok: true,
672 reason: runtime_hygiene_reason().to_string(),
673 counts: RuntimeHygieneCounts {
674 approvals: preview.approvals.len(),
675 checkpoints: preview.checkpoints.len(),
676 total: preview.approvals.len() + preview.checkpoints.len(),
677 },
678 approvals: preview.approvals,
679 checkpoints: preview.checkpoints,
680 })
681 }
682
683 pub fn hygiene_archive(&self) -> Result<RuntimeHygieneArchive> {
684 let preview = self.raw_hygiene_preview()?;
685 let approval_ids = preview
686 .approvals
687 .iter()
688 .map(|approval| approval.id.clone())
689 .collect::<Vec<_>>();
690 let checkpoint_ids = preview
691 .checkpoints
692 .iter()
693 .map(|checkpoint| checkpoint.id.clone())
694 .collect::<Vec<_>>();
695 let reason = runtime_hygiene_reason();
696 let approvals_archived = self.store.approvals().archive(&approval_ids, reason)?;
697 let checkpoints_archived = self.store.checkpoints().archive(&checkpoint_ids, reason)?;
698
699 Ok(RuntimeHygieneArchive {
700 ok: true,
701 reason: reason.to_string(),
702 archived: RuntimeHygieneCounts {
703 approvals: approvals_archived,
704 checkpoints: checkpoints_archived,
705 total: approvals_archived + checkpoints_archived,
706 },
707 matched: RuntimeHygieneCounts {
708 approvals: approval_ids.len(),
709 checkpoints: checkpoint_ids.len(),
710 total: approval_ids.len() + checkpoint_ids.len(),
711 },
712 })
713 }
714
715 pub fn task_detail(&self, id: &str) -> Result<RuntimeTaskDetail> {
716 let task = self
717 .store
718 .tasks()
719 .get(id)?
720 .with_context(|| format!("task not found: {}", id))?;
721 let events = self.store.tasks().events(id)?;
722 let session = match task.conversation_id.as_deref() {
723 Some(session_id) => self.store.sessions().get(session_id)?,
724 None => None,
725 };
726 let messages = match task.conversation_id.as_deref() {
727 Some(session_id) => self.store.messages().list_for_session(session_id)?,
728 None => Vec::new(),
729 };
730 let approvals = self
731 .store
732 .approvals()
733 .list_pending()?
734 .into_iter()
735 .filter(|approval| approval.task_id.as_deref() == Some(id))
736 .collect::<Vec<_>>();
737 let checkpoints = self
738 .store
739 .checkpoints()
740 .list(200)?
741 .into_iter()
742 .filter(|checkpoint| checkpoint.task_id.as_deref() == Some(id))
743 .collect::<Vec<_>>();
744 let processes = self
745 .store
746 .processes()
747 .list(200)?
748 .into_iter()
749 .filter(|process| process.task_id.as_deref() == Some(id))
750 .collect::<Vec<_>>();
751 let tool_runs = self
752 .store
753 .tool_runs()
754 .list(200)?
755 .into_iter()
756 .filter(|tool_run| tool_run.task_id.as_deref() == Some(id))
757 .collect::<Vec<_>>();
758 let compactions = self
759 .store
760 .compactions()
761 .list(200)?
762 .into_iter()
763 .filter(|compaction| compaction.task_id.as_deref() == Some(id))
764 .collect::<Vec<_>>();
765
766 Ok(RuntimeTaskDetail {
767 ok: true,
768 task,
769 events,
770 session,
771 messages,
772 approvals,
773 checkpoints,
774 processes,
775 tool_runs,
776 compactions,
777 })
778 }
779
780 pub fn approval_detail(&self, id: &str) -> Result<RuntimeApprovalDetail> {
781 let approval = self
782 .store
783 .approvals()
784 .get(id)?
785 .with_context(|| format!("approval not found: {}", id))?;
786 let checkpoint = match approval.checkpoint_id.as_deref() {
787 Some(checkpoint_id) => self.store.checkpoints().get(checkpoint_id)?,
788 None => None,
789 };
790 let task = match approval.task_id.as_deref() {
791 Some(task_id) => self.store.tasks().get(task_id)?,
792 None => None,
793 };
794 let pending_action = parse_optional_json(approval.pending_action_json.as_deref());
795 let args = parse_optional_json(approval.args_summary.as_deref());
796 let changed_files = checkpoint
797 .as_ref()
798 .map(|checkpoint| parse_optional_json(Some(&checkpoint.changed_files_json)))
799 .unwrap_or(Value::Null);
800 let affected_paths = affected_paths_from_json(&pending_action, &changed_files);
801
802 Ok(RuntimeApprovalDetail {
803 ok: true,
804 approval,
805 task,
806 checkpoint,
807 pending_action,
808 args,
809 changed_files,
810 affected_paths,
811 })
812 }
813
814 pub fn checkpoint_detail(&self, id: &str) -> Result<RuntimeCheckpointDetail> {
815 let checkpoint = self
816 .store
817 .checkpoints()
818 .get(id)?
819 .with_context(|| format!("checkpoint not found: {}", id))?;
820 let approval = match checkpoint.approval_id.as_deref() {
821 Some(approval_id) => self.store.approvals().get(approval_id)?,
822 None => None,
823 };
824 let task = match checkpoint.task_id.as_deref() {
825 Some(task_id) => self.store.tasks().get(task_id)?,
826 None => None,
827 };
828 let pending_action = parse_optional_json(checkpoint.pending_action_json.as_deref());
829 let changed_files = parse_optional_json(Some(&checkpoint.changed_files_json));
830 let affected_paths = affected_paths_from_json(&pending_action, &changed_files);
831
832 Ok(RuntimeCheckpointDetail {
833 ok: true,
834 checkpoint,
835 task,
836 approval,
837 pending_action,
838 changed_files,
839 affected_paths,
840 })
841 }
842
843 pub fn list_tasks(&self, limit: usize) -> Result<Vec<TaskRecord>> {
844 self.store.tasks().list(limit)
845 }
846
847 pub fn list_processes(&self, limit: usize) -> Result<Vec<ProcessRecord>> {
848 self.store.processes().list(limit)
849 }
850
851 pub fn list_approvals(&self) -> Result<Vec<ApprovalRecord>> {
852 self.store.approvals().list_pending()
853 }
854
855 pub fn list_tool_runs(&self, limit: usize) -> Result<Vec<ToolRunRecord>> {
856 self.store.tool_runs().list(limit)
857 }
858
859 pub fn list_checkpoints(&self, limit: usize) -> Result<Vec<CheckpointRecord>> {
860 self.store.checkpoints().list(limit)
861 }
862
863 pub fn list_plugins(&self) -> Result<Vec<PluginInstallRecord>> {
864 self.store.plugins().list()
865 }
866
867 pub fn process_log(&self, id: &str, tail_bytes: Option<u64>) -> Result<RuntimeProcessLog> {
868 let process = self
869 .store
870 .processes()
871 .get(id)?
872 .with_context(|| format!("process not found: {}", id))?;
873 let path = process
874 .log_path
875 .with_context(|| format!("process has no log path: {}", id))?;
876 let bytes = std::fs::read(&path).with_context(|| format!("failed to read {}", path))?;
877 let tail = tail_bytes.unwrap_or(32 * 1024).min(512 * 1024) as usize;
878 let start = bytes.len().saturating_sub(tail);
879 Ok(RuntimeProcessLog {
880 ok: true,
881 content: String::from_utf8_lossy(&bytes[start..]).into_owned(),
882 })
883 }
884
885 pub fn stop_process(&self, id: &str) -> Result<ProcessRecord> {
886 let process = self
887 .store
888 .processes()
889 .get(id)?
890 .with_context(|| format!("process not found: {}", id))?;
891 kill_pid(process.pid).with_context(|| format!("failed to stop pid {}", process.pid))?;
892 self.store.processes().upsert(NewProcess {
893 id: Some(process.id.clone()),
894 task_id: process.task_id.clone(),
895 pid: process.pid,
896 command: process.command.clone(),
897 cwd: process.cwd.clone(),
898 log_path: process.log_path.clone(),
899 detected_url: process.detected_url.clone(),
900 status: ProcessStatus::Exited,
901 health: Some("stopped".to_string()),
902 })
903 }
904
905 pub fn restart_process(&self, id: &str) -> Result<ProcessRecord> {
906 let process = self
907 .store
908 .processes()
909 .get(id)?
910 .with_context(|| format!("process not found: {}", id))?;
911 let _ = kill_pid(process.pid);
912 let mut command = Command::new("sh");
913 command.arg("-c").arg(&process.command);
914 if let Some(cwd) = process.cwd.as_deref() {
915 command.current_dir(cwd);
916 }
917 if let Some(path) = process.log_path.as_deref() {
918 let file = std::fs::OpenOptions::new()
919 .create(true)
920 .append(true)
921 .open(path)
922 .with_context(|| format!("failed to open process log {}", path))?;
923 let stderr = file.try_clone()?;
924 command
925 .stdout(Stdio::from(file))
926 .stderr(Stdio::from(stderr));
927 }
928 let child = command.spawn()?;
929 self.store.processes().upsert(NewProcess {
930 id: Some(process.id),
931 task_id: process.task_id,
932 pid: child.id(),
933 command: process.command,
934 cwd: process.cwd,
935 log_path: process.log_path,
936 detected_url: process.detected_url,
937 status: ProcessStatus::Running,
938 health: Some("restarted".to_string()),
939 })
940 }
941
942 pub fn open_process(&self, id: &str) -> Result<RuntimeProcessOpen> {
943 let process = self
944 .store
945 .processes()
946 .get(id)?
947 .with_context(|| format!("process not found: {}", id))?;
948 let target = process
949 .detected_url
950 .or(process.log_path)
951 .with_context(|| format!("process has no URL or log path: {}", id))?;
952 crate::utils::open_file(target.clone());
953 Ok(RuntimeProcessOpen { ok: true, target })
954 }
955
956 pub fn resolve_open_target(&self, target: &str) -> Result<String> {
957 if let Some(process) = self.store.processes().get(target)?
958 && let Some(target) = process.detected_url.or(process.log_path)
959 {
960 return Ok(target);
961 }
962 Ok(target.to_string())
963 }
964
965 pub fn ports(&self) -> Result<RuntimePorts> {
966 let output = if cfg!(windows) {
967 Command::new("netstat").arg("-ano").output()?
968 } else {
969 Command::new("sh")
970 .arg("-c")
971 .arg("ss -ltnp 2>/dev/null || netstat -ltnp 2>/dev/null || true")
972 .output()?
973 };
974 Ok(RuntimePorts {
975 ok: true,
976 ports: String::from_utf8_lossy(&output.stdout).into_owned(),
977 })
978 }
979
980 pub fn set_plugin_enabled(&self, id: &str, enabled: bool) -> Result<()> {
981 self.store.plugins().set_enabled(id, enabled)?;
982 super::write_plugin_lockfile()?;
983 Ok(())
984 }
985
986 pub fn set_safety_mode(&self, mode: &str) -> Result<crate::app::SafetyConfig> {
987 let mut config = crate::app::load_config().unwrap_or_default();
988 config.safety.mode = super::SafetyMode::parse(mode)
989 .ok_or_else(|| anyhow::anyhow!("unknown safety mode: {}", mode))?;
990 crate::app::save_config(&config, None)?;
991 Ok(config.safety)
992 }
993
994 pub fn model_info(&self, model: &str) -> Value {
995 let snapshot = crate::domain::ProviderCapabilitySnapshot::from_model_id(model);
996 for (key, value, confidence) in [
997 (
998 "supports_tools",
999 snapshot.supports_tools.to_string(),
1000 "static",
1001 ),
1002 (
1003 "supports_vision",
1004 snapshot.supports_vision.to_string(),
1005 "static",
1006 ),
1007 ("reasoning", snapshot.reasoning.clone(), "static"),
1008 (
1009 "max_context_tokens",
1010 snapshot
1011 .max_context_tokens
1012 .map(|value| value.to_string())
1013 .unwrap_or_else(|| "unknown".to_string()),
1014 "static",
1015 ),
1016 ] {
1017 let _ = self.store.provider_probes().upsert(NewProviderProbe {
1018 provider: snapshot.provider.clone(),
1019 model_id: snapshot.model.clone(),
1020 capability_key: key.to_string(),
1021 capability_value: value,
1022 confidence: confidence.to_string(),
1023 error: None,
1024 });
1025 }
1026 if let Some(profile) = crate::models::lookup_provider(&snapshot.provider) {
1027 for (key, value) in [
1028 (
1029 "max_output_tokens_param",
1030 format!("{:?}", profile.max_tokens_param),
1031 ),
1032 (
1033 "parallel_tool_calls",
1034 (!profile
1035 .disable_parallel_tool_calls_for
1036 .iter()
1037 .any(|disabled| *disabled == snapshot.model))
1038 .to_string(),
1039 ),
1040 (
1041 "reasoning_parameter_shape",
1042 format!("{:?}", profile.reasoning_strategy),
1043 ),
1044 (
1045 "streaming_usage_available",
1046 "provider_dependent".to_string(),
1047 ),
1048 ("token_usage_field_shape", "openai_compatible".to_string()),
1049 ] {
1050 let _ = self.store.provider_probes().upsert(NewProviderProbe {
1051 provider: snapshot.provider.clone(),
1052 model_id: snapshot.model.clone(),
1053 capability_key: key.to_string(),
1054 capability_value: value,
1055 confidence: "static".to_string(),
1056 error: None,
1057 });
1058 }
1059 }
1060 json!({
1061 "id": model,
1062 "provider": snapshot.provider,
1063 "model": snapshot.model,
1064 "supports_tools": snapshot.supports_tools,
1065 "supports_vision": snapshot.supports_vision,
1066 "reasoning": snapshot.reasoning,
1067 "max_context_tokens": snapshot.max_context_tokens,
1068 })
1069 }
1070
1071 pub fn approval_decision(result: ApprovalReplayResult) -> Result<RuntimeApprovalDecision> {
1072 Ok(RuntimeApprovalDecision {
1073 ok: true,
1074 approval: result.approval,
1075 replayed: result.replayed,
1076 summary: result.summary,
1077 })
1078 }
1079
1080 fn raw_hygiene_preview(&self) -> Result<RawRuntimeHygienePreview> {
1081 let checkpoints = self
1082 .store
1083 .checkpoints()
1084 .list_all(10_000)?
1085 .into_iter()
1086 .filter(|checkpoint| checkpoint.archived_at.is_none())
1087 .collect::<Vec<_>>();
1088 let test_checkpoint_ids = checkpoints
1089 .iter()
1090 .filter(|checkpoint| is_runtime_hygiene_checkpoint(checkpoint))
1091 .map(|checkpoint| checkpoint.id.clone())
1092 .collect::<HashSet<_>>();
1093 let approvals = self
1094 .store
1095 .approvals()
1096 .list_all(10_000)?
1097 .into_iter()
1098 .filter(|approval| approval.archived_at.is_none())
1099 .filter(|approval| is_runtime_hygiene_approval(approval, &test_checkpoint_ids))
1100 .collect::<Vec<_>>();
1101 let checkpoints = checkpoints
1102 .into_iter()
1103 .filter(is_runtime_hygiene_checkpoint)
1104 .collect::<Vec<_>>();
1105
1106 Ok(RawRuntimeHygienePreview {
1107 approvals,
1108 checkpoints,
1109 })
1110 }
1111}
1112
1113#[derive(Debug)]
1114struct RawRuntimeHygienePreview {
1115 approvals: Vec<ApprovalRecord>,
1116 checkpoints: Vec<CheckpointRecord>,
1117}
1118
1119pub fn runtime_hygiene_reason() -> &'static str {
1120 "runtime hygiene: test/dev artifact"
1121}
1122
1123pub fn parse_optional_json(raw: Option<&str>) -> Value {
1124 raw.and_then(|value| serde_json::from_str(value).ok())
1125 .unwrap_or(Value::Null)
1126}
1127
1128pub fn affected_paths_from_json(pending_action: &Value, changed_files: &Value) -> Vec<String> {
1129 let mut paths = Vec::new();
1130 collect_path_like_strings(changed_files, &mut paths);
1131 collect_path_like_strings(pending_action, &mut paths);
1132 paths.sort();
1133 paths.dedup();
1134 paths.truncate(25);
1135 paths
1136}
1137
1138fn collect_path_like_strings(value: &Value, paths: &mut Vec<String>) {
1139 match value {
1140 Value::String(value) => {
1141 if looks_path_like(value) {
1142 paths.push(value.clone());
1143 }
1144 },
1145 Value::Array(items) => {
1146 for item in items {
1147 collect_path_like_strings(item, paths);
1148 }
1149 },
1150 Value::Object(object) => {
1151 for (key, value) in object {
1152 match value {
1153 Value::String(text)
1154 if matches!(
1155 key.as_str(),
1156 "path"
1157 | "file"
1158 | "file_path"
1159 | "target"
1160 | "project_path"
1161 | "snapshot_path"
1162 | "cwd"
1163 ) =>
1164 {
1165 if !text.trim().is_empty() {
1166 paths.push(text.clone());
1167 }
1168 },
1169 other => collect_path_like_strings(other, paths),
1170 }
1171 }
1172 },
1173 _ => {},
1174 }
1175}
1176
1177fn looks_path_like(value: &str) -> bool {
1178 let trimmed = value.trim();
1179 trimmed.starts_with('/')
1180 || trimmed.starts_with("./")
1181 || trimmed.starts_with("../")
1182 || trimmed.contains(std::path::MAIN_SEPARATOR)
1183}
1184
1185fn is_runtime_hygiene_checkpoint(checkpoint: &CheckpointRecord) -> bool {
1186 checkpoint.project_path.starts_with("/tmp/mermaid_")
1187}
1188
1189fn is_runtime_hygiene_approval(
1190 approval: &ApprovalRecord,
1191 test_checkpoint_ids: &HashSet<String>,
1192) -> bool {
1193 let is_restore_replay = approval.risk_classification == "restored_action"
1194 && approval.proposed_action.starts_with("restore replay:");
1195 if !is_restore_replay {
1196 return false;
1197 }
1198 match approval.checkpoint_id.as_deref() {
1199 Some(checkpoint_id) => test_checkpoint_ids.contains(checkpoint_id),
1200 None => true,
1201 }
1202}
1203
1204fn kill_pid(pid: u32) -> Result<()> {
1205 if cfg!(windows) {
1206 Command::new("taskkill")
1207 .args(["/PID", &pid.to_string(), "/T", "/F"])
1208 .status()?;
1209 } else {
1210 Command::new("kill").arg(pid.to_string()).status()?;
1211 }
1212 Ok(())
1213}
1214
1215#[cfg(test)]
1216mod tests {
1217 use super::*;
1218 use std::path::PathBuf;
1219
1220 fn temp_db(name: &str) -> PathBuf {
1221 let dir = std::env::temp_dir().join(format!("mermaid_runtime_client_{name}"));
1222 let _ = std::fs::remove_dir_all(&dir);
1223 std::fs::create_dir_all(&dir).expect("create temp dir");
1224 dir.join("runtime.sqlite3")
1225 }
1226
1227 #[test]
1228 fn local_client_lists_tasks_from_store() {
1229 let path = temp_db("tasks");
1230 let service = RuntimeService::from_store(RuntimeStore::open(&path).expect("open store"));
1231 let task = service
1232 .store
1233 .tasks()
1234 .create(super::super::NewTask::new(
1235 "contract task",
1236 "/tmp/mermaid_runtime_client",
1237 "openai/test",
1238 ))
1239 .expect("task");
1240 let tasks = service.list_tasks(10).expect("list");
1241 assert_eq!(
1242 tasks.first().map(|item| item.id.as_str()),
1243 Some(task.id.as_str())
1244 );
1245 let _ = std::fs::remove_dir_all(path.parent().unwrap());
1246 }
1247
1248 #[test]
1249 fn hygiene_preview_matches_test_artifacts_and_archive_is_idempotent() {
1250 let path = temp_db("hygiene");
1251 let store = RuntimeStore::open(&path).expect("open store");
1252 let checkpoint = store
1253 .checkpoints()
1254 .create(super::super::NewCheckpoint {
1255 id: Some("checkpoint-test".to_string()),
1256 task_id: None,
1257 project_path: "/tmp/mermaid_checkpoint_test".to_string(),
1258 snapshot_path: "/data/checkpoints/checkpoint-test".to_string(),
1259 changed_files_json: "[]".to_string(),
1260 pending_action_json: Some("{\"tool\":\"write_file\"}".to_string()),
1261 approval_id: None,
1262 })
1263 .expect("create checkpoint");
1264 let approval = store
1265 .approvals()
1266 .create(super::super::NewApproval {
1267 task_id: None,
1268 proposed_action: "restore replay: write_file".to_string(),
1269 risk_classification: "restored_action".to_string(),
1270 policy_decision: "ask".to_string(),
1271 args_summary: None,
1272 checkpoint_id: Some(checkpoint.id.clone()),
1273 pending_action_json: Some("{\"tool\":\"write_file\"}".to_string()),
1274 })
1275 .expect("create approval");
1276 store
1277 .checkpoints()
1278 .set_approval(&checkpoint.id, &approval.id)
1279 .expect("link approval");
1280
1281 let service = RuntimeService::from_store(store);
1282 let preview = service.hygiene_preview().expect("preview");
1283 assert_eq!(preview.counts.approvals, 1);
1284 assert_eq!(preview.counts.checkpoints, 1);
1285 let archived = service.hygiene_archive().expect("archive");
1286 assert_eq!(archived.archived.total, 2);
1287 let archived_again = service.hygiene_archive().expect("archive again");
1288 assert_eq!(archived_again.archived.total, 0);
1289 let _ = std::fs::remove_dir_all(path.parent().unwrap());
1290 }
1291}