pitchfork_cli/ipc/mod.rs
1use crate::Result;
2use crate::daemon::{Daemon, RunOptions};
3use crate::daemon_id::DaemonId;
4use crate::env;
5use interprocess::local_socket::Name;
6#[cfg(unix)]
7use interprocess::local_socket::{GenericFilePath, ToFsName};
8#[cfg(windows)]
9use interprocess::local_socket::{GenericNamespaced, ToNsName};
10use miette::{Context, IntoDiagnostic};
11use std::path::PathBuf;
12
13pub(crate) mod batch;
14pub(crate) mod client;
15pub(crate) mod server;
16
17// #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, strum::Display, strum::EnumIs)]
18// pub enum IpcMessage {
19// Connect(String),
20// ConnectOK,
21// Run(String, Vec<String>),
22// Stop(String),
23// DaemonAlreadyRunning(String),
24// DaemonAlreadyStopped(String),
25// DaemonStart(Daemon),
26// DaemonStop { name: String },
27// DaemonFailed { name: String, error: String },
28// Response(String),
29// }
30
31#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, strum::Display, strum::EnumIs)]
32#[allow(clippy::large_enum_variant)]
33pub enum IpcRequest {
34 Connect,
35 /// Versioned connect handshake (v2): client sends its version so the supervisor can
36 /// detect mismatches. Kept as a separate variant so the wire format of `Connect`
37 /// (unit variant) stays unchanged for backward compatibility with older supervisors.
38 ConnectV2 {
39 version: String,
40 },
41 Clean,
42 Stop {
43 id: DaemonId,
44 },
45 GetActiveDaemons,
46 GetDisabledDaemons,
47 Run(RunOptions),
48 Enable {
49 id: DaemonId,
50 },
51 Disable {
52 id: DaemonId,
53 },
54 UpdateShellDir {
55 shell_pid: u32,
56 dir: PathBuf,
57 },
58 GetNotifications,
59 /// Notify the supervisor that the slug registry has changed (e.g. `proxy add/remove`).
60 /// The supervisor should re-read slugs and update mDNS records accordingly.
61 SyncMdns,
62 /// Notify the supervisor that settings have changed.
63 /// The supervisor should reload settings from config files.
64 ReloadConfig,
65 /// Enter or replace a project session for a host PID in a directory.
66 ProjectEnter {
67 pid: u32,
68 dir: PathBuf,
69 },
70 /// Leave a project session for a host PID in a directory.
71 ProjectLeave {
72 pid: u32,
73 dir: PathBuf,
74 },
75 /// List all tracked project sessions with live liveness status filled in
76 /// by the supervisor.
77 GetProjectSessions,
78 /// A daemon's log sink reporting a line the supervisor needs to act on:
79 /// one matching the readiness pattern, one that should fire the
80 /// `on_output` hook, or both.
81 ///
82 /// The sink owns the daemon's output stream, so it does the matching and
83 /// tells the supervisor rather than the other way around. Sent by
84 /// `pitchfork log-sink`, not by any user-facing command.
85 SinkOutputLine {
86 id: DaemonId,
87 /// Identifies the start attempt this sink belongs to, so a report from
88 /// a sink still draining a previous attempt cannot satisfy a retry.
89 token: u64,
90 /// Whether this line passed the `on_output` hook's filter and debounce.
91 /// A line reported only because it matched the readiness pattern must
92 /// not fire a hook that filters for something else.
93 fires_hook: bool,
94 line: String,
95 },
96 /// Ask the supervisor for the URL of the web UI, if it is running.
97 /// Reflects the actual bound address, not static config.
98 GetWebUrl,
99 /// Invalid request (failed to deserialize)
100 #[serde(skip)]
101 Invalid {
102 error: String,
103 },
104}
105
106/// A snapshot of a single project session, returned by `GetProjectSessions`.
107///
108/// `liveness_title` is the title recorded at enter time. `alive` and
109/// `current_title` are filled in by the supervisor from its `PROCS` singleton
110/// so the client does not need process introspection.
111#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
112pub struct ProjectSessionInfo {
113 pub pid: u32,
114 pub directory: PathBuf,
115 #[serde(skip_serializing_if = "Option::is_none", default)]
116 pub liveness_title: Option<String>,
117 pub alive: bool,
118 #[serde(skip_serializing_if = "Option::is_none", default)]
119 pub current_title: Option<String>,
120}
121
122#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, strum::Display, strum::EnumIs)]
123pub enum IpcResponse {
124 Ok,
125 /// Successful connect handshake, includes supervisor version for mismatch detection
126 ConnectOk {
127 version: String,
128 },
129 Yes,
130 No,
131 Error(String),
132 Notifications(Vec<(log::LevelFilter, String)>),
133 ActiveDaemons(Vec<Daemon>),
134 DisabledDaemons(Vec<DaemonId>),
135 DaemonAlreadyRunning,
136 DaemonStart {
137 daemon: Daemon,
138 },
139 DaemonFailed {
140 error: String,
141 },
142 /// Port conflict detected with detailed process information
143 PortConflict {
144 port: u16,
145 process: String,
146 pid: u32,
147 },
148 /// No available ports found after exhausting auto-bump attempts
149 NoAvailablePort {
150 start_port: u16,
151 attempts: u32,
152 },
153 DaemonReady {
154 daemon: Daemon,
155 },
156 DaemonFailedWithCode {
157 exit_code: Option<i32>,
158 },
159 /// Process was not running but had a PID record (unexpected exit)
160 DaemonWasNotRunning,
161 /// mDNS sync completed (or was a no-op if LAN mode is disabled)
162 MdnsSynced,
163 /// Settings reloaded from config files
164 ConfigReloaded,
165 /// URL of the running web UI, or `None` if the web UI is not running
166 WebUrl {
167 url: Option<String>,
168 },
169 /// Failed to kill the process (still running)
170 DaemonStopFailed {
171 error: String,
172 },
173 /// Daemon exists but is not running (no PID)
174 DaemonNotRunning,
175 DaemonNotFound,
176 /// Snapshot of all project sessions (response to `GetProjectSessions`).
177 ProjectSessions(Vec<ProjectSessionInfo>),
178}
179fn fs_name(name: &str) -> Result<Name<'_>> {
180 // Unix: use a filesystem path for the AF_UNIX socket.
181 #[cfg(unix)]
182 {
183 let path = env::IPC_SOCK_DIR.join(name).with_extension("sock");
184 let fs_name = path.to_fs_name::<GenericFilePath>().into_diagnostic()?;
185 Ok(fs_name)
186 }
187 // Windows: named pipes use a flat namespace (\\.\pipe\<name>) that
188 // cannot contain path separators. Derive a unique pipe name from the
189 // state directory to preserve test isolation when multiple supervisors
190 // run concurrently with different PITCHFORK_STATE_DIR values.
191 //
192 // Use a hash of the state directory path rather than character replacement
193 // to guarantee injectivity: `C:\a.b` and `C:\a\b` would both flatten to
194 // `C--a-b` with the old approach, causing pipe name collisions.
195 #[cfg(windows)]
196 {
197 let state_dir = env::PITCHFORK_STATE_DIR.to_string_lossy();
198 // FNV-1a hash: deterministic, stable across Rust versions.
199 // DefaultHasher's algorithm is not guaranteed stable, which would
200 // break IPC if the CLI and supervisor were ever compiled with
201 // different toolchains.
202 let mut hash: u64 = 0xcbf29ce484222325;
203 for byte in state_dir.bytes() {
204 hash ^= byte as u64;
205 hash = hash.wrapping_mul(0x100000001b3);
206 }
207 let pipe_name = format!("pitchfork-{hash:016x}-{name}");
208 Ok(pipe_name
209 .to_ns_name::<GenericNamespaced>()
210 .into_diagnostic()?)
211 }
212}
213
214fn serialize<T: serde::Serialize>(msg: &T) -> Result<Vec<u8>> {
215 if *env::IPC_JSON {
216 serde_json::to_vec(msg)
217 .into_diagnostic()
218 .wrap_err("failed to serialize IPC message as JSON")
219 } else {
220 rmp_serde::to_vec(msg)
221 .into_diagnostic()
222 .wrap_err("failed to serialize IPC message as MessagePack")
223 }
224}
225
226fn deserialize<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T> {
227 let mut bytes = bytes.to_vec();
228 bytes.pop();
229 let preview = std::str::from_utf8(&bytes).unwrap_or("<binary>");
230 trace!("msg: {preview:?}");
231 if *env::IPC_JSON {
232 serde_json::from_slice(&bytes)
233 .into_diagnostic()
234 .wrap_err("failed to deserialize IPC JSON response")
235 } else {
236 rmp_serde::from_slice(&bytes)
237 .into_diagnostic()
238 .wrap_err("failed to deserialize IPC MessagePack response")
239 }
240}