tenant.openagents/omega

1//! Omega Rust supervisor for packaged `@openagentsinc/omega-effectd`.
2//!
3//! Authority: OpenAgentsInc/omega#21 (`OMEGA-FA-02`).
4//! Durable run truth stays in omega-effectd on disk. GPUI is not run authority.
5
6mod issue31_nostr;
7mod issue31_provider_handoff;
8mod nostr_websocket_relay;
9mod openagents_binding;
10mod openagents_session;
11mod protocol;
12mod sarah_conversation;
13mod supervisor;
14
15use std::{rc::Rc, sync::Arc};
16
17use anyhow::{Result, anyhow};
18use gpui::{App, Global};
19use smol::lock::Mutex as AsyncMutex;
20
21pub use openagents_binding::{
22    BINDING_RECORD_SCHEMA, BindingEvent, BindingProjection, BindingState,
23    OPENAGENTS_BINDING_CREDENTIAL_KEY, OPENAGENTS_OMEGA_CLIENT_ID, OWNER_SCOPE_REFUSED_MESSAGE,
24    OpenAgentsBinding, apply_binding_transition, binding_record_path, default_binding_data_root,
25    init_openagents_binding, openagents_binding, try_openagents_binding,
26};
27
28pub use issue31_nostr::*;
29pub use issue31_provider_handoff::*;
30pub use nostr_websocket_relay::WebSocketRelayAdapter;
31pub use openagents_session::{
32    OpenAgentsSession, OpenAgentsSessionPhase, VerifiedOpenAgentsSession, init_openagents_session,
33    openagents_session,
34};
35
36pub use protocol::{
37    HealthResult, HostMethod, HostRequestFrame, HostResponseError, HostResponseErrorCode,
38    HostResponseFrame, InitializeResult, PROTOCOL_SCHEMA, PROTOCOL_VERSION, ProtocolError,
39    ProtocolErrorCode, RunSnapshot, SERVICE_VERSION,
40};
41pub use sarah_conversation::{
42    BootstrapResult, ConversationIdentity, GapState, InterruptTurnResult, MockRelayAdapter,
43    RelayTransport, RoomSnapshotResult, RoomStateEvent, SARAH_EVENT_ROOM_EVENT,
44    SARAH_EVENT_ROOM_STATE, SARAH_FRAMED_METHODS, SARAH_METHOD_BOOTSTRAP,
45    SARAH_METHOD_DEVICE_GRANTS, SARAH_METHOD_INTERRUPT_TURN, SARAH_METHOD_READMIT_DEVICE,
46    SARAH_METHOD_RENEW_DEVICE_GRANT, SARAH_METHOD_REVOKE_DEVICE_GRANT,
47    SARAH_METHOD_ROOM_SNAPSHOT, SARAH_METHOD_SEND_MESSAGE,
48    SARAH_METHOD_SESSION_STATUS, Issue31HostProjectionDocuments, Issue31HostProjectionRequest,
49    Issue31HostProjectionSource, Issue31ProviderRosterSource,
50    SarahConversationClient, SarahConversationConfig,
51    SarahConversationError, SendMessageResult, SessionStatusResult, SigningIdentity,
52    asserts_no_khala_sync_client,
53};
54pub use supervisor::{
55    AttentionDecision, MAX_FRAME_BYTES, OmegaEffectdCommand, OmegaEffectdHostFuture,
56    OmegaEffectdHostHandler, OmegaEffectdSupervisor, OmegaEffectdSupervisorOptions,
57    SupervisorError, default_options, fixture_command, resolve_effectd_command,
58};
59
60pub type SharedOmegaEffectdSupervisor = Rc<AsyncMutex<OmegaEffectdSupervisor>>;
61
62enum OmegaEffectdRuntime {
63    Available(SharedOmegaEffectdSupervisor),
64    Unavailable(Arc<str>),
65}
66
67impl Global for OmegaEffectdRuntime {}
68
69pub fn init(cx: &mut App) {
70    init_with_host_handler(None, cx);
71}
72
73pub fn init_with_host_handler(handler: Option<OmegaEffectdHostHandler>, cx: &mut App) {
74    // OMEGA-SW-01: binding is independent of effectd packaging availability.
75    init_openagents_binding(cx);
76
77    if cx.has_global::<OmegaEffectdRuntime>() {
78        return;
79    }
80
81    start_served_acp_surface();
82
83    let runtime = match std::env::current_exe()
84        .map_err(anyhow::Error::from)
85        .and_then(|executable| {
86            resolve_effectd_command(
87                std::env::var_os("OPENAGENTS_OMEGA_EFFECTD_BIN").as_deref(),
88                &executable,
89            )
90        }) {
91        Ok(command) => {
92            let mut supervisor = OmegaEffectdSupervisor::new(default_options(
93                paths::data_dir().join("openagents"),
94                command,
95            ));
96            if let Some(handler) = handler {
97                supervisor.set_host_handler(handler);
98            }
99            OmegaEffectdRuntime::Available(Rc::new(AsyncMutex::new(supervisor)))
100        }
101        Err(error) => OmegaEffectdRuntime::Unavailable(error.to_string().into()),
102    };
103    cx.set_global(runtime);
104}
105
106/// `OMEGA-DELTA-0041`, omega#82. Serve Omega Agent over ACP, if the flag says
107/// so.
108///
109/// The supervisor layer owns this, not GPUI. The socket itself lives in
110/// `crates/omega_acp_server`, which depends on no part of GPUI, and this is its
111/// only production call site — `crates/omega_deltas` fails if a second one
112/// appears in a UI crate. Omega's own windows never open a listener.
113///
114/// Off unless `OMEGA_ACP_SERVER` is exactly `1`, so the shipped default binds
115/// nothing at all and this is a no-op in every normal launch.
116fn start_served_acp_surface() {
117    match omega_acp_server::start_if_enabled() {
118        omega_acp_server::StartOutcome::NotStarted(reason) => {
119            log::debug!(
120                "OMEGA-DELTA-0041: Omega Agent is not served over ACP ({})",
121                reason.token()
122            );
123        }
124        omega_acp_server::StartOutcome::Listening(address) => {
125            log::info!(
126                "OMEGA-DELTA-0041: Omega Agent is served over ACP on {address} \
127                 (loopback, unauthenticated, read-only)"
128            );
129        }
130        omega_acp_server::StartOutcome::Failed(error) => {
131            log::error!(
132                "OMEGA-DELTA-0041: the served ACP surface was asked for and \
133                 could not listen: {error}"
134            );
135        }
136    }
137}
138
139pub fn shared_supervisor(cx: &App) -> Result<SharedOmegaEffectdSupervisor> {
140    match cx.try_global::<OmegaEffectdRuntime>() {
141        Some(OmegaEffectdRuntime::Available(supervisor)) => Ok(supervisor.clone()),
142        Some(OmegaEffectdRuntime::Unavailable(message)) => Err(anyhow!(message.to_string())),
143        None => Err(anyhow!("omega-effectd runtime was not initialized")),
144    }
145}
146
147#[cfg(test)]
148mod tests {
149    use std::ffi::OsStr;
150    use std::path::PathBuf;
151    use std::rc::Rc;
152    use std::sync::{Arc, Mutex};
153    use std::time::Duration;
154
155    use serde_json::json;
156    use tempfile::tempdir;
157
158    use super::*;
159
160    fn fixture_path() -> PathBuf {
161        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("fixtures/fake_effectd.mjs")
162    }
163
164    #[test]
165    fn missing_packaged_component_fails_closed_without_fixture_fallback() {
166        let root = tempdir().expect("tempdir");
167        let executable = root.path().join("Omega.app/Contents/MacOS/omega");
168        std::fs::create_dir_all(executable.parent().expect("executable parent"))
169            .expect("create app executable directory");
170
171        let error = resolve_effectd_command(None, &executable).expect_err("component is absent");
172        let message = error.to_string();
173        assert!(message.contains("packaged omega-effectd component is unavailable"));
174        assert!(!message.contains("fake_effectd"));
175        assert!(!message.contains("openagents/packages"));
176    }
177
178    #[test]
179    fn resolver_accepts_only_explicit_or_packaged_component_paths() {
180        let root = tempdir().expect("tempdir");
181        let executable = root.path().join("Omega.app/Contents/MacOS/omega");
182        let component = root
183            .path()
184            .join("Omega.app/Contents/Resources/omega-effectd/bin/omega-effectd");
185        std::fs::create_dir_all(component.parent().expect("component parent"))
186            .expect("create component directory");
187        std::fs::write(&component, "#!/bin/sh\n").expect("write component");
188
189        let packaged = resolve_effectd_command(None, &executable).expect("packaged component");
190        assert_eq!(packaged.program, component);
191        assert!(packaged.args.is_empty());
192
193        let explicit = root.path().join("explicit-effectd");
194        std::fs::write(&explicit, "#!/bin/sh\n").expect("write explicit component");
195        let overridden = resolve_effectd_command(Some(explicit.as_os_str()), &executable)
196            .expect("explicit component");
197        assert_eq!(overridden.program, explicit);
198
199        let missing = root.path().join("missing-effectd");
200        assert!(
201            resolve_effectd_command(Some(OsStr::new(&missing)), &executable).is_err(),
202            "an explicit missing component must not fall back"
203        );
204    }
205
206    #[test]
207    fn oversized_response_frame_fails_closed_and_stops_child() {
208        smol::block_on(async {
209            let root = tempdir().expect("tempdir");
210            let mut command = fixture_command(&fixture_path());
211            command.args.push("--oversized-health-response".into());
212            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
213                data_root: root.path().to_path_buf(),
214                command,
215                initial_generation: 1,
216                request_timeout: Duration::from_secs(5),
217            });
218
219            supervisor.start().await.expect("start");
220            let error = supervisor.health().await.expect_err("oversized frame");
221            assert!(error.to_string().contains("response frame exceeds"));
222            let stopped = supervisor.health().await.expect_err("child was torn down");
223            assert!(stopped.to_string().contains("not started"));
224        });
225    }
226
227    #[test]
228    fn host_requests_are_multiplexed_with_matching_generation() {
229        smol::block_on(async {
230            let root = tempdir().expect("tempdir");
231            let mut command = fixture_command(&fixture_path());
232            command.args.push("--host-request-health".into());
233            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
234                data_root: root.path().to_path_buf(),
235                command,
236                initial_generation: 7,
237                request_timeout: Duration::from_secs(5),
238            });
239            let observed = Arc::new(Mutex::new(None));
240            supervisor.set_host_handler(Rc::new({
241                let observed = observed.clone();
242                move |request| {
243                    let observed = observed.clone();
244                    Box::pin(async move {
245                        *observed.lock().expect("observed request lock") = Some(request.clone());
246                        Ok(json!({ "workspaceRef": "workspace.omega.supervised" }))
247                    })
248                }
249            }));
250
251            supervisor.start().await.expect("start");
252            supervisor.health().await.expect("health");
253            let request = observed
254                .lock()
255                .expect("observed request lock")
256                .clone()
257                .expect("host request");
258            assert_eq!(request.generation, 7);
259            assert_eq!(request.method, HostMethod::ResolveWorkspace);
260        });
261    }
262
263    #[test]
264    fn missing_host_authority_returns_typed_unavailable_response() {
265        smol::block_on(async {
266            let root = tempdir().expect("tempdir");
267            let mut command = fixture_command(&fixture_path());
268            command
269                .args
270                .push("--unavailable-host-request-health".into());
271            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
272                data_root: root.path().to_path_buf(),
273                command,
274                initial_generation: 1,
275                request_timeout: Duration::from_secs(5),
276            });
277
278            supervisor.start().await.expect("start");
279            supervisor
280                .health()
281                .await
282                .expect("typed unavailable response");
283        });
284    }
285
286    #[test]
287    fn host_authority_timeout_returns_unavailable_without_parking_request() {
288        smol::block_on(async {
289            let root = tempdir().expect("tempdir");
290            let mut command = fixture_command(&fixture_path());
291            command
292                .args
293                .push("--unavailable-host-request-health".into());
294            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
295                data_root: root.path().to_path_buf(),
296                command,
297                initial_generation: 1,
298                request_timeout: Duration::from_secs(5),
299            });
300            supervisor.set_host_request_timeout(Duration::from_millis(10));
301            supervisor.set_host_handler(Rc::new(|_| {
302                Box::pin(futures::future::pending::<
303                    std::result::Result<serde_json::Value, HostResponseError>,
304                >())
305            }));
306
307            supervisor.start().await.expect("start");
308            supervisor.health().await.expect("host timeout response");
309        });
310    }
311
312    #[test]
313    fn stale_host_request_gets_generation_matched_rejection() {
314        smol::block_on(async {
315            let root = tempdir().expect("tempdir");
316            let mut command = fixture_command(&fixture_path());
317            command.args.push("--stale-host-request-health".into());
318            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
319                data_root: root.path().to_path_buf(),
320                command,
321                initial_generation: 2,
322                request_timeout: Duration::from_secs(5),
323            });
324
325            supervisor.start().await.expect("start");
326            supervisor.health().await.expect("stale host rejection");
327        });
328    }
329
330    #[test]
331    fn stale_service_response_fails_closed_and_stops_child() {
332        smol::block_on(async {
333            let root = tempdir().expect("tempdir");
334            let mut command = fixture_command(&fixture_path());
335            command.args.push("--stale-health-response".into());
336            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
337                data_root: root.path().to_path_buf(),
338                command,
339                initial_generation: 2,
340                request_timeout: Duration::from_secs(5),
341            });
342
343            supervisor.start().await.expect("start");
344            let error = supervisor
345                .health()
346                .await
347                .expect_err("stale service response");
348            assert!(error.to_string().contains("stale generation"));
349            assert!(supervisor.health().await.is_err(), "child must be stopped");
350        });
351    }
352
353    #[test]
354    fn oversized_host_response_fails_closed_and_stops_child() {
355        smol::block_on(async {
356            let root = tempdir().expect("tempdir");
357            let mut command = fixture_command(&fixture_path());
358            command.args.push("--host-request-health".into());
359            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
360                data_root: root.path().to_path_buf(),
361                command,
362                initial_generation: 1,
363                request_timeout: Duration::from_secs(5),
364            });
365            supervisor.set_host_handler(Rc::new(|_| {
366                Box::pin(async {
367                    Ok(json!({
368                        "workspaceRef": "x".repeat(MAX_FRAME_BYTES),
369                    }))
370                })
371            }));
372
373            supervisor.start().await.expect("start");
374            let error = supervisor
375                .health()
376                .await
377                .expect_err("oversized host response");
378            assert!(error.to_string().contains("host response frame exceeds"));
379            assert!(supervisor.health().await.is_err(), "child must be stopped");
380        });
381    }
382
383    #[test]
384    fn start_health_restart_stop_and_generation_fence() {
385        smol::block_on(async {
386            let root = tempdir().unwrap();
387            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
388                data_root: root.path().to_path_buf(),
389                command: fixture_command(&fixture_path()),
390                initial_generation: 1,
391                request_timeout: Duration::from_secs(5),
392            });
393
394            let init = supervisor.start().await.expect("start");
395            assert_eq!(init.generation, 1);
396            assert_eq!(init.schema, PROTOCOL_SCHEMA);
397
398            let health = supervisor.health().await.expect("health");
399            assert_eq!(health.status, "running");
400            assert_eq!(health.generation, 1);
401
402            // Persist a run through the fixture file API, then restart.
403            let runs_path = root.path().join("full-auto").join("runs.json");
404            std::fs::create_dir_all(runs_path.parent().unwrap()).unwrap();
405            std::fs::write(
406                &runs_path,
407                serde_json::to_string_pretty(&json!({
408                    "schema": "openagents.desktop.full_auto_run_registry.v1",
409                    "runs": [{
410                        "runRef": "run.full-auto.recovery",
411                        "threadRef": null,
412                        "title": "Recovery proof",
413                        "state": "paused",
414                        "updatedAt": "2026-07-24T00:00:00.000Z"
415                    }]
416                }))
417                .unwrap(),
418            )
419            .unwrap();
420
421            let restarted = supervisor.restart().await.expect("restart");
422            assert_eq!(restarted.generation, 2);
423
424            let runs = supervisor.list_runs().await.expect("list after restart");
425            assert!(
426                runs.iter().any(|run| {
427                    run.run_ref == "run.full-auto.recovery" && run.title == "Recovery proof"
428                }),
429                "durable disk truth must survive restart: {runs:?}"
430            );
431
432            // Stale generation must be refused by the child; supervisor tracks current gen.
433            assert_eq!(supervisor.generation(), 2);
434            supervisor.stop().await.expect("stop");
435        });
436    }
437
438    #[test]
439    fn fa07_control_matrix_and_native_join_survive_restart() {
440        smol::block_on(async {
441            let root = tempdir().unwrap();
442            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
443                data_root: root.path().to_path_buf(),
444                command: fixture_command(&fixture_path()),
445                initial_generation: 1,
446                request_timeout: Duration::from_secs(5),
447            });
448
449            supervisor.start().await.expect("start");
450            let started = supervisor
451                .start_run(json!({
452                    "workspaceRef": "workspace.omega.supervised",
453                    "title": "FA-07 control matrix",
454                    "objective": "Prove pause resume stop and native join.",
455                    "doneCondition": "Controls complete.",
456                    "turnCap": 8,
457                    "projectRef": "project.fa07",
458                    "worktreeRef": "worktree.fa07",
459                    "gitHead": "deadbeef"
460                }))
461                .await
462                .expect("start_run");
463            let run_ref = started
464                .get("runRef")
465                .and_then(|v| v.as_str())
466                .expect("runRef")
467                .to_string();
468            assert_eq!(
469                started
470                    .pointer("/nativeEvidence/projectRef")
471                    .and_then(|v| v.as_str()),
472                Some("project.fa07")
473            );
474
475            let paused = supervisor.pause_run(&run_ref).await.expect("pause");
476            assert_eq!(paused.get("state").and_then(|v| v.as_str()), Some("paused"));
477            let handed_off = supervisor
478                .handoff_run(&run_ref, "claude-local")
479                .await
480                .expect("handoff");
481            assert_eq!(
482                handed_off.get("lane").and_then(|v| v.as_str()),
483                Some("claude-local")
484            );
485            let resumed = supervisor.resume_run(&run_ref).await.expect("resume");
486            assert_eq!(
487                resumed.get("state").and_then(|v| v.as_str()),
488                Some("running")
489            );
490
491            let binding = supervisor
492                .get_native_binding(&run_ref)
493                .await
494                .expect("binding");
495            assert_eq!(
496                binding.get("projectRef").and_then(|v| v.as_str()),
497                Some("project.fa07")
498            );
499            let assessment = supervisor
500                .assess_native_boundary(&run_ref)
501                .await
502                .expect("assessment");
503            assert_eq!(assessment.get("ok").and_then(|v| v.as_bool()), Some(true));
504
505            let sync = supervisor.get_sync_status().await.expect("sync");
506            assert_eq!(
507                sync.get("publishBlocksDispatch").and_then(|v| v.as_bool()),
508                Some(false)
509            );
510
511            supervisor.restart().await.expect("restart");
512            let after = supervisor
513                .get_run(&run_ref)
514                .await
515                .expect("get after restart");
516            assert_eq!(
517                after.get("runRef").and_then(|v| v.as_str()),
518                Some(run_ref.as_str())
519            );
520            assert_eq!(
521                after
522                    .pointer("/nativeEvidence/worktreeRef")
523                    .and_then(|v| v.as_str()),
524                Some("worktree.fa07")
525            );
526
527            let stopped = supervisor.stop_run(&run_ref).await.expect("stop");
528            assert_eq!(
529                stopped.get("state").and_then(|v| v.as_str()),
530                Some("stopped")
531            );
532            supervisor.stop().await.expect("supervisor stop");
533        });
534    }
535
536    #[test]
537    fn ac01_agent_computer_session_survives_restart_without_bearer() {
538        smol::block_on(async {
539            let root = tempdir().unwrap();
540            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
541                data_root: root.path().to_path_buf(),
542                command: fixture_command(&fixture_path()),
543                initial_generation: 1,
544                request_timeout: Duration::from_secs(5),
545            });
546
547            supervisor.start().await.expect("start");
548            let session = supervisor
549                .start_agent_computer_session(json!({
550                    "bearerToken": "secret-fixture-token",
551                    "controlPlaneBaseUrl": "https://openagents.com",
552                    "repoRef": "OpenAgentsInc/openagents",
553                    "objective": "Fixture Agent Computer launch",
554                    "adapter": "codex",
555                    "lane": "cloud-gcp",
556                }))
557                .await
558                .expect("start agent computer");
559            let session_ref = session
560                .get("sessionRef")
561                .and_then(|v| v.as_str())
562                .expect("sessionRef")
563                .to_string();
564            assert_eq!(
565                session.get("environment").and_then(|v| v.as_str()),
566                Some("openagents_cloud")
567            );
568
569            let refreshed = supervisor
570                .refresh_agent_computer_session("secret-fixture-token", &session_ref)
571                .await
572                .expect("refresh");
573            assert_eq!(
574                refreshed.get("state").and_then(|v| v.as_str()),
575                Some("running")
576            );
577
578            let turn = supervisor
579                .run_agent_computer_turn(json!({
580                    "bearerToken": "secret-fixture-token",
581                    "controlPlaneBaseUrl": "https://openagents.com",
582                    "repoRef": "OpenAgentsInc/openagents",
583                    "objective": "Fixture Agent Computer turn",
584                }))
585                .await
586                .expect("run turn");
587            assert_eq!(
588                turn.get("finishReason").and_then(|v| v.as_str()),
589                Some("stop")
590            );
591            assert_eq!(
592                turn.pointer("/session/state").and_then(|v| v.as_str()),
593                Some("completed")
594            );
595
596            supervisor.restart().await.expect("restart");
597            let listed = supervisor
598                .list_agent_computer_sessions()
599                .await
600                .expect("list after restart");
601            let sessions = listed
602                .get("sessions")
603                .and_then(|v| v.as_array())
604                .expect("sessions array");
605            assert!(
606                sessions.iter().any(|row| {
607                    row.get("sessionRef").and_then(|v| v.as_str()) == Some(session_ref.as_str())
608                }),
609                "agent computer session must survive restart: {listed}"
610            );
611
612            let disk =
613                std::fs::read_to_string(root.path().join("agent-computer").join("sessions.json"))
614                    .expect("sessions disk");
615            assert!(!disk.contains("secret-fixture-token"));
616            assert!(!disk.contains("Fixture Agent Computer"));
617
618            supervisor.stop().await.expect("supervisor stop");
619        });
620    }
621
622    #[test]
623    fn sarah_nr06_conversation_methods_via_fixture() {
624        smol::block_on(async {
625            let root = tempdir().expect("tempdir");
626            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
627                data_root: root.path().to_path_buf(),
628                command: fixture_command(&fixture_path()),
629                initial_generation: 1,
630                request_timeout: Duration::from_secs(5),
631            });
632            supervisor.start().await.expect("start");
633
634            let status = supervisor
635                .sarah_session_status()
636                .await
637                .expect("session status");
638            assert_eq!(status.get("signedIn").and_then(|v| v.as_bool()), Some(true));
639            let encoded = status.to_string();
640            assert!(!encoded.contains("token"));
641            assert!(!encoded.contains("bearer"));
642
643            let boot = supervisor.sarah_bootstrap().await.expect("bootstrap");
644            assert_eq!(
645                boot.get("principalRef").and_then(|v| v.as_str()),
646                Some("principal.sarah")
647            );
648            let conversation_ref = boot
649                .get("conversationRef")
650                .and_then(|v| v.as_str())
651                .expect("conversationRef")
652                .to_string();
653            assert!(conversation_ref.starts_with("sarah."));
654
655            let sent = supervisor
656                .sarah_send_message("hello from fixture", "idempotency.fixture.send.1")
657                .await
658                .expect("send");
659            assert_eq!(sent.get("accepted").and_then(|v| v.as_bool()), Some(true));
660            let turn_ref = sent
661                .get("turnRef")
662                .and_then(|v| v.as_str())
663                .expect("turnRef")
664                .to_string();
665
666            let snap = supervisor
667                .sarah_room_snapshot(Some(json!({ "limit": 10 })))
668                .await
669                .expect("snapshot");
670            assert_eq!(
671                snap.get("conversationRef").and_then(|v| v.as_str()),
672                Some(conversation_ref.as_str())
673            );
674            assert!(
675                snap.pointer("/transcript/gapState")
676                    .and_then(|v| v.as_str())
677                    .is_some()
678            );
679            assert!(
680                snap.pointer("/transcript/cursor")
681                    .and_then(|v| v.as_str())
682                    .is_some()
683            );
684
685            let interrupt = supervisor
686                .sarah_interrupt_turn(&turn_ref, "idempotency.fixture.interrupt.1")
687                .await
688                .expect("interrupt");
689            assert_eq!(
690                interrupt.get("pending").and_then(|v| v.as_bool()),
691                Some(true)
692            );
693            assert_eq!(
694                interrupt.get("status").and_then(|v| v.as_str()),
695                Some("pending")
696            );
697
698            // OMEGA-SW-02 cut: no Khala Sync client on this lane.
699            assert!(asserts_no_khala_sync_client());
700            supervisor.stop().await.expect("stop");
701        });
702    }
703
704    #[test]
705    fn attention_decisions_are_typed_and_deduplicated() {
706        smol::block_on(async {
707            let root = tempdir().expect("tempdir");
708            let runs_path = root.path().join("full-auto").join("runs.json");
709            std::fs::create_dir_all(
710                runs_path
711                    .parent()
712                    .expect("runs path should have a parent directory"),
713            )
714            .expect("create runs directory");
715            std::fs::write(
716                &runs_path,
717                serde_json::to_string_pretty(&json!({
718                    "schema": "openagents.desktop.full_auto_run_registry.v1",
719                    "runs": [{
720                        "runRef": "run.full-auto.attention",
721                        "threadRef": null,
722                        "title": "SECRET_OBJECTIVE /Users/owner/private",
723                        "state": "stalled",
724                        "stallCause": "dispatch_overdue",
725                        "updatedAt": "2026-07-24T00:00:00.000Z"
726                    }]
727                }))
728                .expect("encode fixture runs"),
729            )
730            .expect("write fixture runs");
731
732            let mut supervisor = OmegaEffectdSupervisor::new(OmegaEffectdSupervisorOptions {
733                data_root: root.path().to_path_buf(),
734                command: fixture_command(&fixture_path()),
735                initial_generation: 1,
736                request_timeout: Duration::from_secs(5),
737            });
738            supervisor.start().await.expect("start");
739
740            let decision = supervisor
741                .decide_attention("run.full-auto.attention", true, None)
742                .await
743                .expect("attention decision")
744                .expect("stalled run should produce a decision");
745            assert!(decision.notify);
746            assert_eq!(decision.title, "Full Auto stalled");
747            assert!(decision.body.contains("SECRET_OBJECTIVE"));
748
749            let duplicate = supervisor
750                .decide_attention("run.full-auto.attention", true, Some(&decision.dedup_key))
751                .await
752                .expect("deduplicated attention decision");
753            assert!(duplicate.is_none());
754            supervisor.stop().await.expect("stop");
755        });
756    }
757}
758