tenant.openagents/omega

1//! Supervise packaged `omega-effectd` over newline-framed JSON stdio.
2//!
3//! Durable Full Auto run truth lives on disk under the injected data root.
4//! This supervisor owns process life, health, restart, and generation fencing.
5//! It must never become a second durable run authority (GPUI must not rewrite
6//! runs after restart).
7
8use std::ffi::OsStr;
9use std::future::Future;
10use std::path::{Path, PathBuf};
11use std::pin::Pin;
12use std::process::Stdio;
13use std::rc::Rc;
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::time::Duration;
16
17use anyhow::{Context as _, Result, anyhow, bail};
18use futures::io::{AsyncBufReadExt as _, BufReader};
19use futures::{AsyncWriteExt as _, StreamExt as _};
20use serde::Deserialize;
21use serde_json::{Value, json};
22use smol::process::ChildStdin;
23use util::ResultExt as _;
24use util::process::Child;
25use util::redact::redact_command;
26
27use crate::protocol::{
28    HealthResult, HostMethod, HostRequestFrame, HostResponseError, HostResponseErrorCode,
29    HostResponseFrame, InitializeResult, PROTOCOL_SCHEMA, ProtocolErrorCode, ResponseFrame,
30    RunSnapshot, request_frame,
31};
32
33pub use crate::protocol::MAX_FRAME_BYTES;
34const SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(2);
35const AGENT_COMPUTER_TURN_TIMEOUT: Duration = Duration::from_secs(180);
36const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(180);
37const DEFAULT_HOST_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
38const MAX_HOST_ERROR_MESSAGE_BYTES: usize = 1024;
39
40pub type OmegaEffectdHostFuture =
41    Pin<Box<dyn Future<Output = std::result::Result<Value, HostResponseError>> + 'static>>;
42pub type OmegaEffectdHostHandler = Rc<dyn Fn(HostRequestFrame) -> OmegaEffectdHostFuture + 'static>;
43
44#[derive(Debug, Clone)]
45pub struct OmegaEffectdCommand {
46    pub program: PathBuf,
47    pub args: Vec<String>,
48}
49
50#[derive(Debug, Clone)]
51pub struct OmegaEffectdSupervisorOptions {
52    pub data_root: PathBuf,
53    pub command: OmegaEffectdCommand,
54    /// Initial generation. Each successful `restart` increments by one.
55    pub initial_generation: u64,
56    pub request_timeout: Duration,
57}
58
59#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
60#[serde(rename_all = "camelCase", deny_unknown_fields)]
61pub struct AttentionDecision {
62    pub notify: bool,
63    pub dedup_key: String,
64    pub title: String,
65    pub body: String,
66}
67
68#[derive(Debug, thiserror::Error)]
69pub enum SupervisorError {
70    #[error(transparent)]
71    Anyhow(#[from] anyhow::Error),
72    #[error("stale generation")]
73    StaleGeneration,
74    #[error("protocol error ({code:?}): {message}")]
75    Protocol {
76        code: ProtocolErrorCode,
77        message: String,
78    },
79}
80
81pub struct OmegaEffectdSupervisor {
82    options: OmegaEffectdSupervisorOptions,
83    generation: AtomicU64,
84    next_request_id: AtomicU64,
85    child: Option<Child>,
86    stdin: Option<ChildStdin>,
87    stdout: Option<BufReader<smol::process::ChildStdout>>,
88    host_handler: Option<OmegaEffectdHostHandler>,
89    host_request_timeout: Duration,
90}
91
92impl OmegaEffectdSupervisor {
93    pub fn new(options: OmegaEffectdSupervisorOptions) -> Self {
94        let generation = options.initial_generation.max(1);
95        Self {
96            options,
97            generation: AtomicU64::new(generation),
98            next_request_id: AtomicU64::new(1),
99            child: None,
100            stdin: None,
101            stdout: None,
102            host_handler: None,
103            host_request_timeout: DEFAULT_HOST_REQUEST_TIMEOUT,
104        }
105    }
106
107    pub fn set_host_handler(&mut self, handler: OmegaEffectdHostHandler) {
108        self.host_handler = Some(handler);
109    }
110
111    pub fn set_host_request_timeout(&mut self, timeout: Duration) {
112        self.host_request_timeout = timeout;
113    }
114
115    pub fn generation(&self) -> u64 {
116        self.generation.load(Ordering::SeqCst)
117    }
118
119    pub fn data_root(&self) -> &Path {
120        &self.options.data_root
121    }
122
123    pub async fn start(&mut self) -> Result<InitializeResult> {
124        if self.child.is_some() {
125            bail!("omega-effectd is already running");
126        }
127        self.spawn_child().await?;
128        let generation = self.generation();
129        let result = self
130            .request(
131                "initialize",
132                Some(json!({ "generation": generation })),
133                generation,
134            )
135            .await?;
136        serde_json::from_value(result).context("decode initialize result")
137    }
138
139    pub async fn ensure_started(&mut self) -> Result<()> {
140        if self.child.is_none() {
141            self.start().await?;
142        }
143        Ok(())
144    }
145
146    pub async fn health(&mut self) -> Result<HealthResult, SupervisorError> {
147        let result = self.request("health", None, self.generation()).await?;
148        Ok(serde_json::from_value(result).context("decode health result")?)
149    }
150
151    pub async fn list_runs(&mut self) -> Result<Vec<RunSnapshot>, SupervisorError> {
152        let result = self.request("list_runs", None, self.generation()).await?;
153        let runs = result
154            .get("runs")
155            .cloned()
156            .ok_or_else(|| anyhow!("list_runs missing runs"))?;
157        Ok(serde_json::from_value(runs).context("decode list_runs")?)
158    }
159
160    pub async fn get_run(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
161        let result = self
162            .request(
163                "get_run",
164                Some(json!({ "runRef": run_ref })),
165                self.generation(),
166            )
167            .await?;
168        Ok(result
169            .get("run")
170            .cloned()
171            .ok_or_else(|| anyhow!("get_run missing run"))?)
172    }
173
174    pub async fn start_run(&mut self, params: Value) -> Result<Value, SupervisorError> {
175        let result = self
176            .request("start", Some(params), self.generation())
177            .await?;
178        Ok(result
179            .get("run")
180            .cloned()
181            .ok_or_else(|| anyhow!("start missing run"))?)
182    }
183
184    pub async fn pause_run(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
185        self.mutate_run("pause", run_ref).await
186    }
187
188    pub async fn resume_run(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
189        self.mutate_run("resume", run_ref).await
190    }
191
192    pub async fn handoff_run(
193        &mut self,
194        run_ref: &str,
195        target_lane_ref: &str,
196    ) -> Result<Value, SupervisorError> {
197        let result = self
198            .request(
199                "handoff",
200                Some(json!({
201                    "runRef": run_ref,
202                    "targetLaneRef": target_lane_ref,
203                })),
204                self.generation(),
205            )
206            .await?;
207        Ok(result
208            .get("run")
209            .cloned()
210            .ok_or_else(|| anyhow!("handoff missing run"))?)
211    }
212
213    pub async fn stop_run(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
214        self.mutate_run("stop", run_ref).await
215    }
216
217    pub async fn retry_run(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
218        self.mutate_run("retry", run_ref).await
219    }
220
221    pub async fn get_capacity(&mut self) -> Result<Value, SupervisorError> {
222        self.request("get_capacity", None, self.generation()).await
223    }
224
225    pub async fn decide_attention(
226        &mut self,
227        run_ref: &str,
228        permission_granted: bool,
229        previous_dedup_key: Option<&str>,
230    ) -> Result<Option<AttentionDecision>, SupervisorError> {
231        let result = self
232            .request(
233                "decide_attention",
234                Some(json!({
235                    "runRef": run_ref,
236                    "permissionGranted": permission_granted,
237                    "previousDedupKey": previous_dedup_key,
238                })),
239                self.generation(),
240            )
241            .await?;
242        let attention = result.get("attention").cloned().unwrap_or(Value::Null);
243        Ok(serde_json::from_value(attention).context("decode attention decision")?)
244    }
245
246    pub async fn get_report(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
247        let result = self
248            .request(
249                "get_report",
250                Some(json!({ "runRef": run_ref })),
251                self.generation(),
252            )
253            .await?;
254        Ok(result
255            .get("report")
256            .cloned()
257            .ok_or_else(|| anyhow!("get_report missing report"))?)
258    }
259
260    pub async fn get_receipt(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
261        let result = self
262            .request(
263                "get_receipt",
264                Some(json!({ "runRef": run_ref })),
265                self.generation(),
266            )
267            .await?;
268        Ok(result
269            .get("receipt")
270            .cloned()
271            .ok_or_else(|| anyhow!("get_receipt missing receipt"))?)
272    }
273
274    pub async fn apply_control_intent(
275        &mut self,
276        intent_id: &str,
277        run_ref: &str,
278        action: &str,
279    ) -> Result<Value, SupervisorError> {
280        let result = self
281            .request(
282                "apply_control_intent",
283                Some(json!({
284                    "intentId": intent_id,
285                    "runRef": run_ref,
286                    "action": action,
287                })),
288                self.generation(),
289            )
290            .await?;
291        Ok(result
292            .get("outcome")
293            .cloned()
294            .ok_or_else(|| anyhow!("apply_control_intent missing outcome"))?)
295    }
296
297    pub async fn get_sync_status(&mut self) -> Result<Value, SupervisorError> {
298        self.request("get_sync_status", None, self.generation())
299            .await
300    }
301
302    pub async fn publish_projection(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
303        self.request(
304            "publish_projection",
305            Some(json!({ "runRef": run_ref })),
306            self.generation(),
307        )
308        .await
309    }
310
311    pub async fn get_native_binding(&mut self, run_ref: &str) -> Result<Value, SupervisorError> {
312        let result = self
313            .request(
314                "get_native_binding",
315                Some(json!({ "runRef": run_ref })),
316                self.generation(),
317            )
318            .await?;
319        Ok(result.get("binding").cloned().unwrap_or(Value::Null))
320    }
321
322    pub async fn assess_native_boundary(
323        &mut self,
324        run_ref: &str,
325    ) -> Result<Value, SupervisorError> {
326        let result = self
327            .request(
328                "assess_native_boundary",
329                Some(json!({ "runRef": run_ref })),
330                self.generation(),
331            )
332            .await?;
333        Ok(result
334            .get("assessment")
335            .cloned()
336            .ok_or_else(|| anyhow!("assess_native_boundary missing assessment"))?)
337    }
338
339    pub async fn start_agent_computer_session(
340        &mut self,
341        params: Value,
342    ) -> Result<Value, SupervisorError> {
343        let result = self
344            .request(
345                "start_agent_computer_session",
346                Some(params),
347                self.generation(),
348            )
349            .await?;
350        Ok(result
351            .get("session")
352            .cloned()
353            .ok_or_else(|| anyhow!("start_agent_computer_session missing session"))?)
354    }
355
356    pub async fn refresh_agent_computer_session(
357        &mut self,
358        bearer_token: &str,
359        session_ref: &str,
360    ) -> Result<Value, SupervisorError> {
361        let result = self
362            .request(
363                "refresh_agent_computer_session",
364                Some(json!({
365                    "bearerToken": bearer_token,
366                    "sessionRef": session_ref,
367                })),
368                self.generation(),
369            )
370            .await?;
371        Ok(result
372            .get("session")
373            .cloned()
374            .ok_or_else(|| anyhow!("refresh_agent_computer_session missing session"))?)
375    }
376
377    pub async fn run_agent_computer_turn(
378        &mut self,
379        params: Value,
380    ) -> Result<Value, SupervisorError> {
381        self.request_with_timeout(
382            "run_agent_computer_turn",
383            Some(params),
384            self.generation(),
385            AGENT_COMPUTER_TURN_TIMEOUT,
386        )
387        .await
388    }
389
390    pub async fn get_agent_computer_session(
391        &mut self,
392        session_ref: &str,
393    ) -> Result<Value, SupervisorError> {
394        let result = self
395            .request(
396                "get_agent_computer_session",
397                Some(json!({ "sessionRef": session_ref })),
398                self.generation(),
399            )
400            .await?;
401        Ok(result.get("session").cloned().unwrap_or(Value::Null))
402    }
403
404    pub async fn list_agent_computer_sessions(&mut self) -> Result<Value, SupervisorError> {
405        self.request("list_agent_computer_sessions", None, self.generation())
406            .await
407    }
408
409    /// SARAH-NR-06 / OMEGA-SW-03: public-safe session projection (never returns tokens).
410    pub async fn sarah_session_status(&mut self) -> Result<Value, SupervisorError> {
411        self.request("sarah_session_status", None, self.generation())
412            .await
413    }
414
415    /// SARAH-NR-06 / OMEGA-SW-03: principal projection and conversation reference.
416    pub async fn sarah_bootstrap(&mut self) -> Result<Value, SupervisorError> {
417        self.request("sarah_bootstrap", None, self.generation())
418            .await
419    }
420
421    /// SARAH-NR-06 / OMEGA-SW-03: bounded transcript/activity page with cursors and gap state.
422    pub async fn sarah_room_snapshot(
423        &mut self,
424        params: Option<Value>,
425    ) -> Result<Value, SupervisorError> {
426        self.request("sarah_room_snapshot", params, self.generation())
427            .await
428    }
429
430    /// SARAH-NR-06: publish an owner message onto the Nostr conversation.
431    pub async fn sarah_send_message(
432        &mut self,
433        text: &str,
434        idempotency_ref: &str,
435    ) -> Result<Value, SupervisorError> {
436        let generation = self.generation();
437        self.request(
438            "sarah_send_message",
439            Some(json!({
440                "text": text,
441                "idempotencyRef": idempotency_ref,
442                "expectedGeneration": generation,
443            })),
444            generation,
445        )
446        .await
447    }
448
449    /// SARAH-NR-06 / OMEGA-SW-03: publish a cancel_turn control intent (pending until settled).
450    pub async fn sarah_interrupt_turn(
451        &mut self,
452        turn_ref: &str,
453        idempotency_ref: &str,
454    ) -> Result<Value, SupervisorError> {
455        let generation = self.generation();
456        self.request(
457            "sarah_interrupt_turn",
458            Some(json!({
459                "turnRef": turn_ref,
460                "idempotencyRef": idempotency_ref,
461                "expectedGeneration": generation,
462            })),
463            generation,
464        )
465        .await
466    }
467
468    pub async fn sarah_device_grants(&mut self) -> Result<Value, SupervisorError> {
469        self.request("sarah_device_grants", None, self.generation())
470            .await
471    }
472
473    pub async fn sarah_renew_device_grant(
474        &mut self,
475        grant_ref: &str,
476        scopes: &[crate::Issue31PairingScope],
477        expires_at: u64,
478        idempotency_ref: &str,
479    ) -> Result<Value, SupervisorError> {
480        let generation = self.generation();
481        self.request(
482            "sarah_renew_device_grant",
483            Some(json!({
484                "grantRef": grant_ref,
485                "scopes": scopes,
486                "expiresAt": expires_at,
487                "idempotencyRef": idempotency_ref,
488                "expectedGeneration": generation,
489            })),
490            generation,
491        )
492        .await
493    }
494
495    pub async fn sarah_revoke_device_grant(
496        &mut self,
497        grant_ref: &str,
498        reason_ref: &str,
499        idempotency_ref: &str,
500    ) -> Result<Value, SupervisorError> {
501        let generation = self.generation();
502        self.request(
503            "sarah_revoke_device_grant",
504            Some(json!({
505                "grantRef": grant_ref,
506                "reasonRef": reason_ref,
507                "idempotencyRef": idempotency_ref,
508                "expectedGeneration": generation,
509            })),
510            generation,
511        )
512        .await
513    }
514
515    /// Re-admit the device behind a revoked grant so it may pair again.
516    ///
517    /// Revocation fails closed for the device, so this is the owner's only path
518    /// back. It grants nothing on its own — the device must still complete a
519    /// fresh signed pairing handshake.
520    pub async fn sarah_readmit_device(
521        &mut self,
522        grant_ref: &str,
523        idempotency_ref: &str,
524    ) -> Result<Value, SupervisorError> {
525        let generation = self.generation();
526        self.request(
527            "sarah_readmit_device",
528            Some(json!({
529                "grantRef": grant_ref,
530                "idempotencyRef": idempotency_ref,
531                "expectedGeneration": generation,
532            })),
533            generation,
534        )
535        .await
536    }
537
538    async fn mutate_run(&mut self, method: &str, run_ref: &str) -> Result<Value, SupervisorError> {
539        let result = self
540            .request(
541                method,
542                Some(json!({ "runRef": run_ref })),
543                self.generation(),
544            )
545            .await?;
546        Ok(result
547            .get("run")
548            .cloned()
549            .ok_or_else(|| anyhow!("{method} missing run"))?)
550    }
551
552    pub async fn restart(&mut self) -> Result<InitializeResult> {
553        self.stop().await?;
554        let next = self.generation.fetch_add(1, Ordering::SeqCst) + 1;
555        self.generation.store(next, Ordering::SeqCst);
556        self.start().await
557    }
558
559    pub async fn stop(&mut self) -> Result<()> {
560        if let Some(mut child) = self.child.take() {
561            self.stdin.take();
562            self.stdout.take();
563            #[cfg(unix)]
564            {
565                let signal_result = unsafe { libc::kill(child.id() as i32, libc::SIGTERM) };
566                if signal_result != 0 {
567                    let error = std::io::Error::last_os_error();
568                    if error.raw_os_error() != Some(libc::ESRCH) {
569                        return Err(error).context("terminate omega-effectd");
570                    }
571                }
572            }
573
574            #[cfg(unix)]
575            let exited = smol::future::or(async { child.status().await.map(|_| true) }, async {
576                runtime_delay(SHUTDOWN_GRACE_PERIOD).await;
577                Ok(false)
578            })
579            .await
580            .context("wait for omega-effectd shutdown")?;
581
582            #[cfg(not(unix))]
583            let exited = false;
584
585            if !exited {
586                child.kill().context("kill unresponsive omega-effectd")?;
587                child.status().await.context("reap killed omega-effectd")?;
588            }
589        }
590        Ok(())
591    }
592
593    async fn spawn_child(&mut self) -> Result<()> {
594        std::fs::create_dir_all(&self.options.data_root)
595            .with_context(|| format!("create data root {}", self.options.data_root.display()))?;
596
597        let mut command = std::process::Command::new(&self.options.command.program);
598        command.args(&self.options.command.args);
599        command.env(
600            "OPENAGENTS_OMEGA_EFFECTD_DATA_ROOT",
601            &self.options.data_root,
602        );
603
604        let mut child = Child::spawn(command, Stdio::piped(), Stdio::piped(), Stdio::piped())
605            .with_context(|| {
606                format!(
607                    "spawn omega-effectd {}",
608                    redact_command(&format!("{:?}", self.options.command))
609                )
610            })?;
611
612        let stdin = child
613            .stdin
614            .take()
615            .ok_or_else(|| anyhow!("omega-effectd stdin missing"))?;
616        let stdout = child
617            .stdout
618            .take()
619            .ok_or_else(|| anyhow!("omega-effectd stdout missing"))?;
620        if let Some(stderr) = child.stderr.take() {
621            smol::spawn(async move {
622                let mut lines = BufReader::new(stderr).lines();
623                while let Some(line) = lines.next().await {
624                    match line {
625                        Ok(line) => eprintln!("omega-effectd: {}", redact_command(&line)),
626                        Err(error) => {
627                            eprintln!("omega-effectd stderr read failed: {error}");
628                            break;
629                        }
630                    }
631                }
632            })
633            .detach();
634        }
635
636        self.stdin = Some(stdin);
637        self.stdout = Some(BufReader::new(stdout));
638        self.child = Some(child);
639        Ok(())
640    }
641
642    async fn request(
643        &mut self,
644        method: &str,
645        params: Option<Value>,
646        generation: u64,
647    ) -> Result<Value, SupervisorError> {
648        self.request_with_timeout(method, params, generation, self.options.request_timeout)
649            .await
650    }
651
652    async fn request_with_timeout(
653        &mut self,
654        method: &str,
655        params: Option<Value>,
656        generation: u64,
657        timeout: Duration,
658    ) -> Result<Value, SupervisorError> {
659        let id = self
660            .next_request_id
661            .fetch_add(1, Ordering::SeqCst)
662            .to_string();
663        let frame = request_frame(id.clone(), generation, method, params);
664        let line =
665            serde_json::to_string(&frame).map_err(|error| SupervisorError::Anyhow(error.into()))?;
666        if line.len() > MAX_FRAME_BYTES {
667            return Err(SupervisorError::Anyhow(anyhow!(
668                "omega-effectd request frame exceeds {MAX_FRAME_BYTES} bytes"
669            )));
670        }
671        let stdin = self
672            .stdin
673            .as_mut()
674            .ok_or_else(|| anyhow!("omega-effectd not started"))?;
675        stdin
676            .write_all(format!("{line}\n").as_bytes())
677            .await
678            .map_err(|error| SupervisorError::Anyhow(error.into()))?;
679        stdin
680            .flush()
681            .await
682            .map_err(|error| SupervisorError::Anyhow(error.into()))?;
683
684        let response_result = smol::future::or(
685            async {
686                loop {
687                    let line = read_bounded_line(
688                        self.stdout
689                            .as_mut()
690                            .ok_or_else(|| anyhow!("omega-effectd stdout missing"))?,
691                    )
692                    .await?
693                    .ok_or_else(|| anyhow!("omega-effectd closed stdout"))?;
694                    let frame: Value = serde_json::from_str(&line)
695                        .context("decode omega-effectd protocol frame")?;
696                    match frame.get("kind").and_then(Value::as_str) {
697                        Some("host_request") => {
698                            let request: HostRequestFrame = serde_json::from_value(frame)
699                                .context("decode omega-effectd host request")?;
700                            self.respond_to_host_request(request, generation).await?;
701                            continue;
702                        }
703                        Some("event") => continue,
704                        Some("response") => {}
705                        _ => bail!("omega-effectd emitted an invalid frame kind"),
706                    }
707                    let response: ResponseFrame = serde_json::from_value(frame)
708                        .context("decode omega-effectd response frame")?;
709                    if response.schema != PROTOCOL_SCHEMA {
710                        bail!("omega-effectd response used an invalid schema");
711                    }
712                    if response.id != id {
713                        continue;
714                    }
715                    if response.generation != generation {
716                        bail!(
717                            "omega-effectd response used stale generation {}; expected {generation}",
718                            response.generation
719                        );
720                    }
721                    return Ok::<ResponseFrame, anyhow::Error>(response);
722                }
723            },
724            async {
725                runtime_delay(timeout).await;
726                Err(anyhow!("omega-effectd request timed out after {timeout:?}"))
727            },
728        )
729        .await;
730
731        let response = match response_result {
732            Ok(response) => response,
733            Err(error) => {
734                if let Err(stop_error) = self.stop().await {
735                    return Err(SupervisorError::Anyhow(error.context(format!(
736                        "omega-effectd request failed; child teardown also failed: {stop_error:#}"
737                    ))));
738                }
739                return Err(SupervisorError::Anyhow(error));
740            }
741        };
742
743        if !response.ok {
744            let error = response.error.unwrap_or(crate::protocol::ProtocolError {
745                code: ProtocolErrorCode::Internal,
746                message: "request failed without error body".to_string(),
747            });
748            return Err(match error.code {
749                ProtocolErrorCode::StaleGeneration => SupervisorError::StaleGeneration,
750                code => SupervisorError::Protocol {
751                    code,
752                    message: error.message,
753                },
754            });
755        }
756        response
757            .result
758            .ok_or_else(|| SupervisorError::Anyhow(anyhow!("ok response missing result")))
759    }
760
761    async fn respond_to_host_request(
762        &mut self,
763        request: HostRequestFrame,
764        expected_generation: u64,
765    ) -> Result<()> {
766        if request.schema != PROTOCOL_SCHEMA || request.kind != "host_request" {
767            bail!("omega-effectd emitted an invalid host request envelope");
768        }
769        if request.id.is_empty() || request.id.len() > 180 {
770            bail!("omega-effectd emitted an invalid host request id");
771        }
772
773        let response = if request.generation != expected_generation
774            || request.generation != self.generation()
775        {
776            HostResponseFrame::failure(
777                &request,
778                HostResponseError {
779                    code: HostResponseErrorCode::StaleGeneration,
780                    message: format!(
781                        "Host request generation {} does not match active generation {}.",
782                        request.generation,
783                        self.generation()
784                    ),
785                },
786            )
787        } else if request.method == HostMethod::Unsupported {
788            HostResponseFrame::failure(
789                &request,
790                HostResponseError {
791                    code: HostResponseErrorCode::Unsupported,
792                    message: "The requested Omega host method is unsupported.".to_string(),
793                },
794            )
795        } else if let Some(handler) = self.host_handler.clone() {
796            let host_request_timeout = self.host_request_timeout;
797            let result = smol::future::or(handler(request.clone()), async move {
798                runtime_delay(host_request_timeout).await;
799                Err(HostResponseError::unavailable(format!(
800                    "Omega host authority timed out after {host_request_timeout:?}."
801                )))
802            })
803            .await;
804            match result {
805                Ok(result) => HostResponseFrame::success(&request, result),
806                Err(mut error) => {
807                    error.message = truncate_utf8(
808                        &redact_command(&error.message),
809                        MAX_HOST_ERROR_MESSAGE_BYTES,
810                    );
811                    HostResponseFrame::failure(&request, error)
812                }
813            }
814        } else {
815            HostResponseFrame::failure(
816                &request,
817                HostResponseError::unavailable(format!(
818                    "Omega host authority for {:?} is unavailable.",
819                    request.method
820                )),
821            )
822        };
823        self.write_frame(&response).await
824    }
825
826    async fn write_frame(&mut self, frame: &impl serde::Serialize) -> Result<()> {
827        let line = serde_json::to_string(frame).context("encode omega-effectd host response")?;
828        if line.len() > MAX_FRAME_BYTES {
829            bail!("omega-effectd host response frame exceeds {MAX_FRAME_BYTES} bytes");
830        }
831        let stdin = self
832            .stdin
833            .as_mut()
834            .ok_or_else(|| anyhow!("omega-effectd stdin missing"))?;
835        stdin.write_all(line.as_bytes()).await?;
836        stdin.write_all(b"\n").await?;
837        stdin.flush().await?;
838        Ok(())
839    }
840}
841
842impl Drop for OmegaEffectdSupervisor {
843    fn drop(&mut self) {
844        if let Some(mut child) = self.child.take() {
845            child.kill().log_err();
846        }
847    }
848}
849
850#[allow(clippy::disallowed_methods)]
851async fn runtime_delay(duration: Duration) {
852    // The supervisor is also used without a GPUI application context by protocol clients and tests.
853    smol::Timer::after(duration).await;
854}
855
856fn truncate_utf8(message: &str, max_bytes: usize) -> String {
857    if message.len() <= max_bytes {
858        return message.to_string();
859    }
860    let mut boundary = max_bytes;
861    while !message.is_char_boundary(boundary) {
862        boundary -= 1;
863    }
864    message[..boundary].to_string()
865}
866
867async fn read_bounded_line(
868    reader: &mut BufReader<smol::process::ChildStdout>,
869) -> Result<Option<String>> {
870    let mut frame = Vec::new();
871    loop {
872        let (consumed, found_newline) = {
873            let available = reader.fill_buf().await?;
874            if available.is_empty() {
875                if frame.is_empty() {
876                    return Ok(None);
877                }
878                bail!("omega-effectd closed stdout with an incomplete frame");
879            }
880            let consumed = available
881                .iter()
882                .position(|byte| *byte == b'\n')
883                .map_or(available.len(), |index| index + 1);
884            let payload_length = if available.get(consumed.saturating_sub(1)) == Some(&b'\n') {
885                consumed - 1
886            } else {
887                consumed
888            };
889            if frame.len() + payload_length > MAX_FRAME_BYTES {
890                bail!("omega-effectd response frame exceeds {MAX_FRAME_BYTES} bytes");
891            }
892            frame.extend_from_slice(&available[..payload_length]);
893            (consumed, payload_length < consumed)
894        };
895        reader.consume_unpin(consumed);
896        if found_newline {
897            if frame.last() == Some(&b'\r') {
898                frame.pop();
899            }
900            return String::from_utf8(frame)
901                .context("omega-effectd response frame was not UTF-8")
902                .map(Some);
903        }
904    }
905}
906
907pub fn resolve_effectd_command(
908    override_program: Option<&OsStr>,
909    app_executable: &Path,
910) -> Result<OmegaEffectdCommand> {
911    if let Some(program) = override_program.filter(|value| !value.is_empty()) {
912        let program = PathBuf::from(program);
913        if !program.is_file() {
914            bail!(
915                "OPENAGENTS_OMEGA_EFFECTD_BIN does not name a packaged executable: {}",
916                program.display()
917            );
918        }
919        return Ok(OmegaEffectdCommand {
920            program,
921            args: Vec::new(),
922        });
923    }
924
925    let executable_dir = app_executable
926        .parent()
927        .ok_or_else(|| anyhow!("Omega executable has no parent directory"))?;
928    let bundled_program = if executable_dir.file_name() == Some(OsStr::new("MacOS")) {
929        executable_dir
930            .parent()
931            .ok_or_else(|| anyhow!("Omega macOS bundle has no Contents directory"))?
932            .join("Resources/omega-effectd/bin/omega-effectd")
933    } else {
934        executable_dir.join("omega-effectd/bin/omega-effectd")
935    };
936    if !bundled_program.is_file() {
937        bail!(
938            "packaged omega-effectd component is unavailable at {}",
939            bundled_program.display()
940        );
941    }
942    Ok(OmegaEffectdCommand {
943        program: bundled_program,
944        args: Vec::new(),
945    })
946}
947
948/// Shared test helper: fixture command that speaks the framed protocol.
949pub fn fixture_command(fixture: &Path) -> OmegaEffectdCommand {
950    let node = [
951        std::env::var_os("NODE")
952            .map(PathBuf::from)
953            .filter(|path| path.exists()),
954        which_node(),
955        Some(PathBuf::from(
956            "/Users/christopherdavid/.nvm/versions/node/v25.8.2/bin/node",
957        ))
958        .filter(|path| path.exists()),
959        Some(PathBuf::from("/opt/homebrew/bin/node")).filter(|path| path.exists()),
960        Some(PathBuf::from("/usr/local/bin/node")).filter(|path| path.exists()),
961    ]
962    .into_iter()
963    .flatten()
964    .next()
965    .unwrap_or_else(|| PathBuf::from("node"));
966
967    OmegaEffectdCommand {
968        program: node,
969        args: vec![fixture.display().to_string()],
970    }
971}
972
973fn which_node() -> Option<PathBuf> {
974    std::env::var_os("PATH").and_then(|paths| {
975        for dir in std::env::split_paths(&paths) {
976            let candidate = dir.join("node");
977            if candidate.is_file() {
978                return Some(candidate);
979            }
980        }
981        None
982    })
983}
984
985pub fn default_options(
986    data_root: PathBuf,
987    command: OmegaEffectdCommand,
988) -> OmegaEffectdSupervisorOptions {
989    OmegaEffectdSupervisorOptions {
990        data_root,
991        command,
992        initial_generation: 1,
993        request_timeout: DEFAULT_REQUEST_TIMEOUT,
994    }
995}
996