Skip to main content

mermaid_cli/runtime/
client.rs

1use std::collections::HashSet;
2#[cfg(unix)]
3use std::os::unix::process::CommandExt;
4use std::process::{Command, Stdio};
5
6use anyhow::{Context, Result};
7use serde::{Deserialize, Serialize, de::DeserializeOwned};
8use serde_json::{Value, json};
9
10use super::{
11    ApprovalRecord, ApprovalReplayResult, CheckpointManifest, CheckpointRecord, CompactionRecord,
12    MessageRecord, NewProcess, NewProviderProbe, PairingTokenRecord, PluginInstallRecord,
13    ProcessRecord, ProcessStatus, ProviderProbeRecord, RuntimeStore, SessionRecord, TaskRecord,
14    TaskStatus, TaskTimelineEvent, ToolRunRecord, approve_and_replay, deny_approval,
15    request_daemon_json, restore_checkpoint,
16};
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
19#[serde(rename_all = "snake_case")]
20pub enum RuntimeClientSource {
21    Daemon,
22    Local,
23}
24
25impl RuntimeClientSource {
26    pub fn as_str(self) -> &'static str {
27        match self {
28            RuntimeClientSource::Daemon => "daemon",
29            RuntimeClientSource::Local => "local",
30        }
31    }
32}
33
34#[derive(Debug, Clone)]
35enum RuntimeClientMode {
36    PreferDaemon,
37    DaemonOnly,
38    LocalOnly,
39}
40
41#[derive(Debug, Clone)]
42pub struct RuntimeClient {
43    mode: RuntimeClientMode,
44    auth_token: Option<String>,
45}
46
47#[derive(Debug, Clone, Serialize, Deserialize)]
48pub struct RuntimeRead<T> {
49    pub source: RuntimeClientSource,
50    pub value: T,
51}
52
53#[derive(Debug, Clone, Serialize, Deserialize)]
54pub struct RuntimeHealth {
55    pub ok: bool,
56    pub service: String,
57    pub database: String,
58}
59
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct RuntimeSnapshot {
62    pub ok: bool,
63    pub database: String,
64    pub sessions: Vec<SessionRecord>,
65    pub tasks: Vec<TaskRecord>,
66    pub tool_runs: Vec<ToolRunRecord>,
67    pub processes: Vec<ProcessRecord>,
68    pub approvals: Vec<ApprovalRecord>,
69    pub checkpoints: Vec<CheckpointRecord>,
70    pub compactions: Vec<CompactionRecord>,
71    pub plugins: Vec<PluginInstallRecord>,
72    pub provider_probes: Vec<ProviderProbeRecord>,
73    pub pairings: Vec<PairingTokenRecord>,
74    pub safety: crate::app::SafetyConfig,
75}
76
77#[derive(Debug, Clone, Serialize, Deserialize)]
78pub struct RuntimeDashboardCounts {
79    pub pending_approvals: usize,
80    pub running_tasks: usize,
81    pub waiting_tasks: usize,
82    pub blocked_tasks: usize,
83    pub ready_processes: usize,
84    pub recent_checkpoints: usize,
85    pub installed_plugins: usize,
86    pub archived_approvals: usize,
87    pub archived_checkpoints: usize,
88}
89
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct RuntimeDashboard {
92    pub ok: bool,
93    pub health: RuntimeHealth,
94    pub safety: crate::app::SafetyConfig,
95    pub counts: RuntimeDashboardCounts,
96    pub sessions: Vec<SessionRecord>,
97    pub tasks: Vec<TaskRecord>,
98    pub tool_runs: Vec<ToolRunRecord>,
99    pub processes: Vec<ProcessRecord>,
100    pub approvals: Vec<ApprovalRecord>,
101    pub checkpoints: Vec<CheckpointRecord>,
102    pub compactions: Vec<CompactionRecord>,
103    pub plugins: Vec<PluginInstallRecord>,
104    pub provider_probes: Vec<ProviderProbeRecord>,
105    pub pairings: Vec<PairingTokenRecord>,
106}
107
108#[derive(Debug, Clone, Serialize, Deserialize)]
109pub struct RuntimeDiagnostics {
110    #[serde(flatten)]
111    pub snapshot: RuntimeSnapshot,
112    pub mode: String,
113    pub hygiene: RuntimeDiagnosticsHygiene,
114}
115
116#[derive(Debug, Clone, Serialize, Deserialize)]
117pub struct RuntimeDiagnosticsHygiene {
118    pub preview: Value,
119    pub visible: RuntimeDiagnosticsVisible,
120    pub archived: RuntimeDiagnosticsArchived,
121}
122
123#[derive(Debug, Clone, Serialize, Deserialize)]
124pub struct RuntimeDiagnosticsVisible {
125    pub approvals: usize,
126    pub checkpoints: usize,
127}
128
129#[derive(Debug, Clone, Serialize, Deserialize)]
130pub struct RuntimeDiagnosticsArchived {
131    pub approvals: usize,
132    pub checkpoints: usize,
133}
134
135#[derive(Debug, Clone, Serialize, Deserialize)]
136pub struct RuntimeHygienePreview {
137    pub ok: bool,
138    pub reason: String,
139    pub approvals: Vec<ApprovalRecord>,
140    pub checkpoints: Vec<CheckpointRecord>,
141    pub counts: RuntimeHygieneCounts,
142}
143
144#[derive(Debug, Clone, Serialize, Deserialize)]
145pub struct RuntimeHygieneCounts {
146    pub approvals: usize,
147    pub checkpoints: usize,
148    pub total: usize,
149}
150
151#[derive(Debug, Clone, Serialize, Deserialize)]
152pub struct RuntimeHygieneArchive {
153    pub ok: bool,
154    pub reason: String,
155    pub archived: RuntimeHygieneCounts,
156    pub matched: RuntimeHygieneCounts,
157}
158
159#[derive(Debug, Clone, Serialize, Deserialize)]
160pub struct RuntimeTaskDetail {
161    pub ok: bool,
162    pub task: TaskRecord,
163    pub events: Vec<TaskTimelineEvent>,
164    pub session: Option<SessionRecord>,
165    pub messages: Vec<MessageRecord>,
166    pub approvals: Vec<ApprovalRecord>,
167    pub checkpoints: Vec<CheckpointRecord>,
168    pub processes: Vec<ProcessRecord>,
169    pub tool_runs: Vec<ToolRunRecord>,
170    pub compactions: Vec<CompactionRecord>,
171}
172
173#[derive(Debug, Clone, Serialize, Deserialize)]
174pub struct RuntimeApprovalDetail {
175    pub ok: bool,
176    pub approval: ApprovalRecord,
177    pub task: Option<TaskRecord>,
178    pub checkpoint: Option<CheckpointRecord>,
179    pub pending_action: Value,
180    pub args: Value,
181    pub changed_files: Value,
182    pub affected_paths: Vec<String>,
183}
184
185#[derive(Debug, Clone, Serialize, Deserialize)]
186pub struct RuntimeCheckpointDetail {
187    pub ok: bool,
188    pub checkpoint: CheckpointRecord,
189    pub task: Option<TaskRecord>,
190    pub approval: Option<ApprovalRecord>,
191    pub pending_action: Value,
192    pub changed_files: Value,
193    pub affected_paths: Vec<String>,
194}
195
196#[derive(Debug, Clone, Serialize, Deserialize)]
197pub struct RuntimeProcessLog {
198    pub ok: bool,
199    pub content: String,
200}
201
202#[derive(Debug, Clone, Serialize, Deserialize)]
203pub struct RuntimeProcessOpen {
204    pub ok: bool,
205    pub target: String,
206}
207
208#[derive(Debug, Clone, Serialize, Deserialize)]
209pub struct RuntimePorts {
210    pub ok: bool,
211    pub ports: String,
212}
213
214#[derive(Debug, Clone, Serialize, Deserialize)]
215pub struct RuntimeApprovalDecision {
216    pub ok: bool,
217    pub approval: Option<ApprovalRecord>,
218    pub replayed: bool,
219    pub summary: String,
220}
221
222#[derive(Debug, Clone, Serialize, Deserialize)]
223pub struct RuntimeCheckpointRestore {
224    pub ok: bool,
225    pub checkpoint: CheckpointManifest,
226}
227
228#[derive(Debug, Clone, Serialize, Deserialize)]
229struct RuntimeItems<T> {
230    ok: bool,
231    items: Vec<T>,
232}
233
234#[derive(Debug, Clone, Serialize, Deserialize)]
235pub struct RuntimeOne<T> {
236    pub ok: bool,
237    pub item: T,
238}
239
240pub struct RuntimeService {
241    store: RuntimeStore,
242}
243
244impl RuntimeClient {
245    pub fn auto() -> Self {
246        Self {
247            mode: RuntimeClientMode::PreferDaemon,
248            auth_token: None,
249        }
250    }
251
252    pub fn daemon() -> Self {
253        Self {
254            mode: RuntimeClientMode::DaemonOnly,
255            auth_token: None,
256        }
257    }
258
259    pub fn daemon_with_token(token: impl Into<String>) -> Self {
260        Self {
261            mode: RuntimeClientMode::DaemonOnly,
262            auth_token: Some(token.into()),
263        }
264    }
265
266    pub fn local() -> Self {
267        Self {
268            mode: RuntimeClientMode::LocalOnly,
269            auth_token: None,
270        }
271    }
272
273    pub fn health(&self) -> Result<RuntimeRead<RuntimeHealth>> {
274        self.read(json!({"command": "health"}), |service| service.health())
275    }
276
277    pub fn snapshot(&self) -> Result<RuntimeRead<RuntimeSnapshot>> {
278        self.read(json!({"command": "runtime_snapshot"}), |service| {
279            service.snapshot()
280        })
281    }
282
283    pub fn dashboard(&self) -> Result<RuntimeRead<RuntimeDashboard>> {
284        self.read(json!({"command": "runtime_dashboard"}), |service| {
285            service.dashboard()
286        })
287    }
288
289    pub fn diagnostics(&self) -> Result<RuntimeRead<RuntimeDiagnostics>> {
290        self.read(json!({"command": "runtime_diagnostics"}), |service| {
291            service.diagnostics()
292        })
293    }
294
295    pub fn hygiene_preview(&self) -> Result<RuntimeRead<RuntimeHygienePreview>> {
296        self.read(json!({"command": "runtime_hygiene_preview"}), |service| {
297            service.hygiene_preview()
298        })
299    }
300
301    pub fn hygiene_archive(&self) -> Result<RuntimeRead<RuntimeHygieneArchive>> {
302        self.read_inner(
303            json!({"command": "runtime_hygiene_archive"}),
304            true,
305            |service| service.hygiene_archive(),
306        )
307    }
308
309    pub fn task_detail(&self, id: &str) -> Result<RuntimeRead<RuntimeTaskDetail>> {
310        self.read(
311            json!({"command": "runtime_task_detail", "id": id}),
312            |service| service.task_detail(id),
313        )
314    }
315
316    pub fn approval_detail(&self, id: &str) -> Result<RuntimeRead<RuntimeApprovalDetail>> {
317        self.read(
318            json!({"command": "runtime_approval_detail", "id": id}),
319            |service| service.approval_detail(id),
320        )
321    }
322
323    pub fn checkpoint_detail(&self, id: &str) -> Result<RuntimeRead<RuntimeCheckpointDetail>> {
324        self.read(
325            json!({"command": "runtime_checkpoint_detail", "id": id}),
326            |service| service.checkpoint_detail(id),
327        )
328    }
329
330    pub fn list_tasks(&self, limit: usize) -> Result<RuntimeRead<Vec<TaskRecord>>> {
331        self.list(
332            json!({"command": "runtime_tasks", "limit": limit}),
333            |service| service.list_tasks(limit),
334        )
335    }
336
337    pub fn list_processes(&self, limit: usize) -> Result<RuntimeRead<Vec<ProcessRecord>>> {
338        self.list(
339            json!({"command": "runtime_processes", "limit": limit}),
340            |service| service.list_processes(limit),
341        )
342    }
343
344    pub fn list_approvals(&self) -> Result<RuntimeRead<Vec<ApprovalRecord>>> {
345        self.list(json!({"command": "runtime_approvals"}), |service| {
346            service.list_approvals()
347        })
348    }
349
350    pub fn list_tool_runs(&self, limit: usize) -> Result<RuntimeRead<Vec<ToolRunRecord>>> {
351        self.list(
352            json!({"command": "runtime_tool_runs", "limit": limit}),
353            |service| service.list_tool_runs(limit),
354        )
355    }
356
357    pub fn list_checkpoints(&self, limit: usize) -> Result<RuntimeRead<Vec<CheckpointRecord>>> {
358        self.list(
359            json!({"command": "runtime_checkpoints", "limit": limit}),
360            |service| service.list_checkpoints(limit),
361        )
362    }
363
364    pub fn list_plugins(&self) -> Result<RuntimeRead<Vec<PluginInstallRecord>>> {
365        self.list(json!({"command": "runtime_plugins"}), |service| {
366            service.list_plugins()
367        })
368    }
369
370    pub fn process_log(&self, id: &str, tail_bytes: Option<u64>) -> Result<RuntimeProcessLog> {
371        self.action(
372            json!({
373                "command": "logs",
374                "id": id,
375                "tail_bytes": tail_bytes,
376            }),
377            |service| service.process_log(id, tail_bytes),
378        )
379    }
380
381    pub fn stop_process(&self, id: &str) -> Result<RuntimeOne<ProcessRecord>> {
382        // Non-idempotent: `terminate_tree` must not fire twice (RC-G/F25).
383        self.action_authed_non_idempotent(json!({"command": "stop_process", "id": id}), |service| {
384            service
385                .stop_process(id)
386                .map(|item| RuntimeOne { ok: true, item })
387        })
388    }
389
390    pub fn restart_process(&self, id: &str) -> Result<RuntimeOne<ProcessRecord>> {
391        // Non-idempotent: kills then respawns — a duplicate run double-signals
392        // (possibly a since-reused PID) and can spawn two servers (RC-G/F25).
393        self.action_authed_non_idempotent(
394            json!({"command": "restart_process", "id": id}),
395            |service| {
396                service
397                    .restart_process(id)
398                    .map(|item| RuntimeOne { ok: true, item })
399            },
400        )
401    }
402
403    pub fn open_process(&self, id: &str) -> Result<RuntimeProcessOpen> {
404        self.action_authed(json!({"command": "open_process", "id": id}), |service| {
405            service.open_process(id)
406        })
407    }
408
409    pub fn ports(&self) -> Result<RuntimePorts> {
410        self.action(json!({"command": "ports"}), |service| service.ports())
411    }
412
413    pub fn approve(&self, id: &str) -> Result<RuntimeApprovalDecision> {
414        // Non-idempotent: replays the approved action, so a duplicate run could
415        // execute that side effect twice (RC-G/F25).
416        self.action_authed_non_idempotent(json!({"command": "approve", "id": id}), |_service| {
417            RuntimeService::approval_decision(approve_and_replay(id)?)
418        })
419    }
420
421    pub fn deny(&self, id: &str) -> Result<RuntimeApprovalDecision> {
422        self.action_authed(json!({"command": "deny", "id": id}), |_service| {
423            RuntimeService::approval_decision(deny_approval(id)?)
424        })
425    }
426
427    pub fn restore_checkpoint(&self, id: &str) -> Result<RuntimeCheckpointRestore> {
428        // Non-idempotent: rewrites working-tree files from the snapshot — a
429        // duplicate run could clobber edits made between the two runs (RC-G/F25).
430        self.action_authed_non_idempotent(
431            json!({"command": "restore_checkpoint", "id": id}),
432            |_service| {
433                Ok(RuntimeCheckpointRestore {
434                    ok: true,
435                    checkpoint: restore_checkpoint(id)?,
436                })
437            },
438        )
439    }
440
441    pub fn set_plugin_enabled(&self, id: &str, enabled: bool) -> Result<()> {
442        self.action_authed::<Value, _>(
443            json!({"command": "set_plugin_enabled", "id": id, "enabled": enabled}),
444            |service| {
445                service.set_plugin_enabled(id, enabled)?;
446                Ok(json!({"ok": true}))
447            },
448        )?;
449        Ok(())
450    }
451
452    pub fn model_info(&self, model: &str) -> Result<Value> {
453        self.action(
454            json!({"command": "model_info", "model": model}),
455            |service| Ok(json!({"ok": true, "model": service.model_info(model)})),
456        )
457    }
458
459    pub fn set_safety_mode(&self, mode: &str) -> Result<Value> {
460        self.action_authed(
461            json!({"command": "set_safety_mode", "mode": mode}),
462            |service| {
463                let safety = service.set_safety_mode(mode)?;
464                Ok(json!({"ok": true, "safety": safety}))
465            },
466        )
467    }
468
469    fn list<T, F>(&self, body: Value, local: F) -> Result<RuntimeRead<Vec<T>>>
470    where
471        T: DeserializeOwned,
472        F: FnOnce(&RuntimeService) -> Result<Vec<T>>,
473    {
474        self.read(body, |service| {
475            local(service).map(|items| RuntimeItems { ok: true, items })
476        })
477        .map(|read| RuntimeRead {
478            source: read.source,
479            value: read.value.items,
480        })
481    }
482
483    fn read<T, F>(&self, body: Value, local: F) -> Result<RuntimeRead<T>>
484    where
485        T: DeserializeOwned,
486        F: FnOnce(&RuntimeService) -> Result<T>,
487    {
488        // #21: attach the pairing token when the client has one. Reads are now
489        // gated server-side (`command_requires_auth`), so a `daemon_with_token`
490        // client must send its token to read off the socket; a tokenless
491        // `auto()` client sends nothing and `PreferDaemon` falls back to a local
492        // DB read on the daemon's auth rejection. `request_daemon` only attaches
493        // the token if one is present, so tokenless clients are unchanged, and
494        // ungated commands (`health`/`ports`) are served regardless.
495        self.read_inner(body, true, local)
496    }
497
498    fn action<T, F>(&self, body: Value, local: F) -> Result<T>
499    where
500        T: DeserializeOwned,
501        F: FnOnce(&RuntimeService) -> Result<T>,
502    {
503        self.action_inner(body, false, true, local)
504    }
505
506    fn action_authed<T, F>(&self, body: Value, local: F) -> Result<T>
507    where
508        T: DeserializeOwned,
509        F: FnOnce(&RuntimeService) -> Result<T>,
510    {
511        self.action_inner(body, true, true, local)
512    }
513
514    /// Like [`Self::action_authed`], but for a NON-idempotent side effect
515    /// (`stop`/`restart`/`approve`/`restore`). RC-G/F25: under `PreferDaemon` a
516    /// local fallback *re-runs* the action, which is only safe when the daemon
517    /// never received the request (a pre-send connection failure). On a
518    /// post-send or ambiguous failure the daemon MAY have already executed it,
519    /// so we surface an error instead of risking a duplicate side effect.
520    fn action_authed_non_idempotent<T, F>(&self, body: Value, local: F) -> Result<T>
521    where
522        T: DeserializeOwned,
523        F: FnOnce(&RuntimeService) -> Result<T>,
524    {
525        self.action_inner(body, true, false, local)
526    }
527
528    fn read_inner<T, F>(&self, body: Value, authed: bool, local: F) -> Result<RuntimeRead<T>>
529    where
530        T: DeserializeOwned,
531        F: FnOnce(&RuntimeService) -> Result<T>,
532    {
533        match self.mode {
534            RuntimeClientMode::LocalOnly => RuntimeService::open_default()
535                .and_then(|service| local(&service))
536                .map(|value| RuntimeRead {
537                    source: RuntimeClientSource::Local,
538                    value,
539                }),
540            RuntimeClientMode::DaemonOnly => {
541                self.request_daemon(body, authed).map(|value| RuntimeRead {
542                    source: RuntimeClientSource::Daemon,
543                    value,
544                })
545            },
546            RuntimeClientMode::PreferDaemon => match self.request_daemon(body, authed) {
547                Ok(value) => Ok(RuntimeRead {
548                    source: RuntimeClientSource::Daemon,
549                    value,
550                }),
551                Err(_) => RuntimeService::open_default()
552                    .and_then(|service| local(&service))
553                    .map(|value| RuntimeRead {
554                        source: RuntimeClientSource::Local,
555                        value,
556                    }),
557            },
558        }
559    }
560
561    fn action_inner<T, F>(&self, body: Value, authed: bool, idempotent: bool, local: F) -> Result<T>
562    where
563        T: DeserializeOwned,
564        F: FnOnce(&RuntimeService) -> Result<T>,
565    {
566        match self.mode {
567            RuntimeClientMode::LocalOnly => {
568                RuntimeService::open_default().and_then(|service| local(&service))
569            },
570            RuntimeClientMode::DaemonOnly => self.request_daemon(body, authed),
571            RuntimeClientMode::PreferDaemon => match self.request_daemon(body, authed) {
572                Ok(value) => Ok(value),
573                Err(err) => {
574                    // RC-G/F25: an idempotent or read-only action falls back to
575                    // a local run freely. A non-idempotent one falls back ONLY
576                    // when the failure is a pre-send connection failure (the
577                    // request never reached the daemon); otherwise the daemon
578                    // may have already executed it and re-running locally would
579                    // double-fire the side effect (e.g. a second
580                    // `terminate_tree` on a since-reused PID). Conservative
581                    // default: treat any non-connect failure as "may have run".
582                    if idempotent || daemon_failure_is_pre_send(&err) {
583                        RuntimeService::open_default().and_then(|service| local(&service))
584                    } else {
585                        Err(err.context(
586                            "daemon request failed after the action may have already run; \
587                             not re-running it locally to avoid a duplicate side effect — \
588                             retry once the daemon is reachable",
589                        ))
590                    }
591                },
592            },
593        }
594    }
595
596    fn request_daemon<T: DeserializeOwned>(&self, mut body: Value, authed: bool) -> Result<T> {
597        if authed && let Some(token) = self.auth_token.as_ref() {
598            body["auth"] = json!({ "token": token });
599        }
600        let value = request_daemon_json(body)?;
601        serde_json::from_value(value).context("daemon response had unexpected shape")
602    }
603}
604
605impl RuntimeService {
606    pub fn open_default() -> Result<Self> {
607        Ok(Self {
608            store: RuntimeStore::open_default()?,
609        })
610    }
611
612    pub fn from_store(store: RuntimeStore) -> Self {
613        Self { store }
614    }
615
616    pub fn health(&self) -> Result<RuntimeHealth> {
617        Ok(RuntimeHealth {
618            ok: true,
619            service: "mermaidd".to_string(),
620            database: self.store.path().display().to_string(),
621        })
622    }
623
624    pub fn snapshot(&self) -> Result<RuntimeSnapshot> {
625        Ok(RuntimeSnapshot {
626            ok: true,
627            database: self.store.path().display().to_string(),
628            sessions: self.store.sessions().list(100)?,
629            tasks: self.store.tasks().list(100)?,
630            tool_runs: self.store.tool_runs().list(100)?,
631            processes: self.store.processes().list(100)?,
632            approvals: self.store.approvals().list_all(100)?,
633            checkpoints: self.store.checkpoints().list_all(100)?,
634            compactions: self.store.compactions().list(100)?,
635            plugins: self.store.plugins().list()?,
636            provider_probes: self.store.provider_probes().list(None, None)?,
637            // Redact token hashes — this snapshot is served over the local
638            // socket to same-UID processes without auth.
639            pairings: self.store.pairing_tokens().list_redacted()?,
640            safety: crate::app::load_config().unwrap_or_default().safety,
641        })
642    }
643
644    pub fn dashboard(&self) -> Result<RuntimeDashboard> {
645        let sessions = self.store.sessions().list(100)?;
646        let tasks = self.store.tasks().list(200)?;
647        let tool_runs = self.store.tool_runs().list(100)?;
648        let processes = self.store.processes().list(100)?;
649        let approvals = self.store.approvals().list_pending()?;
650        let checkpoints = self.store.checkpoints().list(100)?;
651        let compactions = self.store.compactions().list(100)?;
652        let plugins = self.store.plugins().list()?;
653        let provider_probes = self.store.provider_probes().list(None, None)?;
654        let pairings = self.store.pairing_tokens().list_redacted()?;
655
656        let running_tasks = tasks
657            .iter()
658            .filter(|task| task.status == TaskStatus::Running)
659            .count();
660        let waiting_tasks = tasks
661            .iter()
662            .filter(|task| task.status == TaskStatus::WaitingForApproval)
663            .count();
664        let blocked_tasks = tasks
665            .iter()
666            .filter(|task| matches!(task.status, TaskStatus::Blocked | TaskStatus::Failed))
667            .count();
668        let ready_processes = processes
669            .iter()
670            .filter(|process| process.detected_url.is_some())
671            .count();
672
673        Ok(RuntimeDashboard {
674            ok: true,
675            health: self.health()?,
676            safety: crate::app::load_config().unwrap_or_default().safety,
677            counts: RuntimeDashboardCounts {
678                pending_approvals: approvals.len(),
679                running_tasks,
680                waiting_tasks,
681                blocked_tasks,
682                ready_processes,
683                recent_checkpoints: checkpoints.len(),
684                installed_plugins: plugins.len(),
685                archived_approvals: self.store.approvals().count_archived()?,
686                archived_checkpoints: self.store.checkpoints().count_archived()?,
687            },
688            sessions,
689            tasks,
690            tool_runs,
691            processes,
692            approvals,
693            checkpoints,
694            compactions,
695            plugins,
696            provider_probes,
697            pairings,
698        })
699    }
700
701    pub fn diagnostics(&self) -> Result<RuntimeDiagnostics> {
702        let snapshot = self.snapshot()?;
703        let hygiene = self.hygiene_preview()?;
704        Ok(RuntimeDiagnostics {
705            snapshot,
706            mode: "diagnostics".to_string(),
707            hygiene: RuntimeDiagnosticsHygiene {
708                preview: serde_json::to_value(hygiene)?,
709                visible: RuntimeDiagnosticsVisible {
710                    approvals: self.store.approvals().list_pending()?.len(),
711                    checkpoints: self.store.checkpoints().list(100)?.len(),
712                },
713                archived: RuntimeDiagnosticsArchived {
714                    approvals: self.store.approvals().count_archived()?,
715                    checkpoints: self.store.checkpoints().count_archived()?,
716                },
717            },
718        })
719    }
720
721    pub fn hygiene_preview(&self) -> Result<RuntimeHygienePreview> {
722        let preview = self.raw_hygiene_preview()?;
723        Ok(RuntimeHygienePreview {
724            ok: true,
725            reason: runtime_hygiene_reason().to_string(),
726            counts: RuntimeHygieneCounts {
727                approvals: preview.approvals.len(),
728                checkpoints: preview.checkpoints.len(),
729                total: preview.approvals.len() + preview.checkpoints.len(),
730            },
731            approvals: preview.approvals,
732            checkpoints: preview.checkpoints,
733        })
734    }
735
736    pub fn hygiene_archive(&self) -> Result<RuntimeHygieneArchive> {
737        let preview = self.raw_hygiene_preview()?;
738        let approval_ids = preview
739            .approvals
740            .iter()
741            .map(|approval| approval.id.clone())
742            .collect::<Vec<_>>();
743        let checkpoint_ids = preview
744            .checkpoints
745            .iter()
746            .map(|checkpoint| checkpoint.id.clone())
747            .collect::<Vec<_>>();
748        let reason = runtime_hygiene_reason();
749        let approvals_archived = self.store.approvals().archive(&approval_ids, reason)?;
750        let checkpoints_archived = self.store.checkpoints().archive(&checkpoint_ids, reason)?;
751
752        Ok(RuntimeHygieneArchive {
753            ok: true,
754            reason: reason.to_string(),
755            archived: RuntimeHygieneCounts {
756                approvals: approvals_archived,
757                checkpoints: checkpoints_archived,
758                total: approvals_archived + checkpoints_archived,
759            },
760            matched: RuntimeHygieneCounts {
761                approvals: approval_ids.len(),
762                checkpoints: checkpoint_ids.len(),
763                total: approval_ids.len() + checkpoint_ids.len(),
764            },
765        })
766    }
767
768    pub fn task_detail(&self, id: &str) -> Result<RuntimeTaskDetail> {
769        let task = self
770            .store
771            .tasks()
772            .get(id)?
773            .with_context(|| format!("task not found: {}", id))?;
774        let events = self.store.tasks().events(id)?;
775        let session = match task.conversation_id.as_deref() {
776            Some(session_id) => self.store.sessions().get(session_id)?,
777            None => None,
778        };
779        let messages = match task.conversation_id.as_deref() {
780            Some(session_id) => self.store.messages().list_for_session(session_id)?,
781            None => Vec::new(),
782        };
783        let approvals = self
784            .store
785            .approvals()
786            .list_pending()?
787            .into_iter()
788            .filter(|approval| approval.task_id.as_deref() == Some(id))
789            .collect::<Vec<_>>();
790        let checkpoints = self
791            .store
792            .checkpoints()
793            .list(200)?
794            .into_iter()
795            .filter(|checkpoint| checkpoint.task_id.as_deref() == Some(id))
796            .collect::<Vec<_>>();
797        let processes = self
798            .store
799            .processes()
800            .list(200)?
801            .into_iter()
802            .filter(|process| process.task_id.as_deref() == Some(id))
803            .collect::<Vec<_>>();
804        let tool_runs = self
805            .store
806            .tool_runs()
807            .list(200)?
808            .into_iter()
809            .filter(|tool_run| tool_run.task_id.as_deref() == Some(id))
810            .collect::<Vec<_>>();
811        let compactions = self
812            .store
813            .compactions()
814            .list(200)?
815            .into_iter()
816            .filter(|compaction| compaction.task_id.as_deref() == Some(id))
817            .collect::<Vec<_>>();
818
819        Ok(RuntimeTaskDetail {
820            ok: true,
821            task,
822            events,
823            session,
824            messages,
825            approvals,
826            checkpoints,
827            processes,
828            tool_runs,
829            compactions,
830        })
831    }
832
833    pub fn approval_detail(&self, id: &str) -> Result<RuntimeApprovalDetail> {
834        let approval = self
835            .store
836            .approvals()
837            .get(id)?
838            .with_context(|| format!("approval not found: {}", id))?;
839        let checkpoint = match approval.checkpoint_id.as_deref() {
840            Some(checkpoint_id) => self.store.checkpoints().get(checkpoint_id)?,
841            None => None,
842        };
843        let task = match approval.task_id.as_deref() {
844            Some(task_id) => self.store.tasks().get(task_id)?,
845            None => None,
846        };
847        let pending_action = parse_optional_json(approval.pending_action_json.as_deref());
848        let args = parse_optional_json(approval.args_summary.as_deref());
849        let changed_files = checkpoint
850            .as_ref()
851            .map(|checkpoint| parse_optional_json(Some(&checkpoint.changed_files_json)))
852            .unwrap_or(Value::Null);
853        let affected_paths = affected_paths_from_json(&pending_action, &changed_files);
854
855        Ok(RuntimeApprovalDetail {
856            ok: true,
857            approval,
858            task,
859            checkpoint,
860            pending_action,
861            args,
862            changed_files,
863            affected_paths,
864        })
865    }
866
867    pub fn checkpoint_detail(&self, id: &str) -> Result<RuntimeCheckpointDetail> {
868        let checkpoint = self
869            .store
870            .checkpoints()
871            .get(id)?
872            .with_context(|| format!("checkpoint not found: {}", id))?;
873        let approval = match checkpoint.approval_id.as_deref() {
874            Some(approval_id) => self.store.approvals().get(approval_id)?,
875            None => None,
876        };
877        let task = match checkpoint.task_id.as_deref() {
878            Some(task_id) => self.store.tasks().get(task_id)?,
879            None => None,
880        };
881        let pending_action = parse_optional_json(checkpoint.pending_action_json.as_deref());
882        let changed_files = parse_optional_json(Some(&checkpoint.changed_files_json));
883        let affected_paths = affected_paths_from_json(&pending_action, &changed_files);
884
885        Ok(RuntimeCheckpointDetail {
886            ok: true,
887            checkpoint,
888            task,
889            approval,
890            pending_action,
891            changed_files,
892            affected_paths,
893        })
894    }
895
896    pub fn list_tasks(&self, limit: usize) -> Result<Vec<TaskRecord>> {
897        self.store.tasks().list(limit)
898    }
899
900    pub fn list_processes(&self, limit: usize) -> Result<Vec<ProcessRecord>> {
901        self.store.processes().list(limit)
902    }
903
904    pub fn list_approvals(&self) -> Result<Vec<ApprovalRecord>> {
905        self.store.approvals().list_pending()
906    }
907
908    pub fn list_tool_runs(&self, limit: usize) -> Result<Vec<ToolRunRecord>> {
909        self.store.tool_runs().list(limit)
910    }
911
912    pub fn list_checkpoints(&self, limit: usize) -> Result<Vec<CheckpointRecord>> {
913        self.store.checkpoints().list(limit)
914    }
915
916    pub fn list_plugins(&self) -> Result<Vec<PluginInstallRecord>> {
917        self.store.plugins().list()
918    }
919
920    pub fn process_log(&self, id: &str, tail_bytes: Option<u64>) -> Result<RuntimeProcessLog> {
921        let process = self
922            .store
923            .processes()
924            .get(id)?
925            .with_context(|| format!("process not found: {}", id))?;
926        let path = process
927            .log_path
928            .with_context(|| format!("process has no log path: {}", id))?;
929        let tail = tail_bytes.unwrap_or(32 * 1024).min(512 * 1024);
930        // Seek to the tail rather than reading the whole file: a long-running
931        // dev server can produce a multi-GB log, and we only ever return the
932        // last `tail` bytes. `std::fs::read` would have pinned the whole file
933        // in RAM first (#41).
934        use std::io::{Read, Seek, SeekFrom};
935        let mut file =
936            std::fs::File::open(&path).with_context(|| format!("failed to read {}", path))?;
937        let len = file.metadata().map(|m| m.len()).unwrap_or(0);
938        let start = len.saturating_sub(tail);
939        if start > 0 {
940            file.seek(SeekFrom::Start(start))
941                .with_context(|| format!("failed to seek {}", path))?;
942        }
943        let mut bytes = Vec::new();
944        file.take(tail)
945            .read_to_end(&mut bytes)
946            .with_context(|| format!("failed to read {}", path))?;
947        Ok(RuntimeProcessLog {
948            ok: true,
949            content: String::from_utf8_lossy(&bytes).into_owned(),
950        })
951    }
952
953    pub fn stop_process(&self, id: &str) -> Result<ProcessRecord> {
954        let process = self
955            .store
956            .processes()
957            .get(id)?
958            .with_context(|| format!("process not found: {}", id))?;
959        // Best-effort group kill (SIGTERM → grace → SIGKILL) so workers the dev
960        // server forked die with it; then mark the row stopped regardless.
961        crate::utils::terminate_tree_blocking(process.pid, crate::utils::Grace::Graceful);
962        self.store.processes().upsert(NewProcess {
963            id: Some(process.id.clone()),
964            task_id: process.task_id.clone(),
965            pid: process.pid,
966            command: process.command.clone(),
967            cwd: process.cwd.clone(),
968            log_path: process.log_path.clone(),
969            detected_url: process.detected_url.clone(),
970            status: ProcessStatus::Exited,
971            health: Some("stopped".to_string()),
972        })
973    }
974
975    pub fn restart_process(&self, id: &str) -> Result<ProcessRecord> {
976        let process = self
977            .store
978            .processes()
979            .get(id)?
980            .with_context(|| format!("process not found: {}", id))?;
981        // #63: the command comes from a `processes` row; a tampered DB could swap
982        // in a destructive command. Refuse to respawn it (mirrors exec.rs's
983        // execute_command pre-check) before killing or spawning anything.
984        anyhow::ensure!(
985            !crate::runtime::is_destructive_command(&process.command),
986            "refusing to restart process {id}: command flagged destructive: {:?}",
987            process.command
988        );
989        crate::utils::terminate_tree_blocking(process.pid, crate::utils::Grace::Graceful);
990        // Wait (bounded) for the old PID to actually exit before respawning —
991        // the kill above is fire-and-forget (async signal / taskkill), so an
992        // immediate respawn can race the old process for its listening port.
993        // Poll until it's gone, then fail-open after 5s.
994        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
995        while pid_alive(process.pid) {
996            if std::time::Instant::now() >= deadline {
997                tracing::warn!(
998                    pid = process.pid,
999                    "restart_process: old pid still alive after 5s; respawning anyway"
1000                );
1001                break;
1002            }
1003            std::thread::sleep(std::time::Duration::from_millis(100));
1004        }
1005        let mut command = Command::new("sh");
1006        command.arg("-c").arg(&process.command);
1007        // Lead a new process group so `/stop`/`/restart` can group-kill the
1008        // whole tree (the dev server plus any worker it forks), matching the
1009        // foreground exec path. Windows kills the tree by pid via taskkill /T.
1010        #[cfg(unix)]
1011        command.process_group(0);
1012        if let Some(cwd) = process.cwd.as_deref() {
1013            command.current_dir(cwd);
1014        }
1015        if let Some(path) = process.log_path.as_deref() {
1016            let file = std::fs::OpenOptions::new()
1017                .create(true)
1018                .append(true)
1019                .open(path)
1020                .with_context(|| format!("failed to open process log {}", path))?;
1021            let stderr = file.try_clone()?;
1022            command
1023                .stdout(Stdio::from(file))
1024                .stderr(Stdio::from(stderr));
1025        }
1026        let child = command.spawn()?;
1027        self.store.processes().upsert(NewProcess {
1028            id: Some(process.id),
1029            task_id: process.task_id,
1030            pid: child.id(),
1031            command: process.command,
1032            cwd: process.cwd,
1033            log_path: process.log_path,
1034            detected_url: process.detected_url,
1035            status: ProcessStatus::Running,
1036            health: Some("restarted".to_string()),
1037        })
1038    }
1039
1040    pub fn open_process(&self, id: &str) -> Result<RuntimeProcessOpen> {
1041        let process = self
1042            .store
1043            .processes()
1044            .get(id)?
1045            .with_context(|| format!("process not found: {}", id))?;
1046        let target = process
1047            .detected_url
1048            .or(process.log_path)
1049            .with_context(|| format!("process has no URL or log path: {}", id))?;
1050        validate_open_target(&target)?;
1051        crate::utils::open_file(target.clone());
1052        Ok(RuntimeProcessOpen { ok: true, target })
1053    }
1054
1055    pub fn resolve_open_target(&self, target: &str) -> Result<String> {
1056        if let Some(process) = self.store.processes().get(target)?
1057            && let Some(target) = process.detected_url.or(process.log_path)
1058        {
1059            return Ok(target);
1060        }
1061        Ok(target.to_string())
1062    }
1063
1064    pub fn ports(&self) -> Result<RuntimePorts> {
1065        let output = if cfg!(windows) {
1066            Command::new("netstat").arg("-ano").output()?
1067        } else {
1068            Command::new("sh")
1069                .arg("-c")
1070                .arg("ss -ltnp 2>/dev/null || netstat -ltnp 2>/dev/null || true")
1071                .output()?
1072        };
1073        Ok(RuntimePorts {
1074            ok: true,
1075            ports: String::from_utf8_lossy(&output.stdout).into_owned(),
1076        })
1077    }
1078
1079    pub fn set_plugin_enabled(&self, id: &str, enabled: bool) -> Result<()> {
1080        self.store.plugins().set_enabled(id, enabled)?;
1081        super::write_plugin_lockfile()?;
1082        Ok(())
1083    }
1084
1085    pub fn set_safety_mode(&self, mode: &str) -> Result<crate::app::SafetyConfig> {
1086        let mut config = crate::app::load_config()?;
1087        config.safety.mode = super::SafetyMode::parse(mode)
1088            .ok_or_else(|| anyhow::anyhow!("unknown safety mode: {}", mode))?;
1089        crate::app::save_config(&config, None)?;
1090        Ok(config.safety)
1091    }
1092
1093    pub fn model_info(&self, model: &str) -> Value {
1094        let snapshot = crate::domain::ProviderCapabilitySnapshot::from_model_id(model);
1095        for (key, value, confidence) in [
1096            (
1097                "supports_tools",
1098                snapshot.supports_tools.to_string(),
1099                "static",
1100            ),
1101            (
1102                "supports_vision",
1103                snapshot.supports_vision.to_string(),
1104                "static",
1105            ),
1106            ("reasoning", snapshot.reasoning.clone(), "static"),
1107            (
1108                "max_context_tokens",
1109                snapshot
1110                    .max_context_tokens
1111                    .map(|value| value.to_string())
1112                    .unwrap_or_else(|| "unknown".to_string()),
1113                "static",
1114            ),
1115        ] {
1116            let _ = self.store.provider_probes().upsert(NewProviderProbe {
1117                provider: snapshot.provider.clone(),
1118                model_id: snapshot.model.clone(),
1119                capability_key: key.to_string(),
1120                capability_value: value,
1121                confidence: confidence.to_string(),
1122                error: None,
1123            });
1124        }
1125        if let Some(profile) = crate::models::lookup_provider(&snapshot.provider) {
1126            for (key, value) in [
1127                (
1128                    "max_output_tokens_param",
1129                    format!("{:?}", profile.max_tokens_param),
1130                ),
1131                (
1132                    "parallel_tool_calls",
1133                    (!profile
1134                        .disable_parallel_tool_calls_for
1135                        .iter()
1136                        .any(|disabled| *disabled == snapshot.model))
1137                    .to_string(),
1138                ),
1139                (
1140                    "reasoning_parameter_shape",
1141                    format!("{:?}", profile.reasoning_strategy),
1142                ),
1143                (
1144                    "streaming_usage_available",
1145                    "provider_dependent".to_string(),
1146                ),
1147                ("token_usage_field_shape", "openai_compatible".to_string()),
1148            ] {
1149                let _ = self.store.provider_probes().upsert(NewProviderProbe {
1150                    provider: snapshot.provider.clone(),
1151                    model_id: snapshot.model.clone(),
1152                    capability_key: key.to_string(),
1153                    capability_value: value,
1154                    confidence: "static".to_string(),
1155                    error: None,
1156                });
1157            }
1158        }
1159        json!({
1160            "id": model,
1161            "provider": snapshot.provider,
1162            "model": snapshot.model,
1163            "supports_tools": snapshot.supports_tools,
1164            "supports_vision": snapshot.supports_vision,
1165            "reasoning": snapshot.reasoning,
1166            "max_context_tokens": snapshot.max_context_tokens,
1167        })
1168    }
1169
1170    pub fn approval_decision(result: ApprovalReplayResult) -> Result<RuntimeApprovalDecision> {
1171        Ok(RuntimeApprovalDecision {
1172            ok: true,
1173            approval: result.approval,
1174            replayed: result.replayed,
1175            summary: result.summary,
1176        })
1177    }
1178
1179    fn raw_hygiene_preview(&self) -> Result<RawRuntimeHygienePreview> {
1180        let checkpoints = self
1181            .store
1182            .checkpoints()
1183            .list_all(10_000)?
1184            .into_iter()
1185            .filter(|checkpoint| checkpoint.archived_at.is_none())
1186            .collect::<Vec<_>>();
1187        let test_checkpoint_ids = checkpoints
1188            .iter()
1189            .filter(|checkpoint| is_runtime_hygiene_checkpoint(checkpoint))
1190            .map(|checkpoint| checkpoint.id.clone())
1191            .collect::<HashSet<_>>();
1192        let approvals = self
1193            .store
1194            .approvals()
1195            .list_all(10_000)?
1196            .into_iter()
1197            .filter(|approval| approval.archived_at.is_none())
1198            .filter(|approval| is_runtime_hygiene_approval(approval, &test_checkpoint_ids))
1199            .collect::<Vec<_>>();
1200        let checkpoints = checkpoints
1201            .into_iter()
1202            .filter(is_runtime_hygiene_checkpoint)
1203            .collect::<Vec<_>>();
1204
1205        Ok(RawRuntimeHygienePreview {
1206            approvals,
1207            checkpoints,
1208        })
1209    }
1210}
1211
1212#[derive(Debug)]
1213struct RawRuntimeHygienePreview {
1214    approvals: Vec<ApprovalRecord>,
1215    checkpoints: Vec<CheckpointRecord>,
1216}
1217
1218pub fn runtime_hygiene_reason() -> &'static str {
1219    "runtime hygiene: test/dev artifact"
1220}
1221
1222pub fn parse_optional_json(raw: Option<&str>) -> Value {
1223    raw.and_then(|value| serde_json::from_str(value).ok())
1224        .unwrap_or(Value::Null)
1225}
1226
1227pub fn affected_paths_from_json(pending_action: &Value, changed_files: &Value) -> Vec<String> {
1228    let mut paths = Vec::new();
1229    collect_path_like_strings(changed_files, &mut paths);
1230    collect_path_like_strings(pending_action, &mut paths);
1231    paths.sort();
1232    paths.dedup();
1233    paths.truncate(25);
1234    paths
1235}
1236
1237fn collect_path_like_strings(value: &Value, paths: &mut Vec<String>) {
1238    match value {
1239        Value::String(value) => {
1240            if looks_path_like(value) {
1241                paths.push(value.clone());
1242            }
1243        },
1244        Value::Array(items) => {
1245            for item in items {
1246                collect_path_like_strings(item, paths);
1247            }
1248        },
1249        Value::Object(object) => {
1250            for (key, value) in object {
1251                match value {
1252                    Value::String(text)
1253                        if matches!(
1254                            key.as_str(),
1255                            "path"
1256                                | "file"
1257                                | "file_path"
1258                                | "target"
1259                                | "project_path"
1260                                | "snapshot_path"
1261                                | "cwd"
1262                        ) =>
1263                    {
1264                        if !text.trim().is_empty() {
1265                            paths.push(text.clone());
1266                        }
1267                    },
1268                    other => collect_path_like_strings(other, paths),
1269                }
1270            }
1271        },
1272        _ => {},
1273    }
1274}
1275
1276fn looks_path_like(value: &str) -> bool {
1277    let trimmed = value.trim();
1278    trimmed.starts_with('/')
1279        || trimmed.starts_with("./")
1280        || trimmed.starts_with("../")
1281        || trimmed.contains(std::path::MAIN_SEPARATOR)
1282}
1283
1284fn is_runtime_hygiene_checkpoint(checkpoint: &CheckpointRecord) -> bool {
1285    checkpoint.project_path.starts_with("/tmp/mermaid_")
1286}
1287
1288fn is_runtime_hygiene_approval(
1289    approval: &ApprovalRecord,
1290    test_checkpoint_ids: &HashSet<String>,
1291) -> bool {
1292    let is_restore_replay = approval.risk_classification == "restored_action"
1293        && approval.proposed_action.starts_with("restore replay:");
1294    if !is_restore_replay {
1295        return false;
1296    }
1297    match approval.checkpoint_id.as_deref() {
1298        Some(checkpoint_id) => test_checkpoint_ids.contains(checkpoint_id),
1299        None => true,
1300    }
1301}
1302
1303/// Whether a daemon-request failure happened *before* the request reached the
1304/// daemon — i.e. the Unix-socket `connect()` itself failed (daemon not running,
1305/// socket missing, or connection refused), so the action was never delivered
1306/// and is safe to re-run locally (RC-G/F25).
1307///
1308/// `request_daemon_text` wraps ONLY the `UnixStream::connect` error (with a
1309/// "failed to connect to …" context) and propagates post-connect write/read I/O
1310/// failures bare; a lost reply from a daemon that crashed *after* executing the
1311/// action surfaces as a JSON-parse error, never an I/O error. The connect-phase
1312/// `io::ErrorKind`s below (`NotFound`/`ConnectionRefused`/`PermissionDenied`)
1313/// are therefore unreachable from a post-send failure on this transport, so
1314/// matching them is a precise "never reached the daemon" signal. Everything else
1315/// — post-connect write/read I/O errors, empty/invalid-JSON replies, or a
1316/// daemon-level `ok:false` — is treated conservatively as "the action may have
1317/// already run". On platforms without Unix-socket IPC the transport bails before
1318/// connecting, so the request provably never reached a daemon → always pre-send.
1319fn daemon_failure_is_pre_send(err: &anyhow::Error) -> bool {
1320    #[cfg(not(unix))]
1321    {
1322        let _ = err;
1323        true
1324    }
1325    #[cfg(unix)]
1326    {
1327        match err.downcast_ref::<std::io::Error>() {
1328            Some(io) => matches!(
1329                io.kind(),
1330                std::io::ErrorKind::NotFound
1331                    | std::io::ErrorKind::ConnectionRefused
1332                    | std::io::ErrorKind::PermissionDenied
1333            ),
1334            None => false,
1335        }
1336    }
1337}
1338
1339/// Best-effort synchronous liveness check for a pid. The runtime client is sync,
1340/// so it can't reuse the async `process_running` in `providers::tool::exec`.
1341fn pid_alive(pid: u32) -> bool {
1342    if cfg!(windows) {
1343        Command::new("tasklist")
1344            .args(["/FI", &format!("PID eq {pid}"), "/NH"])
1345            .output()
1346            .map(|out| String::from_utf8_lossy(&out.stdout).contains(&pid.to_string()))
1347            .unwrap_or(false)
1348    } else {
1349        Command::new("kill")
1350            .args(["-0", &pid.to_string()])
1351            .status()
1352            .map(|s| s.success())
1353            .unwrap_or(false)
1354    }
1355}
1356
1357/// Validate a process "open" target before handing it to the OS opener (#63 —
1358/// defense-in-depth for a tampered local `processes` row). The target is either
1359/// a detected URL (a local dev server) or a log-file path mermaid wrote.
1360pub(crate) fn validate_open_target(target: &str) -> Result<()> {
1361    // Gate on the `://` authority so a bare path or Windows drive (`C:\…`, which
1362    // `Url::parse` would read as scheme "c") isn't misclassified as a URL.
1363    if target.contains("://") {
1364        let url = reqwest::Url::parse(target)
1365            .with_context(|| format!("refusing to open unparseable URL target: {target:?}"))?;
1366        anyhow::ensure!(
1367            matches!(url.scheme(), "http" | "https"),
1368            "refusing to open non-http(s) URL target: {target:?}"
1369        );
1370        // Dev-server URLs are loopback by design, so we deliberately do NOT apply
1371        // an SSRF host-class block here — the scheme allowlist is the control.
1372        return Ok(());
1373    }
1374    // Not a URL → a filesystem path (the process log). Open only an existing
1375    // regular file; this also rejects opaque `javascript:`/`data:` URIs, which
1376    // have no `://` and so fail the regular-file check.
1377    let meta = std::fs::metadata(target)
1378        .with_context(|| format!("refusing to open missing target: {target:?}"))?;
1379    anyhow::ensure!(
1380        meta.is_file(),
1381        "refusing to open non-regular-file target: {target:?}"
1382    );
1383    Ok(())
1384}
1385
1386#[cfg(test)]
1387mod tests {
1388    use super::*;
1389    use std::path::PathBuf;
1390
1391    fn temp_db(name: &str) -> PathBuf {
1392        let dir = std::env::temp_dir().join(format!("mermaid_runtime_client_{name}"));
1393        let _ = std::fs::remove_dir_all(&dir);
1394        std::fs::create_dir_all(&dir).expect("create temp dir");
1395        dir.join("runtime.sqlite3")
1396    }
1397
1398    #[test]
1399    fn pid_alive_detects_live_and_dead() {
1400        // Our own process is alive.
1401        assert!(pid_alive(std::process::id()));
1402        // A reaped child is dead (#6 — the restart wait polls this).
1403        #[cfg(unix)]
1404        let mut child = Command::new("true").spawn().expect("spawn true");
1405        #[cfg(windows)]
1406        let mut child = Command::new("cmd")
1407            .args(["/C", "exit 0"])
1408            .spawn()
1409            .expect("spawn exit");
1410        let pid = child.id();
1411        let _ = child.wait();
1412        std::thread::sleep(std::time::Duration::from_millis(250));
1413        assert!(!pid_alive(pid));
1414    }
1415
1416    #[cfg(unix)]
1417    #[test]
1418    fn daemon_failure_pre_send_only_for_connect_kinds() {
1419        // F25: distinguish "never reached the daemon" (safe to re-run locally)
1420        // from "may have already run" (must not re-run a non-idempotent action).
1421        use anyhow::Context as _;
1422        use std::io::{Error as IoError, ErrorKind};
1423
1424        // Connect-phase failures — the request never left the socket. These are
1425        // wrapped exactly as `request_daemon_text` wraps its connect error.
1426        let refused: anyhow::Error = Err::<(), _>(IoError::from(ErrorKind::ConnectionRefused))
1427            .with_context(|| "failed to connect to /run/mermaidd.sock".to_string())
1428            .unwrap_err();
1429        assert!(daemon_failure_is_pre_send(&refused));
1430        let missing: anyhow::Error = Err::<(), _>(IoError::from(ErrorKind::NotFound))
1431            .with_context(|| "failed to connect to /run/mermaidd.sock".to_string())
1432            .unwrap_err();
1433        assert!(daemon_failure_is_pre_send(&missing));
1434
1435        // A lost reply from a daemon that crashed AFTER executing surfaces as a
1436        // JSON-parse error — ambiguous, may have run.
1437        assert!(!daemon_failure_is_pre_send(&anyhow::anyhow!(
1438            "daemon returned invalid JSON"
1439        )));
1440        // A post-connect write failure (broken pipe) is ambiguous — may have run.
1441        let broken = anyhow::Error::from(IoError::from(ErrorKind::BrokenPipe));
1442        assert!(!daemon_failure_is_pre_send(&broken));
1443        // A daemon-level `ok:false` error definitely reached the daemon.
1444        assert!(!daemon_failure_is_pre_send(&anyhow::anyhow!(
1445            "stop_process failed"
1446        )));
1447    }
1448
1449    #[test]
1450    fn validate_open_target_allows_http_rejects_file_and_js() {
1451        assert!(super::validate_open_target("http://localhost:3000").is_ok());
1452        assert!(super::validate_open_target("https://localhost:8443/app").is_ok());
1453        assert!(super::validate_open_target("file:///etc/passwd").is_err());
1454        assert!(super::validate_open_target("javascript:alert(1)").is_err());
1455        assert!(super::validate_open_target("data:text/html,<x>").is_err());
1456        assert!(super::validate_open_target("/no/such/path/xyz.log").is_err());
1457    }
1458
1459    #[test]
1460    fn restart_process_refuses_destructive_command() {
1461        let path = temp_db("restart_destructive");
1462        let service = RuntimeService::from_store(RuntimeStore::open(&path).expect("open store"));
1463        let rec = service
1464            .store
1465            .processes()
1466            .upsert(NewProcess {
1467                id: Some("p-destruct".to_string()),
1468                task_id: None,
1469                pid: 0,
1470                command: "rm -rf /".to_string(),
1471                cwd: None,
1472                log_path: None,
1473                detected_url: None,
1474                status: ProcessStatus::Running,
1475                health: None,
1476            })
1477            .expect("seed process");
1478        // Refused before kill/spawn (pid 0 is never signaled).
1479        let err = service.restart_process(&rec.id).unwrap_err();
1480        assert!(err.to_string().contains("destructive"), "{err}");
1481        let _ = std::fs::remove_dir_all(path.parent().unwrap());
1482    }
1483
1484    #[test]
1485    fn open_process_rejects_file_url_target() {
1486        let path = temp_db("open_file_url");
1487        let service = RuntimeService::from_store(RuntimeStore::open(&path).expect("open store"));
1488        let rec = service
1489            .store
1490            .processes()
1491            .upsert(NewProcess {
1492                id: Some("p-open".to_string()),
1493                task_id: None,
1494                pid: 0,
1495                command: "true".to_string(),
1496                cwd: None,
1497                log_path: None,
1498                detected_url: Some("file:///etc/passwd".to_string()),
1499                status: ProcessStatus::Running,
1500                health: None,
1501            })
1502            .expect("seed process");
1503        // Errors before open_file — nothing is launched.
1504        assert!(service.open_process(&rec.id).is_err());
1505        let _ = std::fs::remove_dir_all(path.parent().unwrap());
1506    }
1507
1508    #[test]
1509    fn local_client_lists_tasks_from_store() {
1510        let path = temp_db("tasks");
1511        let service = RuntimeService::from_store(RuntimeStore::open(&path).expect("open store"));
1512        let task = service
1513            .store
1514            .tasks()
1515            .create(super::super::NewTask::new(
1516                "contract task",
1517                "/tmp/mermaid_runtime_client",
1518                "openai/test",
1519            ))
1520            .expect("task");
1521        let tasks = service.list_tasks(10).expect("list");
1522        assert_eq!(
1523            tasks.first().map(|item| item.id.as_str()),
1524            Some(task.id.as_str())
1525        );
1526        let _ = std::fs::remove_dir_all(path.parent().unwrap());
1527    }
1528
1529    #[test]
1530    fn hygiene_preview_matches_test_artifacts_and_archive_is_idempotent() {
1531        let path = temp_db("hygiene");
1532        let store = RuntimeStore::open(&path).expect("open store");
1533        let checkpoint = store
1534            .checkpoints()
1535            .create(super::super::NewCheckpoint {
1536                id: Some("checkpoint-test".to_string()),
1537                task_id: None,
1538                project_path: "/tmp/mermaid_checkpoint_test".to_string(),
1539                snapshot_path: "/data/checkpoints/checkpoint-test".to_string(),
1540                changed_files_json: "[]".to_string(),
1541                pending_action_json: Some("{\"tool\":\"write_file\"}".to_string()),
1542                approval_id: None,
1543            })
1544            .expect("create checkpoint");
1545        let approval = store
1546            .approvals()
1547            .create(super::super::NewApproval {
1548                task_id: None,
1549                proposed_action: "restore replay: write_file".to_string(),
1550                risk_classification: "restored_action".to_string(),
1551                policy_decision: "ask".to_string(),
1552                args_summary: None,
1553                checkpoint_id: Some(checkpoint.id.clone()),
1554                pending_action_json: Some("{\"tool\":\"write_file\"}".to_string()),
1555            })
1556            .expect("create approval");
1557        store
1558            .checkpoints()
1559            .set_approval(&checkpoint.id, &approval.id)
1560            .expect("link approval");
1561
1562        let service = RuntimeService::from_store(store);
1563        let preview = service.hygiene_preview().expect("preview");
1564        assert_eq!(preview.counts.approvals, 1);
1565        assert_eq!(preview.counts.checkpoints, 1);
1566        let archived = service.hygiene_archive().expect("archive");
1567        assert_eq!(archived.archived.total, 2);
1568        let archived_again = service.hygiene_archive().expect("archive again");
1569        assert_eq!(archived_again.archived.total, 0);
1570        let _ = std::fs::remove_dir_all(path.parent().unwrap());
1571    }
1572}