Skip to repository content2000 lines · 75.7 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T01:53:11.353Z Public web read
NIP-34 coordinate
30617:7649603503856e5148d571eac2766b288a8ff1e9e35d380337a1d2b0015b4f92:omegaMaintainersHidden in public view
References2 branches · 1 tag
Read-only clone
git clone https://openagents.com/git/tenant.openagents/omega.gitBrowse files
omega_host_bridge.rs
1use std::cell::RefCell;
2use std::collections::HashMap as StdHashMap;
3use std::io::Write as _;
4use std::path::{Path, PathBuf};
5use std::rc::Rc;
6use std::sync::{Arc, LazyLock, Mutex};
7use std::time::Duration;
8
9use acp_thread::{AgentThreadEntry, ThreadStatus};
10use chrono::Utc;
11use gpui::{AnyWindowHandle, App, AppContext, AsyncApp, Entity, WeakEntity};
12use omega_effectd::{
13 ConversationIdentity, HostMethod, HostRequestFrame, HostResponseError, HostResponseErrorCode,
14 OmegaEffectdHostHandler, OpenAgentsSession, SARAH_METHOD_BOOTSTRAP, SARAH_METHOD_DEVICE_GRANTS,
15 SARAH_METHOD_INTERRUPT_TURN, SARAH_METHOD_RENEW_DEVICE_GRANT, SARAH_METHOD_REVOKE_DEVICE_GRANT,
16 SARAH_METHOD_ROOM_SNAPSHOT, SARAH_METHOD_SEND_MESSAGE, SARAH_METHOD_SESSION_STATUS,
17 SarahConversationClient, SarahConversationConfig, VerifiedOpenAgentsSession,
18};
19use serde::{Deserialize, Serialize};
20use serde_json::{Value, json};
21use workspace::{AppState, Workspace};
22
23use crate::agent_panel::CreateThreadOptions;
24use crate::{
25 Agent, AgentPanel, AgentThreadSource, ConversationView, ThreadId,
26 agent_connection_store::AgentConnectionStatus,
27};
28
29const SUPERVISED_WORKSPACE_REF: &str = "workspace.omega.supervised";
30const CODEX_LOCAL_LANE: &str = "codex-local";
31const CLAUDE_LOCAL_LANE: &str = "claude-local";
32const CODEX_AGENT_ID: &str = "codex-acp";
33const CLAUDE_AGENT_ID: &str = "claude-acp";
34const THREAD_CONNECT_ATTEMPTS: usize = 100;
35const THREAD_CONNECT_INTERVAL: Duration = Duration::from_millis(100);
36const CLAUDE_TURN_TIMEOUT: Duration = Duration::from_secs(10 * 60);
37const ACP_CANCEL_GRACE_PERIOD: Duration = Duration::from_secs(10);
38const MAX_ASSISTANT_TEXT_BYTES: usize = 24 * 1024;
39const MAX_EVIDENCE_TURNS: usize = 48;
40const MAX_TOTAL_ASSISTANT_TEXT_BYTES: usize = 6 * 1024;
41const CORRELATION_SCHEMA: &str = "openagents.omega.full_auto_host_correlation.v1";
42const CORRELATION_FILE: &str = "full-auto-host-correlation.json";
43
44/// OMEGA-DELTA-0021. Which `omega-effectd` lane run each host-bridge thread
45/// belongs to, so the thread surface can disclose it.
46///
47/// A read-mostly index, not a store. It is derived twice from the correlation
48/// journal — once when the journal is loaded at startup, and again on every
49/// write — so it cannot outlive or disagree with the durable record. omega#77's
50/// falsifier names a *new durable* store for disclosure; this is neither new
51/// nor durable, and deleting the file it mirrors empties it.
52///
53/// It exists because `HostBridgeState` lives inside the host handler's closure
54/// and is not reachable from a render, while the disclosure has to be readable
55/// wherever a thread is drawn.
56static ENGINE_LANE_RUNS: LazyLock<Mutex<StdHashMap<ThreadId, String>>> =
57 LazyLock::new(|| Mutex::new(StdHashMap::new()));
58
59/// The lane index a set of host threads implies.
60fn engine_lane_runs_from(threads: &[HostThread]) -> StdHashMap<ThreadId, String> {
61 threads
62 .iter()
63 .map(|thread| (thread.thread_id, thread.operation_ref.clone()))
64 .collect()
65}
66
67/// Replace the lane index with what the correlation journal now says.
68///
69/// Wholesale replacement rather than insertion: a thread dropped from the
70/// journal must stop being disclosed as a lane run, and an incremental index
71/// would keep disclosing it.
72fn republish_engine_lane_runs(threads: &[HostThread]) {
73 let runs = engine_lane_runs_from(threads);
74 match ENGINE_LANE_RUNS.lock() {
75 Ok(mut index) => *index = runs,
76 Err(poisoned) => *poisoned.into_inner() = runs,
77 }
78}
79
80/// Publish one lane correlation, for a harness that has no engine to run.
81///
82/// `OMEGA-DELTA-0021`'s engine-lane disclosure is normally reached only by a
83/// real `omega-effectd` run, which a rendering harness cannot start. This is
84/// the seam that lets the *rendered* engine-lane line be captured without
85/// weakening the production path: it is behind `test-support`, so no shipped
86/// build can reach it, and it writes the same index the correlation journal
87/// writes.
88#[cfg(any(test, feature = "test-support"))]
89pub fn publish_engine_lane_run_for_tests(thread_id: ThreadId, operation_ref: String) {
90 match ENGINE_LANE_RUNS.lock() {
91 Ok(mut index) => {
92 index.insert(thread_id, operation_ref);
93 }
94 Err(poisoned) => {
95 poisoned.into_inner().insert(thread_id, operation_ref);
96 }
97 }
98}
99
100/// Persist one lane correlation through the production writer, for a harness
101/// that has no engine to run.
102///
103/// [`publish_engine_lane_run_for_tests`] writes the process-local index, which
104/// a restart empties. This writes the durable half — the same
105/// `CorrelationJournal`, in the same schema, at the same path
106/// [`omega_effectd_host_handler`] reads at startup — so a *second process* can
107/// be shown rebuilding the disclosure from disk rather than from a static that
108/// happened to survive. Behind `test-support`, so no shipped build can reach it.
109#[cfg(any(test, feature = "test-support"))]
110pub fn persist_engine_lane_run_for_tests(
111 thread_id: ThreadId,
112 operation_ref: String,
113) -> anyhow::Result<()> {
114 let state = HostBridgeState {
115 workspace: None,
116 threads: vec![HostThread {
117 workspace_ref: SUPERVISED_WORKSPACE_REF.to_string(),
118 lane: CODEX_LOCAL_LANE.to_string(),
119 operation_ref,
120 thread_id,
121 conversation: None,
122 turns: Vec::new(),
123 revision: 1,
124 }],
125 correlation_path: correlation_journal_path(),
126 load_error: None,
127 sarah_conversation: None,
128 };
129 persist_correlation_journal(&state)
130}
131
132/// Where the correlation journal lives.
133fn correlation_journal_path() -> PathBuf {
134 paths::data_dir().join("openagents").join(CORRELATION_FILE)
135}
136
137/// Read the correlation journal and refill the lane index from it.
138///
139/// OMEGA-DELTA-0021's restart edge, in one place. A freshly started process has
140/// an empty lane index, and this is what refills it, so a thread resumed after
141/// a restart still discloses the lane that owns it.
142///
143/// [`omega_effectd_host_handler`] calls this at startup and keeps the threads
144/// for its own state; [`reload_engine_lane_runs_from_disk`] calls it for a
145/// caller that only needs the index. Neither is a copy of the other, so what a
146/// cold process is observed doing here is what the shipped startup does.
147fn load_journal_and_republish(path: &Path) -> (Vec<HostThread>, Option<String>) {
148 let (threads, load_error) = match load_correlation_journal(path) {
149 Ok(threads) => (threads, None),
150 Err(error) => (Vec::new(), Some(error.to_string())),
151 };
152 republish_engine_lane_runs(&threads);
153 (threads, load_error)
154}
155
156/// Refill the lane index from the correlation journal on disk.
157///
158/// Returns the number of lane-bound threads the journal named. The error, if
159/// the journal cannot be read, is returned rather than logged: a caller that
160/// asked for the restart edge explicitly needs to know it did not happen, and
161/// an empty index is indistinguishable from a journal with nothing in it.
162pub fn reload_engine_lane_runs_from_disk() -> anyhow::Result<usize> {
163 let path = correlation_journal_path();
164 let (threads, load_error) = load_journal_and_republish(&path);
165 if let Some(error) = load_error {
166 anyhow::bail!(
167 "correlation journal at {} is unreadable: {error}",
168 path.display()
169 );
170 }
171 Ok(threads.len())
172}
173
174/// The `omega-effectd` lane run this thread belongs to, if it is one.
175///
176/// Returns `None` for every thread the user started themselves, which is what
177/// keeps a hand-driven `codex-acp` thread disclosed as routed rather than as a
178/// delegated lane run.
179pub fn engine_lane_run(thread_id: ThreadId) -> Option<String> {
180 match ENGINE_LANE_RUNS.lock() {
181 Ok(index) => index.get(&thread_id).cloned(),
182 Err(poisoned) => poisoned.into_inner().get(&thread_id).cloned(),
183 }
184}
185
186#[derive(Clone)]
187struct WorkspaceBinding {
188 workspace_ref: String,
189 workspace: WeakEntity<Workspace>,
190 window: AnyWindowHandle,
191}
192
193#[derive(Clone)]
194struct HostThread {
195 workspace_ref: String,
196 lane: String,
197 operation_ref: String,
198 thread_id: ThreadId,
199 conversation: Option<WeakEntity<ConversationView>>,
200 turns: Vec<HostTurn>,
201 revision: u64,
202}
203
204#[derive(Clone, Serialize, Deserialize)]
205#[serde(rename_all = "camelCase", deny_unknown_fields)]
206struct HostTurn {
207 turn_ref: String,
208 lane: String,
209 account_ref: Option<String>,
210 model: Option<String>,
211 provider_session_ref: String,
212 start_entry_index: usize,
213 end_entry_index: Option<usize>,
214 phase: String,
215 disposition: Option<String>,
216 created_at: String,
217 updated_at: String,
218}
219
220#[derive(Clone, Copy, Debug, PartialEq, Eq)]
221enum HostTurnOutcome {
222 Completed,
223 Failed,
224 TimedOut,
225}
226
227struct HostBridgeState {
228 workspace: Option<WorkspaceBinding>,
229 threads: Vec<HostThread>,
230 correlation_path: PathBuf,
231 load_error: Option<String>,
232 sarah_conversation: Option<Arc<Mutex<SarahConversationClient>>>,
233}
234
235#[derive(Serialize, Deserialize)]
236#[serde(rename_all = "camelCase", deny_unknown_fields)]
237struct CorrelationJournal {
238 schema: String,
239 threads: Vec<PersistedHostThread>,
240}
241
242#[derive(Serialize, Deserialize)]
243#[serde(rename_all = "camelCase", deny_unknown_fields)]
244struct PersistedHostThread {
245 workspace_ref: String,
246 lane: String,
247 operation_ref: String,
248 thread_ref: String,
249 turns: Vec<HostTurn>,
250 revision: u64,
251}
252
253#[derive(Deserialize)]
254#[serde(rename_all = "camelCase", deny_unknown_fields)]
255struct ResolveWorkspaceParams {
256 expected_workspace_ref: Option<String>,
257}
258
259#[derive(Deserialize)]
260#[serde(deny_unknown_fields)]
261struct ResolveSyncSessionParams {}
262
263#[derive(Deserialize)]
264#[serde(rename_all = "camelCase", deny_unknown_fields)]
265struct CreateThreadParams {
266 title: String,
267 lane: String,
268 workspace_ref: String,
269 operation_ref: String,
270}
271
272#[derive(Deserialize)]
273#[serde(rename_all = "camelCase", deny_unknown_fields)]
274struct LaneReadinessParams {
275 lane: String,
276 excluding_thread_ref: Option<String>,
277}
278
279#[derive(Deserialize)]
280#[serde(rename_all = "camelCase", deny_unknown_fields)]
281struct DispatchTurnParams {
282 run_ref: String,
283 workspace_ref: String,
284 thread_ref: String,
285 turn_ref: String,
286 message: String,
287 profile: Option<DispatchProfile>,
288}
289
290#[derive(Deserialize)]
291#[serde(rename_all = "camelCase", deny_unknown_fields)]
292struct DispatchProfile {
293 lane: Option<String>,
294 account_ref: Option<String>,
295 model: Option<String>,
296 reasoning_effort: Option<String>,
297}
298
299#[derive(Deserialize)]
300#[serde(rename_all = "camelCase", deny_unknown_fields)]
301struct RefreshEvidenceParams {
302 run_ref: String,
303 thread_ref: String,
304}
305
306#[derive(Deserialize)]
307#[serde(rename_all = "camelCase", deny_unknown_fields)]
308struct InterruptTurnParams {
309 thread_ref: String,
310 turn_ref: Option<String>,
311}
312
313#[derive(Deserialize)]
314#[serde(rename_all = "camelCase", deny_unknown_fields)]
315struct AppendSystemNoteParams {
316 thread_ref: String,
317 note_ref: String,
318 text: String,
319}
320
321pub fn omega_effectd_host_handler(cx: &App) -> OmegaEffectdHostHandler {
322 let async_cx = cx.to_async();
323 let openagents_session = omega_effectd::openagents_session(cx);
324 let correlation_path = correlation_journal_path();
325 // OMEGA-DELTA-0021. This is the restart edge: the lane index is empty in a
326 // freshly started process, and the journal on disk is what refills it, so a
327 // thread resumed after a restart still discloses the lane that owns it.
328 let (threads, load_error) = load_journal_and_republish(&correlation_path);
329 let state = Rc::new(RefCell::new(HostBridgeState {
330 workspace: None,
331 threads,
332 correlation_path,
333 load_error,
334 sarah_conversation: None,
335 }));
336 Rc::new(move |request| {
337 let async_cx = async_cx.clone();
338 let state = state.clone();
339 let openagents_session = openagents_session.clone();
340 Box::pin(async move { handle_request(request, state, openagents_session, async_cx).await })
341 })
342}
343
344fn production_sarah_conversation() -> Result<SarahConversationClient, HostResponseError> {
345 let relay_urls = std::env::var("OPENAGENTS_OMEGA_NOSTR_RELAYS")
346 .map_err(|_| unavailable("OPENAGENTS_OMEGA_NOSTR_RELAYS is not configured."))?
347 .split(',')
348 .map(str::trim)
349 .filter(|relay_url| !relay_url.is_empty())
350 .map(str::to_owned)
351 .collect::<Vec<_>>();
352 let sarah_public_key_hex = std::env::var("OPENAGENTS_OMEGA_SARAH_PUBLIC_KEY_HEX")
353 .map_err(|_| unavailable("OPENAGENTS_OMEGA_SARAH_PUBLIC_KEY_HEX is not configured."))?;
354 let admitted_device_public_key_hexes =
355 std::env::var("OPENAGENTS_OMEGA_NOSTR_DEVICE_PUBLIC_KEYS")
356 .map_err(|_| {
357 unavailable("OPENAGENTS_OMEGA_NOSTR_DEVICE_PUBLIC_KEYS is not configured.")
358 })?
359 .split(',')
360 .map(str::trim)
361 .filter(|public_key| !public_key.is_empty())
362 .map(str::to_owned)
363 .collect::<Vec<_>>();
364 let approved_device_scopes = std::env::var("OPENAGENTS_OMEGA_NOSTR_DEVICE_SCOPES")
365 .map_err(|_| unavailable("OPENAGENTS_OMEGA_NOSTR_DEVICE_SCOPES is not configured."))?
366 .split(',')
367 .map(str::trim)
368 .filter(|scope| !scope.is_empty())
369 .map(omega_effectd::Issue31PairingScope::parse)
370 .collect::<Result<Vec<_>, _>>()
371 .map_err(|error| {
372 unavailable(format!("Issue 31 device scope policy is invalid: {error}"))
373 })?;
374 let community_group_ids = std::env::var("OPENAGENTS_OMEGA_NOSTR_COMMUNITY_GROUP_IDS")
375 .unwrap_or_default()
376 .split(',')
377 .map(str::trim)
378 .filter(|group_id| !group_id.is_empty())
379 .map(str::to_owned)
380 .collect::<Vec<_>>();
381 let community_public_key_hexes = std::env::var("OPENAGENTS_OMEGA_NOSTR_COMMUNITY_PUBLIC_KEYS")
382 .unwrap_or_default()
383 .split(',')
384 .map(str::trim)
385 .filter(|public_key| !public_key.is_empty())
386 .map(str::to_owned)
387 .collect::<Vec<_>>();
388 let identity_service = Arc::new(omega_identity::IdentityService::system(
389 *app_identity::CHANNEL,
390 ));
391 let custody = identity_service
392 .inspect()
393 .map_err(|error| unavailable(format!("Omega identity custody is unavailable: {error}")))?;
394 let owner_public_key_hex = custody
395 .identity
396 .map(|identity| identity.public_key_hex().as_str().to_string())
397 .ok_or_else(|| unavailable("Omega identity custody is not ready."))?;
398 let conversation_digest = std::env::var("OPENAGENTS_OMEGA_SARAH_CONVERSATION_DIGEST")
399 .unwrap_or_else(|_| owner_public_key_hex.chars().take(24).collect());
400 let config = SarahConversationConfig {
401 generation: 1,
402 conversation_digest,
403 identity: ConversationIdentity {
404 owner_public_key_hex,
405 sarah_public_key_hex,
406 account_label: None,
407 binding_state: omega_effectd::BindingState::Unbound,
408 },
409 relay_url: relay_urls.first().cloned(),
410 admitted_device_public_key_hexes,
411 approved_device_scopes,
412 community_group_ids,
413 community_public_key_hexes,
414 };
415 let mut conversation = SarahConversationClient::new_production(
416 config,
417 relay_urls,
418 identity_service,
419 )
420 .map_err(|error| unavailable(format!("Sarah Nostr transport is unavailable: {error}")))?;
421 // omega#49: the host pump publishes the omega#47 snapshot and its Full Auto
422 // detail to every admitted device. Without this the two documents are built
423 // by the desktop panel and never leave the machine, and a paired phone
424 // reports `no_host_projection` for a host that is running work.
425 conversation.set_issue31_host_projection_source(
426 full_auto_ui::issue31_host_projection_source(),
427 );
428 // omega#91: how the host reads its own provider accounts when it decides
429 // which one a connection handoff binds to. Without it no handoff can bind,
430 // and every one the phone opens runs to its deadline and expires.
431 conversation.set_issue31_provider_roster_source(
432 full_auto_ui::issue31_provider_roster_source(),
433 );
434 Ok(conversation)
435}
436
437async fn sarah_request(
438 request: HostRequestFrame,
439 state: &Rc<RefCell<HostBridgeState>>,
440 cx: &mut AsyncApp,
441) -> Result<Value, HostResponseError> {
442 let method = match request.method {
443 HostMethod::SarahSessionStatus => SARAH_METHOD_SESSION_STATUS,
444 HostMethod::SarahBootstrap => SARAH_METHOD_BOOTSTRAP,
445 HostMethod::SarahRoomSnapshot => SARAH_METHOD_ROOM_SNAPSHOT,
446 HostMethod::SarahSendMessage => SARAH_METHOD_SEND_MESSAGE,
447 HostMethod::SarahInterruptTurn => SARAH_METHOD_INTERRUPT_TURN,
448 HostMethod::SarahDeviceGrants => SARAH_METHOD_DEVICE_GRANTS,
449 HostMethod::SarahRenewDeviceGrant => SARAH_METHOD_RENEW_DEVICE_GRANT,
450 HostMethod::SarahRevokeDeviceGrant => SARAH_METHOD_REVOKE_DEVICE_GRANT,
451 _ => return Err(unsupported("Unknown Sarah host method.")),
452 };
453 let conversation = state.borrow().sarah_conversation.clone();
454 let conversation = match conversation {
455 Some(conversation) => conversation,
456 None => match production_sarah_conversation() {
457 Ok(conversation) => {
458 let conversation = Arc::new(Mutex::new(conversation));
459 let mut state = state.borrow_mut();
460 state.sarah_conversation = Some(conversation.clone());
461 conversation
462 }
463 Err(error) => return Err(error),
464 },
465 };
466 let params = request.params;
467 let generation = request.generation;
468 cx.background_spawn(async move {
469 let mut conversation = conversation
470 .lock()
471 .map_err(|_| unavailable("Sarah Nostr transport state is unavailable."))?;
472 conversation
473 .synchronize_process_generation(generation)
474 .map_err(sarah_host_error)?;
475 conversation
476 .handle_request(method, generation, Some(¶ms))
477 .map_err(sarah_host_error)
478 })
479 .await
480}
481
482fn sarah_host_error(error: omega_effectd::SarahConversationError) -> HostResponseError {
483 let code = match error.protocol_code() {
484 omega_effectd::ProtocolErrorCode::StaleGeneration => HostResponseErrorCode::StaleGeneration,
485 omega_effectd::ProtocolErrorCode::InvalidRequest => HostResponseErrorCode::InvalidRequest,
486 omega_effectd::ProtocolErrorCode::UnknownMethod => HostResponseErrorCode::Unsupported,
487 omega_effectd::ProtocolErrorCode::HostUnavailable => HostResponseErrorCode::Unavailable,
488 omega_effectd::ProtocolErrorCode::HostTimeout => HostResponseErrorCode::Unavailable,
489 omega_effectd::ProtocolErrorCode::NotRunning
490 | omega_effectd::ProtocolErrorCode::RunNotFound => HostResponseErrorCode::Unavailable,
491 omega_effectd::ProtocolErrorCode::FrameTooLarge => HostResponseErrorCode::InvalidRequest,
492 omega_effectd::ProtocolErrorCode::Internal => HostResponseErrorCode::Internal,
493 };
494 HostResponseError {
495 code,
496 message: error.to_string(),
497 }
498}
499
500async fn handle_request(
501 request: HostRequestFrame,
502 state: Rc<RefCell<HostBridgeState>>,
503 openagents_session: OpenAgentsSession,
504 mut cx: AsyncApp,
505) -> Result<Value, HostResponseError> {
506 match request.method.clone() {
507 HostMethod::ResolveWorkspace => resolve_workspace(request.params, &state, &mut cx),
508 HostMethod::ResolveSyncSession => {
509 resolve_sync_session(request.params, &openagents_session, &mut cx).await
510 }
511 HostMethod::CreateThread => create_thread(request.params, &state, &mut cx),
512 HostMethod::LaneReadiness => lane_readiness(request.params, &state, &cx),
513 HostMethod::DispatchTurn => dispatch_turn(request.params, &state, &mut cx).await,
514 HostMethod::RefreshEvidence => {
515 refresh_evidence(request.params, request.generation, &state, &cx)
516 }
517 HostMethod::InterruptTurn => interrupt_turn(request.params, &state, &mut cx).await,
518 HostMethod::AppendSystemNote => append_system_note(request.params, &state, &mut cx).await,
519 HostMethod::SarahSessionStatus
520 | HostMethod::SarahBootstrap
521 | HostMethod::SarahRoomSnapshot
522 | HostMethod::SarahSendMessage
523 | HostMethod::SarahInterruptTurn
524 | HostMethod::SarahDeviceGrants => sarah_request(request, &state, &mut cx).await,
525 HostMethod::SarahRenewDeviceGrant | HostMethod::SarahRevokeDeviceGrant => {
526 sarah_request(request, &state, &mut cx).await
527 }
528 HostMethod::Unsupported => Err(unsupported("Unknown Omega host method.")),
529 }
530}
531
532async fn resolve_sync_session(
533 params: Value,
534 session: &OpenAgentsSession,
535 cx: &mut AsyncApp,
536) -> Result<Value, HostResponseError> {
537 let _: ResolveSyncSessionParams = decode_params(params)?;
538 Ok(sync_session_result(session.resolve_verified(cx).await))
539}
540
541fn sync_session_result(session: Option<VerifiedOpenAgentsSession>) -> Value {
542 match session {
543 Some(session) => json!({
544 "available": true,
545 "baseUrl": session.base_url,
546 "accessToken": session.access_token,
547 }),
548 None => json!({ "available": false }),
549 }
550}
551
552fn resolve_workspace(
553 params: Value,
554 state: &Rc<RefCell<HostBridgeState>>,
555 cx: &mut AsyncApp,
556) -> Result<Value, HostResponseError> {
557 require_loaded_journal(state)?;
558 let params: ResolveWorkspaceParams = decode_params(params)?;
559 if let Some(expected) = params.expected_workspace_ref.as_deref() {
560 validate_ref(expected, "expectedWorkspaceRef")?;
561 }
562 if let Some(binding) = state.borrow().workspace.clone()
563 && binding.workspace.upgrade().is_some()
564 {
565 if params
566 .expected_workspace_ref
567 .as_ref()
568 .is_some_and(|expected| expected != &binding.workspace_ref)
569 {
570 return Err(unavailable("The requested workspace is not active."));
571 }
572 return Ok(json!({ "workspaceRef": binding.workspace_ref }));
573 }
574
575 let candidates = cx.update(|cx| {
576 let app_state = AppState::global(cx);
577 app_state
578 .workspace_store
579 .read(cx)
580 .workspaces_with_windows()
581 .filter_map(|(window, workspace)| {
582 let workspace = workspace.upgrade()?;
583 let has_worktree = workspace
584 .read(cx)
585 .project()
586 .read(cx)
587 .worktrees(cx)
588 .next()
589 .is_some();
590 has_worktree.then(|| (window, workspace))
591 })
592 .collect::<Vec<_>>()
593 });
594 let [(window, workspace)] = candidates.as_slice() else {
595 return Err(unavailable(
596 "Omega requires exactly one open workspace with a worktree.",
597 ));
598 };
599 let workspace_ref = params
600 .expected_workspace_ref
601 .unwrap_or_else(|| SUPERVISED_WORKSPACE_REF.to_string());
602 state.borrow_mut().workspace = Some(WorkspaceBinding {
603 workspace_ref: workspace_ref.clone(),
604 workspace: workspace.downgrade(),
605 window: *window,
606 });
607 rebind_persisted_threads(&workspace_ref, state, cx)?;
608 Ok(json!({ "workspaceRef": workspace_ref }))
609}
610
611fn create_thread(
612 params: Value,
613 state: &Rc<RefCell<HostBridgeState>>,
614 cx: &mut AsyncApp,
615) -> Result<Value, HostResponseError> {
616 let params: CreateThreadParams = decode_params(params)?;
617 validate_ref(¶ms.operation_ref, "operationRef")?;
618 validate_ref(¶ms.workspace_ref, "workspaceRef")?;
619 if params.title.trim().is_empty() {
620 return Err(invalid("title must not be empty."));
621 }
622 let agent = agent_for_lane(¶ms.lane)?;
623 if let Some(thread_ref) = state.borrow().threads.iter().find_map(|thread| {
624 (thread.workspace_ref == params.workspace_ref
625 && thread.lane == params.lane
626 && thread.operation_ref == params.operation_ref)
627 .then(|| thread.thread_id.to_key_string())
628 }) {
629 return Ok(json!({ "threadRef": thread_ref }));
630 }
631 let binding = require_workspace(¶ms.workspace_ref, state)?;
632 let workspace = binding
633 .workspace
634 .upgrade()
635 .ok_or_else(|| unavailable("The bound workspace was closed."))?;
636 let (thread_id, conversation) = binding
637 .window
638 .update(cx, |_root, window, cx| {
639 let panel = workspace
640 .read(cx)
641 .panel::<AgentPanel>(cx)
642 .ok_or_else(|| unavailable("The workspace Agent panel is unavailable."))?;
643 let thread_id = panel.update(cx, |panel, cx| {
644 panel.create_thread_with_options(
645 CreateThreadOptions {
646 title: Some(params.title.clone().into()),
647 agent: Some(agent),
648 ..Default::default()
649 },
650 AgentThreadSource::AgentPanel,
651 window,
652 cx,
653 )
654 });
655 let conversation = panel
656 .read(cx)
657 .conversation_view_for_id(&thread_id, cx)
658 .cloned()
659 .ok_or_else(|| internal("The created thread was not retained."))?;
660 Ok((thread_id, conversation))
661 })
662 .map_err(|error| unavailable(format!("The workspace window is unavailable: {error}")))??;
663 let thread_ref = thread_id.to_key_string();
664 state.borrow_mut().threads.push(HostThread {
665 workspace_ref: params.workspace_ref,
666 lane: params.lane,
667 operation_ref: params.operation_ref,
668 thread_id,
669 conversation: Some(conversation.downgrade()),
670 turns: Vec::new(),
671 revision: 1,
672 });
673 persist_state(state)?;
674 Ok(json!({ "threadRef": thread_ref }))
675}
676
677fn lane_readiness(
678 params: Value,
679 state: &Rc<RefCell<HostBridgeState>>,
680 cx: &AsyncApp,
681) -> Result<Value, HostResponseError> {
682 let params: LaneReadinessParams = decode_params(params)?;
683 validate_lane(¶ms.lane)?;
684 if let Some(thread_ref) = params.excluding_thread_ref.as_deref() {
685 validate_ref(thread_ref, "excludingThreadRef")?;
686 }
687 if !is_supported_lane(¶ms.lane) {
688 return Ok(json!({
689 "known": false,
690 "admitted": false,
691 "fullAuto": false,
692 "state": "unavailable",
693 }));
694 }
695 let workspace_ready = state
696 .borrow()
697 .workspace
698 .as_ref()
699 .is_some_and(|binding| binding.workspace.upgrade().is_some());
700 let dispatch_thread_exists =
701 params
702 .excluding_thread_ref
703 .as_ref()
704 .is_some_and(|excluding_thread_ref| {
705 state.borrow().threads.iter().any(|thread| {
706 thread.lane == params.lane
707 && thread.thread_id.to_key_string() == *excluding_thread_ref
708 })
709 });
710 let (authority_ready, authority_state) = match params.lane.as_str() {
711 CODEX_LOCAL_LANE => {
712 external_agent_lane_readiness(state, cx, CODEX_AGENT_ID, dispatch_thread_exists)?
713 }
714 CLAUDE_LOCAL_LANE => {
715 external_agent_lane_readiness(state, cx, CLAUDE_AGENT_ID, dispatch_thread_exists)?
716 }
717 _ => return Err(unsupported("The provider lane is not supported.")),
718 };
719 let busy = state.borrow().threads.iter().any(|host_thread| {
720 if host_thread.lane != params.lane {
721 return false;
722 }
723 if params
724 .excluding_thread_ref
725 .as_ref()
726 .is_some_and(|thread_ref| thread_ref == &host_thread.thread_id.to_key_string())
727 {
728 return false;
729 }
730 let Some(conversation) = host_thread
731 .conversation
732 .as_ref()
733 .and_then(WeakEntity::upgrade)
734 else {
735 return false;
736 };
737 cx.update(|cx| {
738 conversation
739 .read(cx)
740 .root_thread(cx)
741 .is_some_and(|thread| thread.read(cx).status() == ThreadStatus::Generating)
742 })
743 });
744 let admitted = workspace_ready && authority_ready;
745 Ok(json!({
746 "known": true,
747 "admitted": admitted,
748 "fullAuto": admitted,
749 "state": if admitted && !busy { "available" } else if busy { "busy" } else { authority_state },
750 }))
751}
752
753async fn dispatch_turn(
754 params: Value,
755 state: &Rc<RefCell<HostBridgeState>>,
756 cx: &mut AsyncApp,
757) -> Result<Value, HostResponseError> {
758 let params: DispatchTurnParams = decode_params(params)?;
759 validate_ref(¶ms.run_ref, "runRef")?;
760 validate_ref(¶ms.workspace_ref, "workspaceRef")?;
761 validate_ref(¶ms.thread_ref, "threadRef")?;
762 validate_ref(¶ms.turn_ref, "turnRef")?;
763 if params.message.is_empty() {
764 return Err(invalid("message must not be empty."));
765 }
766 require_workspace(¶ms.workspace_ref, state)?;
767 let lane = params
768 .profile
769 .as_ref()
770 .and_then(|profile| profile.lane.clone())
771 .unwrap_or_else(|| CODEX_LOCAL_LANE.to_string());
772 if !is_supported_lane(&lane) {
773 return Ok(json!({
774 "accepted": false,
775 "reason": "unsupported_lane",
776 "failureCause": "lane_unavailable",
777 }));
778 }
779 validate_lane(&lane)?;
780 if let Some(profile) = params.profile.as_ref() {
781 if let Some(account_ref) = profile.account_ref.as_deref() {
782 validate_ref(account_ref, "profile.accountRef")?;
783 }
784 if let Some(model) = profile.model.as_deref() {
785 validate_ref(model, "profile.model")?;
786 }
787 if let Some(reasoning_effort) = profile.reasoning_effort.as_deref() {
788 validate_ref(reasoning_effort, "profile.reasoningEffort")?;
789 }
790 }
791 if params
792 .profile
793 .as_ref()
794 .is_some_and(|profile| profile.model.is_some() || profile.reasoning_effort.is_some())
795 {
796 return Ok(json!({
797 "accepted": false,
798 "reason": "profile_override_unavailable",
799 "failureCause": "lane_unavailable",
800 }));
801 }
802 let conversation = {
803 let bridge = state.borrow();
804 let host_thread = bridge
805 .threads
806 .iter()
807 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
808 .ok_or_else(|| unavailable("The requested Agent thread is not bound."))?;
809 if host_thread.workspace_ref != params.workspace_ref {
810 return Err(unavailable("The thread belongs to a different workspace."));
811 }
812 if host_thread.lane != lane {
813 return Err(invalid(
814 "The dispatch lane does not match the Agent thread lane.",
815 ));
816 }
817 host_thread
818 .conversation
819 .as_ref()
820 .and_then(WeakEntity::upgrade)
821 .ok_or_else(|| unavailable("The requested Agent thread was closed."))?
822 };
823 let thread = wait_for_root_thread(&conversation, cx).await?;
824 let (start_entry_index, provider_session_ref) = cx.update(|cx| {
825 let thread = thread.read(cx);
826 if thread.status() == ThreadStatus::Generating {
827 return Err(unavailable("The Agent thread already has a running turn."));
828 }
829 Ok((thread.entries().len(), thread.session_id().0.to_string()))
830 })?;
831 validate_ref(&provider_session_ref, "providerSessionRef")?;
832 let turn_timeout = turn_timeout_for_lane(&lane);
833 let now = Utc::now().to_rfc3339();
834 {
835 let mut bridge = state.borrow_mut();
836 let host_thread = bridge
837 .threads
838 .iter_mut()
839 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
840 .ok_or_else(|| unavailable("The requested Agent thread is not bound."))?;
841 if host_thread
842 .turns
843 .iter()
844 .any(|turn| turn.turn_ref == params.turn_ref)
845 {
846 return Ok(json!({ "accepted": false, "reason": "duplicate_turn_ref" }));
847 }
848 host_thread.turns.push(HostTurn {
849 turn_ref: params.turn_ref.clone(),
850 lane,
851 account_ref: params
852 .profile
853 .as_ref()
854 .and_then(|profile| profile.account_ref.clone()),
855 model: params
856 .profile
857 .as_ref()
858 .and_then(|profile| profile.model.clone()),
859 provider_session_ref,
860 start_entry_index,
861 end_entry_index: None,
862 phase: "streaming".to_string(),
863 disposition: None,
864 created_at: now.clone(),
865 updated_at: now,
866 });
867 host_thread.revision += 1;
868 }
869 persist_state(state)?;
870 let send = thread.update(cx, |thread, cx| {
871 thread.send(vec![params.message.into()], cx)
872 });
873 let turn_ref = params.turn_ref;
874 let state_for_completion = state.clone();
875 let thread_for_completion = thread.clone();
876 cx.spawn(async move |cx| {
877 let outcome = if let Some(timeout) = turn_timeout {
878 let timer = cx.background_executor().timer(timeout);
879 match futures::future::select(send, timer).await {
880 futures::future::Either::Left((result, _)) => {
881 if result.is_ok() {
882 HostTurnOutcome::Completed
883 } else {
884 HostTurnOutcome::Failed
885 }
886 }
887 futures::future::Either::Right(((), _)) => {
888 log::warn!("Omega Full Auto Claude turn exceeded its host deadline");
889 let cancel = cx.update(|cx| {
890 thread_for_completion.update(cx, |thread, cx| {
891 (thread.status() == ThreadStatus::Generating).then(|| thread.cancel(cx))
892 })
893 });
894 if let Some(cancel) = cancel {
895 let cancel_deadline =
896 cx.background_executor().timer(ACP_CANCEL_GRACE_PERIOD);
897 if matches!(
898 futures::future::select(cancel, cancel_deadline).await,
899 futures::future::Either::Right(_)
900 ) {
901 log::warn!(
902 "Omega Full Auto Claude cancellation exceeded its grace period"
903 );
904 }
905 }
906 HostTurnOutcome::TimedOut
907 }
908 }
909 } else if send.await.is_ok() {
910 HostTurnOutcome::Completed
911 } else {
912 HostTurnOutcome::Failed
913 };
914 let (end_entry_index, had_error) = cx.update(|cx| {
915 let thread = thread_for_completion.read(cx);
916 (thread.entries().len(), thread.had_error())
917 });
918 let outcome = if outcome == HostTurnOutcome::Completed && had_error {
919 HostTurnOutcome::Failed
920 } else {
921 outcome
922 };
923 if let Err(error) =
924 record_turn_completion(&state_for_completion, &turn_ref, end_entry_index, outcome)
925 {
926 log::error!("failed to persist Omega Full Auto host correlation: {error:#}");
927 }
928 })
929 .detach();
930 Ok(json!({ "accepted": true }))
931}
932
933fn refresh_evidence(
934 params: Value,
935 generation: u64,
936 state: &Rc<RefCell<HostBridgeState>>,
937 cx: &AsyncApp,
938) -> Result<Value, HostResponseError> {
939 let params: RefreshEvidenceParams = decode_params(params)?;
940 validate_ref(¶ms.run_ref, "runRef")?;
941 validate_ref(¶ms.thread_ref, "threadRef")?;
942 let (conversation, mut turns, mut revision) = {
943 let bridge = state.borrow();
944 let Some(thread) = bridge
945 .threads
946 .iter()
947 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
948 else {
949 return Ok(json!({ "present": false, "revision": 0, "live": null, "turns": [] }));
950 };
951 (
952 thread.conversation.as_ref().and_then(WeakEntity::upgrade),
953 thread.turns.clone(),
954 thread.revision,
955 )
956 };
957 if turns.len() > MAX_EVIDENCE_TURNS {
958 turns.drain(..turns.len() - MAX_EVIDENCE_TURNS);
959 }
960 let Some(conversation) = conversation else {
961 return Ok(json!({ "present": false, "revision": revision, "live": null, "turns": [] }));
962 };
963 let Some(thread) = cx.update(|cx| conversation.read(cx).root_thread(cx)) else {
964 return Ok(json!({ "present": true, "revision": revision, "live": null, "turns": [] }));
965 };
966 let (status, entry_count, assistant_texts) = cx.update(|cx| {
967 let thread = thread.read(cx);
968 let mut texts = turns
969 .iter()
970 .map(|turn| {
971 let end_entry_index = turn.end_entry_index.unwrap_or(thread.entries().len());
972 let text = thread
973 .entries()
974 .iter()
975 .skip(turn.start_entry_index)
976 .take(end_entry_index.saturating_sub(turn.start_entry_index))
977 .filter_map(|entry| match entry {
978 AgentThreadEntry::AssistantMessage(message) => {
979 Some(message.to_markdown(cx))
980 }
981 _ => None,
982 })
983 .collect::<Vec<_>>()
984 .join("\n");
985 truncate_utf8(&text, MAX_ASSISTANT_TEXT_BYTES)
986 })
987 .collect::<Vec<_>>();
988 let mut remaining_bytes = MAX_TOTAL_ASSISTANT_TEXT_BYTES;
989 for text in texts.iter_mut().rev() {
990 *text = truncate_utf8(text, text.len().min(remaining_bytes));
991 remaining_bytes = remaining_bytes.saturating_sub(text.len());
992 }
993 (thread.status(), thread.entries().len(), texts)
994 });
995 if status == ThreadStatus::Idle
996 && let Some(turn) = turns
997 .iter_mut()
998 .rev()
999 .find(|turn| turn.disposition.is_none())
1000 {
1001 turn.phase = "completed".to_string();
1002 turn.disposition = Some("completed".to_string());
1003 turn.end_entry_index = Some(entry_count);
1004 turn.updated_at = Utc::now().to_rfc3339();
1005 {
1006 let mut bridge = state.borrow_mut();
1007 if let Some(host_thread) = bridge
1008 .threads
1009 .iter_mut()
1010 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
1011 && let Some(stored_turn) = host_thread
1012 .turns
1013 .iter_mut()
1014 .find(|stored_turn| stored_turn.turn_ref == turn.turn_ref)
1015 && stored_turn.disposition.is_none()
1016 {
1017 *stored_turn = turn.clone();
1018 host_thread.revision += 1;
1019 revision = host_thread.revision;
1020 }
1021 }
1022 persist_state(state)?;
1023 }
1024 let active_turn = turns.iter().rev().find(|turn| turn.disposition.is_none());
1025 let last_turn = turns.last();
1026 let live = if let Some(turn) = active_turn {
1027 json!({ "state": "turn_running", "turnRef": turn.turn_ref })
1028 } else if let Some(turn) = last_turn {
1029 match turn.disposition.as_deref() {
1030 Some("completed") => json!({ "state": "turn_completed", "turnRef": turn.turn_ref }),
1031 Some("failed") => json!({
1032 "state": "blocked",
1033 "turnRef": turn.turn_ref,
1034 "reason": "agent_turn_failed",
1035 }),
1036 Some("timed_out") => json!({
1037 "state": "blocked",
1038 "turnRef": turn.turn_ref,
1039 "reason": "provider_turn_timed_out",
1040 }),
1041 Some("owner_interrupted") => json!({
1042 "state": "blocked",
1043 "turnRef": turn.turn_ref,
1044 "reason": "owner_interrupted",
1045 }),
1046 _ => Value::Null,
1047 }
1048 } else {
1049 Value::Null
1050 };
1051 let turn_records = turns
1052 .iter()
1053 .zip(assistant_texts)
1054 .map(|(turn, assistant_text)| {
1055 let user_message_key = message_key(&turn.turn_ref, "user");
1056 let assistant_message_key = message_key(&turn.turn_ref, "assistant");
1057 let assistant_segments = if assistant_text.is_empty() {
1058 Vec::new()
1059 } else {
1060 vec![json!({
1061 "key": assistant_message_key,
1062 "text": assistant_text,
1063 })]
1064 };
1065 json!({
1066 "schema": "openagents.desktop.local_turn_record.v1",
1067 "threadRef": params.thread_ref,
1068 "turnRef": turn.turn_ref,
1069 "lane": turn.lane,
1070 "userMessageKey": user_message_key,
1071 "assistantMessageKey": assistant_message_key,
1072 "accountRef": turn.account_ref,
1073 "providerSessionRef": turn.provider_session_ref,
1074 "model": turn.model,
1075 "phase": turn.phase,
1076 "persistedCursor": assistant_text.len(),
1077 "assistantText": assistant_text,
1078 "assistantSegments": assistant_segments,
1079 "recoveryGeneration": generation,
1080 "disposition": turn.disposition,
1081 "createdAt": turn.created_at,
1082 "updatedAt": turn.updated_at,
1083 })
1084 })
1085 .collect::<Vec<_>>();
1086 Ok(json!({
1087 "present": true,
1088 "revision": revision,
1089 "live": live,
1090 "turns": turn_records,
1091 }))
1092}
1093
1094async fn interrupt_turn(
1095 params: Value,
1096 state: &Rc<RefCell<HostBridgeState>>,
1097 cx: &mut AsyncApp,
1098) -> Result<Value, HostResponseError> {
1099 let params: InterruptTurnParams = decode_params(params)?;
1100 validate_ref(¶ms.thread_ref, "threadRef")?;
1101 if let Some(turn_ref) = params.turn_ref.as_deref() {
1102 validate_ref(turn_ref, "turnRef")?;
1103 }
1104 let (conversation, turn_ref) = {
1105 let bridge = state.borrow();
1106 let host_thread = bridge
1107 .threads
1108 .iter()
1109 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
1110 .ok_or_else(|| unavailable("The requested Agent thread is not bound."))?;
1111 let turn_ref = params.turn_ref.clone().or_else(|| {
1112 host_thread
1113 .turns
1114 .iter()
1115 .rev()
1116 .find(|turn| turn.disposition.is_none())
1117 .map(|turn| turn.turn_ref.clone())
1118 });
1119 (
1120 host_thread
1121 .conversation
1122 .as_ref()
1123 .and_then(WeakEntity::upgrade)
1124 .ok_or_else(|| unavailable("The requested Agent thread was closed."))?,
1125 turn_ref,
1126 )
1127 };
1128 let Some(turn_ref) = turn_ref else {
1129 return Ok(json!({ "interrupted": false }));
1130 };
1131 let thread = wait_for_root_thread(&conversation, cx).await?;
1132 let cancel = thread.update(cx, |thread, cx| {
1133 if thread.status() != ThreadStatus::Generating {
1134 None
1135 } else {
1136 Some(thread.cancel(cx))
1137 }
1138 });
1139 let Some(cancel) = cancel else {
1140 return Ok(json!({ "interrupted": false }));
1141 };
1142 cancel.await;
1143 let end_entry_index = cx.update(|cx| thread.read(cx).entries().len());
1144 {
1145 let mut bridge = state.borrow_mut();
1146 if let Some(host_thread) = bridge
1147 .threads
1148 .iter_mut()
1149 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
1150 && let Some(turn) = host_thread
1151 .turns
1152 .iter_mut()
1153 .find(|turn| turn.turn_ref == turn_ref)
1154 {
1155 turn.phase = "interrupted".to_string();
1156 turn.disposition = Some("owner_interrupted".to_string());
1157 turn.end_entry_index = Some(end_entry_index);
1158 turn.updated_at = Utc::now().to_rfc3339();
1159 host_thread.revision += 1;
1160 }
1161 }
1162 persist_state(state)?;
1163 Ok(json!({ "interrupted": true }))
1164}
1165
1166/// OMEGA-DELTA-0045. Write a host-authored note into the thread the owner
1167/// reads.
1168///
1169/// This used to refuse — `unavailable("Agent threads do not expose an
1170/// owner-visible system-note authority.")` — because `AgentThreadEntry` had no
1171/// variant a non-model disclosure could be. The refusal was typed and honest,
1172/// and it was still the rc11 silence: the engine emits a provider-handoff note
1173/// naming both lanes, the host dropped it, and a run that changed which model
1174/// was spending the owner's budget left nothing in the transcript. FA-07 gate
1175/// 5 exists to forbid exactly that.
1176///
1177/// The note goes to the thread named in `threadRef` and nowhere else, so a
1178/// handoff addressed to the *target* thread cannot be filed against the source
1179/// one. `noteRef` makes the append idempotent: the engine may retry after a
1180/// response it never saw, and the owner must not be shown the same disclosure
1181/// twice — nor a rewritten one, which is why the id wins over the newer text.
1182///
1183/// The idempotence is scoped to the live thread, not to the correlation
1184/// journal. That is the scope that matters: the note is an entry in the
1185/// thread, so a thread that is gone has no owner reading it and nothing to
1186/// double.
1187async fn append_system_note(
1188 params: Value,
1189 state: &Rc<RefCell<HostBridgeState>>,
1190 cx: &mut AsyncApp,
1191) -> Result<Value, HostResponseError> {
1192 let params: AppendSystemNoteParams = decode_params(params)?;
1193 validate_ref(¶ms.thread_ref, "threadRef")?;
1194 validate_ref(¶ms.note_ref, "noteRef")?;
1195 if params.text.is_empty() {
1196 return Err(invalid("text must not be empty."));
1197 }
1198 let conversation = {
1199 let bridge = state.borrow();
1200 bridge
1201 .threads
1202 .iter()
1203 .find(|thread| thread.thread_id.to_key_string() == params.thread_ref)
1204 .ok_or_else(|| unavailable("The requested Agent thread is not bound."))?
1205 .conversation
1206 .as_ref()
1207 .and_then(WeakEntity::upgrade)
1208 .ok_or_else(|| unavailable("The requested Agent thread was closed."))?
1209 };
1210 let thread = wait_for_root_thread(&conversation, cx).await?;
1211 let appended = thread.update(cx, |thread, cx| {
1212 thread.push_system_note(
1213 acp_thread::SystemNote {
1214 id: acp_thread::SystemNoteId(params.note_ref.as_str().into()),
1215 text: params.text.into(),
1216 },
1217 cx,
1218 )
1219 });
1220 Ok(json!({ "appended": appended }))
1221}
1222
1223async fn wait_for_root_thread(
1224 conversation: &Entity<ConversationView>,
1225 cx: &mut AsyncApp,
1226) -> Result<Entity<acp_thread::AcpThread>, HostResponseError> {
1227 for _ in 0..THREAD_CONNECT_ATTEMPTS {
1228 if let Some(thread) = cx.update(|cx| conversation.read(cx).root_thread(cx)) {
1229 return Ok(thread);
1230 }
1231 cx.background_executor()
1232 .timer(THREAD_CONNECT_INTERVAL)
1233 .await;
1234 }
1235 Err(unavailable(
1236 "The Agent thread did not connect before the host deadline.",
1237 ))
1238}
1239
1240fn require_workspace(
1241 workspace_ref: &str,
1242 state: &Rc<RefCell<HostBridgeState>>,
1243) -> Result<WorkspaceBinding, HostResponseError> {
1244 let binding = state
1245 .borrow()
1246 .workspace
1247 .clone()
1248 .ok_or_else(|| unavailable("No Omega workspace is bound."))?;
1249 if binding.workspace_ref != workspace_ref {
1250 return Err(unavailable("The requested workspace is not bound."));
1251 }
1252 if binding.workspace.upgrade().is_none() {
1253 return Err(unavailable("The bound workspace was closed."));
1254 }
1255 Ok(binding)
1256}
1257
1258fn require_loaded_journal(state: &Rc<RefCell<HostBridgeState>>) -> Result<(), HostResponseError> {
1259 if let Some(error) = state.borrow().load_error.as_deref() {
1260 Err(internal(format!(
1261 "Omega Full Auto host correlation could not be loaded: {error}"
1262 )))
1263 } else {
1264 Ok(())
1265 }
1266}
1267
1268fn rebind_persisted_threads(
1269 workspace_ref: &str,
1270 state: &Rc<RefCell<HostBridgeState>>,
1271 cx: &mut AsyncApp,
1272) -> Result<(), HostResponseError> {
1273 let thread_ids = state
1274 .borrow()
1275 .threads
1276 .iter()
1277 .filter(|thread| {
1278 thread.workspace_ref == workspace_ref
1279 && thread
1280 .conversation
1281 .as_ref()
1282 .and_then(WeakEntity::upgrade)
1283 .is_none()
1284 })
1285 .map(|thread| (thread.thread_id, thread.lane.clone()))
1286 .collect::<Vec<_>>();
1287 if thread_ids.is_empty() {
1288 return Ok(());
1289 }
1290
1291 let binding = require_workspace(workspace_ref, state)?;
1292 let workspace = binding
1293 .workspace
1294 .upgrade()
1295 .ok_or_else(|| unavailable("The bound workspace was closed."))?;
1296 let rebound = binding
1297 .window
1298 .update(cx, |_root, window, cx| {
1299 let panel = workspace
1300 .read(cx)
1301 .panel::<AgentPanel>(cx)
1302 .ok_or_else(|| unavailable("The workspace Agent panel is unavailable."))?;
1303 let mut rebound = Vec::with_capacity(thread_ids.len());
1304 for (thread_id, lane) in thread_ids {
1305 let agent = agent_for_lane(&lane)?;
1306 let conversation = panel.update(cx, |panel, cx| {
1307 if panel.conversation_view_for_id(&thread_id, cx).is_none() {
1308 panel.load_agent_thread(
1309 agent,
1310 thread_id,
1311 None,
1312 None,
1313 false,
1314 AgentThreadSource::AgentPanel,
1315 window,
1316 cx,
1317 );
1318 }
1319 panel.conversation_view_for_id(&thread_id, cx).cloned()
1320 });
1321 let conversation = conversation.ok_or_else(|| {
1322 unavailable("The persisted Full Auto Agent thread could not be restored.")
1323 })?;
1324 rebound.push((thread_id, conversation.downgrade()));
1325 }
1326 Ok(rebound)
1327 })
1328 .map_err(|error| unavailable(format!("The workspace window is unavailable: {error}")))??;
1329
1330 let mut bridge = state.borrow_mut();
1331 for (thread_id, conversation) in rebound {
1332 if let Some(thread) = bridge
1333 .threads
1334 .iter_mut()
1335 .find(|thread| thread.thread_id == thread_id)
1336 {
1337 thread.conversation = Some(conversation);
1338 }
1339 }
1340 Ok(())
1341}
1342
1343fn load_correlation_journal(path: &Path) -> anyhow::Result<Vec<HostThread>> {
1344 let bytes = match std::fs::read(path) {
1345 Ok(bytes) => bytes,
1346 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
1347 Err(error) => return Err(error.into()),
1348 };
1349 let journal: CorrelationJournal = serde_json::from_slice(&bytes)?;
1350 anyhow::ensure!(
1351 journal.schema == CORRELATION_SCHEMA,
1352 "unsupported correlation schema {}",
1353 journal.schema
1354 );
1355 journal
1356 .threads
1357 .into_iter()
1358 .map(|thread| {
1359 Ok(HostThread {
1360 workspace_ref: thread.workspace_ref,
1361 lane: thread.lane,
1362 operation_ref: thread.operation_ref,
1363 thread_id: ThreadId::from_key_string(&thread.thread_ref)?,
1364 conversation: None,
1365 turns: thread.turns,
1366 revision: thread.revision,
1367 })
1368 })
1369 .collect()
1370}
1371
1372fn persist_state(state: &Rc<RefCell<HostBridgeState>>) -> Result<(), HostResponseError> {
1373 persist_correlation_journal(&state.borrow()).map_err(|error| {
1374 internal(format!(
1375 "Omega Full Auto host correlation could not be persisted: {error:#}"
1376 ))
1377 })
1378}
1379
1380fn record_turn_completion(
1381 state: &Rc<RefCell<HostBridgeState>>,
1382 turn_ref: &str,
1383 end_entry_index: usize,
1384 outcome: HostTurnOutcome,
1385) -> anyhow::Result<()> {
1386 {
1387 let mut bridge = state.borrow_mut();
1388 if let Some(host_thread) = bridge
1389 .threads
1390 .iter_mut()
1391 .find(|thread| thread.turns.iter().any(|turn| turn.turn_ref == turn_ref))
1392 {
1393 let changed = if let Some(turn) = host_thread
1394 .turns
1395 .iter_mut()
1396 .find(|turn| turn.turn_ref == turn_ref)
1397 {
1398 if turn.disposition.is_none() {
1399 match outcome {
1400 HostTurnOutcome::Completed => {
1401 turn.phase = "completed".to_string();
1402 turn.disposition = Some("completed".to_string());
1403 }
1404 HostTurnOutcome::Failed => {
1405 turn.phase = "failed".to_string();
1406 turn.disposition = Some("failed".to_string());
1407 }
1408 HostTurnOutcome::TimedOut => {
1409 turn.phase = "failed".to_string();
1410 turn.disposition = Some("timed_out".to_string());
1411 }
1412 }
1413 turn.end_entry_index = Some(end_entry_index);
1414 turn.updated_at = Utc::now().to_rfc3339();
1415 true
1416 } else {
1417 false
1418 }
1419 } else {
1420 false
1421 };
1422 if changed {
1423 host_thread.revision += 1;
1424 }
1425 }
1426 }
1427 persist_correlation_journal(&state.borrow())
1428}
1429
1430fn turn_timeout_for_lane(lane: &str) -> Option<Duration> {
1431 (lane == CLAUDE_LOCAL_LANE).then_some(CLAUDE_TURN_TIMEOUT)
1432}
1433
1434fn persist_correlation_journal(state: &HostBridgeState) -> anyhow::Result<()> {
1435 // OMEGA-DELTA-0021. Every mutation of `state.threads` reaches disk through
1436 // here, so republishing here is what keeps the lane index and the durable
1437 // journal from drifting apart within a session.
1438 republish_engine_lane_runs(&state.threads);
1439 let journal = CorrelationJournal {
1440 schema: CORRELATION_SCHEMA.to_string(),
1441 threads: state
1442 .threads
1443 .iter()
1444 .map(|thread| PersistedHostThread {
1445 workspace_ref: thread.workspace_ref.clone(),
1446 lane: thread.lane.clone(),
1447 operation_ref: thread.operation_ref.clone(),
1448 thread_ref: thread.thread_id.to_key_string(),
1449 turns: thread.turns.clone(),
1450 revision: thread.revision,
1451 })
1452 .collect(),
1453 };
1454 let parent = state
1455 .correlation_path
1456 .parent()
1457 .ok_or_else(|| anyhow::anyhow!("correlation path has no parent"))?;
1458 std::fs::create_dir_all(parent)?;
1459 let temporary_path = state.correlation_path.with_extension("json.tmp");
1460 let bytes = serde_json::to_vec_pretty(&journal)?;
1461 let mut temporary_file = std::fs::OpenOptions::new()
1462 .create(true)
1463 .truncate(true)
1464 .write(true)
1465 .open(&temporary_path)?;
1466 temporary_file.write_all(&bytes)?;
1467 temporary_file.sync_all()?;
1468 std::fs::rename(&temporary_path, &state.correlation_path)?;
1469 std::fs::File::open(parent)?.sync_all()?;
1470 Ok(())
1471}
1472
1473fn agent_for_lane(lane: &str) -> Result<Agent, HostResponseError> {
1474 match lane {
1475 CODEX_LOCAL_LANE => Ok(Agent::Custom {
1476 id: CODEX_AGENT_ID.into(),
1477 }),
1478 CLAUDE_LOCAL_LANE => Ok(Agent::Custom {
1479 id: CLAUDE_AGENT_ID.into(),
1480 }),
1481 _ => Err(unsupported(format!("The {lane} lane is not supported."))),
1482 }
1483}
1484
1485fn is_supported_lane(lane: &str) -> bool {
1486 matches!(lane, CODEX_LOCAL_LANE | CLAUDE_LOCAL_LANE)
1487}
1488
1489fn external_agent_lane_readiness(
1490 state: &Rc<RefCell<HostBridgeState>>,
1491 cx: &AsyncApp,
1492 agent_id: &'static str,
1493 dispatch_thread_exists: bool,
1494) -> Result<(bool, &'static str), HostResponseError> {
1495 let binding = state
1496 .borrow()
1497 .workspace
1498 .clone()
1499 .ok_or_else(|| unavailable("No Omega workspace is bound."))?;
1500 let workspace = binding
1501 .workspace
1502 .upgrade()
1503 .ok_or_else(|| unavailable("The bound workspace was closed."))?;
1504 let agent = Agent::Custom {
1505 id: agent_id.into(),
1506 };
1507 let (registered, connection_status) = cx.update(|cx| {
1508 let project = workspace.read(cx).project().clone();
1509 let registered = project
1510 .read(cx)
1511 .agent_server_store()
1512 .read(cx)
1513 .external_agents()
1514 .any(|registered_agent_id| registered_agent_id.as_ref() == agent_id);
1515 let connection_status = workspace
1516 .read(cx)
1517 .panel::<AgentPanel>(cx)
1518 .map(|panel| {
1519 panel
1520 .read(cx)
1521 .connection_store()
1522 .read(cx)
1523 .connection_status(&agent, cx)
1524 })
1525 .unwrap_or(AgentConnectionStatus::Disconnected);
1526 (registered, connection_status)
1527 });
1528 Ok(external_agent_authority_state(
1529 registered,
1530 connection_status,
1531 dispatch_thread_exists,
1532 ))
1533}
1534
1535fn external_agent_authority_state(
1536 registered: bool,
1537 connection_status: AgentConnectionStatus,
1538 dispatch_thread_exists: bool,
1539) -> (bool, &'static str) {
1540 if !registered {
1541 return (false, "unavailable");
1542 }
1543 match connection_status {
1544 AgentConnectionStatus::Connected => (true, "available"),
1545 // Dispatch waits for this exact retained thread's root ACP session, so
1546 // treating its bootstrap as unavailable would strand the run at zero turns.
1547 AgentConnectionStatus::Connecting if dispatch_thread_exists => (true, "available"),
1548 AgentConnectionStatus::Connecting => (false, "connecting"),
1549 AgentConnectionStatus::Disconnected => (true, "available"),
1550 }
1551}
1552
1553fn decode_params<T: for<'de> Deserialize<'de>>(params: Value) -> Result<T, HostResponseError> {
1554 serde_json::from_value(params)
1555 .map_err(|error| invalid(format!("Invalid host request parameters: {error}")))
1556}
1557
1558fn validate_ref(value: &str, name: &str) -> Result<(), HostResponseError> {
1559 if value.is_empty() || value.len() > 180 {
1560 Err(invalid(format!("{name} must contain 1 to 180 bytes.")))
1561 } else {
1562 Ok(())
1563 }
1564}
1565
1566fn validate_lane(value: &str) -> Result<(), HostResponseError> {
1567 if value.is_empty() || value.len() > 64 {
1568 Err(invalid("lane must contain 1 to 64 bytes."))
1569 } else {
1570 Ok(())
1571 }
1572}
1573
1574fn truncate_utf8(value: &str, max_bytes: usize) -> String {
1575 if value.len() <= max_bytes {
1576 return value.to_string();
1577 }
1578 let mut boundary = max_bytes;
1579 while !value.is_char_boundary(boundary) {
1580 boundary -= 1;
1581 }
1582 value[..boundary].to_string()
1583}
1584
1585fn message_key(turn_ref: &str, role: &str) -> String {
1586 let name = format!("{turn_ref}.{role}");
1587 format!(
1588 "omega.{role}.{}",
1589 uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_OID, name.as_bytes())
1590 )
1591}
1592
1593fn invalid(message: impl Into<String>) -> HostResponseError {
1594 HostResponseError {
1595 code: HostResponseErrorCode::InvalidRequest,
1596 message: message.into(),
1597 }
1598}
1599
1600fn unsupported(message: impl Into<String>) -> HostResponseError {
1601 HostResponseError {
1602 code: HostResponseErrorCode::Unsupported,
1603 message: message.into(),
1604 }
1605}
1606
1607fn unavailable(message: impl Into<String>) -> HostResponseError {
1608 HostResponseError::unavailable(message)
1609}
1610
1611fn internal(message: impl Into<String>) -> HostResponseError {
1612 HostResponseError {
1613 code: HostResponseErrorCode::Internal,
1614 message: message.into(),
1615 }
1616}
1617
1618#[cfg(test)]
1619mod tests {
1620 use super::*;
1621 use omega_front_door::{ExecutorClass, ExecutorDisclosure};
1622 use tempfile::tempdir;
1623
1624 /// `ENGINE_LANE_RUNS` is process-wide and every publication replaces it
1625 /// wholesale, so two tests publishing at once would clobber each other's
1626 /// assertions. Every test that publishes takes this first.
1627 static LANE_INDEX_GUARD: Mutex<()> = Mutex::new(());
1628
1629 fn lane_index_guard() -> std::sync::MutexGuard<'static, ()> {
1630 LANE_INDEX_GUARD.lock().unwrap_or_else(|poisoned| {
1631 LANE_INDEX_GUARD.clear_poison();
1632 poisoned.into_inner()
1633 })
1634 }
1635
1636 #[test]
1637 fn validates_bounded_refs_and_params() {
1638 assert!(validate_ref("turn.full-auto.1", "turnRef").is_ok());
1639 assert!(validate_ref("", "turnRef").is_err());
1640 assert!(validate_ref(&"x".repeat(181), "turnRef").is_err());
1641 assert!(decode_params::<ResolveWorkspaceParams>(json!({})).is_ok());
1642 assert!(decode_params::<ResolveWorkspaceParams>(json!({ "extra": true })).is_err());
1643 assert!(decode_params::<ResolveSyncSessionParams>(json!({})).is_ok());
1644 assert!(decode_params::<ResolveSyncSessionParams>(json!({ "token": "no" })).is_err());
1645 }
1646
1647 #[test]
1648 fn sync_session_host_result_is_unavailable_or_runtime_only_verified_material() {
1649 assert_eq!(sync_session_result(None), json!({ "available": false }));
1650 assert_eq!(
1651 sync_session_result(Some(VerifiedOpenAgentsSession {
1652 base_url: "https://openagents.com".to_string(),
1653 access_token: "runtime-only-fixture".to_string(),
1654 })),
1655 json!({
1656 "available": true,
1657 "baseUrl": "https://openagents.com",
1658 "accessToken": "runtime-only-fixture",
1659 })
1660 );
1661 }
1662
1663 #[test]
1664 fn assistant_evidence_truncation_preserves_utf8_boundaries() {
1665 let text = format!("{}é", "x".repeat(MAX_ASSISTANT_TEXT_BYTES));
1666 let truncated = truncate_utf8(&text, MAX_ASSISTANT_TEXT_BYTES + 1);
1667 assert!(truncated.is_char_boundary(truncated.len()));
1668 assert!(truncated.len() <= MAX_ASSISTANT_TEXT_BYTES + 1);
1669 }
1670
1671 #[test]
1672 fn evidence_keys_remain_bounded_for_maximum_turn_refs() {
1673 let turn_ref = format!("{}é", "x".repeat(178));
1674 assert_eq!(turn_ref.len(), 180);
1675
1676 let user_key = message_key(&turn_ref, "user");
1677 let assistant_key = message_key(&turn_ref, "assistant");
1678
1679 assert!(user_key.len() <= 180);
1680 assert!(assistant_key.len() <= 180);
1681 assert_ne!(user_key, assistant_key);
1682 assert_eq!(user_key, message_key(&turn_ref, "user"));
1683 }
1684
1685 #[test]
1686 fn correlation_journal_round_trips_without_conversation_entities() {
1687 let _lane_index = lane_index_guard();
1688 let directory = tempdir().expect("tempdir");
1689 let path = directory.path().join(CORRELATION_FILE);
1690 let thread_id = ThreadId::new();
1691 let state = HostBridgeState {
1692 workspace: None,
1693 threads: vec![HostThread {
1694 workspace_ref: "workspace.omega.supervised".to_string(),
1695 lane: CLAUDE_LOCAL_LANE.to_string(),
1696 operation_ref: "operation.full-auto.1".to_string(),
1697 thread_id,
1698 conversation: None,
1699 turns: vec![HostTurn {
1700 turn_ref: "turn.full-auto.1".to_string(),
1701 lane: CODEX_LOCAL_LANE.to_string(),
1702 account_ref: Some("account.owner".to_string()),
1703 model: None,
1704 provider_session_ref: "session.native.1".to_string(),
1705 start_entry_index: 4,
1706 end_entry_index: Some(8),
1707 phase: "completed".to_string(),
1708 disposition: Some("completed".to_string()),
1709 created_at: "2026-07-24T12:00:00Z".to_string(),
1710 updated_at: "2026-07-24T12:01:00Z".to_string(),
1711 }],
1712 revision: 3,
1713 }],
1714 correlation_path: path.clone(),
1715 load_error: None,
1716 sarah_conversation: None,
1717 };
1718
1719 persist_correlation_journal(&state).expect("persist correlation journal");
1720 let restored = load_correlation_journal(&path).expect("load correlation journal");
1721
1722 assert_eq!(restored.len(), 1);
1723 assert_eq!(restored[0].thread_id, thread_id);
1724 assert_eq!(restored[0].lane, CLAUDE_LOCAL_LANE);
1725 assert_eq!(restored[0].operation_ref, "operation.full-auto.1");
1726 assert!(restored[0].conversation.is_none());
1727 assert_eq!(restored[0].revision, 3);
1728 assert_eq!(restored[0].turns[0].turn_ref, "turn.full-auto.1");
1729 assert_eq!(
1730 restored[0].turns[0].disposition.as_deref(),
1731 Some("completed")
1732 );
1733 }
1734
1735 /// OMEGA-DELTA-0021, omega#77's exit: a thread still names its executor
1736 /// after a restart.
1737 ///
1738 /// The restart is real rather than mimed. The lane index is process state,
1739 /// so it is emptied here exactly as a process exit empties it, and the only
1740 /// thing that survives into the assertion is the file on disk. A disclosure
1741 /// that lived only in memory fails this test at the `engine_lane_run` call.
1742 #[test]
1743 fn a_restarted_process_still_discloses_the_lane_that_owns_a_thread() {
1744 let _lane_index = lane_index_guard();
1745 let directory = tempdir().expect("tempdir");
1746 let path = directory.path().join(CORRELATION_FILE);
1747 let thread_id = ThreadId::new();
1748 let state = HostBridgeState {
1749 workspace: None,
1750 threads: vec![HostThread {
1751 workspace_ref: SUPERVISED_WORKSPACE_REF.to_string(),
1752 lane: CODEX_LOCAL_LANE.to_string(),
1753 operation_ref: "operation.full-auto.77".to_string(),
1754 thread_id,
1755 conversation: None,
1756 turns: Vec::new(),
1757 revision: 1,
1758 }],
1759 correlation_path: path.clone(),
1760 load_error: None,
1761 sarah_conversation: None,
1762 };
1763 persist_correlation_journal(&state).expect("persist correlation journal");
1764
1765 // The process ends here. Everything below is a cold start.
1766 republish_engine_lane_runs(&[]);
1767 assert_eq!(
1768 engine_lane_run(thread_id),
1769 None,
1770 "a cold process must know nothing until it reads the journal"
1771 );
1772
1773 let restored = load_correlation_journal(&path).expect("load correlation journal");
1774 republish_engine_lane_runs(&restored);
1775
1776 let run = engine_lane_run(thread_id).expect("the reloaded journal names this thread's run");
1777 assert_eq!(run, "operation.full-auto.77");
1778
1779 let disclosure = crate::omega_executor_disclosure::delegated_to_run(
1780 ExecutorDisclosure {
1781 class: ExecutorClass::ExternalAcp,
1782 agent_id: CODEX_AGENT_ID.to_string(),
1783 provider: None,
1784 model: None,
1785 run_ref: None,
1786 route: Some(omega_front_door::RouteReason::PinHonored),
1787 },
1788 run,
1789 );
1790 assert_eq!(disclosure.class, ExecutorClass::EngineLane);
1791 assert!(disclosure.is_coherent());
1792
1793 let line = disclosure.label();
1794 assert!(line.contains(CODEX_AGENT_ID), "{line:?}");
1795 assert!(line.contains("engine_lane"), "{line:?}");
1796 assert!(line.contains("operation.full-auto.77"), "{line:?}");
1797 }
1798
1799 /// A thread the user started themselves is not a lane run, and must not be
1800 /// disclosed as one. Without this, the index's default answer would
1801 /// silently attribute every hand-driven thread to Full Auto.
1802 #[test]
1803 fn a_thread_that_is_not_a_lane_run_is_not_disclosed_as_one() {
1804 let _lane_index = lane_index_guard();
1805 let hand_started = ThreadId::new();
1806 republish_engine_lane_runs(&[HostThread {
1807 workspace_ref: SUPERVISED_WORKSPACE_REF.to_string(),
1808 lane: CLAUDE_LOCAL_LANE.to_string(),
1809 operation_ref: "operation.full-auto.1".to_string(),
1810 thread_id: ThreadId::new(),
1811 conversation: None,
1812 turns: Vec::new(),
1813 revision: 1,
1814 }]);
1815 assert_eq!(engine_lane_run(hand_started), None);
1816 }
1817
1818 #[test]
1819 fn completed_turn_is_persisted_after_releasing_mutable_state_borrow() {
1820 let _lane_index = lane_index_guard();
1821 let directory = tempdir().expect("tempdir");
1822 let path = directory.path().join(CORRELATION_FILE);
1823 let thread_id = ThreadId::new();
1824 let state = Rc::new(RefCell::new(HostBridgeState {
1825 workspace: None,
1826 threads: vec![HostThread {
1827 workspace_ref: "workspace.omega.supervised".to_string(),
1828 lane: CODEX_LOCAL_LANE.to_string(),
1829 operation_ref: "operation.full-auto.1".to_string(),
1830 thread_id,
1831 conversation: None,
1832 turns: vec![HostTurn {
1833 turn_ref: "turn.full-auto.1".to_string(),
1834 lane: CODEX_LOCAL_LANE.to_string(),
1835 account_ref: None,
1836 model: None,
1837 provider_session_ref: "session.native.1".to_string(),
1838 start_entry_index: 2,
1839 end_entry_index: None,
1840 phase: "streaming".to_string(),
1841 disposition: None,
1842 created_at: "2026-07-24T12:00:00Z".to_string(),
1843 updated_at: "2026-07-24T12:00:00Z".to_string(),
1844 }],
1845 revision: 1,
1846 }],
1847 correlation_path: path.clone(),
1848 load_error: None,
1849 sarah_conversation: None,
1850 }));
1851
1852 record_turn_completion(&state, "turn.full-auto.1", 7, HostTurnOutcome::Completed)
1853 .expect("record completed turn");
1854
1855 let restored = load_correlation_journal(&path).expect("load correlation journal");
1856 let turn = &restored[0].turns[0];
1857 assert_eq!(restored[0].revision, 2);
1858 assert_eq!(turn.phase, "completed");
1859 assert_eq!(turn.disposition.as_deref(), Some("completed"));
1860 assert_eq!(turn.end_entry_index, Some(7));
1861 }
1862
1863 #[test]
1864 fn claude_turns_have_a_bounded_host_deadline() {
1865 assert_eq!(
1866 turn_timeout_for_lane(CLAUDE_LOCAL_LANE),
1867 Some(CLAUDE_TURN_TIMEOUT)
1868 );
1869 assert_eq!(turn_timeout_for_lane(CODEX_LOCAL_LANE), None);
1870 }
1871
1872 #[test]
1873 fn timed_out_turn_is_persisted_as_a_terminal_failure() {
1874 let _lane_index = lane_index_guard();
1875 let directory = tempdir().expect("tempdir");
1876 let path = directory.path().join(CORRELATION_FILE);
1877 let thread_id = ThreadId::new();
1878 let state = Rc::new(RefCell::new(HostBridgeState {
1879 workspace: None,
1880 threads: vec![HostThread {
1881 workspace_ref: "workspace.omega.supervised".to_string(),
1882 lane: CLAUDE_LOCAL_LANE.to_string(),
1883 operation_ref: "operation.full-auto.timeout".to_string(),
1884 thread_id,
1885 conversation: None,
1886 turns: vec![HostTurn {
1887 turn_ref: "turn.full-auto.timeout".to_string(),
1888 lane: CLAUDE_LOCAL_LANE.to_string(),
1889 account_ref: None,
1890 model: None,
1891 provider_session_ref: "session.native.timeout".to_string(),
1892 start_entry_index: 2,
1893 end_entry_index: None,
1894 phase: "streaming".to_string(),
1895 disposition: None,
1896 created_at: "2026-07-24T12:00:00Z".to_string(),
1897 updated_at: "2026-07-24T12:00:00Z".to_string(),
1898 }],
1899 revision: 1,
1900 }],
1901 correlation_path: path.clone(),
1902 load_error: None,
1903 sarah_conversation: None,
1904 }));
1905
1906 record_turn_completion(
1907 &state,
1908 "turn.full-auto.timeout",
1909 5,
1910 HostTurnOutcome::TimedOut,
1911 )
1912 .expect("record timed-out turn");
1913
1914 let restored = load_correlation_journal(&path).expect("load correlation journal");
1915 let turn = &restored[0].turns[0];
1916 assert_eq!(turn.phase, "failed");
1917 assert_eq!(turn.disposition.as_deref(), Some("timed_out"));
1918 assert_eq!(turn.end_entry_index, Some(5));
1919 }
1920
1921 #[test]
1922 fn correlation_journal_rejects_unknown_schema_and_invalid_thread_refs() {
1923 let directory = tempdir().expect("tempdir");
1924 let path = directory.path().join(CORRELATION_FILE);
1925 std::fs::write(
1926 &path,
1927 serde_json::to_vec(&json!({
1928 "schema": "openagents.omega.full_auto_host_correlation.v0",
1929 "threads": [],
1930 }))
1931 .expect("serialize fixture"),
1932 )
1933 .expect("write fixture");
1934 assert!(load_correlation_journal(&path).is_err());
1935
1936 std::fs::write(
1937 &path,
1938 serde_json::to_vec(&json!({
1939 "schema": CORRELATION_SCHEMA,
1940 "threads": [{
1941 "workspaceRef": "workspace.omega.supervised",
1942 "lane": "codex-local",
1943 "operationRef": "operation.full-auto.1",
1944 "threadRef": "not-a-uuid",
1945 "turns": [],
1946 "revision": 1,
1947 }],
1948 }))
1949 .expect("serialize fixture"),
1950 )
1951 .expect("write fixture");
1952 assert!(load_correlation_journal(&path).is_err());
1953 }
1954
1955 #[test]
1956 fn local_lanes_resolve_to_their_real_agent_authorities() {
1957 assert_eq!(
1958 agent_for_lane(CODEX_LOCAL_LANE).expect("codex lane"),
1959 Agent::Custom {
1960 id: CODEX_AGENT_ID.into(),
1961 }
1962 );
1963 assert_eq!(
1964 agent_for_lane(CLAUDE_LOCAL_LANE).expect("claude lane"),
1965 Agent::Custom {
1966 id: CLAUDE_AGENT_ID.into(),
1967 }
1968 );
1969 assert!(agent_for_lane("claude-cloud").is_err());
1970 }
1971
1972 #[test]
1973 fn external_agent_readiness_allows_dispatch_to_bootstrap_its_connection() {
1974 assert_eq!(
1975 external_agent_authority_state(false, AgentConnectionStatus::Disconnected, false,),
1976 (false, "unavailable")
1977 );
1978 assert_eq!(
1979 external_agent_authority_state(true, AgentConnectionStatus::Disconnected, false,),
1980 (true, "available")
1981 );
1982 assert_eq!(
1983 external_agent_authority_state(true, AgentConnectionStatus::Connecting, true),
1984 (true, "available")
1985 );
1986 assert_eq!(
1987 external_agent_authority_state(true, AgentConnectionStatus::Connecting, false),
1988 (false, "connecting")
1989 );
1990 assert_eq!(
1991 external_agent_authority_state(true, AgentConnectionStatus::Connected, true),
1992 (true, "available")
1993 );
1994 assert_eq!(
1995 external_agent_authority_state(true, AgentConnectionStatus::Disconnected, true),
1996 (true, "available")
1997 );
1998 }
1999}
2000