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