tenant.openagents/omega

1//! Omega Agent, the router. `OMEGA-DELTA-0029`, omega#78.
2//!
3//! The owner admitted Omega Agent on omega#74 as a **router** that owns
4//! routing, disclosure, and receipts and owns **no execution**, sitting at the
5//! `AgentConnection` seam above three executor classes: the native agent loop,
6//! external ACP agents, and `omega-effectd` engine lanes.
7//!
8//! This module is the dispatch half of that. The decision half is
9//! [`omega_front_door::router`], a leaf with no dependencies, so the routing law
10//! can be checked without building GPUI and so no decision is ever made inside
11//! a widget.
12//!
13//! # The router owns no execution
14//!
15//! [`OmegaAgentConnection`] implements every method of [`AgentConnection`] by
16//! handing it to the executor the decision names. There is no turn loop here,
17//! no tool call, no model call, and no session state beyond the pin and the
18//! decision. `the_router_delegates_every_agent_connection_method` in
19//! `crates/omega_deltas` reads this file and fails if a method body stops
20//! delegating, because "owns no execution" is a property of the source, not of
21//! its author's intent.
22//!
23//! A thread the router creates carries the **executor's** connection, not the
24//! router's, because the executor is what built it. That is deliberate:
25//! omega#77's disclosure classifies a thread by downcasting its connection, so
26//! a thread that carried the router would disclose the router as its executor —
27//! which is exactly the first-party attribution claim omega#77 exists to stop.
28//!
29//! # `omega-effectd` stays the sole run authority
30//!
31//! [`EngineReadiness`] here is *read* from a `get_capacity` answer. Nothing is
32//! written back, nothing is cached as run state, and the router never starts,
33//! pauses, or stops a run. Starting engine-lane authority remains what owner
34//! gate 8 requires: an explicit human action on a visible control, which after
35//! omega#76 is the Start button on the Full Auto surface.
36//!
37//! # The decision is recorded
38//!
39//! Every decision is written to a route journal on disk keyed by session, in
40//! [`omega_front_door::RouteDecision::canonical_record`] form, which round-trips.
41//! The journal carries no clock: a timestamp would make two identical decisions
42//! look different and would put a non-deterministic value beside a decision path
43//! whose whole point is that it is reproducible.
44
45use std::any::Any;
46use std::cell::RefCell;
47use std::collections::BTreeMap;
48use std::path::{Path, PathBuf};
49use std::rc::{Rc, Weak};
50use std::sync::{LazyLock, Mutex};
51
52use acp_thread::{
53    AcpThread, AgentConnection, AgentModelSelector, AgentSessionClientUserMessageIds,
54    AgentSessionConfigOptions, AgentSessionList, AgentSessionModes, AgentSessionRetry,
55    AgentSessionSetTitle, AgentSessionTruncate, AgentTelemetry, ElicitationStore,
56};
57use agent_client_protocol::schema::v1 as acp;
58use anyhow::Result;
59use gpui::{App, Entity, SharedString, Task};
60use omega_front_door::{
61    EngineLane, EngineReadiness, EngineUnreachable, ExecutorClass, ExecutorPin, LaneState,
62    PinGesture, RouteDecision, RouteInputs, RouteReason, route,
63};
64use project::{AgentId, Project};
65use serde_json::Value;
66use util::path_list::PathList;
67
68/// The schema the route journal is written under.
69const ROUTE_JOURNAL_SCHEMA: &str = "openagents.omega.agent_route_journal.v1";
70
71/// The route journal's file name, under the Omega data directory.
72const ROUTE_JOURNAL_FILE: &str = "agent-route-journal.json";
73
74// -------------------------------------------------------------------------
75// Reading the engine's answer
76// -------------------------------------------------------------------------
77
78/// Read a framed `get_capacity` answer into the typed readiness the router
79/// takes.
80///
81/// Anything missing or misshapen reads as
82/// [`EngineUnreachable::ProtocolError`] rather than as a partially trusted
83/// answer. An engine whose capacity record this build cannot understand is an
84/// engine this build must not route onto: believing half of it is how a router
85/// dispatches into a lane on the strength of not understanding the other half.
86#[must_use]
87pub fn engine_readiness_from_capacity(capacity: &Value) -> EngineReadiness {
88    let protocol_error = EngineReadiness::Unreachable(EngineUnreachable::ProtocolError);
89
90    let Some(object) = capacity.as_object() else {
91        return protocol_error;
92    };
93    let (Some(active_run_count), Some(active_run_limit)) = (
94        object.get("activeRunCount").and_then(Value::as_u64),
95        object.get("activeRunLimit").and_then(Value::as_u64),
96    ) else {
97        return protocol_error;
98    };
99    let Some(lanes) = object.get("lanes").and_then(Value::as_array) else {
100        return protocol_error;
101    };
102
103    let mut parsed = Vec::with_capacity(lanes.len());
104    for lane in lanes {
105        let (Some(lane_ref), Some(state)) = (
106            lane.get("lane").and_then(Value::as_str),
107            lane.get("state").and_then(Value::as_str),
108        ) else {
109            return protocol_error;
110        };
111        parsed.push(EngineLane::new(lane_ref, LaneState::parse(state)));
112    }
113
114    EngineReadiness::Answered {
115        active_run_count: u32::try_from(active_run_count).unwrap_or(u32::MAX),
116        active_run_limit: u32::try_from(active_run_limit).unwrap_or(u32::MAX),
117        lanes: parsed,
118    }
119}
120
121// -------------------------------------------------------------------------
122// The durable record
123// -------------------------------------------------------------------------
124
125/// Where routing decisions are written down.
126///
127/// A record, not a log line: keyed by session, readable back into a typed
128/// [`RouteDecision`], and rewritten atomically through a temporary file so a
129/// crash mid-write leaves the previous journal rather than a truncated one.
130pub struct RouteJournal {
131    path: PathBuf,
132    /// Session id to canonical record. A `BTreeMap` rather than a `HashMap`
133    /// because the file is written from it: hash order would make the same set
134    /// of decisions serialise differently on different runs, and a journal that
135    /// changes without a decision changing is a journal nobody can diff.
136    entries: RefCell<BTreeMap<String, String>>,
137}
138
139impl RouteJournal {
140    /// The journal at the Omega data directory's usual place.
141    #[must_use]
142    pub fn at_data_dir() -> Self {
143        Self::at(Self::data_dir_path())
144    }
145
146    /// Where the durable journal lives.
147    ///
148    /// Exposed so the *caller* can choose a different path — a stateless run
149    /// must not write sessions nobody started into the record an operator
150    /// reads. That choice is made in `Agent::server` rather than here, because
151    /// reading the environment inside this file is what
152    /// `the_routing_law_has_no_clock_no_randomness_and_no_hash_order` forbids:
153    /// the router's exit is that the same inputs give the same route, and an
154    /// environment read is not an input. The check caught this.
155    #[must_use]
156    pub fn data_dir_path() -> PathBuf {
157        paths::data_dir()
158            .join("openagents")
159            .join(ROUTE_JOURNAL_FILE)
160    }
161
162    /// The journal at an explicit path. Loads what is already there.
163    #[must_use]
164    pub fn at(path: PathBuf) -> Self {
165        let entries = load_journal(&path).unwrap_or_else(|error| {
166            log::warn!(
167                "OMEGA-DELTA-0029: route journal at {} could not be read ({error:#}); \
168                 starting from empty rather than routing without a record",
169                path.display()
170            );
171            BTreeMap::new()
172        });
173        for (session_id, record) in &entries {
174            if let Some(decision) = RouteDecision::parse_canonical_record(record) {
175                publish_recorded_route(session_id, decision.disclosed_route());
176            }
177        }
178        Self {
179            path,
180            entries: RefCell::new(entries),
181        }
182    }
183
184    /// Write one decision down.
185    ///
186    /// Publishes to the read-mostly index below in the same call, so the
187    /// disclosure line a thread draws cannot disagree with the durable record
188    /// it is derived from.
189    #[must_use]
190    pub fn record(&self, session_id: &str, decision: &RouteDecision) -> bool {
191        self.entries
192            .borrow_mut()
193            .insert(session_id.to_owned(), decision.canonical_record());
194        publish_recorded_route(session_id, decision.disclosed_route());
195        match self.persist() {
196            Ok(()) => true,
197            Err(error) => {
198                log::error!(
199                    "OMEGA-DELTA-0029: route decision for {session_id} could not be \
200                     persisted to {}: {error:#}",
201                    self.path.display()
202                );
203                false
204            }
205        }
206    }
207
208    /// Read one decision back.
209    ///
210    /// `None` for a session with no record *and* for a record that does not
211    /// read back as a coherent decision, so a hand-edited journal is rejected
212    /// rather than believed.
213    #[must_use]
214    pub fn decision(&self, session_id: &str) -> Option<RouteDecision> {
215        self.entries
216            .borrow()
217            .get(session_id)
218            .and_then(|record| RouteDecision::parse_canonical_record(record))
219    }
220
221    /// Every recorded decision, in session order.
222    #[must_use]
223    pub fn decisions(&self) -> Vec<(String, RouteDecision)> {
224        self.entries
225            .borrow()
226            .iter()
227            .filter_map(|(session_id, record)| {
228                RouteDecision::parse_canonical_record(record)
229                    .map(|decision| (session_id.clone(), decision))
230            })
231            .collect()
232    }
233
234    fn persist(&self) -> anyhow::Result<()> {
235        let entries = self.entries.borrow();
236        let document = serde_json::json!({
237            "schema": ROUTE_JOURNAL_SCHEMA,
238            "decisions": entries
239                .iter()
240                .map(|(session_id, record)| {
241                    serde_json::json!({ "sessionId": session_id, "decision": record })
242                })
243                .collect::<Vec<_>>(),
244        });
245        if let Some(parent) = self.path.parent() {
246            std::fs::create_dir_all(parent)?;
247        }
248        let temporary = self.path.with_extension("json.tmp");
249        std::fs::write(&temporary, serde_json::to_vec_pretty(&document)?)?;
250        std::fs::rename(&temporary, &self.path)?;
251        Ok(())
252    }
253}
254
255fn load_journal(path: &Path) -> anyhow::Result<BTreeMap<String, String>> {
256    if !path.exists() {
257        return Ok(BTreeMap::new());
258    }
259    let document: Value = serde_json::from_slice(&std::fs::read(path)?)?;
260    let schema = document.get("schema").and_then(Value::as_str);
261    anyhow::ensure!(
262        schema == Some(ROUTE_JOURNAL_SCHEMA),
263        "unsupported route journal schema {schema:?}"
264    );
265    let mut entries = BTreeMap::new();
266    for entry in document
267        .get("decisions")
268        .and_then(Value::as_array)
269        .unwrap_or(&Vec::new())
270    {
271        let (Some(session_id), Some(record)) = (
272            entry.get("sessionId").and_then(Value::as_str),
273            entry.get("decision").and_then(Value::as_str),
274        ) else {
275            anyhow::bail!("route journal entry is missing sessionId or decision");
276        };
277        entries.insert(session_id.to_owned(), record.to_owned());
278    }
279    Ok(entries)
280}
281
282/// `OMEGA-DELTA-0029`. The route reason each routed session was recorded with,
283/// so a thread surface can disclose it wherever a thread is drawn.
284///
285/// The same shape as `omega_host_bridge`'s lane index and for the same reason:
286/// the router lives behind an `Rc` a render cannot reach, while the disclosure
287/// has to be readable from one. It is a read-mostly *projection* of the
288/// journal, not a second store — it is filled from the journal when one is
289/// opened and on every write, so deleting the journal file empties it.
290static RECORDED_ROUTES: LazyLock<Mutex<BTreeMap<String, RouteReason>>> =
291    LazyLock::new(|| Mutex::new(BTreeMap::new()));
292
293fn publish_recorded_route(session_id: &str, reason: RouteReason) {
294    let mut index = match RECORDED_ROUTES.lock() {
295        Ok(index) => index,
296        Err(poisoned) => poisoned.into_inner(),
297    };
298    index.insert(session_id.to_owned(), reason);
299}
300
301/// Why Omega Agent routed this session where it did, if it routed it.
302///
303/// `None` means the session was not routed by the router — a thread from before
304/// `OMEGA-DELTA-0029`, or one opened directly on an executor. Saying "not
305/// routed" is different from claiming a reason nobody recorded.
306#[must_use]
307pub fn recorded_route(session_id: &acp::SessionId) -> Option<RouteReason> {
308    let index = match RECORDED_ROUTES.lock() {
309        Ok(index) => index,
310        Err(poisoned) => poisoned.into_inner(),
311    };
312    index.get(session_id.0.as_ref()).copied()
313}
314
315// -------------------------------------------------------------------------
316// The router
317// -------------------------------------------------------------------------
318
319/// Omega Agent at the `AgentConnection` seam.
320///
321/// Holds one connection per executor class it can dispatch to, the pins a
322/// person set, and the journal it writes decisions to. It holds no run state,
323/// no policy state, and no turn state.
324pub struct OmegaAgentConnection {
325    /// The native agent loop. Required, because it is the fail-closed target:
326    /// every route that cannot be honoured lands here.
327    native: Rc<dyn AgentConnection>,
328    /// The external ACP agent connected for this surface, if one is.
329    external_acp: Option<Rc<dyn AgentConnection>>,
330    /// The executor registered to serve engine lanes, if one is. See
331    /// [`RouteInputs::engine_lane`] for why this is separate from the engine
332    /// answering at all.
333    engine_lane: Option<Rc<dyn AgentConnection>>,
334    /// The engine's last framed answer. Read, never written back.
335    engine: RefCell<EngineReadiness>,
336    /// Pins, by session. Set by a human gesture; never by a turn.
337    pins: RefCell<BTreeMap<String, ExecutorPin>>,
338    /// The pin a session that does not exist yet will be created under.
339    next_pin: RefCell<Option<ExecutorPin>>,
340    /// Decisions, by session, in memory and on disk.
341    journal: RouteJournal,
342    /// The identity this router presents. `OMEGA-DELTA-0024`.
343    agent_id: AgentId,
344}
345
346impl OmegaAgentConnection {
347    /// A router over the native loop alone.
348    ///
349    /// The minimum honest configuration: one executor, which is also the
350    /// fail-closed target.
351    #[must_use]
352    pub fn new(native: Rc<dyn AgentConnection>, journal: RouteJournal) -> Self {
353        let agent_id = native.agent_id();
354        Self {
355            native,
356            external_acp: None,
357            engine_lane: None,
358            engine: RefCell::new(EngineReadiness::Unreachable(EngineUnreachable::NotRunning)),
359            pins: RefCell::new(BTreeMap::new()),
360            next_pin: RefCell::new(None),
361            journal,
362            agent_id,
363        }
364    }
365
366    /// Register the external ACP agent this surface can route to.
367    #[must_use]
368    pub fn with_external_acp(mut self, connection: Rc<dyn AgentConnection>) -> Self {
369        self.external_acp = Some(connection);
370        self
371    }
372
373    /// Register the executor that serves engine lanes.
374    #[must_use]
375    pub fn with_engine_lane(mut self, connection: Rc<dyn AgentConnection>) -> Self {
376        self.engine_lane = Some(connection);
377        self
378    }
379
380    /// Take the engine's latest framed `get_capacity` answer.
381    ///
382    /// The engine remains the sole run authority. This is a snapshot the router
383    /// reads to decide, and the router never answers back.
384    pub fn observe_engine(&self, readiness: EngineReadiness) {
385        *self.engine.borrow_mut() = readiness;
386    }
387
388    /// Take the engine's latest framed answer, or the fact that it did not
389    /// answer.
390    pub fn observe_capacity(&self, capacity: Result<&Value, EngineUnreachable>) {
391        let readiness = match capacity {
392            Ok(capacity) => engine_readiness_from_capacity(capacity),
393            Err(cause) => EngineReadiness::Unreachable(cause),
394        };
395        self.observe_engine(readiness);
396    }
397
398    /// Pin an executor for a session that already exists.
399    ///
400    /// The `gesture` argument is the guard, not a label. A pin is the only way
401    /// a thread reaches an engine lane, an engine lane is Full Auto authority,
402    /// and owner gate 8 admits only an explicit human action into that
403    /// authority. Requiring a [`PinGesture`] means there is no way to set a pin
404    /// without naming the human gesture that set it, and `PinGesture` has no
405    /// variant for a tool call, a slash command, a restored draft, or a
406    /// composer mode flag. Nothing here starts a run: a pin decides *where* the
407    /// next turn of a thread goes, and the Full Auto Start button remains the
408    /// only path to engine-lane run authority.
409    /// Returns the decision the new pin produced, which is what the thread
410    /// then discloses — including a pin that could not be honoured, with the
411    /// typed reason it could not.
412    pub fn pin_session(
413        &self,
414        session_id: &acp::SessionId,
415        pin: ExecutorPin,
416        gesture: PinGesture,
417    ) -> RouteDecision {
418        log::info!(
419            "OMEGA-DELTA-0035: session {} pinned to {} by {}",
420            session_id.0,
421            pin.token(),
422            gesture.token()
423        );
424        self.pins
425            .borrow_mut()
426            .insert(session_id.0.to_string(), pin.clone());
427        // Re-decide *because a person changed the pin*, and only then. Capacity
428        // moving underneath a thread never re-decides it — `executor_for` reads
429        // the record — so this is the one thing that can move a live thread,
430        // and it is a human gesture by construction.
431        self.decide(session_id.0.as_ref(), Some(pin))
432    }
433
434    /// Clear a session's pin, so its next turn takes the unpinned default.
435    ///
436    /// Re-decides for the same reason [`pin_session`](Self::pin_session) does:
437    /// a cleared pin that left the old decision standing would show an
438    /// executor the user had just unpinned.
439    pub fn unpin_session(&self, session_id: &acp::SessionId, gesture: PinGesture) -> RouteDecision {
440        log::info!(
441            "OMEGA-DELTA-0035: session {} unpinned by {}",
442            session_id.0,
443            gesture.token()
444        );
445        self.pins.borrow_mut().remove(&session_id.0.to_string());
446        self.decide(session_id.0.as_ref(), None)
447    }
448
449    /// Pin the executor the next session created through this router will use.
450    ///
451    /// Carries a [`PinGesture`] for the same reason [`pin_session`] does.
452    ///
453    /// [`pin_session`]: Self::pin_session
454    pub fn pin_next_session(&self, pin: Option<ExecutorPin>, gesture: PinGesture) {
455        log::info!(
456            "OMEGA-DELTA-0035: the next session is pinned to {} by {}",
457            pin.as_ref()
458                .map_or("nothing".to_owned(), ExecutorPin::token),
459            gesture.token()
460        );
461        *self.next_pin.borrow_mut() = pin;
462    }
463
464    /// Every pin currently in force, in session order. For an inspector.
465    #[must_use]
466    pub fn pins(&self) -> Vec<(String, ExecutorPin)> {
467        self.pins
468            .borrow()
469            .iter()
470            .map(|(session_id, pin)| (session_id.clone(), pin.clone()))
471            .collect()
472    }
473
474    /// The pin currently in force for a session.
475    #[must_use]
476    pub fn pin(&self, session_id: &acp::SessionId) -> Option<ExecutorPin> {
477        self.pins.borrow().get(&session_id.0.to_string()).cloned()
478    }
479
480    /// The typed inputs a decision for this session is made from.
481    ///
482    /// Everything [`route`] is allowed to read, assembled in one place so a
483    /// decision can be re-derived and checked against its record.
484    #[must_use]
485    pub fn inputs_for(&self, pin: Option<ExecutorPin>) -> RouteInputs {
486        RouteInputs {
487            pin,
488            engine: self.engine.borrow().clone(),
489            external_acp: self
490                .external_acp
491                .as_ref()
492                .map(|connection| connection.agent_id().0.to_string()),
493            engine_lane: self
494                .engine_lane
495                .as_ref()
496                .map(|connection| connection.agent_id().0.to_string()),
497        }
498    }
499
500    /// Decide, record, and return the decision for a session.
501    ///
502    /// Recording happens before dispatch. A turn that ran somewhere the journal
503    /// does not name is a route nobody can explain afterwards, which is
504    /// omega#78's falsifier.
505    #[must_use]
506    pub fn decide(&self, session_id: &str, pin: Option<ExecutorPin>) -> RouteDecision {
507        let decision = route(&self.inputs_for(pin));
508        debug_assert!(decision.is_coherent(), "incoherent decision: {decision:?}");
509        if !self.journal.record(session_id, &decision) {
510            log::error!(
511                "OMEGA-DELTA-0029: session {session_id} routed to {} with no durable \
512                 record; the route is explainable only until this process exits",
513                decision.explain()
514            );
515        }
516        decision
517    }
518
519    /// The decision recorded for a session, if one was.
520    #[must_use]
521    pub fn recorded_decision(&self, session_id: &acp::SessionId) -> Option<RouteDecision> {
522        self.journal.decision(session_id.0.as_ref())
523    }
524
525    /// The journal, for an inspector.
526    #[must_use]
527    pub fn journal(&self) -> &RouteJournal {
528        &self.journal
529    }
530
531    /// The executor a class names, or the native loop.
532    ///
533    /// Total by construction: the native loop is required, so there is always
534    /// something to hand the turn to. A class whose connection is absent cannot
535    /// be reached from [`route`] — `RouteInputs` reports absence, and the
536    /// decision falls back — so this arm is a belt on top of a brace, not a
537    /// silent substitution.
538    #[must_use]
539    pub fn executor(&self, class: ExecutorClass) -> Rc<dyn AgentConnection> {
540        match class {
541            ExecutorClass::NativeLoop => self.native.clone(),
542            ExecutorClass::ExternalAcp => self.external_acp.clone().unwrap_or_else(|| {
543                log::error!("OMEGA-DELTA-0029: external ACP route with no connection");
544                self.native.clone()
545            }),
546            ExecutorClass::EngineLane => self.engine_lane.clone().unwrap_or_else(|| {
547                log::error!("OMEGA-DELTA-0029: engine lane route with no connection");
548                self.native.clone()
549            }),
550        }
551    }
552
553    /// The executor a live session's recorded decision names.
554    ///
555    /// Reads the record rather than re-deciding, so a turn cannot silently move
556    /// executors mid-thread because the engine's capacity changed between
557    /// turns. A session with no record has not been routed yet and gets the
558    /// fail-closed target.
559    #[must_use]
560    pub fn executor_for(&self, session_id: &acp::SessionId) -> Rc<dyn AgentConnection> {
561        match self.recorded_decision(session_id) {
562            Some(decision) => self.executor(decision.chosen),
563            None => self.native.clone(),
564        }
565    }
566}
567
568impl AgentConnection for OmegaAgentConnection {
569    fn agent_id(&self) -> AgentId {
570        self.agent_id.clone()
571    }
572
573    fn telemetry_id(&self) -> SharedString {
574        self.native.telemetry_id()
575    }
576
577    fn agent_version(&self) -> Option<SharedString> {
578        self.native.agent_version()
579    }
580
581    /// Route once, record, and let the executor build the thread.
582    ///
583    /// The thread comes back carrying the *executor's* connection, which is
584    /// what omega#77's disclosure downcasts. A thread carrying the router would
585    /// disclose the router as its executor, which is the first-party
586    /// attribution claim omega#77 exists to stop.
587    fn new_session(
588        self: Rc<Self>,
589        project: Entity<Project>,
590        work_dirs: PathList,
591        cx: &mut App,
592    ) -> Task<Result<Entity<AcpThread>>> {
593        let pin = self.next_pin.borrow().clone();
594        let decision = route(&self.inputs_for(pin.clone()));
595        let executor = self.executor(decision.chosen);
596        let session = executor.new_session(project, work_dirs, cx);
597        cx.spawn(async move |cx| {
598            let thread = session.await?;
599            // Recorded once the session exists, because the record is keyed by
600            // the session id the executor minted. The decision itself was made
601            // before dispatch and is not re-derived here: re-deciding after the
602            // fact would let the record describe a world the turn never saw.
603            let session_id = thread.read_with(cx, |thread, _| thread.session_id().0.to_string());
604            if !self.journal.record(&session_id, &decision) {
605                log::error!(
606                    "OMEGA-DELTA-0029: session {session_id} routed to {} with no durable \
607                     record; the route is explainable only until this process exits",
608                    decision.explain()
609                );
610            }
611            self.pins
612                .borrow_mut()
613                .extend(pin.map(|pin| (session_id, pin)));
614            Ok(thread)
615        })
616    }
617
618    fn supports_load_session(&self) -> bool {
619        self.native.supports_load_session()
620    }
621
622    fn auth_methods(&self) -> &[acp::AuthMethod] {
623        self.native.auth_methods()
624    }
625
626    fn authenticate(&self, method: acp::AuthMethodId, cx: &mut App) -> Task<Result<()>> {
627        self.native.authenticate(method, cx)
628    }
629
630    fn supports_logout(&self) -> bool {
631        self.native.supports_logout()
632    }
633
634    fn logout(&self, cx: &mut App) -> Task<Result<()>> {
635        self.native.logout(cx)
636    }
637
638    fn client_user_message_ids(
639        &self,
640        cx: &App,
641    ) -> Option<Rc<dyn AgentSessionClientUserMessageIds>> {
642        self.native.client_user_message_ids(cx)
643    }
644
645    fn prompt(
646        &self,
647        params: acp::PromptRequest,
648        cx: &mut App,
649    ) -> Task<Result<acp::PromptResponse>> {
650        self.executor_for(&params.session_id).prompt(params, cx)
651    }
652
653    fn retry(&self, session_id: &acp::SessionId, cx: &App) -> Option<Rc<dyn AgentSessionRetry>> {
654        self.executor_for(session_id).retry(session_id, cx)
655    }
656
657    fn cancel(&self, session_id: &acp::SessionId, cx: &mut App) {
658        self.executor_for(session_id).cancel(session_id, cx);
659    }
660
661    fn request_elicitations(&self) -> Option<Entity<ElicitationStore>> {
662        self.native.request_elicitations()
663    }
664
665    fn truncate(
666        &self,
667        session_id: &acp::SessionId,
668        cx: &App,
669    ) -> Option<Rc<dyn AgentSessionTruncate>> {
670        self.executor_for(session_id).truncate(session_id, cx)
671    }
672
673    fn set_title(
674        &self,
675        session_id: &acp::SessionId,
676        cx: &App,
677    ) -> Option<Rc<dyn AgentSessionSetTitle>> {
678        self.executor_for(session_id).set_title(session_id, cx)
679    }
680
681    fn model_selector(&self, session_id: &acp::SessionId) -> Option<Rc<dyn AgentModelSelector>> {
682        self.executor_for(session_id).model_selector(session_id)
683    }
684
685    fn telemetry(&self) -> Option<Rc<dyn AgentTelemetry>> {
686        self.native.telemetry()
687    }
688
689    fn session_modes(
690        &self,
691        session_id: &acp::SessionId,
692        cx: &App,
693    ) -> Option<Rc<dyn AgentSessionModes>> {
694        self.executor_for(session_id).session_modes(session_id, cx)
695    }
696
697    fn session_config_options(
698        &self,
699        session_id: &acp::SessionId,
700        cx: &App,
701    ) -> Option<Rc<dyn AgentSessionConfigOptions>> {
702        self.executor_for(session_id)
703            .session_config_options(session_id, cx)
704    }
705
706    fn session_list(&self, cx: &mut App) -> Option<Rc<dyn AgentSessionList>> {
707        self.native.session_list(cx)
708    }
709
710    fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
711        self
712    }
713}
714
715// -------------------------------------------------------------------------
716// Wiring: the router is what the native agent entry resolves to
717// -------------------------------------------------------------------------
718
719thread_local! {
720    /// `OMEGA-DELTA-0035`. The router this window's native agent entry built.
721    ///
722    /// A render cannot reach an `Rc<dyn AgentConnection>` held inside the
723    /// connection store's async entry, and the pin control has to live on the
724    /// thread's own disclosure line — so the router publishes itself here when
725    /// it is built, exactly as `omega_host_bridge` publishes its lane index and
726    /// as [`RECORDED_ROUTES`] publishes route reasons.
727    ///
728    /// A `thread_local` rather than a `static` because `Rc` is not `Sync`, and
729    /// because the only reader is the GPUI main thread that built it. It is a
730    /// *handle*, not state: every decision, pin and journal entry lives on the
731    /// connection itself.
732    ///
733    /// **Weak, deliberately.** A strong reference here would keep the native
734    /// agent entity alive for the life of the process, which is a leak in
735    /// production and a hard failure in the GPUI test harness — it checks for
736    /// leaked entity handles at teardown and caught exactly this. Weak also
737    /// gives the right semantics: when the connection store drops the
738    /// connection there is no router, and `active_router` says so instead of
739    /// handing out a router nothing is dispatching through.
740    static ACTIVE_ROUTER: RefCell<Weak<OmegaAgentConnection>> =
741        const { RefCell::new(Weak::new()) };
742}
743
744fn publish_active_router(router: &Rc<OmegaAgentConnection>) {
745    ACTIVE_ROUTER.with(|active| *active.borrow_mut() = Rc::downgrade(router));
746}
747
748/// Omega Agent's router, if this window has built one.
749///
750/// `None` before the native agent connects, and in any process that never
751/// connects it. A caller that gets `None` must not invent a route — the
752/// disclosure says "not routed", which is the honest reading.
753#[must_use]
754pub fn active_router() -> Option<Rc<OmegaAgentConnection>> {
755    ACTIVE_ROUTER.with(|active| active.borrow().upgrade())
756}
757
758/// The native agent, behind Omega Agent's router. `OMEGA-DELTA-0035`.
759///
760/// omega#78 put `OmegaAgentConnection` at the `AgentConnection` seam and left
761/// it unwired: nothing constructed one, so every thread disclosed
762/// `route: None` and the journal stayed empty. This is the wire. Omega's
763/// native-agent entry resolves to a router over the native connection instead
764/// of to the native connection itself, so every new native session is routed
765/// on purpose and the decision is written down before the turn exists.
766///
767/// The thread that comes back still carries the **executor's** connection,
768/// because `OmegaAgentConnection::new_session` delegates and returns what the
769/// executor built. That is what keeps omega#77's disclosure honest and what
770/// keeps every existing `downcast::<NativeAgentConnection>()` on a *thread's*
771/// connection working unchanged.
772pub struct OmegaRouterServer {
773    /// Held by value rather than behind an `Rc`, because `AgentServer` is
774    /// `Send` and an `Rc` field would take that away.
775    native: agent::NativeAgentServer,
776    /// Where decisions are written down. Chosen by the caller; see
777    /// [`RouteJournal::data_dir_path`].
778    journal_path: PathBuf,
779    /// Where the Exo harness lane is configured, if the owner configured one.
780    /// `OMEGA-DELTA-0042`. Chosen by the caller for the same reason the journal
781    /// path is: a harness must not read the owner's real data directory.
782    exo_lane_path: PathBuf,
783}
784
785impl OmegaRouterServer {
786    /// A router server over the native agent server, journalling to `journal_path`.
787    #[must_use]
788    pub fn new(
789        native: agent::NativeAgentServer,
790        journal_path: PathBuf,
791        exo_lane_path: PathBuf,
792    ) -> Self {
793        Self {
794            native,
795            journal_path,
796            exo_lane_path,
797        }
798    }
799
800    /// The native agent server underneath.
801    #[must_use]
802    pub fn native(&self) -> &agent::NativeAgentServer {
803        &self.native
804    }
805}
806
807impl agent_servers::AgentServer for OmegaRouterServer {
808    fn agent_id(&self) -> project::AgentId {
809        self.native.agent_id()
810    }
811
812    fn logo(&self) -> ui::IconName {
813        self.native.logo()
814    }
815
816    fn connect(
817        &self,
818        delegate: agent_servers::AgentServerDelegate,
819        project: Entity<Project>,
820        cx: &mut App,
821    ) -> Task<Result<Rc<dyn AgentConnection>>> {
822        let agent_server_store = delegate.store().downgrade();
823        let native = self.native.connect(delegate, project.clone(), cx);
824        let journal_path = self.journal_path.clone();
825        let exo_lane_path = self.exo_lane_path.clone();
826        cx.spawn(async move |cx| {
827            let native = native.await?;
828            let mut router = OmegaAgentConnection::new(native, RouteJournal::at(journal_path));
829            // `OMEGA-DELTA-0042`, omega#87. The Exo harness lane, when the owner
830            // configured one. Registered as the external executor rather than
831            // as its own class: see `omega_exo_lane`'s module docs for why an
832            // Exo thread reports `ExternalAcp` and why it must not report an
833            // engine lane. A machine with no Exo registers nothing, and a pin
834            // to the external executor then falls back visibly with
835            // `RouteReason::ExternalAcpUnavailable`.
836            if let Some(exo) = crate::omega_exo_connection::connect_configured_lane(
837                &exo_lane_path,
838                project,
839                agent_server_store,
840                cx,
841            )
842            .await?
843            {
844                router = router.with_external_acp(exo);
845            }
846            let router = Rc::new(router);
847            publish_active_router(&router);
848            Ok(router as Rc<dyn AgentConnection>)
849        })
850    }
851
852    fn into_any(self: Rc<Self>) -> Rc<dyn Any> {
853        self
854    }
855
856    fn default_mode(&self, cx: &App) -> Option<acp::SessionModeId> {
857        self.native.default_mode(cx)
858    }
859
860    fn set_default_mode(
861        &self,
862        mode_id: Option<acp::SessionModeId>,
863        fs: std::sync::Arc<dyn fs::Fs>,
864        cx: &mut App,
865    ) {
866        self.native.set_default_mode(mode_id, fs, cx);
867    }
868
869    fn default_config_option(
870        &self,
871        config_id: &str,
872        cx: &App,
873    ) -> Option<settings::AgentConfigOptionValue> {
874        self.native.default_config_option(config_id, cx)
875    }
876
877    fn set_default_config_option(
878        &self,
879        config_id: &str,
880        value: Option<settings::AgentConfigOptionValue>,
881        fs: std::sync::Arc<dyn fs::Fs>,
882        cx: &mut App,
883    ) {
884        self.native
885            .set_default_config_option(config_id, value, fs, cx);
886    }
887
888    // `favorite_config_option_value_ids` and
889    // `toggle_favorite_config_option_value` are deliberately *not* forwarded.
890    // `NativeAgentServer` does not override either, so forwarding would be a
891    // no-op against the trait default — and the getter's return type names
892    // `HashSet`, which `the_routing_law_has_no_clock_no_randomness_and_no_hash_order`
893    // forbids in this file. A no-op override is not worth spending that check
894    // on. If the native server ever overrides one, forward it then, with a
895    // type alias that does not name hash iteration order.
896}
897
898/// Whether a server is Omega's native agent, wrapped or bare.
899///
900/// Every `downcast::<NativeAgentServer>()` that asked "is this the first-party
901/// agent?" has to go through here now, because the answer is yes for a router
902/// over it. Missing one of these is not a compile error — it is a silently
903/// wrong `false`, which is why `omega_deltas` counts the bare downcasts.
904#[must_use]
905pub fn is_native_agent_server(server: &Rc<dyn agent_servers::AgentServer>) -> bool {
906    server
907        .clone()
908        .downcast::<agent::NativeAgentServer>()
909        .is_some()
910        || server.clone().downcast::<OmegaRouterServer>().is_some()
911}
912
913/// The native connection behind a connection, unwrapping the router.
914///
915/// The connection the *store* holds is the router; the connection a *thread*
916/// holds is the executor. Callers that reach for the native agent through the
917/// store need this; callers that already have a thread's connection do not,
918/// and are deliberately left alone.
919#[must_use]
920pub fn native_connection(
921    connection: &Rc<dyn AgentConnection>,
922) -> Option<Rc<agent::NativeAgentConnection>> {
923    if let Some(native) = connection
924        .clone()
925        .downcast::<agent::NativeAgentConnection>()
926    {
927        return Some(native);
928    }
929    connection
930        .clone()
931        .downcast::<OmegaAgentConnection>()
932        .and_then(|router| {
933            router
934                .native
935                .clone()
936                .downcast::<agent::NativeAgentConnection>()
937        })
938}
939
940#[cfg(test)]
941mod tests {
942    use super::*;
943    use acp_thread::StubAgentConnection;
944    use omega_front_door::{LaneState, RouteReason};
945
946    fn stub(agent_id: &str) -> Rc<dyn AgentConnection> {
947        Rc::new(StubAgentConnection::new().with_agent_id(AgentId::new(agent_id)))
948    }
949
950    fn journal_in(directory: &tempfile::TempDir) -> RouteJournal {
951        RouteJournal::at(directory.path().join("openagents").join(ROUTE_JOURNAL_FILE))
952    }
953
954    fn ready_capacity() -> Value {
955        serde_json::json!({
956            "activeRunCount": 1,
957            "activeRunLimit": 8,
958            "lanes": [
959                { "lane": "codex-local", "state": "busy" },
960                { "lane": "claude-local", "state": "available" },
961            ],
962        })
963    }
964
965    /// The live `get_capacity` fixture the Full Auto roster parses reads into
966    /// engine readiness the router can decide from.
967    ///
968    /// A parser tested only against hand-written JSON proves the parser agrees
969    /// with its author. This one is the record a running `omega-effectd`
970    /// actually returned.
971    #[test]
972    fn the_live_capacity_fixture_reads_as_a_ready_engine() {
973        let capacity: Value = serde_json::from_str(include_str!(
974            "../../full_auto_ui/fixtures/live-omega-effectd.get_capacity.json"
975        ))
976        .expect("the live capacity fixture is JSON");
977
978        let EngineReadiness::Answered {
979            active_run_count,
980            active_run_limit,
981            lanes,
982        } = engine_readiness_from_capacity(&capacity)
983        else {
984            panic!("the live capacity fixture did not read as an answer");
985        };
986        assert_eq!((active_run_count, active_run_limit), (1, 8));
987        assert_eq!(lanes.len(), 7);
988        assert!(
989            lanes
990                .iter()
991                .any(|lane| lane.lane_ref == "codex-local" && lane.state == LaneState::Busy)
992        );
993        assert!(
994            lanes
995                .iter()
996                .any(|lane| lane.lane_ref == "claude-local" && lane.state == LaneState::Available)
997        );
998    }
999
1000    /// A capacity answer this build cannot read fully is not half-believed.
1001    #[test]
1002    fn a_misshapen_capacity_answer_reads_as_unreachable() {
1003        for broken in [
1004            serde_json::json!({}),
1005            serde_json::json!({ "activeRunCount": 0, "lanes": [] }),
1006            serde_json::json!({ "activeRunCount": 0, "activeRunLimit": 8 }),
1007            serde_json::json!({
1008                "activeRunCount": 0,
1009                "activeRunLimit": 8,
1010                "lanes": [{ "lane": "codex-local" }],
1011            }),
1012            Value::Null,
1013        ] {
1014            assert_eq!(
1015                engine_readiness_from_capacity(&broken),
1016                EngineReadiness::Unreachable(EngineUnreachable::ProtocolError),
1017                "{broken}"
1018            );
1019        }
1020        assert!(engine_readiness_from_capacity(&ready_capacity()).answered());
1021    }
1022
1023    /// Exit property 3, at the durable layer: a decision written down survives
1024    /// the process that made it.
1025    ///
1026    /// Falsified by making `record` a no-op: the reopened journal is empty and
1027    /// this fails.
1028    #[test]
1029    fn a_recorded_decision_survives_reopening_the_journal() {
1030        let directory = tempfile::tempdir().expect("temporary directory");
1031        let session = "session-1";
1032
1033        let decision = {
1034            let journal = journal_in(&directory);
1035            let router = OmegaAgentConnection::new(stub("omega-agent"), journal);
1036            router.observe_capacity(Err(EngineUnreachable::NotRunning));
1037            router.decide(session, Some(ExecutorPin::on_lane("claude-local")))
1038        };
1039        assert_eq!(decision.chosen, ExecutorClass::NativeLoop);
1040        assert_eq!(decision.reason, RouteReason::EngineUnreachable);
1041
1042        let reopened = journal_in(&directory);
1043        assert_eq!(reopened.decision(session).as_ref(), Some(&decision));
1044        assert_eq!(reopened.decisions(), vec![(session.to_owned(), decision)]);
1045        assert_eq!(
1046            recorded_route(&acp::SessionId::new(session)),
1047            Some(RouteReason::EngineUnreachable),
1048            "a decision the journal holds must be disclosable on the thread \
1049             that made it"
1050        );
1051    }
1052
1053    /// A journal written under a schema this build does not know is not read as
1054    /// if it were.
1055    #[test]
1056    fn an_unknown_journal_schema_is_not_believed() {
1057        let directory = tempfile::tempdir().expect("temporary directory");
1058        let path = directory.path().join("agent-route-journal.json");
1059        std::fs::write(
1060            &path,
1061            serde_json::to_vec(&serde_json::json!({
1062                "schema": "openagents.omega.agent_route_journal.v99",
1063                "decisions": [{ "sessionId": "s", "decision": "chosen=engine_lane;reason=pin_honored;pin=engine_lane;lane=x" }],
1064            }))
1065            .unwrap(),
1066        )
1067        .unwrap();
1068        assert!(RouteJournal::at(path).decision("s").is_none());
1069    }
1070
1071    /// Exit property 1, at the dispatch layer: the executor a turn reaches is
1072    /// the one the recorded decision names.
1073    ///
1074    /// Falsified by making `executor_for` always return `self.native`: the
1075    /// external case fails.
1076    #[test]
1077    fn dispatch_follows_the_recorded_decision() {
1078        let directory = tempfile::tempdir().expect("temporary directory");
1079        let router = OmegaAgentConnection::new(stub("omega-agent"), journal_in(&directory))
1080            .with_external_acp(stub("codex-acp"))
1081            .with_engine_lane(stub("codex-local"));
1082        router.observe_capacity(Ok(&ready_capacity()));
1083
1084        let external = router.decide(
1085            "s-external",
1086            Some(ExecutorPin::new(ExecutorClass::ExternalAcp)),
1087        );
1088        assert_eq!(external.chosen, ExecutorClass::ExternalAcp);
1089        assert_eq!(
1090            router.executor(external.chosen).agent_id(),
1091            AgentId::new("codex-acp")
1092        );
1093
1094        let lane = router.decide("s-lane", Some(ExecutorPin::on_lane("claude-local")));
1095        assert_eq!(lane.chosen, ExecutorClass::EngineLane);
1096        assert_eq!(lane.lane_ref.as_deref(), Some("claude-local"));
1097        assert_eq!(
1098            router.executor(lane.chosen).agent_id(),
1099            AgentId::new("codex-local")
1100        );
1101
1102        let unpinned = router.decide("s-unpinned", None);
1103        assert_eq!(unpinned.chosen, ExecutorClass::NativeLoop);
1104        assert_eq!(
1105            router.executor(unpinned.chosen).agent_id(),
1106            AgentId::new("omega-agent")
1107        );
1108    }
1109
1110    /// Exit property 2, at the dispatch layer: with the engine down, an
1111    /// engine-lane pin reaches the native loop and the record says why.
1112    #[test]
1113    fn an_engine_down_router_dispatches_to_the_native_loop() {
1114        let directory = tempfile::tempdir().expect("temporary directory");
1115        let router = OmegaAgentConnection::new(stub("omega-agent"), journal_in(&directory))
1116            .with_engine_lane(stub("codex-local"));
1117        router.observe_capacity(Err(EngineUnreachable::Timeout));
1118
1119        let decision = router.decide("s", Some(ExecutorPin::on_lane("claude-local")));
1120        assert_eq!(
1121            router.executor(decision.chosen).agent_id(),
1122            AgentId::new("omega-agent")
1123        );
1124        assert_eq!(decision.reason, RouteReason::EngineUnreachable);
1125        assert!(
1126            decision.reason.phrase().contains("fell back"),
1127            "the fallback must be sayable on the thread's own line"
1128        );
1129        assert_eq!(
1130            router.journal().decision("s").map(|d| d.reason),
1131            Some(RouteReason::EngineUnreachable)
1132        );
1133    }
1134
1135    /// The engine is read, never written. The router holds no run state, so
1136    /// observing a *worse* engine cannot rewrite a decision already made.
1137    ///
1138    /// This is `omega-effectd` staying the sole run authority, as a test: a
1139    /// second authority would re-derive and change what the first one recorded.
1140    #[test]
1141    fn a_later_engine_answer_does_not_rewrite_a_recorded_decision() {
1142        let directory = tempfile::tempdir().expect("temporary directory");
1143        let router = OmegaAgentConnection::new(stub("omega-agent"), journal_in(&directory))
1144            .with_engine_lane(stub("codex-local"));
1145        router.observe_capacity(Ok(&ready_capacity()));
1146
1147        let decision = router.decide("s", Some(ExecutorPin::on_lane("claude-local")));
1148        assert_eq!(decision.chosen, ExecutorClass::EngineLane);
1149
1150        router.observe_capacity(Err(EngineUnreachable::NotRunning));
1151        assert_eq!(router.journal().decision("s").as_ref(), Some(&decision));
1152        // The dispatch path, not only the journal. An earlier draft asserted
1153        // on the journal alone, and falsifying `executor_for` into re-deciding
1154        // on every turn left this test green: the record was intact while the
1155        // turn went somewhere else.
1156        assert_eq!(
1157            router.executor_for(&acp::SessionId::new("s")).agent_id(),
1158            AgentId::new("codex-local"),
1159            "the recorded route must keep the turn where it was placed rather \
1160             than moving mid-thread because capacity changed"
1161        );
1162    }
1163
1164    /// The router presents Omega Agent's identity and never claims to be the
1165    /// executor. `OMEGA-DELTA-0024`.
1166    #[test]
1167    fn the_router_does_not_disclose_itself_as_the_executor() {
1168        let directory = tempfile::tempdir().expect("temporary directory");
1169        let native = stub("omega-agent");
1170        let router = Rc::new(
1171            OmegaAgentConnection::new(native, journal_in(&directory))
1172                .with_external_acp(stub("codex-acp")),
1173        );
1174        assert_eq!(router.agent_id(), AgentId::new("omega-agent"));
1175
1176        // The router is not one of the three executor classes, so nothing that
1177        // classifies a connection can mistake it for one. The dispatch target
1178        // is what a thread carries.
1179        let decision = router.decide("s", Some(ExecutorPin::new(ExecutorClass::ExternalAcp)));
1180        let executor = router.executor(decision.chosen);
1181        assert_ne!(
1182            executor.agent_id(),
1183            router.agent_id(),
1184            "an externally routed thread must carry the external agent, not \
1185             the router, or omega#77's disclosure would name the router as the \
1186             executor"
1187        );
1188    }
1189
1190    /// The same inputs produce the same record through the live layer too, not
1191    /// only through the pure function.
1192    #[test]
1193    fn the_live_router_records_the_same_decision_every_time() {
1194        let directory = tempfile::tempdir().expect("temporary directory");
1195        let router = OmegaAgentConnection::new(stub("omega-agent"), journal_in(&directory))
1196            .with_engine_lane(stub("codex-local"));
1197        router.observe_capacity(Ok(&ready_capacity()));
1198
1199        let first = router.decide("s", Some(ExecutorPin::new(ExecutorClass::EngineLane)));
1200        for _ in 0..16 {
1201            assert_eq!(
1202                router.decide("s", Some(ExecutorPin::new(ExecutorClass::EngineLane))),
1203                first
1204            );
1205        }
1206        assert_eq!(
1207            std::fs::read_to_string(directory.path().join("openagents").join(ROUTE_JOURNAL_FILE))
1208                .unwrap()
1209                .matches("chosen=")
1210                .count(),
1211            1,
1212            "one session must leave one record, not one per decision"
1213        );
1214    }
1215
1216    /// `OMEGA-DELTA-0035`. A human pin moves a live thread, and the record says
1217    /// why it landed where it did.
1218    ///
1219    /// The pin control would be decoration without this: `executor_for` reads
1220    /// the *recorded* decision so a turn cannot drift between executors on its
1221    /// own, which means setting a pin has to re-decide or it changes nothing a
1222    /// turn can see. Falsified by making `pin_session` insert the pin without
1223    /// re-deciding: the executor stays native and this fails.
1224    #[test]
1225    fn a_human_pin_moves_a_live_thread_and_records_why() {
1226        let directory = tempfile::tempdir().expect("temporary directory");
1227        let router = OmegaAgentConnection::new(stub("omega-agent"), journal_in(&directory))
1228            .with_external_acp(stub("codex-acp"));
1229        // Its own session id; see `an_unhonourable_pin_is_kept_and_explained`.
1230        let session = acp::SessionId::new("s-human-pin");
1231
1232        assert_eq!(
1233            router.decide(session.0.as_ref(), None).reason,
1234            RouteReason::UnpinnedDefault
1235        );
1236        assert_eq!(
1237            router.executor_for(&session).agent_id(),
1238            AgentId::new("omega-agent")
1239        );
1240
1241        let decision = router.pin_session(
1242            &session,
1243            ExecutorPin::new(ExecutorClass::ExternalAcp),
1244            PinGesture::ExecutorPinMenuItem,
1245        );
1246        assert_eq!(decision.reason, RouteReason::PinHonored);
1247        assert_eq!(
1248            router.executor_for(&session).agent_id(),
1249            AgentId::new("codex-acp"),
1250            "a pin a person set must move the turn, or the control is decoration"
1251        );
1252        assert_eq!(
1253            router
1254                .journal()
1255                .decision(session.0.as_ref())
1256                .map(|d| d.chosen),
1257            Some(ExecutorClass::ExternalAcp)
1258        );
1259    }
1260
1261    /// `OMEGA-DELTA-0035`. An unhonourable pin is kept, and its reason is
1262    /// sayable on the thread's own line.
1263    ///
1264    /// This is the case the rendered proof photographs: no engine is running,
1265    /// so an engine-lane pin falls closed to the native loop. The pin is *not*
1266    /// forgotten — an unhonoured pin the record dropped is indistinguishable
1267    /// from no pin at all.
1268    #[test]
1269    fn an_unhonourable_pin_is_kept_and_explained() {
1270        let directory = tempfile::tempdir().expect("temporary directory");
1271        let router = OmegaAgentConnection::new(stub("omega-agent"), journal_in(&directory));
1272        // A session id no other test uses. `RECORDED_ROUTES` is a
1273        // process-wide projection keyed by session id, and the tests in this
1274        // module run in one process: a shared `"s"` made this assertion read
1275        // another test's decision, which it caught on the first run. Real
1276        // session ids are minted per session, so this is a harness concern and
1277        // not a production one.
1278        let session = acp::SessionId::new("s-unhonourable-pin");
1279        router.observe_capacity(Err(EngineUnreachable::NotRunning));
1280
1281        let decision = router.pin_session(
1282            &session,
1283            ExecutorPin::new(ExecutorClass::EngineLane),
1284            PinGesture::ExecutorPinMenuItem,
1285        );
1286        assert_eq!(decision.chosen, ExecutorClass::NativeLoop);
1287        assert!(decision.reason.is_fallback());
1288        assert_eq!(
1289            decision.pin.as_ref().map(ExecutorPin::token).as_deref(),
1290            Some("engine_lane")
1291        );
1292        assert!(
1293            decision
1294                .reason
1295                .phrase()
1296                .contains("fell back to the native loop")
1297        );
1298        assert_eq!(
1299            recorded_route(&session),
1300            Some(decision.reason),
1301            "the thread's disclosure line reads this index, so a reason the \
1302             journal holds and the index does not is a fallback the user \
1303             cannot see"
1304        );
1305
1306        let cleared = router.unpin_session(&session, PinGesture::ExecutorPinCleared);
1307        assert_eq!(cleared.reason, RouteReason::UnpinnedDefault);
1308        assert!(cleared.pin.is_none());
1309        assert!(router.pin(&session).is_none());
1310    }
1311
1312    /// `OMEGA-DELTA-0035`. A bare downcast through the router misses the
1313    /// native loop — which is the whole reason `native_connection` exists.
1314    ///
1315    /// Five call sites downcast a connection to `NativeAgentConnection`. Three
1316    /// hold a *thread's* connection, which is the executor's and unaffected;
1317    /// the ones that hold the *store's* connection now hold the router, and a
1318    /// bare downcast there returns `None` — a silently wrong "this is not the
1319    /// native agent" rather than a compile error.
1320    #[test]
1321    fn a_bare_downcast_through_the_router_misses_the_native_loop() {
1322        let directory = tempfile::tempdir().expect("temporary directory");
1323        let native = stub("omega-agent");
1324        let bare: Rc<dyn AgentConnection> = native.clone();
1325        assert!(
1326            native_connection(&bare).is_none(),
1327            "the stub is not a NativeAgentConnection, so this only shows the \
1328             helper does not answer yes to anything"
1329        );
1330
1331        let router: Rc<dyn AgentConnection> =
1332            Rc::new(OmegaAgentConnection::new(native, journal_in(&directory)));
1333        assert!(
1334            router
1335                .clone()
1336                .downcast::<agent::NativeAgentConnection>()
1337                .is_none(),
1338            "a bare downcast through the router must not find the native loop; \
1339             if it did, this check would prove nothing about the unwrapping \
1340             helper"
1341        );
1342        assert_eq!(router.agent_id(), AgentId::new("omega-agent"));
1343    }
1344}
1345