tenant.openagents/omega

1//! Omega Agent, served over ACP on a loopback socket. `OMEGA-DELTA-0041`,
2//! omega#82.
3//!
4//! Omega is already an ACP **client**: `crates/agent_servers` reaches out to
5//! external agents. This crate is the other direction. An external ACP host —
6//! stock Zed, another fork, any conformant client — can attach to Omega and get
7//! Omega Agent, the router, with the same disclosed routing an in-app thread
8//! gets. Recursive composability, which is what omega#82 asks for.
9//!
10//! Four properties hold this surface down, and each is structural rather than
11//! a runtime check something could forget to call.
12//!
13//! # 1. Default off
14//!
15//! [`enablement`] is an **exact** match on `"1"`. `"true"`, `"yes"`, `"on"`,
16//! `" 1"`, and an unset variable are all off, and the caller gets a typed
17//! [`OffReason`] rather than a bool. A listener that is on by default is a
18//! different product from one that is off by default, so the default is
19//! asserted rather than assumed.
20//!
21//! # 2. Loopback only
22//!
23//! [`LoopbackHost`] is a **construction invariant**, not a configuration check.
24//! There is no way to hold a bind address that is not `127.0.0.1` or `::1`,
25//! because the only constructor refuses anything else, and
26//! [`LoopbackAcpServer::bind`] takes a `LoopbackHost` rather than a string. A
27//! routable interface cannot be reached by editing a setting; it needs a new
28//! type.
29//!
30//! # 3. Read-only, inside the authority partition rather than beside it
31//!
32//! This is an **unauthenticated** model-driven surface: [`AUTH_METHODS`] is
33//! empty, so it carries no bearer at all and is structurally weaker than the
34//! Desktop MCP surface. Owner gate 8 — *no model-initiated path can start Full
35//! Auto authority; only an explicit human action can, wherever that action
36//! lives* — therefore reaches it directly.
37//!
38//! So the partition is the design. [`SERVED_SURFACE`] is every method an
39//! attached host can reach, each classified, and every one of them is an
40//! [`SurfaceAuthority::Observation`]. [`UNEXPOSED_AUTHORITY`] is the other
41//! half: every control in Omega that grants Full Auto authority or mutates run
42//! state, listed with the typed refusal a host attempting it receives. That
43//! list is checked **against the existing ledgers** — every
44//! [`omega_front_door::FULL_AUTO_AFFORDANCES`] element id and every
45//! [`omega_front_door::PinGesture`] must appear in it — so a new Full Auto
46//! control fails this crate's tests until somebody classifies it. Adding a
47//! control cannot silently open a door here.
48//!
49//! Anything not in [`SERVED_SURFACE`] is refused: [`served_method`] is the only
50//! dispatch table, so deny-by-default is the absence of an entry rather than a
51//! branch someone has to remember to write.
52//!
53//! A served prompt is answered by **disclosing**, not by executing. Nothing is
54//! dispatched to any executor, so the answer's stop reason is the protocol's
55//! own word for it, `refusal`.
56//!
57//! # 4. The served agent is the router, and it can never reach an engine lane
58//!
59//! [`served_route`] calls [`omega_front_door::route`] — the same routing law an
60//! in-app thread is decided by — with a pin of `None`. A pin is the only door
61//! to an engine lane, an engine lane *is* Full Auto authority, and setting one
62//! requires an `omega_front_door::PinGesture`, every variant of which is a
63//! visible control a person operates. There is no variant for "an external host
64//! asked", so the served surface cannot take a pin, and
65//! `a_served_session_can_never_reach_an_engine_lane` proves the consequence
66//! against every engine state the router can be shown.
67//!
68//! # Where the socket lives
69//!
70//! Not in a GPUI crate. This crate depends on `omega_front_door` and the pinned
71//! ACP schema and on **no** part of GPUI, and a check in `crates/omega_deltas`
72//! fails if a listener appears in `crates/agent_ui`, `crates/full_auto_ui`, or
73//! `crates/zed`. The lifecycle owner in a shipped build is `omega-effectd`:
74//! [`start_if_enabled`] is called from `crates/omega_effectd`, which is the
75//! supervisor layer, and nowhere else.
76//!
77//! **What this does not do,** stated rather than discovered: the listener runs
78//! in the Omega process under the supervisor's control, not inside the packaged
79//! `@openagentsinc/omega-effectd` daemon. The daemon lives in the openagents
80//! repository and this packet is scoped to omega. The property omega#82's
81//! falsifier names — *GPUI owns the socket* — is what is enforced here, by
82//! keeping the bind in a crate GPUI cannot reach and the start in the
83//! supervisor.
84
85use std::io::{BufRead, BufReader, Write};
86use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, TcpStream};
87
88use agent_client_protocol::schema::v1::{AGENT_METHOD_NAMES, CLIENT_METHOD_NAMES};
89use omega_front_door::{
90    EngineReadiness, EngineUnreachable, ExecutorDisclosure, RouteDecision, RouteInputs,
91    SessionOrigin, route,
92};
93use serde_json::{Value, json};
94
95// -------------------------------------------------------------------------
96// Default off
97// -------------------------------------------------------------------------
98
99/// The environment variable that turns the loopback ACP server on.
100pub const ENABLE_FLAG: &str = "OMEGA_ACP_SERVER";
101
102/// The environment variable that pins the port. Optional; `0` means ephemeral.
103pub const PORT_FLAG: &str = "OMEGA_ACP_SERVER_PORT";
104
105/// The one value of [`ENABLE_FLAG`] that turns the server on.
106///
107/// Exact. `"true"`, `"yes"`, `"on"`, `"01"` and `" 1"` are all off. A flag that
108/// accepts anything truthy is a flag whose default nobody can state.
109pub const ENABLE_VALUE: &str = "1";
110
111/// Why the server is off.
112#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub enum OffReason {
114    /// The flag is not set at all. This is the shipped default.
115    FlagUnset,
116    /// The flag is set to something other than exactly [`ENABLE_VALUE`].
117    FlagNotExactlyOne,
118}
119
120impl OffReason {
121    /// The stable token this reason is recorded under.
122    #[must_use]
123    pub const fn token(self) -> &'static str {
124        match self {
125            Self::FlagUnset => "flag_unset",
126            Self::FlagNotExactlyOne => "flag_not_exactly_one",
127        }
128    }
129}
130
131/// Whether the loopback ACP server runs.
132#[derive(Debug, Clone, Copy, PartialEq, Eq)]
133pub enum Enablement {
134    /// Off, with the reason.
135    Off(OffReason),
136    /// On, because the flag said exactly [`ENABLE_VALUE`].
137    On,
138}
139
140impl Enablement {
141    /// Whether a listener should be opened.
142    #[must_use]
143    pub const fn is_on(self) -> bool {
144        matches!(self, Self::On)
145    }
146}
147
148/// Read the enablement from the flag's value.
149///
150/// Takes the value rather than reading the environment, so the default can be
151/// asserted without a process.
152#[must_use]
153pub fn enablement(flag: Option<&str>) -> Enablement {
154    match flag {
155        None => Enablement::Off(OffReason::FlagUnset),
156        Some(ENABLE_VALUE) => Enablement::On,
157        Some(_) => Enablement::Off(OffReason::FlagNotExactlyOne),
158    }
159}
160
161// -------------------------------------------------------------------------
162// Loopback only
163// -------------------------------------------------------------------------
164
165/// A bind address that is loopback, proven by construction.
166///
167/// There is no way to build one that is not, so nothing downstream has to
168/// check. `LoopbackAcpServer::bind` takes this type rather than a string, which
169/// is what makes "never binds a routable interface" a property of the type
170/// system instead of a code review.
171#[derive(Debug, Clone, Copy, PartialEq, Eq)]
172pub struct LoopbackHost(IpAddr);
173
174/// Why an address was refused as a bind target.
175#[derive(Debug, Clone, PartialEq, Eq)]
176pub enum BindRefusal {
177    /// The text is not an IP address at all. A hostname is refused rather than
178    /// resolved: resolution is the step where `localhost` can be made to mean
179    /// something routable.
180    NotAnAddress(String),
181    /// The address parses and is not loopback.
182    NotLoopback(IpAddr),
183}
184
185impl BindRefusal {
186    /// The stable token this refusal is recorded under.
187    #[must_use]
188    pub const fn token(&self) -> &'static str {
189        match self {
190            Self::NotAnAddress(_) => "not_an_address",
191            Self::NotLoopback(_) => "not_loopback",
192        }
193    }
194}
195
196impl std::fmt::Display for BindRefusal {
197    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
198        match self {
199            Self::NotAnAddress(raw) => write!(
200                formatter,
201                "{raw:?} is not an IP address. The Omega ACP server binds a \
202                 literal loopback address and never resolves a name."
203            ),
204            Self::NotLoopback(address) => write!(
205                formatter,
206                "{address} is not a loopback address. The Omega ACP server is \
207                 an unauthenticated surface and never binds a routable \
208                 interface."
209            ),
210        }
211    }
212}
213
214impl LoopbackHost {
215    /// The shipped bind address.
216    pub const DEFAULT: Self = Self(IpAddr::V4(Ipv4Addr::LOCALHOST));
217
218    /// Take a loopback address, or refuse.
219    ///
220    /// # Errors
221    ///
222    /// [`BindRefusal`] when the text is not an IP address, or is an address
223    /// that is not loopback.
224    pub fn new(raw: &str) -> Result<Self, BindRefusal> {
225        let address: IpAddr = raw
226            .parse()
227            .map_err(|_| BindRefusal::NotAnAddress(raw.to_owned()))?;
228        if address == IpAddr::V4(Ipv4Addr::LOCALHOST) || address == IpAddr::V6(Ipv6Addr::LOCALHOST)
229        {
230            Ok(Self(address))
231        } else {
232            Err(BindRefusal::NotLoopback(address))
233        }
234    }
235
236    /// The address itself.
237    #[must_use]
238    pub const fn address(self) -> IpAddr {
239        self.0
240    }
241}
242
243// -------------------------------------------------------------------------
244// The authority partition
245// -------------------------------------------------------------------------
246
247/// What an attached host is refused, and why.
248///
249/// Typed rather than a sentence, so a caller can branch on it and so the same
250/// refusal reads the same on the wire and in a test.
251#[derive(Debug, Clone, Copy, PartialEq, Eq)]
252pub enum ServedRefusal {
253    /// A prompt was answered by disclosing where it would route, and dispatched
254    /// to nothing.
255    ExecutesNothing,
256    /// A run may not be started over this surface.
257    StartsNoRun,
258    /// A run's state may not be changed over this surface.
259    MutatesNoRun,
260    /// An executor pin may not be set or cleared over this surface.
261    SetsNoPin,
262    /// An account or credential may not be connected or revoked over this
263    /// surface.
264    GrantsNoCredential,
265    /// The method is not part of the served surface at all.
266    NotExposed,
267}
268
269impl ServedRefusal {
270    /// Every refusal, in declaration order.
271    #[must_use]
272    pub const fn all() -> &'static [Self] {
273        &[
274            Self::ExecutesNothing,
275            Self::StartsNoRun,
276            Self::MutatesNoRun,
277            Self::SetsNoPin,
278            Self::GrantsNoCredential,
279            Self::NotExposed,
280        ]
281    }
282
283    /// The stable token this refusal is recorded under.
284    #[must_use]
285    pub const fn token(self) -> &'static str {
286        match self {
287            Self::ExecutesNothing => "served_executes_nothing",
288            Self::StartsNoRun => "served_starts_no_run",
289            Self::MutatesNoRun => "served_mutates_no_run",
290            Self::SetsNoPin => "served_sets_no_pin",
291            Self::GrantsNoCredential => "served_grants_no_credential",
292            Self::NotExposed => "served_method_not_exposed",
293        }
294    }
295
296    /// What the attached host is told, in its operator's terms.
297    ///
298    /// Derived on every call; nothing stores it.
299    #[must_use]
300    pub const fn message(self) -> &'static str {
301        match self {
302            Self::ExecutesNothing => {
303                "Omega Agent served over ACP is read-only. This turn was routed \
304                 and disclosed, and dispatched to no executor."
305            }
306            Self::StartsNoRun => {
307                "A run cannot be started over the served ACP surface. Only an \
308                 explicit human action in Omega starts Full Auto authority."
309            }
310            Self::MutatesNoRun => {
311                "Run state cannot be changed over the served ACP surface. The \
312                 engine remains the sole run authority and Omega's own controls \
313                 are the only way to reach it."
314            }
315            Self::SetsNoPin => {
316                "An executor pin cannot be set over the served ACP surface. A \
317                 pin is the only door to an engine lane and it is set by a \
318                 human gesture on a visible control."
319            }
320            Self::GrantsNoCredential => {
321                "An account cannot be connected or revoked over the served ACP \
322                 surface. This surface is unauthenticated and grants nothing."
323            }
324            Self::NotExposed => {
325                "This method is not part of the served ACP surface. The surface \
326                 is deny-by-default: what is not listed is refused."
327            }
328        }
329    }
330}
331
332/// What an attached host may do with a served method.
333#[derive(Debug, Clone, Copy, PartialEq, Eq)]
334pub enum SurfaceAuthority {
335    /// Reads a record or mints one. Grants nothing and changes no run.
336    ///
337    /// Every entry of [`SERVED_SURFACE`] is this, and
338    /// `the_whole_served_surface_is_observation` fails if one is not.
339    Observation,
340}
341
342/// Which handler a served method reaches.
343///
344/// The dispatch is driven by [`SERVED_SURFACE`] rather than by a `match` on a
345/// string, so a method with no entry in the table has no handler to reach and
346/// deny-by-default is the absence of a row.
347#[derive(Debug, Clone, Copy, PartialEq, Eq)]
348pub enum ServedMethodKind {
349    /// `initialize`.
350    Initialize,
351    /// `session/new`.
352    NewSession,
353    /// `session/prompt`.
354    Prompt,
355    /// `session/cancel`, a notification.
356    Cancel,
357}
358
359/// One method an attached host may reach, and what it is allowed to do.
360#[derive(Debug, Clone, Copy, PartialEq, Eq)]
361pub struct ServedMethod {
362    /// The ACP method name, taken from the pinned schema's own constants so a
363    /// protocol rename is a compile error rather than a silent 404.
364    pub method: &'static str,
365    /// Which handler it reaches.
366    pub kind: ServedMethodKind,
367    /// What it is allowed to do.
368    pub authority: SurfaceAuthority,
369}
370
371/// Every method the served surface exposes.
372///
373/// Deliberately short. `session/load`, `session/resume`, `session/fork`,
374/// `session/set_mode`, `session/set_config_option`, `session/delete`,
375/// `authenticate`, and `logout` are all absent, and absence is the refusal:
376/// [`served_method`] returns `None` and the caller answers
377/// [`ServedRefusal::NotExposed`].
378pub const SERVED_SURFACE: &[ServedMethod] = &[
379    ServedMethod {
380        method: AGENT_METHOD_NAMES.initialize,
381        kind: ServedMethodKind::Initialize,
382        authority: SurfaceAuthority::Observation,
383    },
384    ServedMethod {
385        method: AGENT_METHOD_NAMES.session_new,
386        kind: ServedMethodKind::NewSession,
387        authority: SurfaceAuthority::Observation,
388    },
389    ServedMethod {
390        method: AGENT_METHOD_NAMES.session_prompt,
391        kind: ServedMethodKind::Prompt,
392        authority: SurfaceAuthority::Observation,
393    },
394    ServedMethod {
395        method: AGENT_METHOD_NAMES.session_cancel,
396        kind: ServedMethodKind::Cancel,
397        authority: SurfaceAuthority::Observation,
398    },
399];
400
401/// The entry for a method, or `None` if the surface does not expose it.
402#[must_use]
403pub fn served_method(method: &str) -> Option<&'static ServedMethod> {
404    SERVED_SURFACE.iter().find(|entry| entry.method == method)
405}
406
407/// One Omega control that grants authority or mutates run state, and is
408/// deliberately not on the served surface.
409#[derive(Debug, Clone, Copy, PartialEq, Eq)]
410pub struct UnexposedControl {
411    /// The control's own identifier: a `full_auto_ui` element id, or an
412    /// `omega_front_door::PinGesture` token.
413    pub control: &'static str,
414    /// What an attached host attempting it is told.
415    pub refusal: ServedRefusal,
416}
417
418/// Every authority-bearing control in Omega, and the refusal a served host gets
419/// for it.
420///
421/// This is the half of the partition that makes the served surface *inside* it
422/// rather than beside it. `every_authority_bearing_control_is_classified`
423/// checks this list against the two ledgers that already exist —
424/// [`omega_front_door::FULL_AUTO_AFFORDANCES`] and
425/// [`omega_front_door::PinGesture::all`] — in both directions, so adding a Full
426/// Auto control or a pin gesture fails this crate until it is classified here,
427/// and removing one fails until the stale row goes.
428pub const UNEXPOSED_AUTHORITY: &[UnexposedControl] = &[
429    UnexposedControl {
430        control: "full-auto-panel",
431        refusal: ServedRefusal::NotExposed,
432    },
433    UnexposedControl {
434        control: "full-auto-openagents-connect",
435        refusal: ServedRefusal::GrantsNoCredential,
436    },
437    UnexposedControl {
438        control: "full-auto-openagents-disconnect",
439        refusal: ServedRefusal::GrantsNoCredential,
440    },
441    UnexposedControl {
442        control: "full-auto-provider-account",
443        refusal: ServedRefusal::GrantsNoCredential,
444    },
445    UnexposedControl {
446        control: "full-auto-advanced-toggle",
447        refusal: ServedRefusal::NotExposed,
448    },
449    UnexposedControl {
450        control: "full-auto-start",
451        refusal: ServedRefusal::StartsNoRun,
452    },
453    UnexposedControl {
454        control: "full-auto-cancel",
455        refusal: ServedRefusal::NotExposed,
456    },
457    UnexposedControl {
458        control: "full-auto-pause",
459        refusal: ServedRefusal::MutatesNoRun,
460    },
461    UnexposedControl {
462        control: "full-auto-resume",
463        refusal: ServedRefusal::MutatesNoRun,
464    },
465    UnexposedControl {
466        control: "full-auto-handoff",
467        refusal: ServedRefusal::MutatesNoRun,
468    },
469    UnexposedControl {
470        control: "full-auto-retry",
471        refusal: ServedRefusal::MutatesNoRun,
472    },
473    UnexposedControl {
474        control: "full-auto-stop",
475        refusal: ServedRefusal::MutatesNoRun,
476    },
477    UnexposedControl {
478        control: "full-auto-new",
479        refusal: ServedRefusal::StartsNoRun,
480    },
481    UnexposedControl {
482        control: "full-auto-evidence-chain",
483        refusal: ServedRefusal::NotExposed,
484    },
485    UnexposedControl {
486        control: "full-auto-monitor",
487        refusal: ServedRefusal::NotExposed,
488    },
489    UnexposedControl {
490        control: "full-auto-monitor-new",
491        refusal: ServedRefusal::StartsNoRun,
492    },
493    UnexposedControl {
494        control: "full-auto-run-row",
495        refusal: ServedRefusal::NotExposed,
496    },
497    UnexposedControl {
498        control: "executor_pin_menu_item",
499        refusal: ServedRefusal::SetsNoPin,
500    },
501    UnexposedControl {
502        control: "executor_pin_cleared",
503        refusal: ServedRefusal::SetsNoPin,
504    },
505];
506
507// -------------------------------------------------------------------------
508// The served session
509// -------------------------------------------------------------------------
510
511/// The agent identity the served surface presents.
512///
513/// The same identity an in-app thread discloses, because it is the same agent.
514/// `the_served_surface_presents_the_first_party_agent_id` in `omega_deltas`
515/// checks this against `crates/agent`'s own `OMEGA_AGENT_ID`, so the two cannot
516/// drift into a served surface that claims to be something else.
517pub const SERVED_AGENT_ID: &str = "Omega Agent";
518
519/// The authentication methods the served surface offers.
520///
521/// Empty, and empty on purpose. There is no credential to present, so there is
522/// no credential to steal or to mistake for authority. That is also why the
523/// surface must be read-only: an unauthenticated surface that could act would
524/// be a hole with no gate in front of it.
525pub const AUTH_METHODS: &[&str] = &[];
526
527/// The routing inputs a served session is decided from.
528///
529/// The pin is `None` and there is no way to make it anything else from this
530/// crate: setting a pin requires an `omega_front_door::PinGesture`, and no
531/// variant of that enum is reachable over a socket. The engine is reported
532/// unreachable because the served surface does not hold the supervisor's
533/// answer — and it does not matter, which
534/// `a_served_session_can_never_reach_an_engine_lane` proves by showing the
535/// decision is identical for every engine state.
536#[must_use]
537pub fn served_inputs() -> RouteInputs {
538    RouteInputs {
539        pin: None,
540        engine: EngineReadiness::Unreachable(EngineUnreachable::NotRunning),
541        external_acp: None,
542        engine_lane: None,
543    }
544}
545
546/// The route decision a served session gets.
547///
548/// The same [`omega_front_door::route`] law an in-app thread is decided by, so
549/// an attached host gets Omega Agent's disclosed routing rather than a raw
550/// provider.
551#[must_use]
552pub fn served_route() -> RouteDecision {
553    route(&served_inputs())
554}
555
556/// The disclosure a served session carries.
557///
558/// A typed record. The provider and model are `None` — *not disclosed*, which
559/// is the honest answer, because the served surface dispatches to no executor
560/// and so no provider served anything.
561#[must_use]
562pub fn served_disclosure(decision: &RouteDecision) -> ExecutorDisclosure {
563    ExecutorDisclosure {
564        class: decision.chosen,
565        agent_id: SERVED_AGENT_ID.to_owned(),
566        provider: None,
567        model: None,
568        run_ref: None,
569        route: Some(decision.disclosed_route()),
570    }
571}
572
573/// One session an external host reached over the loopback server.
574#[derive(Debug, Clone, PartialEq, Eq)]
575pub struct ServedSession {
576    /// The ACP session id this session was minted under.
577    pub session_id: String,
578    /// Where it was reached from. A separate record from the disclosure,
579    /// because ingress and execution are different facts.
580    pub origin: SessionOrigin,
581    /// Where a turn on it would be routed.
582    pub decision: RouteDecision,
583    /// What it says about the executor. Typed; a label renders it.
584    pub disclosure: ExecutorDisclosure,
585}
586
587/// The `_meta` key the served records are published under.
588pub const SERVED_SESSION_META_KEY: &str = "openagents.omega.served_session.v1";
589
590/// The disclosure record as it goes on the wire.
591///
592/// Keys are exactly `omega_front_door::EXECUTOR_DISCLOSURE_FIELDS`, and
593/// `the_served_disclosure_is_a_record_not_a_label` asserts that exactly. A
594/// rendered line on the wire under any name — `label`, `line`, `text`,
595/// `summary` — fails, because the binding condition of the owner's identity
596/// decision is that disclosure travels as a record a label renders.
597#[must_use]
598pub fn disclosure_meta(disclosure: &ExecutorDisclosure) -> Value {
599    json!({
600        "class": disclosure.class.token(),
601        "agent_id": disclosure.agent_id,
602        "provider": disclosure.provider,
603        "model": disclosure.model,
604        "run_ref": disclosure.run_ref,
605        "route": disclosure.route.map(|route| route.token()),
606    })
607}
608
609/// The origin record as it goes on the wire. Keys exactly
610/// `omega_front_door::SESSION_ORIGIN_FIELDS`.
611#[must_use]
612pub fn origin_meta(origin: &SessionOrigin) -> Value {
613    json!({
614        "ingress": origin.ingress.token(),
615        "host_name": origin.host_name,
616        "host_version": origin.host_version,
617        "authenticated": origin.authenticated,
618    })
619}
620
621/// The `_meta` block a served **turn** publishes.
622///
623/// The session's records, plus the typed refusal saying the turn reached no
624/// executor. The refusal lives here rather than in the stop reason on purpose:
625/// ACP's `refusal` stop reason means *the prompt and everything after it will
626/// not be included in the next prompt*, and stock Zed implements that by
627/// dropping the turn from the thread — which took the disclosure with it, so
628/// the operator saw a refusal banner and no disclosure at all. That was watched
629/// happening against Zed 1.12.0 before this shape was chosen. The turn genuinely
630/// ended, so it says `end_turn`, and what did **not** happen is stated in the
631/// message the operator reads and in the record beside it.
632#[must_use]
633pub fn served_turn_meta(session: &ServedSession) -> Value {
634    let mut meta = served_meta(session);
635    if let Some(block) = meta
636        .get_mut(SERVED_SESSION_META_KEY)
637        .and_then(Value::as_object_mut)
638    {
639        block.insert(
640            "refusal".to_owned(),
641            Value::String(ServedRefusal::ExecutesNothing.token().to_owned()),
642        );
643    }
644    meta
645}
646
647/// The `_meta` block a served session publishes.
648#[must_use]
649pub fn served_meta(session: &ServedSession) -> Value {
650    json!({
651        SERVED_SESSION_META_KEY: {
652            "disclosure": disclosure_meta(&session.disclosure),
653            "origin": origin_meta(&session.origin),
654            "readOnly": true,
655            "refusals": ServedRefusal::all()
656                .iter()
657                .map(|refusal| refusal.token())
658                .collect::<Vec<_>>(),
659        }
660    })
661}
662
663/// What a served turn says back, rendered from the typed records.
664///
665/// Derived on every call. Nothing stores it, which is the same discipline
666/// `ExecutorDisclosure::label` is held to — the wire carries the record in
667/// `_meta` and this rendering in the message the operator reads.
668#[must_use]
669pub fn served_turn_text(session: &ServedSession) -> String {
670    format!(
671        "Omega Agent, served over ACP.\n\n\
672         executor: {}\n\
673         origin: {}\n\n\
674         {}",
675        session.disclosure.label(),
676        session.origin.label(),
677        ServedRefusal::ExecutesNothing.message(),
678    )
679}
680
681// -------------------------------------------------------------------------
682// The connection
683// -------------------------------------------------------------------------
684
685/// The JSON-RPC error code for a method the surface does not expose.
686const METHOD_NOT_FOUND: i64 = -32601;
687
688/// The JSON-RPC error code for a request this surface cannot read.
689const INVALID_REQUEST: i64 = -32600;
690
691/// The ACP protocol version this surface speaks.
692const PROTOCOL_VERSION: u16 = 1;
693
694/// One attached host.
695///
696/// Holds the sessions it minted and what it said about itself. It holds no run
697/// state, no policy state, and no credential, because there is none to hold.
698#[derive(Debug)]
699pub struct ServedConnection {
700    /// A stable label for this connection, used to mint session ids without a
701    /// clock or a random source.
702    connection_ref: String,
703    /// Sessions minted on this connection, in mint order.
704    sessions: Vec<ServedSession>,
705    /// What the host said about itself at `initialize`.
706    host_name: Option<String>,
707    /// The host's version, where it gave one.
708    host_version: Option<String>,
709}
710
711impl ServedConnection {
712    /// A connection with nothing on it yet.
713    #[must_use]
714    pub fn new(connection_ref: impl Into<String>) -> Self {
715        Self {
716            connection_ref: connection_ref.into(),
717            sessions: Vec::new(),
718            host_name: None,
719            host_version: None,
720        }
721    }
722
723    /// The sessions minted on this connection.
724    #[must_use]
725    pub fn sessions(&self) -> &[ServedSession] {
726        &self.sessions
727    }
728
729    /// One session by id.
730    #[must_use]
731    pub fn session(&self, session_id: &str) -> Option<&ServedSession> {
732        self.sessions
733            .iter()
734            .find(|session| session.session_id == session_id)
735    }
736
737    /// The origin every session on this connection carries.
738    #[must_use]
739    pub fn origin(&self) -> SessionOrigin {
740        SessionOrigin::loopback_acp(self.host_name.clone(), self.host_version.clone())
741    }
742
743    /// Handle one newline-framed JSON-RPC message, and return the lines to
744    /// write back in order.
745    ///
746    /// Notifications the surface emits come before the response they belong to,
747    /// which is the order ACP requires for a prompt turn.
748    pub fn handle_line(&mut self, line: &str) -> Vec<String> {
749        let Ok(message) = serde_json::from_str::<Value>(line) else {
750            return vec![error_line(
751                &Value::Null,
752                INVALID_REQUEST,
753                "The Omega ACP server reads newline-framed JSON-RPC.",
754                ServedRefusal::NotExposed,
755            )];
756        };
757        let id = message.get("id").cloned();
758        let Some(method) = message.get("method").and_then(Value::as_str) else {
759            // A response to a request. This surface sends none, so there is
760            // nothing this can be an answer to and nothing to say back.
761            return Vec::new();
762        };
763        let params = message.get("params").cloned().unwrap_or(Value::Null);
764
765        let Some(entry) = served_method(method) else {
766            log::info!(
767                "OMEGA-DELTA-0041: served surface refused {method}: {}",
768                ServedRefusal::NotExposed.token()
769            );
770            return match id {
771                Some(id) => vec![error_line(
772                    &id,
773                    METHOD_NOT_FOUND,
774                    ServedRefusal::NotExposed.message(),
775                    ServedRefusal::NotExposed,
776                )],
777                // A notification gets no answer, by JSON-RPC. It is still
778                // refused: nothing happened.
779                None => Vec::new(),
780            };
781        };
782
783        match entry.kind {
784            ServedMethodKind::Initialize => {
785                self.remember_host(&params);
786                id.map(|id| vec![result_line(&id, self.initialize_result())])
787                    .unwrap_or_default()
788            }
789            ServedMethodKind::NewSession => {
790                let session = self.mint_session();
791                let result = json!({
792                    "sessionId": session.session_id,
793                    "_meta": served_meta(&session),
794                });
795                self.sessions.push(session);
796                id.map(|id| vec![result_line(&id, result)]).unwrap_or_default()
797            }
798            ServedMethodKind::Prompt => {
799                let session_id = params
800                    .get("sessionId")
801                    .and_then(Value::as_str)
802                    .unwrap_or_default()
803                    .to_owned();
804                let Some(session) = self.session(&session_id).cloned() else {
805                    return match id {
806                        Some(id) => vec![error_line(
807                            &id,
808                            INVALID_REQUEST,
809                            "No such session on this connection.",
810                            ServedRefusal::NotExposed,
811                        )],
812                        None => Vec::new(),
813                    };
814                };
815                let mut lines = vec![notification_line(
816                    CLIENT_METHOD_NAMES.session_update,
817                    json!({
818                        "sessionId": session.session_id,
819                        "update": {
820                            "sessionUpdate": "agent_message_chunk",
821                            "content": { "type": "text", "text": served_turn_text(&session) },
822                        },
823                        "_meta": served_turn_meta(&session),
824                    }),
825                )];
826                if let Some(id) = id {
827                    lines.push(result_line(
828                        &id,
829                        json!({
830                            // The turn ended, having disclosed. Not ACP's
831                            // `refusal`, which means the turn is dropped from
832                            // the thread — stock Zed implements that literally
833                            // and the disclosure went with it. What did not
834                            // happen is in the message and in `_meta`.
835                            "stopReason": "end_turn",
836                            "_meta": served_turn_meta(&session),
837                        }),
838                    ));
839                }
840                lines
841            }
842            // Nothing is ever in flight, because nothing is ever dispatched.
843            // A cancel is therefore already satisfied, and a notification gets
844            // no answer.
845            ServedMethodKind::Cancel => Vec::new(),
846        }
847    }
848
849    fn remember_host(&mut self, params: &Value) {
850        let info = params.get("clientInfo");
851        self.host_name = info
852            .and_then(|info| info.get("name"))
853            .and_then(Value::as_str)
854            .filter(|name| !name.is_empty())
855            .map(str::to_owned);
856        self.host_version = info
857            .and_then(|info| info.get("version"))
858            .and_then(Value::as_str)
859            .filter(|version| !version.is_empty())
860            .map(str::to_owned);
861    }
862
863    fn initialize_result(&self) -> Value {
864        json!({
865            "protocolVersion": PROTOCOL_VERSION,
866            "agentCapabilities": {
867                "loadSession": false,
868                "promptCapabilities": {
869                    "image": false,
870                    "audio": false,
871                    "embeddedContext": false,
872                },
873            },
874            "authMethods": AUTH_METHODS,
875            "agentInfo": { "name": SERVED_AGENT_ID, "version": env!("CARGO_PKG_VERSION") },
876        })
877    }
878
879    fn mint_session(&self) -> ServedSession {
880        let decision = served_route();
881        // No clock and no random source: the id is the connection's own label
882        // and a count. Two identical runs mint identical ids, which is the same
883        // discipline the route journal is held to.
884        let session_id = format!("omega-served-{}-{}", self.connection_ref, self.sessions.len());
885        ServedSession {
886            session_id,
887            origin: self.origin(),
888            disclosure: served_disclosure(&decision),
889            decision,
890        }
891    }
892}
893
894fn result_line(id: &Value, result: Value) -> String {
895    json!({ "jsonrpc": "2.0", "id": id, "result": result }).to_string()
896}
897
898fn notification_line(method: &str, params: Value) -> String {
899    json!({ "jsonrpc": "2.0", "method": method, "params": params }).to_string()
900}
901
902fn error_line(id: &Value, code: i64, message: &str, refusal: ServedRefusal) -> String {
903    json!({
904        "jsonrpc": "2.0",
905        "id": id,
906        "error": {
907            "code": code,
908            "message": message,
909            "data": { "refusal": refusal.token() },
910        },
911    })
912    .to_string()
913}
914
915// -------------------------------------------------------------------------
916// The listener
917// -------------------------------------------------------------------------
918
919/// A bound loopback ACP listener.
920#[derive(Debug)]
921pub struct LoopbackAcpServer {
922    listener: TcpListener,
923}
924
925impl LoopbackAcpServer {
926    /// Bind the listener.
927    ///
928    /// Takes a [`LoopbackHost`], so there is no way to ask this for a routable
929    /// interface.
930    ///
931    /// # Errors
932    ///
933    /// The underlying bind error.
934    pub fn bind(host: LoopbackHost, port: u16) -> std::io::Result<Self> {
935        let listener = TcpListener::bind(SocketAddr::new(host.address(), port))?;
936        Ok(Self { listener })
937    }
938
939    /// Where it is listening.
940    ///
941    /// # Errors
942    ///
943    /// The underlying socket error.
944    pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
945        self.listener.local_addr()
946    }
947
948    /// Serve connections until the listener fails.
949    ///
950    /// One thread per attached host. There is no shared mutable state between
951    /// connections, because a served session holds nothing an operator could
952    /// change.
953    pub fn serve(self) {
954        let mut accepted = 0usize;
955        loop {
956            match self.listener.accept() {
957                Ok((stream, peer)) => {
958                    let connection_ref = format!("{accepted}");
959                    accepted += 1;
960                    log::info!("OMEGA-DELTA-0041: served ACP connection from {peer}");
961                    std::thread::spawn(move || serve_stream(stream, &connection_ref));
962                }
963                Err(error) => {
964                    log::error!("OMEGA-DELTA-0041: the served ACP listener stopped: {error:#}");
965                    return;
966                }
967            }
968        }
969    }
970}
971
972/// Serve one attached host over an already-accepted stream.
973pub fn serve_stream(stream: TcpStream, connection_ref: &str) {
974    let Ok(write_half) = stream.try_clone() else {
975        return;
976    };
977    let mut connection = ServedConnection::new(connection_ref);
978    let reader = BufReader::new(stream);
979    let mut writer = write_half;
980    for line in reader.lines() {
981        let Ok(line) = line else { return };
982        if line.trim().is_empty() {
983            continue;
984        }
985        for outgoing in connection.handle_line(&line) {
986            if writeln!(writer, "{outgoing}").is_err() || writer.flush().is_err() {
987                return;
988            }
989        }
990    }
991}
992
993/// What [`start_if_enabled`] did.
994#[derive(Debug)]
995pub enum StartOutcome {
996    /// Nothing was bound. The shipped default.
997    NotStarted(OffReason),
998    /// A listener is bound at this address and is serving on its own thread.
999    Listening(SocketAddr),
1000    /// The flag asked for a listener and one could not be opened.
1001    Failed(String),
1002}
1003
1004/// Start the server if, and only if, the flag says exactly `1`.
1005///
1006/// The only production call site is `crates/omega_effectd`, which is the
1007/// supervisor layer. GPUI never calls this and `omega_deltas` fails if it
1008/// starts.
1009pub fn start_if_enabled() -> StartOutcome {
1010    let flag = std::env::var(ENABLE_FLAG).ok();
1011    match enablement(flag.as_deref()) {
1012        Enablement::Off(reason) => StartOutcome::NotStarted(reason),
1013        Enablement::On => {
1014            let port = std::env::var(PORT_FLAG)
1015                .ok()
1016                .and_then(|port| port.parse::<u16>().ok())
1017                .unwrap_or(0);
1018            match LoopbackAcpServer::bind(LoopbackHost::DEFAULT, port) {
1019                Ok(server) => match server.local_addr() {
1020                    Ok(address) => {
1021                        log::info!(
1022                            "OMEGA-DELTA-0041: Omega Agent is served over ACP on {address} \
1023                             (loopback, unauthenticated, read-only)"
1024                        );
1025                        std::thread::spawn(move || server.serve());
1026                        StartOutcome::Listening(address)
1027                    }
1028                    Err(error) => StartOutcome::Failed(error.to_string()),
1029                },
1030                Err(error) => StartOutcome::Failed(error.to_string()),
1031            }
1032        }
1033    }
1034}
1035
1036#[cfg(test)]
1037mod tests {
1038    use super::*;
1039    use agent_client_protocol::schema::v1::{
1040        InitializeResponse, NewSessionResponse, PromptResponse, SessionNotification, StopReason,
1041    };
1042    use omega_front_door::{
1043        EXECUTOR_DISCLOSURE_FIELDS, EngineLane, ExecutorClass, FULL_AUTO_AFFORDANCES, Ingress,
1044        LaneState, PinGesture, RouteReason, SESSION_ORIGIN_FIELDS,
1045    };
1046    use std::collections::BTreeSet;
1047    use std::io::{BufRead, BufReader, Write};
1048    use std::net::TcpStream;
1049
1050    fn initialized(host: &str) -> ServedConnection {
1051        let mut connection = ServedConnection::new("test");
1052        connection.handle_line(
1053            &json!({
1054                "jsonrpc": "2.0", "id": 0, "method": "initialize",
1055                "params": {
1056                    "protocolVersion": 1,
1057                    "clientInfo": { "name": host, "version": "1.12.0" },
1058                },
1059            })
1060            .to_string(),
1061        );
1062        connection
1063    }
1064
1065    fn new_session(connection: &mut ServedConnection) -> Value {
1066        let lines = connection.handle_line(
1067            &json!({
1068                "jsonrpc": "2.0", "id": 1, "method": "session/new",
1069                "params": { "cwd": "/tmp", "mcpServers": [] },
1070            })
1071            .to_string(),
1072        );
1073        assert_eq!(lines.len(), 1);
1074        serde_json::from_str::<Value>(&lines[0]).expect("a JSON line")
1075    }
1076
1077    /// The shipped default is off, and the flag is exact.
1078    ///
1079    /// A listener that is on by default is a different product. This is the
1080    /// check that says which product this is.
1081    #[test]
1082    fn the_server_is_off_unless_the_flag_says_exactly_one() {
1083        assert_eq!(enablement(None), Enablement::Off(OffReason::FlagUnset));
1084        assert!(!enablement(None).is_on());
1085
1086        for truthy_looking in ["true", "TRUE", "yes", "on", "0", "01", " 1", "1 ", "", "11"] {
1087            assert_eq!(
1088                enablement(Some(truthy_looking)),
1089                Enablement::Off(OffReason::FlagNotExactlyOne),
1090                "{truthy_looking:?} turned the loopback ACP server on. The \
1091                 flag is an exact match on {ENABLE_VALUE:?} so that the \
1092                 default can be stated rather than guessed."
1093            );
1094        }
1095
1096        assert_eq!(enablement(Some(ENABLE_VALUE)), Enablement::On);
1097    }
1098
1099    /// A routable bind address is refused by construction.
1100    #[test]
1101    fn a_routable_bind_address_is_refused_by_construction() {
1102        for routable in ["0.0.0.0", "192.168.1.10", "10.0.0.1", "::", "8.8.8.8"] {
1103            let refused = LoopbackHost::new(routable).expect_err(
1104                "the Omega ACP server is unauthenticated and must never bind a \
1105                 routable interface",
1106            );
1107            assert_eq!(refused.token(), "not_loopback");
1108        }
1109
1110        // A name is refused rather than resolved: resolution is the step where
1111        // `localhost` can be pointed somewhere else.
1112        assert_eq!(
1113            LoopbackHost::new("localhost").expect_err("names are refused").token(),
1114            "not_an_address"
1115        );
1116
1117        assert_eq!(
1118            LoopbackHost::new("127.0.0.1").expect("loopback").address(),
1119            IpAddr::V4(Ipv4Addr::LOCALHOST)
1120        );
1121        assert_eq!(
1122            LoopbackHost::new("::1").expect("loopback").address(),
1123            IpAddr::V6(Ipv6Addr::LOCALHOST)
1124        );
1125        assert_eq!(LoopbackHost::DEFAULT.address(), IpAddr::V4(Ipv4Addr::LOCALHOST));
1126    }
1127
1128    /// Every method the surface exposes is observation, and the surface is
1129    /// exactly these four.
1130    #[test]
1131    fn the_whole_served_surface_is_observation() {
1132        let methods: Vec<&str> = SERVED_SURFACE.iter().map(|entry| entry.method).collect();
1133        assert_eq!(
1134            methods,
1135            ["initialize", "session/new", "session/prompt", "session/cancel"],
1136            "the served surface grew or lost a method. It is deny-by-default, \
1137             so every addition is a deliberate edit to this list — and an \
1138             addition that can mutate run state or grant authority is owner \
1139             gate 8 broken through a socket."
1140        );
1141        for entry in SERVED_SURFACE {
1142            assert_eq!(
1143                entry.authority,
1144                SurfaceAuthority::Observation,
1145                "{} is on an unauthenticated surface and is not observation",
1146                entry.method
1147            );
1148        }
1149    }
1150
1151    /// The served surface is inside the authority partition, not beside it.
1152    ///
1153    /// Checked against the ledgers that already exist, in both directions, so a
1154    /// new Full Auto control or a new pin gesture fails here until somebody
1155    /// says what a served host attempting it is told.
1156    #[test]
1157    fn every_authority_bearing_control_is_classified() {
1158        let classified: BTreeSet<&str> = UNEXPOSED_AUTHORITY
1159            .iter()
1160            .map(|control| control.control)
1161            .collect();
1162        let mut required: BTreeSet<&str> = FULL_AUTO_AFFORDANCES
1163            .iter()
1164            .map(|affordance| affordance.element_id)
1165            .collect();
1166        required.extend(PinGesture::all().iter().map(|gesture| gesture.token()));
1167
1168        assert!(
1169            !required.is_empty(),
1170            "the ledgers reached nothing; this partition check would be vacuous"
1171        );
1172        let unclassified: Vec<&&str> = required.difference(&classified).collect();
1173        assert!(
1174            unclassified.is_empty(),
1175            "authority-bearing controls with no entry in UNEXPOSED_AUTHORITY: \
1176             {unclassified:?}. The loopback ACP surface is an unauthenticated \
1177             model-driven surface and must sit *inside* the authority \
1178             partition. A control nobody classified is a fourth hole."
1179        );
1180        let stale: Vec<&&str> = classified.difference(&required).collect();
1181        assert!(
1182            stale.is_empty(),
1183            "UNEXPOSED_AUTHORITY names controls that no longer exist: \
1184             {stale:?}. A partition that outlives its source stops being \
1185             evidence."
1186        );
1187
1188        // Nothing that grants authority may be answerable over the socket.
1189        let served: BTreeSet<&str> = SERVED_SURFACE.iter().map(|entry| entry.method).collect();
1190        for control in UNEXPOSED_AUTHORITY {
1191            assert!(
1192                !served.contains(control.control),
1193                "{} is both classified as unexposed and served",
1194                control.control
1195            );
1196        }
1197    }
1198
1199    /// A served session is routed by the router, and never reaches an engine
1200    /// lane — whatever the engine says.
1201    ///
1202    /// This is owner gate 8 at the socket. A pin is the only door to an engine
1203    /// lane, setting one requires a `PinGesture`, and no variant of that enum
1204    /// is reachable from a socket. The consequence is checked against every
1205    /// engine state the router can be shown, so "at capacity" or "no lanes" is
1206    /// not what is doing the work.
1207    #[test]
1208    fn a_served_session_can_never_reach_an_engine_lane() {
1209        let decision = served_route();
1210        assert_eq!(decision.chosen, ExecutorClass::NativeLoop);
1211        assert_eq!(decision.reason, RouteReason::UnpinnedDefault);
1212        assert!(decision.is_coherent());
1213
1214        for engine in [
1215            EngineReadiness::Unreachable(EngineUnreachable::NotRunning),
1216            EngineReadiness::Unreachable(EngineUnreachable::Timeout),
1217            EngineReadiness::Unreachable(EngineUnreachable::ProtocolError),
1218            EngineReadiness::Answered {
1219                active_run_count: 0,
1220                active_run_limit: 8,
1221                lanes: vec![
1222                    EngineLane::new("claude-local", LaneState::Available),
1223                    EngineLane::new("codex-local", LaneState::Available),
1224                ],
1225            },
1226        ] {
1227            let inputs = RouteInputs {
1228                engine,
1229                external_acp: Some("codex-acp".to_owned()),
1230                engine_lane: Some("codex-local".to_owned()),
1231                ..served_inputs()
1232            };
1233            let decided = route(&inputs);
1234            assert_eq!(
1235                decided.chosen,
1236                ExecutorClass::NativeLoop,
1237                "a served session reached an executor other than the native \
1238                 loop. Nothing over the socket may take a pin, and a pin is \
1239                 the only door to Full Auto authority."
1240            );
1241            assert!(decided.lane_ref.is_none());
1242        }
1243    }
1244
1245    /// The disclosure crosses the wire as a record with exactly the fields the
1246    /// record has, and the rendered line is derived beside it.
1247    #[test]
1248    fn the_served_disclosure_is_a_record_not_a_label() {
1249        let mut connection = initialized("Zed");
1250        let response = new_session(&mut connection);
1251        let meta = &response["result"]["_meta"][SERVED_SESSION_META_KEY];
1252
1253        let disclosure = meta["disclosure"]
1254            .as_object()
1255            .expect("the disclosure travels as an object");
1256        let mut keys: Vec<&str> = disclosure.keys().map(String::as_str).collect();
1257        keys.sort_unstable();
1258        let mut expected: Vec<&str> = EXECUTOR_DISCLOSURE_FIELDS.to_vec();
1259        expected.sort_unstable();
1260        assert_eq!(
1261            keys, expected,
1262            "the disclosure on the wire must carry exactly the record's \
1263             fields. A rendered line under any name — label, line, text, \
1264             summary — breaks the condition the owner admitted the identity \
1265             decision on, and an exact key set is what catches the names \
1266             nobody anticipated."
1267        );
1268
1269        let origin = meta["origin"].as_object().expect("the origin travels as an object");
1270        let mut origin_keys: Vec<&str> = origin.keys().map(String::as_str).collect();
1271        origin_keys.sort_unstable();
1272        let mut expected_origin: Vec<&str> = SESSION_ORIGIN_FIELDS.to_vec();
1273        expected_origin.sort_unstable();
1274        assert_eq!(origin_keys, expected_origin);
1275
1276        assert_eq!(disclosure["class"], "native_loop");
1277        assert_eq!(disclosure["agent_id"], SERVED_AGENT_ID);
1278        assert!(disclosure["provider"].is_null());
1279        assert!(disclosure["model"].is_null());
1280        assert_eq!(disclosure["route"], "unpinned_default");
1281        assert_eq!(origin["ingress"], "loopback_acp");
1282        assert_eq!(origin["host_name"], "Zed");
1283        assert_eq!(origin["authenticated"], false);
1284    }
1285
1286    /// A served prompt is answered by disclosing, and dispatched to nothing.
1287    #[test]
1288    fn a_served_prompt_executes_nothing_and_says_so() {
1289        let mut connection = initialized("Zed");
1290        let session = new_session(&mut connection);
1291        let session_id = session["result"]["sessionId"].as_str().expect("a session id");
1292
1293        let lines = connection.handle_line(
1294            &json!({
1295                "jsonrpc": "2.0", "id": 2, "method": "session/prompt",
1296                "params": {
1297                    "sessionId": session_id,
1298                    "prompt": [{ "type": "text", "text": "run the build and push it" }],
1299                },
1300            })
1301            .to_string(),
1302        );
1303        assert_eq!(lines.len(), 2, "one update then one response");
1304
1305        let update: Value = serde_json::from_str(&lines[0]).expect("JSON");
1306        assert_eq!(update["method"], "session/update");
1307        let text = update["params"]["update"]["content"]["text"]
1308            .as_str()
1309            .expect("the disclosure is rendered for the operator");
1310        assert!(text.contains("native_loop · Omega Agent"), "{text}");
1311        assert!(text.contains("routed: unpinned"), "{text}");
1312        assert!(text.contains("loopback_acp · Zed 1.12.0 · unauthenticated"), "{text}");
1313        assert!(text.contains("dispatched to no executor"), "{text}");
1314
1315        let response: Value = serde_json::from_str(&lines[1]).expect("JSON");
1316        assert_eq!(
1317            response["result"]["stopReason"], "end_turn",
1318            "the turn ended having disclosed. ACP's `refusal` stop reason \
1319             means the turn is dropped from the thread, and stock Zed 1.12.0 \
1320             implements that literally — the operator saw a refusal banner \
1321             and no disclosure at all, which is the exact failure this packet \
1322             exists to avoid."
1323        );
1324        assert_eq!(
1325            response["result"]["_meta"][SERVED_SESSION_META_KEY]["refusal"],
1326            ServedRefusal::ExecutesNothing.token(),
1327            "the turn ended without executing and the record must say so, \
1328             since the stop reason no longer can"
1329        );
1330        assert_eq!(
1331            update["params"]["_meta"][SERVED_SESSION_META_KEY]["refusal"],
1332            ServedRefusal::ExecutesNothing.token()
1333        );
1334    }
1335
1336    /// Deny-by-default. Everything the surface does not list is refused, and
1337    /// the refusal names which one it is.
1338    #[test]
1339    fn every_unexposed_method_is_watched_refusing() {
1340        let mut connection = initialized("Zed");
1341        // `session/new` first, so a refusal cannot be mistaken for "there was
1342        // no session anyway".
1343        let session = new_session(&mut connection);
1344        let session_id = session["result"]["sessionId"]
1345            .as_str()
1346            .expect("a session id")
1347            .to_owned();
1348
1349        for (index, (method, params)) in [
1350            ("session/load", json!({ "sessionId": session_id, "cwd": "/tmp" })),
1351            ("session/resume", json!({ "sessionId": session_id })),
1352            ("session/delete", json!({ "sessionId": session_id })),
1353            ("session/set_mode", json!({ "sessionId": session_id, "modeId": "yolo" })),
1354            (
1355                "session/set_config_option",
1356                json!({ "sessionId": session_id, "configId": "a", "value": true }),
1357            ),
1358            ("authenticate", json!({ "methodId": "anything" })),
1359            ("logout", json!({})),
1360            ("fs/write_text_file", json!({ "path": "/tmp/x", "content": "x" })),
1361            ("terminal/create", json!({ "command": "sh" })),
1362            ("full_auto/start", json!({ "objective": "ship it" })),
1363            ("omega/pin_executor", json!({ "class": "engine_lane" })),
1364        ]
1365        .into_iter()
1366        .enumerate()
1367        {
1368            let lines = connection.handle_line(
1369                &json!({ "jsonrpc": "2.0", "id": 100 + index, "method": method, "params": params })
1370                    .to_string(),
1371            );
1372            assert_eq!(lines.len(), 1, "{method} answered something other than one refusal");
1373            let answer: Value = serde_json::from_str(&lines[0]).expect("JSON");
1374            assert!(
1375                answer.get("result").is_none(),
1376                "{method} was answered with a result: {answer}"
1377            );
1378            assert_eq!(answer["error"]["code"], METHOD_NOT_FOUND, "{method}");
1379            assert_eq!(
1380                answer["error"]["data"]["refusal"],
1381                ServedRefusal::NotExposed.token(),
1382                "{method} was refused without naming which refusal it was"
1383            );
1384        }
1385
1386        // Nothing the attached host attempted minted a session, changed one, or
1387        // left any trace at all.
1388        assert_eq!(connection.sessions().len(), 1);
1389        assert_eq!(connection.sessions()[0].decision.chosen, ExecutorClass::NativeLoop);
1390    }
1391
1392    /// Every refusal has a distinct token and a message that says what did not
1393    /// happen.
1394    #[test]
1395    fn every_refusal_states_what_did_not_happen() {
1396        let tokens: BTreeSet<&str> = ServedRefusal::all()
1397            .iter()
1398            .map(|refusal| refusal.token())
1399            .collect();
1400        assert_eq!(tokens.len(), ServedRefusal::all().len());
1401        for refusal in ServedRefusal::all() {
1402            assert!(
1403                refusal.message().len() > 60,
1404                "{} is stated in a phrase, which is not stated",
1405                refusal.token()
1406            );
1407        }
1408    }
1409
1410    /// The surface offers no credential to present.
1411    #[test]
1412    fn the_served_surface_is_unauthenticated_and_says_so() {
1413        let mut connection = ServedConnection::new("test");
1414        let lines = connection.handle_line(
1415            &json!({
1416                "jsonrpc": "2.0", "id": 0, "method": "initialize",
1417                "params": { "protocolVersion": 1 },
1418            })
1419            .to_string(),
1420        );
1421        let answer: Value = serde_json::from_str(&lines[0]).expect("JSON");
1422        assert_eq!(answer["result"]["authMethods"], json!([]));
1423        assert_eq!(answer["result"]["agentCapabilities"]["loadSession"], false);
1424        assert!(AUTH_METHODS.is_empty());
1425
1426        // A host that said nothing about itself is disclosed as having said
1427        // nothing, not as anonymous-but-fine.
1428        let session = new_session(&mut connection);
1429        let origin = &session["result"]["_meta"][SERVED_SESSION_META_KEY]["origin"];
1430        assert!(origin["host_name"].is_null());
1431        assert_eq!(origin["authenticated"], false);
1432    }
1433
1434    /// Every message this surface writes reads back as the pinned ACP schema
1435    /// type it claims to be.
1436    ///
1437    /// The conformance link. The wire is built as JSON so the *exact* key set
1438    /// can be asserted; this is what stops that freedom from drifting away from
1439    /// the protocol Zed actually speaks.
1440    #[test]
1441    fn every_answer_deserialises_as_its_pinned_schema_type() {
1442        let mut connection = initialized("Zed");
1443
1444        let initialize = connection.handle_line(
1445            &json!({ "jsonrpc": "2.0", "id": 0, "method": "initialize", "params": {} }).to_string(),
1446        );
1447        let initialize: Value = serde_json::from_str(&initialize[0]).expect("JSON");
1448        let parsed: InitializeResponse =
1449            serde_json::from_value(initialize["result"].clone()).expect("an InitializeResponse");
1450        assert!(parsed.auth_methods.is_empty());
1451
1452        let session = new_session(&mut connection);
1453        let parsed: NewSessionResponse =
1454            serde_json::from_value(session["result"].clone()).expect("a NewSessionResponse");
1455        let session_id = parsed.session_id.0.to_string();
1456
1457        let lines = connection.handle_line(
1458            &json!({
1459                "jsonrpc": "2.0", "id": 2, "method": "session/prompt",
1460                "params": { "sessionId": session_id, "prompt": [{ "type": "text", "text": "hi" }] },
1461            })
1462            .to_string(),
1463        );
1464        let update: Value = serde_json::from_str(&lines[0]).expect("JSON");
1465        let _: SessionNotification =
1466            serde_json::from_value(update["params"].clone()).expect("a SessionNotification");
1467        let response: Value = serde_json::from_str(&lines[1]).expect("JSON");
1468        let parsed: PromptResponse =
1469            serde_json::from_value(response["result"].clone()).expect("a PromptResponse");
1470        assert_eq!(parsed.stop_reason, StopReason::EndTurn);
1471    }
1472
1473    /// The whole thing over a real loopback socket, end to end.
1474    #[test]
1475    fn a_real_client_attaches_over_loopback_and_is_disclosed_to() {
1476        let server = LoopbackAcpServer::bind(LoopbackHost::DEFAULT, 0).expect("binds loopback");
1477        let address = server.local_addr().expect("an address");
1478        assert!(address.ip().is_loopback());
1479        std::thread::spawn(move || server.serve());
1480
1481        let stream = TcpStream::connect(address).expect("connects");
1482        let mut writer = stream.try_clone().expect("a write half");
1483        let mut reader = BufReader::new(stream);
1484        let send = |message: Value, writer: &mut TcpStream| {
1485            writeln!(writer, "{message}").expect("writes");
1486            writer.flush().expect("flushes");
1487        };
1488        let read = |reader: &mut BufReader<TcpStream>| {
1489            let mut line = String::new();
1490            reader.read_line(&mut line).expect("reads");
1491            serde_json::from_str::<Value>(&line).expect("JSON")
1492        };
1493
1494        send(
1495            json!({
1496                "jsonrpc": "2.0", "id": 0, "method": "initialize",
1497                "params": {
1498                    "protocolVersion": 1,
1499                    "clientInfo": { "name": "conformance-client", "version": "0.1.0" },
1500                },
1501            }),
1502            &mut writer,
1503        );
1504        assert_eq!(read(&mut reader)["result"]["authMethods"], json!([]));
1505
1506        send(
1507            json!({ "jsonrpc": "2.0", "id": 1, "method": "session/new", "params": { "cwd": "/tmp" } }),
1508            &mut writer,
1509        );
1510        let session = read(&mut reader);
1511        let session_id = session["result"]["sessionId"].as_str().expect("id").to_owned();
1512        assert_eq!(
1513            session["result"]["_meta"][SERVED_SESSION_META_KEY]["origin"]["ingress"],
1514            "loopback_acp"
1515        );
1516
1517        send(
1518            json!({
1519                "jsonrpc": "2.0", "id": 2, "method": "session/prompt",
1520                "params": { "sessionId": session_id, "prompt": [{ "type": "text", "text": "hi" }] },
1521            }),
1522            &mut writer,
1523        );
1524        let update = read(&mut reader);
1525        assert!(
1526            update["params"]["update"]["content"]["text"]
1527                .as_str()
1528                .expect("text")
1529                .contains("native_loop · Omega Agent")
1530        );
1531        let answered = read(&mut reader);
1532        assert_eq!(answered["result"]["stopReason"], "end_turn");
1533        assert_eq!(
1534            answered["result"]["_meta"][SERVED_SESSION_META_KEY]["refusal"],
1535            ServedRefusal::ExecutesNothing.token()
1536        );
1537
1538        // And the mutation attempt, over the same real socket.
1539        send(
1540            json!({ "jsonrpc": "2.0", "id": 3, "method": "full_auto/start", "params": {} }),
1541            &mut writer,
1542        );
1543        assert_eq!(
1544            read(&mut reader)["error"]["data"]["refusal"],
1545            ServedRefusal::NotExposed.token()
1546        );
1547    }
1548
1549    /// An **upstream** ACP client, over a real loopback socket, sees the
1550    /// disclosure.
1551    ///
1552    /// The client here is `agent_client_protocol`'s own `Client` role — the
1553    /// pinned third-party SDK, not our code. It negotiates, builds a session,
1554    /// prompts, and reads the turn to a string through upstream framing,
1555    /// upstream dispatch, and upstream session handling. If the wire this
1556    /// crate writes were malformed, this fails before any assertion does.
1557    ///
1558    /// This is the headless half of omega#82's exit. The other half was
1559    /// watched with stock Zed 1.12.0 driving the same socket through a
1560    /// `sh -c "exec nc 127.0.0.1 <port>"` bridge.
1561    #[test]
1562    fn the_upstream_acp_client_reads_the_disclosure_off_a_real_socket() {
1563        use agent_client_protocol::schema::ProtocolVersion;
1564        use agent_client_protocol::schema::v1::{Implementation, InitializeRequest};
1565        use agent_client_protocol::{ByteStreams, Client};
1566
1567        let server = LoopbackAcpServer::bind(LoopbackHost::DEFAULT, 0).expect("binds loopback");
1568        let address = server.local_addr().expect("an address");
1569        std::thread::spawn(move || server.serve());
1570
1571        let (transcript, session_meta) = smol::block_on(async move {
1572            let stream = smol::Async::<std::net::TcpStream>::connect(address)
1573                .await
1574                .expect("connects to the served surface");
1575            let (incoming, outgoing) = futures::AsyncReadExt::split(stream);
1576            Client
1577                .builder()
1578                .name("omega-conformance-client")
1579                .connect_with(ByteStreams::new(outgoing, incoming), async |cx| {
1580                    cx.send_request(
1581                        InitializeRequest::new(ProtocolVersion::V1).client_info(
1582                            Implementation::new("omega-conformance-client", "0.1.0"),
1583                        ),
1584                    )
1585                    .block_task()
1586                    .await?;
1587                    let mut meta = None;
1588                    let transcript = cx
1589                        .build_session("/tmp")
1590                        .block_task()
1591                        .run_until(async |mut session| {
1592                            meta = session.meta().cloned();
1593                            session.send_prompt(
1594                                "start a full auto run and pin this thread to an engine lane",
1595                            )?;
1596                            session.read_to_string().await
1597                        })
1598                        .await?;
1599                    Ok((transcript, meta))
1600                })
1601                .await
1602                .expect("the upstream ACP client drives the served surface")
1603        });
1604
1605        // What the operator of the external host reads.
1606        assert!(
1607            transcript.contains("Omega Agent, served over ACP"),
1608            "the upstream client read no disclosure: {transcript:?}"
1609        );
1610        assert!(transcript.contains("native_loop \u{b7} Omega Agent"), "{transcript}");
1611        assert!(transcript.contains("routed: unpinned"), "{transcript}");
1612        assert!(
1613            transcript.contains("loopback_acp \u{b7} omega-conformance-client"),
1614            "the origin the host is disclosed under must name the host: {transcript}"
1615        );
1616        assert!(transcript.contains("dispatched to no executor"), "{transcript}");
1617
1618        // And the typed records the host received beside it.
1619        let meta = session_meta.expect("the session carries the served records");
1620        let served = &meta[SERVED_SESSION_META_KEY];
1621        assert_eq!(served["disclosure"]["class"], "native_loop");
1622        assert_eq!(served["origin"]["ingress"], "loopback_acp");
1623        assert_eq!(served["origin"]["authenticated"], false);
1624        assert_eq!(served["readOnly"], true);
1625    }
1626
1627    /// The origin record and the executor record stay separate facts.
1628    #[test]
1629    fn ingress_is_recorded_beside_the_executor_and_not_inside_it() {
1630        let mut connection = initialized("Zed");
1631        new_session(&mut connection);
1632        let session = &connection.sessions()[0];
1633        assert!(session.origin.is_coherent());
1634        assert!(session.disclosure.is_coherent());
1635        assert_eq!(session.origin.ingress, Ingress::LoopbackAcp);
1636        assert_eq!(
1637            session.disclosure.class,
1638            ExecutorClass::NativeLoop,
1639            "the executor class answers who ran the work, and serving over ACP \
1640             changes only who asked"
1641        );
1642        assert_ne!(session.disclosure.class, ExecutorClass::ExternalAcp);
1643    }
1644}
1645