tenant.openagents/omega

1//! Driving Exo. `OMEGA-DELTA-0042`, omega#87.
2//!
3//! The law lives in `crates/omega_exo_lane`, which is a leaf. This file starts
4//! `exo acp` on standard input and output. `AcpConnection` puts each text delta,
5//! tool call, tool result, and completion record into the existing `AcpThread`.
6//!
7//! # How the lane is reached
8//!
9//! Through the router omega#78 already built, and through nothing else. The Exo
10//! connection registers as [`OmegaAgentConnection`]'s external executor, so a
11//! thread reaches it exactly when a person pins `ExecutorClass::ExternalAcp` on
12//! the disclosure line's pin control — a `PinGesture`, which has no variant for
13//! a tool call, a slash command, a restored draft, or a model turn. Nothing here
14//! adds a way to start authority: an unpinned thread still runs on the native
15//! loop, and when no Exo lane is configured the pin falls back with
16//! `RouteReason::ExternalAcpUnavailable`, which the user sees on the line.
17//!
18//! # What runs before every turn
19//!
20//! Four refusals, in order, all of them from live observation rather than from
21//! configuration Omega wrote down earlier:
22//!
23//! 1. **Where Exo is.** `EXO_EXOHARNESS_URL`, which Exo reads from the
24//!    environment and which redirects it from the state root on disk to an
25//!    unauthenticated HTTP server, is parsed through [`LoopbackEndpoint`] and
26//!    refused unless it names this machine.
27//! 2. **Which Exo.** `git` in the checkout answers with a remote, a commit, and
28//!    a tree, and [`EXO_PIN`] admits or refuses it. This is what stops the
29//!    omega#86 mistake — the other Exo — and an upstream that moved under a
30//!    no-backwards-compatibility house rule.
31//! 3. **Which bytes.** The binary is hashed and compared against the owner's
32//!    pin ledger entry for `exo`, when the owner froze one.
33//! 4. **Which capability.** Omega reads the agent and conversation. The normal
34//!    path refuses self-modification. A one-use grant can authorize one exact
35//!    draft after a visible human confirmation.
36//!
37//! # What is deliberately absent
38//!
39//! No listener, no bind, no port, no proxy. Omega never stands between Exo's
40//! unauthenticated endpoint and anything else — [`LoopbackEndpoint`] exists so
41//! that a future caller who *does* need an address cannot build a non-loopback
42//! one, and Tier A does not need an address at all, because the CLI talks to
43//! the state root on disk. `omega_deltas` asserts the absence, because "we did
44//! not add a listener" is the kind of thing that stays true only while somebody
45//! is checking.
46
47use std::any::Any;
48use std::cell::{Cell, RefCell};
49use std::path::PathBuf;
50use std::process::Stdio;
51use std::rc::Rc;
52use std::sync::atomic::{AtomicU64, Ordering};
53use std::time::{SystemTime, UNIX_EPOCH};
54
55use acp_thread::{AcpThread, AgentConnection};
56use agent_client_protocol::schema::v1 as acp;
57use agent_servers::AcpConnection;
58use anyhow::{Context as _, Result, anyhow, bail};
59use gpui::{App, AsyncApp, Entity, SharedString, Task, WeakEntity};
60use omega_exo_lane::{
61    EXO_HARNESS_ID, EXO_PIN, ExoAgent, ExoCommand, ExoConversation, ExoLaneIdentity,
62    ExoModelBinding, ExoMount, ExoRoot, ExoSelfModificationConsentOrigin, ExoSelfModificationGrant,
63    ExoSelfModificationGrantRequest, ExoSelfModificationReceipt, LoopbackEndpoint,
64    ObservedExoCapabilityState, ObservedExoCheckout, ObservedReadWriteMount, ObservedToolModule,
65    admits_bytes,
66};
67use omega_harness::MeasuredDigest;
68use project::{
69    AgentId, Project,
70    agent_server_store::{AgentServerCommand, AgentServerStore},
71};
72use util::path_list::PathList;
73
74/// Where the lane's configuration lives, under the Omega data directory.
75const EXO_LANE_FILE: &str = "omega-exo-lane.json";
76
77/// The schema that file carries.
78const EXO_LANE_SCHEMA: &str = "openagents.omega.exo_lane.v1";
79
80static NEXT_EXO_CONNECTION_GENERATION: AtomicU64 = AtomicU64::new(1);
81
82/// Everything Omega needs to reach one Exo install.
83///
84/// Every field names something Exo owns. None of them is something Omega
85/// writes: the binary was built by whoever installed Exo, the checkout is
86/// theirs, and the state root is Exo's — Omega only ever passes its path on a
87/// command line. See `ExoRoot`.
88#[derive(Clone, Debug, PartialEq, Eq)]
89pub struct ExoLaneConfig {
90    /// The `exo` binary.
91    pub binary: PathBuf,
92    /// The checkout the binary was built from, for the pin check.
93    pub checkout: PathBuf,
94    /// Exo's state root.
95    pub root: ExoRoot,
96    /// The agent slug the lane sends to.
97    pub agent: String,
98    /// The conversation slug the lane sends to.
99    pub conversation: String,
100}
101
102impl ExoLaneConfig {
103    /// The lane file's path under the Omega data directory.
104    #[must_use]
105    pub fn data_dir_path() -> PathBuf {
106        paths::data_dir().join("openagents").join(EXO_LANE_FILE)
107    }
108
109    /// Read the lane file, if the owner wrote one.
110    ///
111    /// A machine with no Exo has no lane, and that is the ordinary case rather
112    /// than an error: `None` means the router has no external executor
113    /// registered and a pin to one falls back visibly. A file that exists and
114    /// cannot be read is logged and treated as no lane, because a half-read
115    /// configuration is how a lane ends up pointed at the wrong `.exo`.
116    #[must_use]
117    pub fn load(path: &std::path::Path) -> Option<Self> {
118        let file = std::fs::read_to_string(path).ok()?;
119        let value: serde_json::Value = serde_json::from_str(&file)
120            .inspect_err(|error| {
121                log::warn!(
122                    "OMEGA-DELTA-0042: the Exo lane file is not JSON ({error}); no Exo lane"
123                );
124            })
125            .ok()?;
126        if value.get("schema").and_then(serde_json::Value::as_str) != Some(EXO_LANE_SCHEMA) {
127            log::warn!("OMEGA-DELTA-0042: the Exo lane file carries an unsupported schema");
128            return None;
129        }
130        let field = |name: &str| {
131            value
132                .get(name)
133                .and_then(serde_json::Value::as_str)
134                .filter(|value| !value.trim().is_empty())
135                .map(str::to_owned)
136        };
137        Some(Self {
138            binary: PathBuf::from(field("binary")?),
139            checkout: PathBuf::from(field("checkout")?),
140            root: ExoRoot::at(field("root")?),
141            agent: field("agent")?,
142            conversation: field("conversation")?,
143        })
144    }
145}
146
147/// Everything a turn needs, separated from the connection that owns it.
148///
149/// Held behind an `Rc` so a turn can be *moved into the spawned task* rather
150/// than run on the thread that draws the window. `AgentConnection::prompt`
151/// takes `&self`, so without this split the only way to reach the Exo process
152/// from `prompt` would be to run it inline — which is exactly the blocking call
153/// the workspace's clippy configuration disallows, and it disallows it because
154/// an Exo turn calls a model and takes seconds.
155struct ExoDriver {
156    config: ExoLaneConfig,
157    /// What Exo said about itself, cached so a disclosure render does not spawn
158    /// a process. Refreshed by every turn, because the turn re-reads the agent
159    /// anyway to check its capability.
160    identity: RefCell<Option<ExoLaneIdentity>>,
161    /// The digest the owner froze for `exo`, if any. Read once at construction
162    /// from the harness pin ledger; a claim, never a measurement.
163    frozen_digest: Option<String>,
164    /// This connection process's generation. A grant cannot survive a
165    /// reconnect because a new connection receives a new generation.
166    generation: u64,
167}
168
169#[derive(Clone, Copy, Debug, PartialEq, Eq)]
170pub enum ExoInspectionPhase {
171    NotLoaded,
172    Refreshing,
173    Ready,
174    Unavailable,
175}
176
177#[derive(Clone, Debug, PartialEq, Eq)]
178pub struct ExoInspectionSnapshot {
179    pub phase: ExoInspectionPhase,
180    pub observed: Option<ObservedExoCapabilityState>,
181    pub identity: Option<ExoLaneIdentity>,
182    pub networking: Option<bool>,
183    pub refreshed_at_ms: Option<u64>,
184    pub error: Option<String>,
185}
186
187impl Default for ExoInspectionSnapshot {
188    fn default() -> Self {
189        Self {
190            phase: ExoInspectionPhase::NotLoaded,
191            observed: None,
192            identity: None,
193            networking: None,
194            refreshed_at_ms: None,
195            error: None,
196        }
197    }
198}
199
200#[derive(Clone, Copy, Debug, PartialEq, Eq)]
201pub enum ExoTurnPhase {
202    Idle,
203    Inspecting,
204    Working,
205    Cancelling,
206    Completed,
207    Cancelled,
208    Refused,
209    Failed,
210}
211
212impl ExoTurnPhase {
213    #[must_use]
214    pub const fn label(self) -> &'static str {
215        match self {
216            Self::Idle => "Ready",
217            Self::Inspecting => "Checking runtime",
218            Self::Working => "Working",
219            Self::Cancelling => "Cancelling",
220            Self::Completed => "Completed",
221            Self::Cancelled => "Cancelled",
222            Self::Refused => "Refused",
223            Self::Failed => "Failed",
224        }
225    }
226}
227
228#[derive(Clone, Debug, PartialEq, Eq)]
229pub struct ExoTurnSnapshot {
230    pub phase: ExoTurnPhase,
231    pub detail: Option<String>,
232    pub exo_session_id: Option<String>,
233    pub exo_turn_id: Option<String>,
234    pub latest_event_id: Option<String>,
235    pub updated_at_ms: u64,
236}
237
238impl Default for ExoTurnSnapshot {
239    fn default() -> Self {
240        Self {
241            phase: ExoTurnPhase::Idle,
242            detail: None,
243            exo_session_id: None,
244            exo_turn_id: None,
245            latest_event_id: None,
246            updated_at_ms: now_ms(),
247        }
248    }
249}
250
251#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
252pub struct ExoTierCReceipt {
253    pub schema: String,
254    pub recorded_at_ms: u64,
255    pub authority: Option<ExoSelfModificationReceipt>,
256    pub observed: ObservedExoCapabilityState,
257    pub turn_ref: String,
258    pub requested_objective: Option<String>,
259    pub outcome: String,
260    pub exo_session_id: Option<String>,
261    pub exo_turn_id: Option<String>,
262    pub latest_event_id: Option<String>,
263    pub verification: String,
264    pub reconnect: String,
265    pub rollback: String,
266}
267
268/// One Exo install, behind Omega Agent's router.
269pub struct ExoHarnessConnection {
270    driver: Rc<ExoDriver>,
271    acp: Rc<AcpConnection>,
272    pending_grant: RefCell<Option<ExoSelfModificationGrant>>,
273    tier_c_receipt: Rc<RefCell<Option<ExoTierCReceipt>>>,
274    active_tier_c_turn: Rc<Cell<bool>>,
275    inspection: Rc<RefCell<ExoInspectionSnapshot>>,
276    turn: Rc<RefCell<ExoTurnSnapshot>>,
277}
278
279impl ExoHarnessConnection {
280    /// A connection to the Exo the lane file names.
281    #[must_use]
282    pub fn new(
283        config: ExoLaneConfig,
284        frozen_digest: Option<String>,
285        acp: Rc<AcpConnection>,
286    ) -> Self {
287        let previous_receipt = load_latest_tier_c_receipt(&config.agent, &config.conversation);
288        Self {
289            driver: Rc::new(ExoDriver {
290                config,
291                identity: RefCell::new(None),
292                frozen_digest,
293                generation: NEXT_EXO_CONNECTION_GENERATION.fetch_add(1, Ordering::Relaxed),
294            }),
295            acp,
296            pending_grant: RefCell::new(None),
297            tier_c_receipt: Rc::new(RefCell::new(previous_receipt)),
298            active_tier_c_turn: Rc::new(Cell::new(false)),
299            inspection: Rc::new(RefCell::new(ExoInspectionSnapshot::default())),
300            turn: Rc::new(RefCell::new(ExoTurnSnapshot::default())),
301        }
302    }
303
304    /// What this lane discloses, as far as it has been able to observe.
305    ///
306    /// `None` before the first turn: nothing has asked Exo who it is yet, and
307    /// inventing an executor name would be the fabrication the disclosure
308    /// record exists to prevent.
309    #[must_use]
310    pub fn identity(&self) -> Option<ExoLaneIdentity> {
311        self.driver.identity.borrow().clone()
312    }
313
314    /// The lane's configuration.
315    #[must_use]
316    pub fn config(&self) -> &ExoLaneConfig {
317        &self.driver.config
318    }
319
320    #[must_use]
321    pub fn tier_c_receipt(&self) -> Option<ExoTierCReceipt> {
322        self.tier_c_receipt.borrow().clone()
323    }
324
325    #[must_use]
326    pub fn inspection(&self) -> ExoInspectionSnapshot {
327        self.inspection.borrow().clone()
328    }
329
330    #[must_use]
331    pub fn turn(&self) -> ExoTurnSnapshot {
332        self.turn.borrow().clone()
333    }
334
335    fn set_turn(&self, phase: ExoTurnPhase, detail: Option<String>) {
336        *self.turn.borrow_mut() = ExoTurnSnapshot {
337            phase,
338            detail,
339            exo_session_id: None,
340            exo_turn_id: None,
341            latest_event_id: None,
342            updated_at_ms: now_ms(),
343        };
344    }
345
346    pub fn refresh_inspection(&self, cx: &mut App) -> Task<Result<()>> {
347        {
348            let mut inspection = self.inspection.borrow_mut();
349            inspection.phase = ExoInspectionPhase::Refreshing;
350            inspection.error = None;
351        }
352        let driver = self.driver.clone();
353        let inspection = self.inspection.clone();
354        cx.spawn(async move |_| match driver.observe().await {
355            Ok(observed) => {
356                *inspection.borrow_mut() = ExoInspectionSnapshot {
357                    phase: ExoInspectionPhase::Ready,
358                    observed: Some(observed.observed),
359                    identity: Some(observed.identity),
360                    networking: Some(observed.networking),
361                    refreshed_at_ms: Some(now_ms()),
362                    error: None,
363                };
364                Ok(())
365            }
366            Err(error) => {
367                let message = error.to_string();
368                let mut snapshot = inspection.borrow_mut();
369                snapshot.phase = ExoInspectionPhase::Unavailable;
370                snapshot.refreshed_at_ms = Some(now_ms());
371                snapshot.error = Some(message.clone());
372                Err(anyhow!(message))
373            }
374        })
375    }
376
377    /// Observe the exact Exo capability state for a dedicated confirmation
378    /// dialog. This does not mint authority.
379    pub async fn self_modification_request(
380        &self,
381        objective: String,
382        turn_ref: String,
383    ) -> Result<ExoSelfModificationGrantRequest> {
384        let observed = self.driver.observe().await?.observed;
385        let capabilities = observed.requested_capabilities();
386        if capabilities.is_empty() {
387            bail!("this Exo agent has no self-modification capability to authorize");
388        }
389        Ok(ExoSelfModificationGrantRequest {
390            objective,
391            turn_ref,
392            observed,
393            capabilities,
394            expires_at_ms: now_ms().saturating_add(60_000),
395        })
396    }
397
398    /// Mint the one-use grant after the visible confirmation action returns.
399    pub fn confirm_self_modification(
400        &self,
401        request: ExoSelfModificationGrantRequest,
402    ) -> Result<()> {
403        let grant = ExoSelfModificationGrant::mint(
404            request,
405            ExoSelfModificationConsentOrigin::HumanConfirmationDialog,
406            now_ms(),
407        )
408        .map_err(|refusal| anyhow!("Exo self-modification grant refused: {refusal:?}"))?;
409        *self.pending_grant.borrow_mut() = Some(grant);
410        Ok(())
411    }
412}
413
414struct ObservedTurn {
415    observed: ObservedExoCapabilityState,
416    identity: ExoLaneIdentity,
417    networking: bool,
418}
419
420fn now_ms() -> u64 {
421    SystemTime::now()
422        .duration_since(UNIX_EPOCH)
423        .map_or(0, |duration| {
424            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
425        })
426}
427
428fn set_turn_snapshot(turn: &RefCell<ExoTurnSnapshot>, phase: ExoTurnPhase, detail: Option<String>) {
429    *turn.borrow_mut() = ExoTurnSnapshot {
430        phase,
431        detail,
432        exo_session_id: None,
433        exo_turn_id: None,
434        latest_event_id: None,
435        updated_at_ms: now_ms(),
436    };
437}
438
439fn set_terminal_turn_snapshot(
440    turn: &RefCell<ExoTurnSnapshot>,
441    phase: ExoTurnPhase,
442    detail: Option<String>,
443    response: &acp::PromptResponse,
444) {
445    let meta = response.meta.as_ref();
446    *turn.borrow_mut() = ExoTurnSnapshot {
447        phase,
448        detail,
449        exo_session_id: meta_value(meta, "exo.session_id"),
450        exo_turn_id: meta_value(meta, "exo.turn_id"),
451        latest_event_id: meta_value(meta, "exo.latest_event_id"),
452        updated_at_ms: now_ms(),
453    };
454}
455
456fn resolve_module_host_path(module: &str, mounts: &[ExoMount]) -> Option<PathBuf> {
457    let path = std::path::Path::new(module);
458    if path.is_file() {
459        return Some(path.to_path_buf());
460    }
461    mounts.iter().find_map(|mount| {
462        let suffix = path.strip_prefix(&mount.mount_path).ok()?;
463        Some(PathBuf::from(&mount.host_path).join(suffix))
464    })
465}
466
467#[must_use]
468pub fn exo_prompt_objective(prompt: &[acp::ContentBlock]) -> String {
469    prompt
470        .iter()
471        .filter_map(|block| match block {
472            acp::ContentBlock::Text(text) => Some(text.text.as_str()),
473            _ => None,
474        })
475        .collect::<Vec<_>>()
476        .join("\n")
477}
478
479pub fn exo_turn_ref(session_id: &acp::SessionId, prompt: &[acp::ContentBlock]) -> Result<String> {
480    let canonical = serde_json::to_vec(&(session_id, prompt))
481        .context("encoding the exact Exo turn for its authority grant")?;
482    Ok(format!(
483        "exo-turn:{}",
484        MeasuredDigest::measure(&canonical).as_str()
485    ))
486}
487
488impl ExoDriver {
489    /// Run one `exo` command and return its stdout.
490    ///
491    /// Async, and spawned rather than blocking, because an Exo turn calls a
492    /// model: it takes seconds, and running it on the thread that draws the
493    /// window would freeze Omega for the length of somebody else's inference.
494    async fn run(&self, command: &ExoCommand) -> Result<String> {
495        let argv = command.argv(&self.config.root);
496        let output = smol::process::Command::new(&self.config.binary)
497            .args(&argv)
498            .stdin(Stdio::null())
499            .output()
500            .await
501            .with_context(|| {
502                format!(
503                    "running {} {}",
504                    self.config.binary.display(),
505                    argv.join(" ")
506                )
507            })?;
508        if !output.status.success() {
509            bail!(
510                "exo {} exited {}: {}",
511                command.shape(),
512                output.status,
513                String::from_utf8_lossy(&output.stderr).trim()
514            );
515        }
516        Ok(String::from_utf8_lossy(&output.stdout).into_owned())
517    }
518
519    /// Refuse an Exo that is not on this machine.
520    ///
521    /// Exo reads `EXO_EXOHARNESS_URL` from its environment and, when it is set,
522    /// stops using the state root on disk and talks to an `exo serve` over
523    /// HTTP instead. That server has **no authentication** — a client may send
524    /// a bearer token and the server never checks it — and it exposes Exo's
525    /// whole request protocol, secrets included. Loopback is the entire
526    /// boundary, and Exo's own documentation says so.
527    ///
528    /// Omega never passes that flag; [`ExoCommand`] cannot express it. But the
529    /// lane inherits Omega's environment, which is how Exo is meant to receive
530    /// its own configuration, and a variable set there would redirect the lane
531    /// off-machine without a single Omega command line changing. So the
532    /// environment is read and parsed through [`LoopbackEndpoint`], whose only
533    /// constructor refuses anything that is not this machine.
534    fn check_endpoint(&self) -> Result<()> {
535        let Ok(url) = std::env::var("EXO_EXOHARNESS_URL") else {
536            // Unset is the ordinary case and the safe one: the CLI reads the
537            // state root on disk and no socket exists at all.
538            return Ok(());
539        };
540        let endpoint = LoopbackEndpoint::parse(&url).map_err(|refusal| {
541            anyhow!(
542                "{refusal}: the Exo lane will not talk to {url}. Exo's server has no \
543                 authentication and full access to its secrets."
544            )
545        })?;
546        log::info!("OMEGA-DELTA-0042: the Exo lane is pointed at {endpoint}, which is loopback");
547        Ok(())
548    }
549
550    /// Refuse an Exo that is not the pinned one.
551    async fn check_pin(&self) -> Result<(ObservedExoCheckout, MeasuredDigest)> {
552        let git = async |args: &[&str]| -> Result<String> {
553            let output = smol::process::Command::new("git")
554                .arg("-C")
555                .arg(&self.config.checkout)
556                .args(args)
557                .stdin(Stdio::null())
558                .output()
559                .await
560                .context("reading the Exo checkout with git")?;
561            if !output.status.success() {
562                bail!("git {args:?} failed in the Exo checkout");
563            }
564            Ok(String::from_utf8_lossy(&output.stdout).trim().to_owned())
565        };
566        let observed = ObservedExoCheckout {
567            upstream: git(&["config", "--get", "remote.origin.url"]).await?,
568            commit: git(&["rev-parse", "HEAD"]).await?,
569            tree: git(&["rev-parse", "HEAD^{tree}"]).await?,
570        };
571        EXO_PIN.admits(&observed).map_err(|mismatch| {
572            anyhow!(
573                "{mismatch}: the Exo lane is pinned to {} at {}",
574                EXO_PIN.upstream,
575                EXO_PIN.source_commit
576            )
577        })?;
578
579        let bytes = smol::fs::read(&self.config.binary)
580            .await
581            .context("reading the exo binary")?;
582        let digest = MeasuredDigest::measure(&bytes);
583        admits_bytes(self.frozen_digest.as_deref(), &digest)
584            .map_err(|mismatch| anyhow!("{mismatch}"))?;
585        Ok((observed, digest))
586    }
587
588    /// Read the exact agent and conversation capability state.
589    async fn observe(&self) -> Result<ObservedTurn> {
590        self.check_endpoint()?;
591        let (checkout, binary_digest) = self.check_pin().await?;
592        let shown = self
593            .run(&ExoCommand::ShowAgent {
594                agent: self.config.agent.clone(),
595            })
596            .await?;
597        let agent = ExoAgent::parse(&shown).map_err(|error| {
598            anyhow!("{error}; the Omega lane refuses an Exo agent it cannot read")
599        })?;
600        let shown_conversation = self
601            .run(&ExoCommand::ShowConversation {
602                agent: self.config.agent.clone(),
603                conversation: self.config.conversation.clone(),
604            })
605            .await?;
606        let conversation = ExoConversation::parse(&shown_conversation).map_err(|error| {
607            anyhow!("{error}; the Omega lane refuses an Exo conversation it cannot read")
608        })?;
609
610        let bindings = ExoModelBinding::read_table(&self.run(&ExoCommand::ListModels).await?);
611        let identity = ExoLaneIdentity::resolve(&agent, &bindings);
612        *self.identity.borrow_mut() = Some(identity.clone());
613        let mut mounts = agent.mounts.clone();
614        mounts.extend(conversation.mounts.clone());
615        let read_write_mounts = mounts
616            .iter()
617            .filter(|mount| mount.read_write)
618            .map(|mount| ObservedReadWriteMount {
619                host_path: mount.host_path.clone(),
620                mount_path: mount.mount_path.clone(),
621            })
622            .collect::<Vec<_>>();
623        let mut tool_modules = Vec::new();
624        for module in &agent.tool_module_paths {
625            let host_path = resolve_module_host_path(module, &mounts).ok_or_else(|| {
626                anyhow!(
627                    "the Exo tool module {module} is not a readable host file or inside an observed mount"
628                )
629            })?;
630            let bytes = smol::fs::read(&host_path)
631                .await
632                .with_context(|| format!("reading Exo tool module {}", host_path.display()))?;
633            tool_modules.push(ObservedToolModule {
634                path: module.clone(),
635                digest: MeasuredDigest::measure(&bytes).as_str().to_owned(),
636            });
637        }
638        tool_modules.sort();
639        let mut read_write_mounts = read_write_mounts;
640        read_write_mounts.sort();
641        Ok(ObservedTurn {
642            observed: ObservedExoCapabilityState {
643                source_commit: checkout.commit,
644                source_tree: checkout.tree,
645                binary_digest: binary_digest.as_str().to_owned(),
646                agent: agent.slug,
647                conversation: conversation.slug,
648                generation: self.generation,
649                agent_authored_tools: agent.agent_authored_tools,
650                tool_modules,
651                read_write_mounts,
652            },
653            identity,
654            networking: agent.networking,
655        })
656    }
657
658    /// Re-observe the exact executable and agent immediately before a streamed
659    /// ACP turn. A self-modifying turn needs a matching one-use grant.
660    async fn preflight(&self) -> Result<ObservedTurn> {
661        self.observe().await
662    }
663}
664
665/// The Exo lane the owner configured, if they configured one.
666///
667/// Called once, when the router is built. A machine with no lane file gets
668/// `None` and the router registers no external executor — which is the ordinary
669/// case, and is why the return type is an `Option` rather than a `Result` that
670/// every caller would have to decide to ignore.
671pub async fn connect_configured_lane(
672    lane_path: &std::path::Path,
673    project: Entity<Project>,
674    agent_server_store: WeakEntity<AgentServerStore>,
675    cx: &mut AsyncApp,
676) -> Result<Option<Rc<dyn AgentConnection>>> {
677    let Some(config) = ExoLaneConfig::load(lane_path) else {
678        return Ok(None);
679    };
680    let frozen = frozen_exo_digest();
681    log::info!(
682        "OMEGA-DELTA-0042: Exo harness lane configured at {} ({} {}), pin {}",
683        config.root.as_str(),
684        config.agent,
685        config.conversation,
686        if frozen.is_some() {
687            "frozen"
688        } else {
689            "unfrozen"
690        }
691    );
692    let command = AgentServerCommand {
693        path: config.binary.clone(),
694        args: vec![
695            "--root".to_owned(),
696            config.root.as_str().to_owned(),
697            "acp".to_owned(),
698            config.agent.clone(),
699            config.conversation.clone(),
700        ],
701        env: None,
702    };
703    let acp = Rc::new(
704        AcpConnection::stdio(
705            AgentId::new(EXO_HARNESS_ID),
706            project,
707            command,
708            agent_server_store,
709            None,
710            Default::default(),
711            cx,
712        )
713        .await?,
714    );
715    Ok(Some(Rc::new(ExoHarnessConnection::new(
716        config, frozen, acp,
717    ))))
718}
719
720/// The digest the owner froze for `exo`, from the harness pin ledger.
721///
722/// Read here rather than inside the connection so the connection takes a value
723/// and stays testable without a filesystem. A ledger that cannot be read is not
724/// an empty ledger — but it is also not this file's decision, so the absence is
725/// logged and the pin check proceeds unfrozen, exactly as it does on a machine
726/// where the owner never froze Exo.
727fn frozen_exo_digest() -> Option<String> {
728    let path = paths::data_dir()
729        .join("openagents")
730        .join("external_agents")
731        .join(omega_harness::HARNESS_PIN_LEDGER_FILE_NAME);
732    let file = std::fs::read_to_string(path).ok()?;
733    match omega_harness::decode_harness_pin_ledger(&file) {
734        Ok(ledger) => ledger.pin(EXO_HARNESS_ID).map(|pin| pin.digest.clone()),
735        Err(error) => {
736            log::warn!("OMEGA-DELTA-0042: the harness pin ledger could not be read ({error})");
737            None
738        }
739    }
740}
741
742fn meta_value(meta: Option<&acp::Meta>, key: &str) -> Option<String> {
743    meta.and_then(|meta| meta.get(key))
744        .and_then(serde_json::Value::as_str)
745        .map(str::to_owned)
746}
747
748fn tier_c_receipt_path() -> PathBuf {
749    paths::data_dir()
750        .join("openagents")
751        .join("exo-self-modification-receipts.jsonl")
752}
753
754fn load_latest_tier_c_receipt(agent: &str, conversation: &str) -> Option<ExoTierCReceipt> {
755    std::fs::read_to_string(tier_c_receipt_path())
756        .ok()?
757        .lines()
758        .rev()
759        .filter_map(|line| serde_json::from_str::<ExoTierCReceipt>(line).ok())
760        .find(|receipt| {
761            receipt.observed.agent == agent && receipt.observed.conversation == conversation
762        })
763}
764
765fn refused_tier_c_receipt(
766    observed: ObservedExoCapabilityState,
767    turn_ref: String,
768    requested_objective: Option<String>,
769    outcome: String,
770) -> ExoTierCReceipt {
771    ExoTierCReceipt {
772        schema: "openagents.omega.exo_self_modification_receipt.v1".to_owned(),
773        recorded_at_ms: now_ms(),
774        authority: None,
775        observed,
776        turn_ref,
777        requested_objective,
778        outcome,
779        exo_session_id: None,
780        exo_turn_id: None,
781        latest_event_id: None,
782        verification: "Omega refused the turn before it crossed ACP".to_owned(),
783        reconnect: "not applicable; the turn did not start".to_owned(),
784        rollback: "not applicable; the turn did not start".to_owned(),
785    }
786}
787
788async fn persist_tier_c_receipt(receipt: ExoTierCReceipt) -> Result<()> {
789    let path = tier_c_receipt_path();
790    smol::unblock(move || -> Result<()> {
791        use std::io::Write as _;
792        if let Some(parent) = path.parent() {
793            std::fs::create_dir_all(parent)
794                .with_context(|| format!("creating {}", parent.display()))?;
795        }
796        let mut file = std::fs::OpenOptions::new()
797            .create(true)
798            .append(true)
799            .open(&path)
800            .with_context(|| format!("opening {}", path.display()))?;
801        serde_json::to_writer(&mut file, &receipt)?;
802        file.write_all(b"\n")?;
803        file.sync_data()?;
804        Ok(())
805    })
806    .await
807}
808
809impl AgentConnection for ExoHarnessConnection {
810    /// A display label, and only that. omega#77 classifies by downcast because
811    /// this is renameable; see `omega_executor_disclosure`.
812    fn agent_id(&self) -> AgentId {
813        AgentId::new(
814            self.identity()
815                .map_or_else(|| EXO_HARNESS_ID.to_owned(), |identity| identity.agent_id()),
816        )
817    }
818
819    fn telemetry_id(&self) -> SharedString {
820        EXO_HARNESS_ID.into()
821    }
822
823    fn agent_version(&self) -> Option<SharedString> {
824        Some(EXO_PIN.version.into())
825    }
826
827    fn new_session(
828        self: Rc<Self>,
829        project: Entity<Project>,
830        work_dirs: PathList,
831        cx: &mut App,
832    ) -> Task<Result<Entity<AcpThread>>> {
833        let inner_session = self.acp.clone().new_session(project, work_dirs, cx);
834        let facade: Rc<dyn AgentConnection> = self;
835        cx.spawn(async move |cx| {
836            let thread = inner_session.await?;
837            thread.update(cx, |thread, _| {
838                thread.replace_connection(facade);
839            });
840            Ok(thread)
841        })
842    }
843
844    fn auth_methods(&self) -> &[acp::AuthMethod] {
845        &[]
846    }
847
848    fn authenticate(&self, _method: acp::AuthMethodId, _cx: &mut App) -> Task<Result<()>> {
849        Task::ready(Err(anyhow!(
850            "Exo holds its own credentials; Omega never edits them"
851        )))
852    }
853
854    fn prompt(
855        &self,
856        params: acp::PromptRequest,
857        cx: &mut App,
858    ) -> Task<Result<acp::PromptResponse>> {
859        self.set_turn(
860            ExoTurnPhase::Inspecting,
861            Some("Checking the pinned Exo runtime before this turn".to_owned()),
862        );
863        let driver = Rc::clone(&self.driver);
864        let acp = Rc::clone(&self.acp);
865        let pending_grant = self.pending_grant.take();
866        let receipt_cell = self.tier_c_receipt.clone();
867        let active_tier_c_turn = self.active_tier_c_turn.clone();
868        let inspection_cell = self.inspection.clone();
869        let turn_cell = self.turn.clone();
870        cx.spawn(async move |cx| {
871            let turn_ref = exo_turn_ref(&params.session_id, &params.prompt)?;
872            let observed = match driver.preflight().await {
873                Ok(observed) => observed,
874                Err(error) => {
875                    let message = error.to_string();
876                    {
877                        let mut inspection = inspection_cell.borrow_mut();
878                        inspection.phase = ExoInspectionPhase::Unavailable;
879                        inspection.refreshed_at_ms = Some(now_ms());
880                        inspection.error = Some(message.clone());
881                    }
882                    set_turn_snapshot(
883                        &turn_cell,
884                        ExoTurnPhase::Failed,
885                        Some(format!("Runtime check failed: {message}")),
886                    );
887                    return Err(error);
888                }
889            };
890            *inspection_cell.borrow_mut() = ExoInspectionSnapshot {
891                phase: ExoInspectionPhase::Ready,
892                observed: Some(observed.observed.clone()),
893                identity: Some(observed.identity.clone()),
894                networking: Some(observed.networking),
895                refreshed_at_ms: Some(now_ms()),
896                error: None,
897            };
898            let capabilities = observed.observed.requested_capabilities();
899            let authority_receipt = if capabilities.is_empty() {
900                None
901            } else {
902                let Some(grant) = pending_grant else {
903                    let message = "this Exo turn can modify itself; use the dedicated confirmation control for this exact draft";
904                    let receipt = refused_tier_c_receipt(
905                        observed.observed.clone(),
906                        turn_ref.clone(),
907                        None,
908                        format!("refused: {message}"),
909                    );
910                    *receipt_cell.borrow_mut() = Some(receipt.clone());
911                    persist_tier_c_receipt(receipt).await?;
912                    set_turn_snapshot(
913                        &turn_cell,
914                        ExoTurnPhase::Refused,
915                        Some(message.to_owned()),
916                    );
917                    bail!("{message}");
918                };
919                let requested_objective = Some(grant.request().objective.clone());
920                match grant.consume(&observed.observed, &turn_ref, now_ms()) {
921                    Ok(authority) => Some(authority),
922                    Err(refusal) => {
923                        let message = format!(
924                            "Exo self-modification grant refused: {refusal:?}"
925                        );
926                        let receipt = refused_tier_c_receipt(
927                            observed.observed.clone(),
928                            turn_ref.clone(),
929                            requested_objective,
930                            format!("refused: {refusal:?}"),
931                        );
932                        *receipt_cell.borrow_mut() = Some(receipt.clone());
933                        persist_tier_c_receipt(receipt).await?;
934                        set_turn_snapshot(
935                            &turn_cell,
936                            ExoTurnPhase::Refused,
937                            Some(message.clone()),
938                        );
939                        bail!("{message}");
940                    }
941                }
942            };
943            let is_tier_c_turn = authority_receipt.is_some();
944            if let Some(authority) = authority_receipt {
945                *receipt_cell.borrow_mut() = Some(ExoTierCReceipt {
946                    schema: "openagents.omega.exo_self_modification_receipt.v1".to_owned(),
947                    recorded_at_ms: now_ms(),
948                    observed: authority.observed.clone(),
949                    turn_ref: authority.turn_ref.clone(),
950                    requested_objective: Some(authority.objective.clone()),
951                    authority: Some(authority),
952                    outcome: "sent".to_owned(),
953                    exo_session_id: None,
954                    exo_turn_id: None,
955                    latest_event_id: None,
956                    verification: "waiting for Exo's durable completion receipt".to_owned(),
957                    reconnect: "not reported by Exo ACP".to_owned(),
958                    rollback: "not reported by Exo ACP".to_owned(),
959                });
960                active_tier_c_turn.set(true);
961                let receipt = receipt_cell.borrow().clone();
962                if let Some(receipt) = receipt {
963                    if let Err(error) = persist_tier_c_receipt(receipt).await {
964                        active_tier_c_turn.set(false);
965                        return Err(error);
966                    }
967                }
968            }
969            set_turn_snapshot(
970                &turn_cell,
971                ExoTurnPhase::Working,
972                Some("Streaming this turn over ACP".to_owned()),
973            );
974            let prompt = cx.update(|cx| acp.prompt(params, cx));
975            let response = prompt.await;
976            match &response {
977                Ok(response) => match response.stop_reason {
978                    acp::StopReason::Cancelled => set_terminal_turn_snapshot(
979                        &turn_cell,
980                        ExoTurnPhase::Cancelled,
981                        Some("Exo confirmed cancellation".to_owned()),
982                        response,
983                    ),
984                    _ => set_terminal_turn_snapshot(
985                        &turn_cell,
986                        ExoTurnPhase::Completed,
987                        Some("Exo completed the streamed turn".to_owned()),
988                        response,
989                    ),
990                },
991                Err(error) => set_turn_snapshot(
992                    &turn_cell,
993                    ExoTurnPhase::Failed,
994                    Some(error.to_string()),
995                ),
996            }
997            if is_tier_c_turn {
998                if let Some(receipt) = receipt_cell.borrow_mut().as_mut() {
999                    match &response {
1000                        Ok(response) => {
1001                            receipt.outcome = match response.stop_reason {
1002                                acp::StopReason::Cancelled => "cancelled",
1003                                _ => "completed",
1004                            }
1005                            .to_owned();
1006                            receipt.recorded_at_ms = now_ms();
1007                            let meta = response.meta.as_ref();
1008                            receipt.exo_session_id = meta_value(meta, "exo.session_id");
1009                            receipt.exo_turn_id = meta_value(meta, "exo.turn_id");
1010                            receipt.latest_event_id = meta_value(meta, "exo.latest_event_id");
1011                            receipt.verification =
1012                                if receipt.latest_event_id.is_some() {
1013                                    "Exo returned its durable latest event reference".to_owned()
1014                                } else {
1015                                    "Exo returned no durable event reference".to_owned()
1016                                };
1017                        }
1018                        Err(error) => {
1019                            receipt.recorded_at_ms = now_ms();
1020                            receipt.outcome = format!("failed: {error}");
1021                            receipt.verification =
1022                                "the ACP turn failed before verification".to_owned();
1023                        }
1024                    }
1025                }
1026                let receipt = receipt_cell.borrow().clone();
1027                let persisted = if let Some(receipt) = receipt {
1028                    persist_tier_c_receipt(receipt).await
1029                } else {
1030                    Ok(())
1031                };
1032                active_tier_c_turn.set(false);
1033                persisted?;
1034            }
1035            response
1036        })
1037    }
1038
1039    fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
1040        self.set_turn(
1041            ExoTurnPhase::Cancelling,
1042            Some("Waiting for Exo to confirm cancellation".to_owned()),
1043        );
1044        let cancelled_pending_grant = self.pending_grant.borrow_mut().take().map(|grant| {
1045            let request = grant.request();
1046            refused_tier_c_receipt(
1047                request.observed.clone(),
1048                request.turn_ref.clone(),
1049                Some(request.objective.clone()),
1050                "refused: cancelled before send".to_owned(),
1051            )
1052        });
1053        if let Some(receipt) = cancelled_pending_grant {
1054            *self.tier_c_receipt.borrow_mut() = Some(receipt.clone());
1055            cx.spawn(async move |_| {
1056                if let Err(error) = persist_tier_c_receipt(receipt).await {
1057                    log::error!("omega#87: failed to persist the cancellation receipt: {error}");
1058                }
1059            })
1060            .detach();
1061        } else if self.active_tier_c_turn.get() {
1062            if let Some(receipt) = self.tier_c_receipt.borrow_mut().as_mut() {
1063                receipt.recorded_at_ms = now_ms();
1064                receipt.outcome = "cancellation requested".to_owned();
1065            }
1066            if let Some(receipt) = self.tier_c_receipt.borrow().clone() {
1067                cx.spawn(async move |_| {
1068                    if let Err(error) = persist_tier_c_receipt(receipt).await {
1069                        log::error!(
1070                            "omega#87: failed to persist the cancellation receipt: {error}"
1071                        );
1072                    }
1073                })
1074                .detach();
1075            }
1076        }
1077        self.acp.cancel(session_id, cx);
1078    }
1079
1080    fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
1081        self
1082    }
1083}
1084
1085#[cfg(test)]
1086mod tests {
1087    use super::*;
1088
1089    fn lane_file(schema: &str) -> String {
1090        serde_json::json!({
1091            "schema": schema,
1092            "binary": "/opt/exo/target/release/exo",
1093            "checkout": "/opt/exo",
1094            "root": "/opt/exo/.exo",
1095            "agent": "omega-lane",
1096            "conversation": "tier-a",
1097        })
1098        .to_string()
1099    }
1100
1101    fn write(contents: &str) -> (tempfile::TempDir, PathBuf) {
1102        let dir = tempfile::tempdir().expect("a temp dir");
1103        let path = dir.path().join(EXO_LANE_FILE);
1104        std::fs::write(&path, contents).expect("write");
1105        (dir, path)
1106    }
1107
1108    #[test]
1109    fn a_lane_file_names_one_exo_install() {
1110        let (_dir, path) = write(&lane_file(EXO_LANE_SCHEMA));
1111        let config = ExoLaneConfig::load(&path).expect("a lane");
1112        assert_eq!(config.agent, "omega-lane");
1113        assert_eq!(config.conversation, "tier-a");
1114        assert_eq!(config.root.as_str(), "/opt/exo/.exo");
1115    }
1116
1117    #[test]
1118    fn an_unread_inspector_claims_no_runtime_facts() {
1119        let inspection = ExoInspectionSnapshot::default();
1120        assert_eq!(inspection.phase, ExoInspectionPhase::NotLoaded);
1121        assert_eq!(inspection.observed, None);
1122        assert_eq!(inspection.identity, None);
1123        assert_eq!(inspection.networking, None);
1124        assert_eq!(inspection.refreshed_at_ms, None);
1125        assert_eq!(inspection.error, None);
1126    }
1127
1128    #[test]
1129    fn every_exo_turn_phase_has_a_distinct_visible_label() {
1130        let phases = [
1131            ExoTurnPhase::Idle,
1132            ExoTurnPhase::Inspecting,
1133            ExoTurnPhase::Working,
1134            ExoTurnPhase::Cancelling,
1135            ExoTurnPhase::Completed,
1136            ExoTurnPhase::Cancelled,
1137            ExoTurnPhase::Refused,
1138            ExoTurnPhase::Failed,
1139        ];
1140        let labels = phases.map(ExoTurnPhase::label);
1141        for (index, label) in labels.iter().enumerate() {
1142            assert!(!label.is_empty());
1143            assert!(
1144                !labels[..index].contains(label),
1145                "duplicate Exo turn label: {label}"
1146            );
1147        }
1148    }
1149
1150    #[test]
1151    fn a_turn_snapshot_records_terminal_state_and_detail() {
1152        let turn = RefCell::new(ExoTurnSnapshot::default());
1153        let response = acp::PromptResponse::new(acp::StopReason::EndTurn).meta(
1154            [
1155                ("exo.session_id".into(), serde_json::json!("session-1")),
1156                ("exo.turn_id".into(), serde_json::json!("turn-1")),
1157                ("exo.latest_event_id".into(), serde_json::json!("event-1")),
1158            ]
1159            .into_iter()
1160            .collect::<serde_json::Map<String, serde_json::Value>>(),
1161        );
1162        set_terminal_turn_snapshot(
1163            &turn,
1164            ExoTurnPhase::Completed,
1165            Some("durable event observed".into()),
1166            &response,
1167        );
1168        let turn = turn.borrow();
1169        assert_eq!(turn.phase, ExoTurnPhase::Completed);
1170        assert_eq!(turn.detail.as_deref(), Some("durable event observed"));
1171        assert_eq!(turn.exo_session_id.as_deref(), Some("session-1"));
1172        assert_eq!(turn.exo_turn_id.as_deref(), Some("turn-1"));
1173        assert_eq!(turn.latest_event_id.as_deref(), Some("event-1"));
1174        assert!(turn.updated_at_ms > 0);
1175    }
1176
1177    /// A machine with no Exo has no lane. This is the ordinary case, and it
1178    /// must not be an error.
1179    #[test]
1180    fn no_lane_file_means_no_lane() {
1181        let dir = tempfile::tempdir().expect("a temp dir");
1182        assert_eq!(ExoLaneConfig::load(&dir.path().join(EXO_LANE_FILE)), None);
1183    }
1184
1185    /// A file this build cannot read is no lane rather than a partially trusted
1186    /// one. A lane pointed at the wrong `.exo` is worse than no lane.
1187    #[test]
1188    fn an_unreadable_lane_file_is_no_lane() {
1189        for contents in [
1190            "not json".to_owned(),
1191            lane_file("openagents.omega.exo_lane.v0"),
1192            serde_json::json!({ "schema": EXO_LANE_SCHEMA, "agent": "a" }).to_string(),
1193            lane_file(EXO_LANE_SCHEMA).replace("omega-lane", ""),
1194        ] {
1195            let (_dir, path) = write(&contents);
1196            assert_eq!(ExoLaneConfig::load(&path), None, "{contents}");
1197        }
1198    }
1199
1200    #[test]
1201    fn a_tier_c_grant_is_bound_to_the_exact_session_and_prompt() {
1202        let session_a = acp::SessionId::new("session-a");
1203        let session_b = acp::SessionId::new("session-b");
1204        let prompt_a = vec![acp::ContentBlock::Text(acp::TextContent::new("edit"))];
1205        let prompt_b = vec![acp::ContentBlock::Text(acp::TextContent::new("edit more"))];
1206        let reference = exo_turn_ref(&session_a, &prompt_a).expect("turn ref");
1207        assert_ne!(
1208            reference,
1209            exo_turn_ref(&session_b, &prompt_a).expect("session-bound ref")
1210        );
1211        assert_ne!(
1212            reference,
1213            exo_turn_ref(&session_a, &prompt_b).expect("prompt-bound ref")
1214        );
1215    }
1216
1217    #[test]
1218    fn a_tier_c_receipt_round_trips_all_authority_and_outcome_fields() {
1219        let observed = ObservedExoCapabilityState {
1220            source_commit: "commit".into(),
1221            source_tree: "tree".into(),
1222            binary_digest: "sha256:binary".into(),
1223            agent: "agent".into(),
1224            conversation: "conversation".into(),
1225            generation: 12,
1226            agent_authored_tools: true,
1227            tool_modules: Vec::new(),
1228            read_write_mounts: Vec::new(),
1229        };
1230        let authority = ExoSelfModificationReceipt {
1231            objective: "Update and verify Exo.".into(),
1232            turn_ref: "turn".into(),
1233            generation: observed.generation,
1234            expires_at_ms: 200,
1235            origin: ExoSelfModificationConsentOrigin::HumanConfirmationDialog,
1236            capabilities: observed.requested_capabilities(),
1237            observed: observed.clone(),
1238        };
1239        let receipt = ExoTierCReceipt {
1240            schema: "openagents.omega.exo_self_modification_receipt.v1".into(),
1241            recorded_at_ms: 150,
1242            authority: Some(authority),
1243            observed,
1244            turn_ref: "turn".into(),
1245            requested_objective: Some("Update and verify Exo.".into()),
1246            outcome: "completed".into(),
1247            exo_session_id: Some("session".into()),
1248            exo_turn_id: Some("turn".into()),
1249            latest_event_id: Some("event".into()),
1250            verification: "verified".into(),
1251            reconnect: "not reported by Exo ACP".into(),
1252            rollback: "not reported by Exo ACP".into(),
1253        };
1254        let encoded = serde_json::to_string(&receipt).expect("serialize receipt");
1255        let decoded = serde_json::from_str::<ExoTierCReceipt>(&encoded).expect("read receipt");
1256        assert_eq!(decoded, receipt);
1257    }
1258
1259    #[test]
1260    fn a_refusal_receipt_has_no_authority_or_durable_exo_turn() {
1261        let observed = ObservedExoCapabilityState {
1262            source_commit: "commit".into(),
1263            source_tree: "tree".into(),
1264            binary_digest: "sha256:binary".into(),
1265            agent: "agent".into(),
1266            conversation: "conversation".into(),
1267            generation: 12,
1268            agent_authored_tools: true,
1269            tool_modules: Vec::new(),
1270            read_write_mounts: Vec::new(),
1271        };
1272        let receipt =
1273            refused_tier_c_receipt(observed, "turn".into(), None, "refused: no grant".into());
1274        assert!(receipt.authority.is_none());
1275        assert!(receipt.exo_turn_id.is_none());
1276        assert_eq!(
1277            receipt.verification,
1278            "Omega refused the turn before it crossed ACP"
1279        );
1280    }
1281
1282    #[test]
1283    fn a_sandbox_module_resolves_only_through_an_observed_mount() {
1284        let mount = ExoMount {
1285            host_path: "/host/exo".into(),
1286            mount_path: "/workspace/exo".into(),
1287            read_write: true,
1288        };
1289        assert_eq!(
1290            resolve_module_host_path("/workspace/exo/tools/guardian.ts", &[mount]),
1291            Some(PathBuf::from("/host/exo/tools/guardian.ts"))
1292        );
1293        assert_eq!(
1294            resolve_module_host_path("/other/tools/guardian.ts", &[]),
1295            None
1296        );
1297    }
1298
1299    /// The lane, against a real Exo. `#[ignore]`d because it needs one: a
1300    /// built `exo` at the pinned commit, a configured agent and conversation,
1301    /// and whatever credential that agent's model binding resolves to. Exo
1302    /// `#[ignore]`s its own heavy integration cells for the same reason.
1303    ///
1304    /// Point it at an install with the same lane file the product reads:
1305    ///
1306    /// ```text
1307    /// OMEGA_EXO_LANE_FILE=/path/to/omega-exo-lane.json \
1308    ///   cargo test -p agent_ui drives_a_real_exo -- --ignored --nocapture
1309    /// ```
1310    ///
1311    /// This is the acceptance path end to end: the router's external executor
1312    /// builds a thread, a turn runs through Exo's CLI, the reply lands in the
1313    /// thread, and the thread discloses who ran it.
1314    #[gpui::test]
1315    #[ignore = "needs a local Exo install at the pinned commit"]
1316    async fn drives_a_real_exo(cx: &mut gpui::TestAppContext) {
1317        use crate::omega_executor_disclosure::ThreadExecutorDisclosure as _;
1318
1319        let Ok(lane_file) = std::env::var("OMEGA_EXO_LANE_FILE") else {
1320            panic!("set OMEGA_EXO_LANE_FILE to a lane file; see this test's docs");
1321        };
1322        // The turn runs a real process against a real model, so this test
1323        // waits on wall-clock I/O rather than on the deterministic scheduler.
1324        // That is the point of it: the same await that parks here is what keeps
1325        // the window drawing in production, and the workspace's clippy
1326        // configuration disallows the blocking call that would not park.
1327        cx.executor().allow_parking();
1328        crate::test_support::init_test(cx);
1329        let fs = fs::FakeFs::new(cx.executor());
1330        let lane_path = PathBuf::from(lane_file);
1331        let process_cwd = lane_path
1332            .parent()
1333            .expect("the lane file has a parent")
1334            .to_path_buf();
1335        fs.insert_tree(&process_cwd, serde_json::json!({})).await;
1336        let project = Project::test(fs, [process_cwd.as_path()], cx).await;
1337
1338        let agent_server_store =
1339            project.read_with(cx, |project, _| project.agent_server_store().downgrade());
1340        let connect_project = project.clone();
1341        let connection = cx
1342            .update(|cx| {
1343                cx.spawn(async move |cx| {
1344                    connect_configured_lane(&lane_path, connect_project, agent_server_store, cx)
1345                        .await
1346                })
1347            })
1348            .await
1349            .expect("the Exo ACP lane connects")
1350            .expect("the lane file names an Exo install");
1351        let thread = cx
1352            .update(|cx| {
1353                connection
1354                    .clone()
1355                    .new_session(project, PathList::new(&[process_cwd]), cx)
1356            })
1357            .await
1358            .expect("a session on the Exo lane");
1359
1360        let session_id = thread.read_with(cx, |thread, _| thread.session_id().clone());
1361        let tool_marker = "OMEGA-EXO-TOOL";
1362        let marker = "OMEGA-EXO-PANE-READY";
1363        let request = acp::PromptRequest::new(
1364            session_id,
1365            vec![acp::ContentBlock::Text(acp::TextContent::new(format!(
1366                "Run one tool that prints {tool_marker}, then reply with exactly {marker}."
1367            )))],
1368        );
1369        let response = cx
1370            .update(|cx| connection.prompt(request, cx))
1371            .await
1372            .expect("Exo ran the turn");
1373        assert_eq!(response.stop_reason, acp::StopReason::EndTurn);
1374
1375        let rendered = thread.read_with(cx, |thread, cx| thread.to_markdown(cx));
1376        assert!(
1377            rendered.contains(marker),
1378            "the Exo reply did not reach the thread: {rendered}"
1379        );
1380        assert!(
1381            rendered.contains(tool_marker),
1382            "the Exo tool result did not reach the thread: {rendered}"
1383        );
1384
1385        let disclosure = thread.read_with(cx, |thread, cx| thread.omega_executor_disclosure(cx));
1386        assert!(disclosure.is_coherent(), "{disclosure:?}");
1387        assert_eq!(disclosure.class, omega_exo_lane::EXO_EXECUTOR_CLASS);
1388        assert_eq!(disclosure.run_ref, None);
1389        let exo = connection
1390            .clone()
1391            .downcast::<ExoHarnessConnection>()
1392            .expect("the configured lane is Exo");
1393        let identity = exo.identity().expect("Exo told the lane who it is");
1394        let inspection = exo.inspection();
1395        assert_eq!(inspection.phase, ExoInspectionPhase::Ready);
1396        assert_eq!(inspection.identity.as_ref(), Some(&identity));
1397        assert!(
1398            inspection.observed.is_some(),
1399            "the workspace inspector did not retain the turn's exact preflight"
1400        );
1401        let turn = exo.turn();
1402        assert_eq!(turn.phase, ExoTurnPhase::Completed, "{turn:?}");
1403        assert!(turn.exo_session_id.is_some(), "{turn:?}");
1404        assert!(turn.exo_turn_id.is_some(), "{turn:?}");
1405        assert!(turn.latest_event_id.is_some(), "{turn:?}");
1406        assert!(disclosure.agent_id.starts_with("exo/"), "{disclosure:?}");
1407        // The parts that only the Exo arm of `classify_connection` can supply.
1408        // Without them this assertion set passes on the shared external-agent
1409        // fallback, which discloses no model at all — and a disclosure that
1410        // says "model not disclosed" about an executor that *did* disclose one
1411        // is the silent failure this test exists to catch. It was silent here
1412        // once, on 2026-07-26, until this line was added.
1413        assert_eq!(
1414            disclosure.model, identity.model,
1415            "the disclosure dropped the model Exo reported: {disclosure:?}"
1416        );
1417        assert!(
1418            disclosure.model.is_some(),
1419            "Exo reported a model and the thread did not disclose it: {disclosure:?}"
1420        );
1421        println!("executor disclosure: {}", disclosure.label());
1422        println!("exo identity: {identity:?}");
1423    }
1424}
1425