Skip to repository content1691 lines · 72.7 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T05:16:08.537Z 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
issue31_delivery.rs
1//! Hand the omega#47 documents to the issue-31 host pump (omega#49).
2//!
3//! `issue31_adjunct::publish_issue31_host_snapshot` already builds both
4//! documents from one reading of live host state, and until now nothing sent
5//! them anywhere. The pump that gift-wraps owner-private records lives in
6//! `omega_effectd`, which this crate depends on — so the reading crosses the
7//! seam as data rather than by calling upward.
8//!
9//! Two things are deliberately NOT done here:
10//!
11//! - the delivery binding is not stated. The pump adds `recordType`,
12//! `hostPublicKeyHex`, `devicePublicKeyHex`, `grantRef` and
13//! `expectedGeneration`, because those are facts about who may read the
14//! snapshot and this module knows only what the runs are. A document that
15//! arrived already claiming them is refused by the pump.
16//! - no reading is invented. When the host has not observed its Full Auto
17//! state, `latest_issue31_live_reading` is `None` and the pump publishes
18//! nothing, so the phone reads `no_host_projection`. An empty reading — a
19//! host that looked and found no runs — is a `Some` carrying zero runs, and
20//! the phone renders it as a host that is running nothing. The two are not
21//! the same claim and are never collapsed.
22
23use std::sync::{Arc, Mutex, OnceLock};
24
25use omega_effectd::{
26 Issue31HostProjectionDocuments, Issue31HostProjectionRequest, Issue31HostProjectionSource,
27 Issue31ProviderRosterAccount, Issue31ProviderRosterSource,
28};
29use serde_json::Value;
30use sha2::{Digest, Sha256};
31
32use crate::issue31_adjunct::{
33 Issue31FullAutoLiveSources, Issue31HostIdentitySource, Issue31HostProjectionError,
34 publish_issue31_host_snapshot,
35};
36use crate::provider_roster::parse_provider_accounts;
37
38/// This host's clock, in epoch milliseconds. The single reading every
39/// observation is stamped from.
40fn unix_millis() -> u64 {
41 std::time::SystemTime::now()
42 .duration_since(std::time::UNIX_EPOCH)
43 .map(|elapsed| u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX))
44 .unwrap_or_default()
45}
46
47/// One complete reading of the live Full Auto surface, owned rather than
48/// borrowed so it can be cached between the panel that reads it and the pump
49/// that publishes it.
50///
51/// `host_ref` is absent on purpose: the host reference belongs to the device's
52/// grant, not to the run registry, so it is supplied per delivery.
53///
54/// `Default` is deliberately not derived. A default reading is one nobody
55/// measured, and it would carry `generated_at_ms: 0` — a stamp that decodes,
56/// projects, and reads on the phone as a host observation taken in 1970. The
57/// only readings that exist are ones a daemon answered.
58#[derive(Clone, Debug)]
59pub struct Issue31FullAutoReading {
60 /// When the daemon was actually read, in epoch milliseconds.
61 ///
62 /// Private, and there is no production path that sets it (omega#97). The
63 /// host's own clock stamps it once inside `observed`, so a caller cannot
64 /// supply one: a reading whose stamp its author chose is exactly how a
65 /// recorded fixture becomes host authority, and omega#49's exit forbids
66 /// that in as many words. Making it unwritable is stronger than forbidding
67 /// it in a comment, because a comment does not survive a rebase.
68 generated_at_ms: u64,
69 pub host_generation: u64,
70 /// One `get_run` record per run the daemon listed.
71 pub run_details: Vec<Value>,
72 /// The `get_capacity` record.
73 pub capacity: Value,
74 /// One `(get_report, get_receipt)` pair per run with evidence.
75 pub evidence: Vec<(Value, Value)>,
76}
77
78impl Issue31FullAutoReading {
79 /// The only constructor a production path can reach.
80 ///
81 /// `generated_at_ms` is one reading of this host's clock, taken here and
82 /// stamped once. There is no parameter for it, so a client-supplied,
83 /// fixture-supplied, or replayed value cannot enter — it is discarded by
84 /// being inexpressible rather than by being checked.
85 ///
86 /// Taken *after* the daemon answered, on purpose. Stamping before the reads
87 /// would put the stamp behind a run that began during them, and the
88 /// contract refuses a run whose start is newer than the snapshot; stamping
89 /// after can only over-state freshness by the length of one poll, which
90 /// under-states no run's unattended duration.
91 pub fn observed(
92 host_generation: u64,
93 run_details: Vec<Value>,
94 capacity: Value,
95 evidence: Vec<(Value, Value)>,
96 ) -> Self {
97 Self {
98 generated_at_ms: unix_millis(),
99 host_generation,
100 run_details,
101 capacity,
102 evidence,
103 }
104 }
105
106 /// When this host read its daemon.
107 pub fn generated_at_ms(&self) -> u64 {
108 self.generated_at_ms
109 }
110
111 /// A reading stamped at a recorded instant, for tests over captured daemon
112 /// bytes only.
113 ///
114 /// `#[cfg(test)]` is the whole point: the projection has to be testable
115 /// against real recorded `get_run` output at a known instant, and no
116 /// shipped build may contain a way to state a stamp the host did not
117 /// measure. This function does not exist in a release binary.
118 #[cfg(test)]
119 fn at_recorded_instant(
120 generated_at_ms: u64,
121 host_generation: u64,
122 run_details: Vec<Value>,
123 capacity: Value,
124 evidence: Vec<(Value, Value)>,
125 ) -> Self {
126 Self {
127 generated_at_ms,
128 host_generation,
129 run_details,
130 capacity,
131 evidence,
132 }
133 }
134
135 /// A stable identifier for exactly this reading.
136 ///
137 /// The mobile contract binds the detail projection to the snapshot that
138 /// advertised it, so the reference has to change when — and only when —
139 /// the observed state changes. Deriving it from the content means a pump
140 /// pass that saw the same world republishes nothing, and a pass that saw a
141 /// different world cannot reuse the old label.
142 ///
143 /// `handoffs` is passed in rather than held because the handoff ledger is
144 /// durable host state owned by the pump, not part of the panel's reading
145 /// of the daemon (omega#91). It is folded in here all the same: a handoff
146 /// that opened, bound, or ended is a different world, and a snapshot
147 /// reference that did not move would leave the phone holding two different
148 /// details under one label.
149 pub fn snapshot_ref(&self, handoffs: &[Value]) -> String {
150 let mut hasher = Sha256::new();
151 hasher.update(self.host_generation.to_string().as_bytes());
152 for run in &self.run_details {
153 hasher.update(run.to_string().as_bytes());
154 }
155 hasher.update(self.capacity.to_string().as_bytes());
156 for handoff in handoffs {
157 hasher.update(handoff.to_string().as_bytes());
158 }
159 for (report, receipt) in &self.evidence {
160 hasher.update(report.to_string().as_bytes());
161 hasher.update(receipt.to_string().as_bytes());
162 }
163 format!("snapshot.omega.issue31.{:x}", hasher.finalize())
164 .chars()
165 .take(48)
166 .collect()
167 }
168}
169
170/// Build the two omega#47 documents for one admitted device.
171pub fn issue31_host_projection_documents(
172 reading: &Issue31FullAutoReading,
173 request: &Issue31HostProjectionRequest<'_>,
174) -> Result<Issue31HostProjectionDocuments, Issue31HostProjectionError> {
175 // Projected against the exact stamp these documents will carry, not
176 // against the pump's own clock. A handoff opened after this reading was
177 // taken belongs to the next snapshot, and the contract says so: a row whose
178 // request time is newer than `generatedAtMs` is refused (omega#91).
179 let handoffs = request.handoffs.projected(reading.generated_at_ms()).rows;
180 let snapshot_ref = reading.snapshot_ref(&handoffs);
181 let sources = Issue31FullAutoLiveSources {
182 host_ref: request.host_ref,
183 snapshot_ref: &snapshot_ref,
184 generated_at_ms: reading.generated_at_ms(),
185 host_generation: reading.host_generation,
186 run_details: &reading.run_details,
187 capacity: &reading.capacity,
188 handoffs: &handoffs,
189 evidence: &reading.evidence,
190 };
191 // The grant is what makes this reader's role active. Without one the
192 // snapshot states an unknown role and offers no actions, which is the
193 // contract's way of saying "this host cannot presently vouch for you".
194 let identity = Issue31HostIdentitySource {
195 source_ref: "source.omega.issue31-pairing",
196 observed_at_ms: reading.generated_at_ms(),
197 owner_grant_ref: Some(request.grant_ref),
198 record_refs: &["record.omega.host-announcement", "record.omega.owner-grant"],
199 permitted_action_refs: &[
200 "action.omega.device.renew",
201 "action.omega.device.revoke",
202 ],
203 };
204 let publication = publish_issue31_host_snapshot(&sources, &identity)?;
205 Ok(Issue31HostProjectionDocuments {
206 host: publication.host_document,
207 detail: publication.detail_document,
208 })
209}
210
211fn reading_cache() -> &'static Mutex<Option<Issue31FullAutoReading>> {
212 static CACHE: OnceLock<Mutex<Option<Issue31FullAutoReading>>> = OnceLock::new();
213 CACHE.get_or_init(|| Mutex::new(None))
214}
215
216/// Record what the host just observed. Called by whoever polls the daemon.
217pub fn set_issue31_live_reading(reading: Issue31FullAutoReading) {
218 if let Ok(mut cache) = reading_cache().lock() {
219 *cache = Some(reading);
220 }
221}
222
223/// The most recent observation, or `None` if the host has never looked.
224pub fn latest_issue31_live_reading() -> Option<Issue31FullAutoReading> {
225 reading_cache().lock().ok().and_then(|cache| cache.clone())
226}
227
228/// How the Sarah host pump reads this host's provider roster (omega#91).
229///
230/// Routed through the exact `parse_provider_accounts` the desktop roster panel
231/// renders, so the account a handoff binds to, and the lane that account
232/// serves, are the same ones Omega shows on the desktop. A handoff that chose
233/// an account the panel does not list would be a second opinion about the
234/// host's own capacity.
235///
236/// `None` when the host has never read its Full Auto state at all — the same
237/// silence `latest_issue31_live_reading` means, propagated rather than
238/// flattened into "this host holds no accounts".
239pub fn issue31_provider_roster_source() -> Issue31ProviderRosterSource {
240 Arc::new(|| {
241 latest_issue31_live_reading().map(|reading| {
242 parse_provider_accounts(&reading.capacity)
243 .into_iter()
244 .map(|account| Issue31ProviderRosterAccount {
245 account_ref: account.account_ref,
246 provider: account.provider,
247 lane_ref: account.lane,
248 readiness: account.readiness,
249 })
250 .collect()
251 })
252 })
253}
254
255/// The source the Sarah host pump publishes from.
256pub fn issue31_host_projection_source() -> Issue31HostProjectionSource {
257 Arc::new(|request| match latest_issue31_live_reading() {
258 None => Ok(None),
259 Some(reading) => issue31_host_projection_documents(&reading, request)
260 .map(Some)
261 .map_err(|error| error.to_string()),
262 })
263}
264
265#[cfg(test)]
266mod tests {
267 use super::*;
268 use omega_effectd::Issue31ProviderHandoffLedger;
269 use serde_json::json;
270
271 /// Exactly what a running `omega-effectd` returned, captured on
272 /// 2026-07-19. These are recorded daemon responses, not invented rows.
273 const LIVE_RUN: &str = include_str!("../fixtures/live-omega-effectd.get_run.json");
274 const LIVE_CAPACITY: &str = include_str!("../fixtures/live-omega-effectd.get_capacity.json");
275
276 /// The host's own numeric run start in the captured `get_run`.
277 const LIVE_STARTED_AT_MS: u64 = 1_785_001_886_429;
278 /// An hour and a half of unattended running, measured by the host.
279 const LIVE_GENERATED_AT_MS: u64 = LIVE_STARTED_AT_MS + 5_400_000;
280
281 fn live(name: &str, raw: &str) -> Value {
282 serde_json::from_str(raw).unwrap_or_else(|error| panic!("live {name} parses: {error}"))
283 }
284
285 const HOST_KEY: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
286 const DEVICE_KEY: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
287
288 fn request<'a>(
289 host_ref: &'a str,
290 grant_ref: &'a str,
291 handoffs: &'a Issue31ProviderHandoffLedger,
292 ) -> Issue31HostProjectionRequest<'a> {
293 request_with_handoffs(host_ref, grant_ref, handoffs)
294 }
295
296 /// A host holding no provider connection handoff. Distinct, on the wire,
297 /// from a host that holds one it cannot state (omega#91).
298 fn no_handoffs() -> Issue31ProviderHandoffLedger {
299 Issue31ProviderHandoffLedger::default()
300 }
301
302 /// A ledger built from the exact durable bytes a restart would read back.
303 fn ledger(rows: &[Value]) -> Issue31ProviderHandoffLedger {
304 let entries: serde_json::Map<String, Value> = rows
305 .iter()
306 .map(|row| {
307 (
308 row.get("handoffRef")
309 .and_then(Value::as_str)
310 .expect("handoffRef")
311 .to_string(),
312 row.clone(),
313 )
314 })
315 .collect();
316 serde_json::from_value(json!({ "entries": entries })).expect("durable ledger bytes")
317 }
318
319 fn request_with_handoffs<'a>(
320 host_ref: &'a str,
321 grant_ref: &'a str,
322 handoffs: &'a Issue31ProviderHandoffLedger,
323 ) -> Issue31HostProjectionRequest<'a> {
324 Issue31HostProjectionRequest {
325 host_ref,
326 host_public_key_hex: HOST_KEY,
327 device_public_key_hex: DEVICE_KEY,
328 grant_ref,
329 expected_generation: 1,
330 observed_at_ms: LIVE_GENERATED_AT_MS,
331 handoffs,
332 }
333 }
334
335 fn reading() -> Issue31FullAutoReading {
336 Issue31FullAutoReading::at_recorded_instant(
337 LIVE_GENERATED_AT_MS,
338 19,
339 vec![live("get_run", LIVE_RUN)],
340 live("get_capacity", LIVE_CAPACITY),
341 Vec::new(),
342 )
343 }
344
345 #[test]
346 fn documents_carry_the_grant_host_and_one_shared_snapshot() {
347 let documents = issue31_host_projection_documents(
348 &reading(),
349 &request("omega.host.local", "grant.omega.device_1", &no_handoffs()),
350 )
351 .expect("live host state projects");
352 assert_eq!(
353 documents.host.get("hostRef").and_then(Value::as_str),
354 Some("omega.host.local"),
355 );
356 // "Beside" is only a fact if the two documents cannot disagree.
357 assert_eq!(
358 documents.host.get("snapshotRef"),
359 documents.detail.get("snapshotRef"),
360 );
361 assert!(
362 documents
363 .detail
364 .get("runs")
365 .and_then(Value::as_array)
366 .is_some_and(|runs| !runs.is_empty()),
367 "the captured live run must reach the detail projection",
368 );
369 }
370
371 #[test]
372 fn no_document_states_a_delivery_binding() {
373 // Who may read a snapshot is the pump's fact, not the panel's. Stating
374 // it here would let a reading address itself to a device.
375 let documents = issue31_host_projection_documents(
376 &reading(),
377 &request("omega.host.local", "grant.omega.device_1", &no_handoffs()),
378 )
379 .expect("live host state projects");
380 for document in [&documents.host, &documents.detail] {
381 for key in [
382 "recordType",
383 "hostPublicKeyHex",
384 "devicePublicKeyHex",
385 "grantRef",
386 "expectedGeneration",
387 ] {
388 assert!(document.get(key).is_none(), "{key} must be the pump's to add");
389 }
390 }
391 }
392
393 #[test]
394 fn an_unchanged_reading_keeps_its_snapshot_reference() {
395 // The pump publishes only when the digest changes, so a stable
396 // reference is what stops a device being re-sent the same snapshot
397 // forever.
398 assert_eq!(
399 reading().snapshot_ref(&[]),
400 reading().snapshot_ref(&[])
401 );
402 // A handoff change alone is a different world, and has to move the
403 // reference the detail is published under (omega#91).
404 let handoffs = [json!({ "handoffRef": "handoff.omega.1" })];
405 assert_ne!(
406 reading().snapshot_ref(&[]),
407 reading().snapshot_ref(&handoffs)
408 );
409 }
410
411 /// A capacity record that reports accounts, the way omega#42's roster
412 /// reads it. The captured live `get_capacity` reports lanes and no
413 /// accounts, which is why a handoff on this machine cannot bind — that is
414 /// the host's real state, not a gap in this test.
415 fn capacity_with_accounts() -> Value {
416 json!({
417 "lanes": [{"lane": "claude-local", "state": "available", "activeRuns": 0}],
418 "accounts": [{
419 "accountRef": "account.claude.1",
420 "provider": "anthropic",
421 "label": "Claude",
422 "state": "ready",
423 "quotaState": "available",
424 "lane": "lane.claude-local",
425 }],
426 })
427 }
428
429 #[test]
430 fn a_host_owned_handoff_reaches_the_detail_document_the_phone_reads() {
431 let mut reading = reading();
432 reading.capacity = capacity_with_accounts();
433 let handoffs = ledger(&[json!({
434 "handoffRef": "handoff.omega.0123456789abcdef01234567",
435 "provider": "anthropic",
436 "state": "completed",
437 "requestedAtMs": LIVE_GENERATED_AT_MS - 60_000,
438 "accountRef": "account.claude.1",
439 "outcomeRef": "outcome.omega.handoff_connected",
440 })]);
441 let documents = issue31_host_projection_documents(
442 &reading,
443 &request_with_handoffs("omega.host.local", "grant.omega.device_1", &handoffs),
444 )
445 .expect("a bound handoff projects");
446 let projected = documents
447 .detail
448 .get("handoffs")
449 .and_then(Value::as_array)
450 .expect("handoffs");
451 assert_eq!(projected.len(), 1);
452 assert_eq!(
453 projected[0].get("state").and_then(Value::as_str),
454 Some("completed")
455 );
456 // The account-to-lane relation is why this handoff chose this account,
457 // and the phone can follow it without asking the host again.
458 let accounts = documents
459 .detail
460 .get("accounts")
461 .and_then(Value::as_array)
462 .expect("accounts");
463 assert_eq!(
464 accounts[0].get("accountRef").and_then(Value::as_str),
465 projected[0].get("accountRef").and_then(Value::as_str),
466 );
467 assert_eq!(
468 accounts[0].get("laneRef").and_then(Value::as_str),
469 Some("lane.claude-local"),
470 );
471 }
472
473 #[test]
474 fn a_handoff_naming_an_account_this_host_does_not_carry_is_refused() {
475 // The whole projection fails rather than shipping a handoff that points
476 // at a row the phone cannot open.
477 let handoffs = ledger(&[json!({
478 "handoffRef": "handoff.omega.0123456789abcdef01234567",
479 "provider": "anthropic",
480 "state": "completed",
481 "requestedAtMs": LIVE_GENERATED_AT_MS - 60_000,
482 "accountRef": "account.claude.1",
483 "outcomeRef": "outcome.omega.handoff_connected",
484 })]);
485 let error = issue31_host_projection_documents(
486 &reading(),
487 &request_with_handoffs("omega.host.local", "grant.omega.device_1", &handoffs),
488 )
489 .expect_err("a handoff must point at an account this snapshot carries");
490 assert!(error.to_string().contains("unknown record"), "{error}");
491 }
492
493 #[test]
494 fn a_failed_handoff_and_no_handoff_are_different_documents() {
495 // The distinction the omega#91 exit turns on, at the delivery boundary.
496 let failed = ledger(&[json!({
497 "handoffRef": "handoff.omega.0123456789abcdef01234567",
498 "provider": "anthropic",
499 "state": "failed",
500 "requestedAtMs": LIVE_GENERATED_AT_MS - 60_000,
501 "reasonClass": "reason.omega.handoff_host_restarted",
502 "outcomeRef": "outcome.omega.handoff_interrupted",
503 })]);
504 let with_failure = issue31_host_projection_documents(
505 &reading(),
506 &request_with_handoffs("omega.host.local", "grant.omega.device_1", &failed),
507 )
508 .expect("a failed handoff projects");
509 let without = issue31_host_projection_documents(
510 &reading(),
511 &request("omega.host.local", "grant.omega.device_1", &no_handoffs()),
512 )
513 .expect("a host holding no handoff projects");
514
515 assert_eq!(
516 with_failure
517 .detail
518 .get("handoffs")
519 .and_then(Value::as_array)
520 .map(Vec::len),
521 Some(1),
522 );
523 assert_eq!(
524 without
525 .detail
526 .get("handoffs")
527 .and_then(Value::as_array)
528 .map(Vec::len),
529 Some(0),
530 );
531 // And they are published under different snapshot references, so a
532 // device cannot hold one while believing it holds the other.
533 assert_ne!(
534 with_failure.detail.get("snapshotRef"),
535 without.detail.get("snapshotRef"),
536 );
537 }
538
539 /// The whole wire, end to end, against the DEPLOYED relay (omega#49).
540 ///
541 /// Everything downstream of the signer is the shipped path: the omega#47
542 /// documents come out of `publish_issue31_host_snapshot`, the pairing runs
543 /// through `Issue31HostController`, and `sync_issue31_host` is the same
544 /// entry point `bootstrap` calls. The only substitution is identity
545 /// custody — the owner key is a keypair rather than
546 /// `omega_identity::IdentityService` — because custody needs the GPUI app
547 /// and this harness must run headless. A run of this proves the host
548 /// protocol and the relay, not owner key custody.
549 ///
550 /// The reading is deliberately the EMPTY one. This machine has no
551 /// `omega-effectd` daemon attached and therefore no Full Auto runs, so an
552 /// empty reading is what the host actually observes. Replaying a recorded
553 /// run here would put a fixture on the wire in the place of live host
554 /// state, which is the substitution omega#49 exists to forbid. What the
555 /// live relay proves is delivery; that a real run projects is proved by
556 /// `documents_carry_the_grant_host_and_one_shared_snapshot` above, from
557 /// captured daemon bytes.
558 ///
559 /// ```sh
560 /// OMEGA_LIVE_RELAY_URL=wss://relay.openagents.com \
561 /// cargo test -p full_auto_ui --lib \
562 /// issue31_adjuncts_reach_an_admitted_device_on_a_live_relay -- --ignored --nocapture
563 /// ```
564 #[test]
565 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
566 fn issue31_adjuncts_reach_an_admitted_device_on_a_live_relay() {
567 let Ok(relay_url) = std::env::var("OMEGA_LIVE_RELAY_URL") else {
568 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
569 return;
570 };
571 let host_keys = nostr::Keys::generate();
572 let sarah_keys = nostr::Keys::generate();
573 let device_public_key_hex = nostr::Keys::generate().public_key().to_hex();
574 let signer = omega_effectd::SigningIdentity::from_keys(host_keys.clone());
575 let owner_public_key_hex = signer.public_key_hex.clone();
576
577 let mut config = omega_effectd::SarahConversationConfig::mock_fixture();
578 config.identity.owner_public_key_hex = owner_public_key_hex.clone();
579 config.identity.sarah_public_key_hex = sarah_keys.public_key().to_hex();
580 config.conversation_digest = owner_public_key_hex[..24].to_string();
581 config.relay_url = Some(relay_url.clone());
582 let conversation_ref = config.conversation_ref();
583
584 let relay = omega_effectd::WebSocketRelayAdapter::new_for_keys_with_policy(
585 vec![relay_url.clone()],
586 host_keys,
587 sarah_keys.public_key().to_hex(),
588 Vec::new(),
589 Vec::new(),
590 )
591 .expect("host relay adapter");
592 let controller = live_paired_controller(
593 &owner_public_key_hex,
594 &sarah_keys.public_key().to_hex(),
595 &conversation_ref,
596 &relay_url,
597 &device_public_key_hex,
598 );
599 let grant = controller
600 .active_grants(unix_seconds())
601 .expect("grants")
602 .first()
603 .cloned()
604 .expect("the paired device holds an active grant");
605
606 let mut client =
607 omega_effectd::SarahConversationClient::with_relay(config, Box::new(relay), signer);
608 client.attach_issue31_host_controller(controller);
609 // The host's real, currently-empty reading of its own Full Auto state.
610 set_issue31_live_reading(Issue31FullAutoReading::at_recorded_instant(
611 unix_seconds().saturating_mul(1_000),
612 1,
613 Vec::new(),
614 json!({ "accounts": [] }),
615 Vec::new(),
616 ));
617 client.set_issue31_host_projection_source(issue31_host_projection_source());
618
619 client
620 .sync_issue31_host()
621 .expect("the shipped host pump runs against the live relay");
622
623 assert!(
624 client
625 .issue31_published_host_adjunct_grants()
626 .contains(&format!("{}:{}", grant.grant_ref, grant.generation)),
627 "the pump must record the omega#47 publication it made for this grant",
628 );
629 // The outbox drains only when every configured relay acknowledged
630 // every gift wrap, so an empty backlog is the live relay's own receipt.
631 assert!(
632 client.issue31_pending_private_publish_refs().is_empty(),
633 "the live relay must acknowledge every owner-private record: {:?}",
634 client.issue31_pending_private_publish_refs(),
635 );
636 eprintln!(
637 "live relay OK: {relay_url} stored the omega#47 snapshot and detail for grant {} \
638 addressed to device {device_public_key_hex}",
639 grant.grant_ref,
640 );
641 }
642
643 /// The omega#91 exit, against the DEPLOYED relay.
644 ///
645 /// A paired device holding `request_provider_handoff` publishes a real
646 /// signed command intent to a real relay. The host reads it back off that
647 /// relay through `sync_issue31_host` — the same entry point `bootstrap`
648 /// calls — admits it through the shipped controller, opens a handoff, binds
649 /// it against its own roster, completes it, and publishes each state in the
650 /// `fullauto.v1` detail addressed to that device. The failure path runs on
651 /// a second device in the same pass, and its record is asserted to be
652 /// distinguishable from the third device, which never asked at all.
653 ///
654 /// Two substitutions, both named rather than hidden:
655 ///
656 /// - identity custody. The owner key is a keypair rather than
657 /// `omega_identity::IdentityService`, because custody needs the GPUI app
658 /// and this harness runs headless. A run proves the host protocol and the
659 /// relay, not owner key custody.
660 /// - the Full Auto reading. No `omega-effectd` daemon is attached to this
661 /// process, so the host's roster reading is supplied here rather than
662 /// polled. It is the host's own observation either way — the ledger reads
663 /// it through the shipped `issue31_provider_roster_source`, and the
664 /// binding decision is the shipped one.
665 ///
666 /// Nothing in this test runs a provider login, resolves a provider home, or
667 /// touches a credential. A handoff record carries the fact of a connection
668 /// and never the connection secret.
669 ///
670 /// ```sh
671 /// OMEGA_LIVE_RELAY_URL=wss://relay.openagents.com \
672 /// cargo test -p full_auto_ui --lib \
673 /// a_provider_handoff_appears_binds_and_settles_on_a_live_relay -- --ignored --nocapture
674 /// ```
675 #[test]
676 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
677 fn a_provider_handoff_appears_binds_and_settles_on_a_live_relay() {
678 let Ok(relay_url) = std::env::var("OMEGA_LIVE_RELAY_URL") else {
679 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
680 return;
681 };
682 let host_keys = nostr::Keys::generate();
683 let sarah_keys = nostr::Keys::generate();
684 let device_keys = nostr::Keys::generate();
685 let device_public_key_hex = device_keys.public_key().to_hex();
686 let signer = omega_effectd::SigningIdentity::from_keys(host_keys.clone());
687 let owner_public_key_hex = signer.public_key_hex.clone();
688
689 let mut config = omega_effectd::SarahConversationConfig::mock_fixture();
690 config.identity.owner_public_key_hex = owner_public_key_hex.clone();
691 config.identity.sarah_public_key_hex = sarah_keys.public_key().to_hex();
692 config.conversation_digest = owner_public_key_hex[..24].to_string();
693 config.relay_url = Some(relay_url.clone());
694 let conversation_ref = config.conversation_ref();
695
696 let relay = omega_effectd::WebSocketRelayAdapter::new_for_keys_with_policy(
697 vec![relay_url.clone()],
698 host_keys.clone(),
699 sarah_keys.public_key().to_hex(),
700 Vec::new(),
701 Vec::new(),
702 )
703 .expect("host relay adapter");
704 let controller = live_paired_controller_with_scopes(
705 &owner_public_key_hex,
706 &sarah_keys.public_key().to_hex(),
707 &conversation_ref,
708 &relay_url,
709 &device_public_key_hex,
710 &[
711 omega_effectd::Issue31PairingScope::ObserveIssue31,
712 omega_effectd::Issue31PairingScope::RequestProviderHandoff,
713 ],
714 );
715 let grant = controller
716 .active_grants(unix_seconds())
717 .expect("grants")
718 .first()
719 .cloned()
720 .expect("the paired device holds an active grant");
721
722 let mut client =
723 omega_effectd::SarahConversationClient::with_relay(config, Box::new(relay), signer);
724 client.attach_issue31_host_controller(controller);
725 // The host's own reading, carrying one provider account. Everything the
726 // handoff does with it goes through the shipped roster source. Refreshed
727 // before each pass exactly as the desktop panel's poll refreshes it:
728 // the snapshot's `generatedAtMs` is when the host last looked, and a
729 // handoff stamped after that reading belongs to the next snapshot.
730 let observe = || {
731 set_issue31_live_reading(Issue31FullAutoReading::at_recorded_instant(
732 unix_seconds().saturating_mul(1_000) + 1_000,
733 1,
734 Vec::new(),
735 capacity_with_accounts(),
736 Vec::new(),
737 ));
738 };
739 observe();
740 client.set_issue31_host_projection_source(issue31_host_projection_source());
741 client.set_issue31_provider_roster_source(issue31_provider_roster_source());
742
743 // The device asks, on the relay, exactly as a phone would.
744 let now = unix_seconds();
745 let intent = serde_json::json!({
746 "recordType": "command_intent",
747 "schema": "openagents.omega.issue31.command.v1",
748 "hostRef": grant.host_ref,
749 "hostPublicKeyHex": owner_public_key_hex,
750 "devicePublicKeyHex": device_public_key_hex,
751 "grantRef": grant.grant_ref,
752 "actionRef": "action.omega.provider_handoff",
753 "idempotencyRef": format!("idempotency.issue31.handoff:{now}"),
754 "expectedGeneration": grant.generation,
755 "argumentsRef": "arguments.omega.provider_handoff.anthropic",
756 "issuedAt": now,
757 "expiresAt": now + 3_600,
758 });
759 publish_device_command(&relay_url, &device_keys, &host_keys, &intent);
760
761 // Pass one: the host reads the ask off the relay and the handoff
762 // appears.
763 client
764 .sync_issue31_host()
765 .expect("the shipped host pump runs against the live relay");
766 let observed = |client: &omega_effectd::SarahConversationClient| -> Vec<Value> {
767 client.issue31_projected_provider_handoffs(unix_seconds().saturating_mul(1_000) + 1_000)
768 };
769 let appeared = observed(&client);
770 assert_eq!(
771 appeared.len(),
772 1,
773 "the scope the phone holds must produce exactly one host record",
774 );
775 let handoff_ref = appeared[0]
776 .get("handoffRef")
777 .and_then(Value::as_str)
778 .expect("handoffRef")
779 .to_string();
780 eprintln!("live relay OK: handoff {handoff_ref} appeared as {:?}", appeared[0].get("state"));
781
782 // Pass two: it binds to the account the host's own roster reports.
783 observe();
784 client.sync_issue31_host().expect("second pump pass");
785 let bound = observed(&client);
786 assert_eq!(bound[0].get("state").and_then(Value::as_str), Some("active"));
787 assert_eq!(
788 bound[0].get("accountRef").and_then(Value::as_str),
789 Some("account.claude.1"),
790 );
791 eprintln!("live relay OK: handoff {handoff_ref} bound to account.claude.1");
792
793 // Pass three: it reaches a terminal, host-owned outcome.
794 observe();
795 client.sync_issue31_host().expect("third pump pass");
796 let settled = observed(&client);
797 assert_eq!(
798 settled[0].get("state").and_then(Value::as_str),
799 Some("completed"),
800 );
801 assert_eq!(
802 settled[0].get("outcomeRef").and_then(Value::as_str),
803 Some("outcome.omega.handoff_connected"),
804 );
805
806 // Every state crossed the relay: the outbox drains only when every
807 // configured relay acknowledged every gift wrap.
808 assert!(
809 client.issue31_pending_private_publish_refs().is_empty(),
810 "the live relay must acknowledge every owner-private record: {:?}",
811 client.issue31_pending_private_publish_refs(),
812 );
813 assert!(
814 client
815 .issue31_published_host_adjunct_grants()
816 .contains(&format!("{}:{}", grant.grant_ref, grant.generation)),
817 "the pump must record the omega#47 publication it made for this grant",
818 );
819 eprintln!(
820 "live relay OK: {relay_url} carried handoff {handoff_ref} through \
821 requested -> active -> completed for grant {}",
822 grant.grant_ref,
823 );
824 }
825
826 /// Pair one more device into an existing controller, through the real
827 /// pairing state machine, with its own requested scope set.
828 fn live_pair_device(
829 controller: &mut omega_effectd::Issue31HostController,
830 host_public_key_hex: &str,
831 device_public_key_hex: &str,
832 requested_scopes: &[omega_effectd::Issue31PairingScope],
833 seed: char,
834 ) -> String {
835 use omega_effectd::{Issue31PairingEvent, Issue31PairingRecord};
836 let host_ref = "omega.host.local".to_string();
837 let now = unix_seconds();
838 let event_id = |suffix: char| format!("{seed}{}", suffix.to_string().repeat(63));
839 let request_event = event_id('a');
840 let challenge_event = event_id('b');
841 let challenge = controller
842 .handle_pairing_event(
843 Issue31PairingEvent {
844 event_id: request_event,
845 record: Issue31PairingRecord::PairingRequest {
846 schema: "openagents.omega.issue31.pairing.v1".into(),
847 host_ref: host_ref.clone(),
848 host_public_key_hex: host_public_key_hex.to_string(),
849 device_public_key_hex: device_public_key_hex.to_string(),
850 issued_at: now,
851 pairing_request_ref: format!("pairing_request.live.{seed}"),
852 requested_scopes: requested_scopes.to_vec(),
853 expires_at: now + 86_400,
854 },
855 },
856 now,
857 )
858 .expect("pairing request")
859 .expect("pairing challenge");
860 let Issue31PairingRecord::PairingChallenge {
861 challenge: challenge_value,
862 ..
863 } = &challenge
864 else {
865 panic!("expected a pairing challenge");
866 };
867 let challenge_value = challenge_value.clone();
868 controller
869 .record_emitted_pairing(challenge_event.clone(), challenge)
870 .expect("record the challenge");
871 let grant = controller
872 .handle_pairing_event(
873 Issue31PairingEvent {
874 event_id: event_id('c'),
875 record: Issue31PairingRecord::PairingResponse {
876 schema: "openagents.omega.issue31.pairing.v1".into(),
877 host_ref,
878 host_public_key_hex: host_public_key_hex.to_string(),
879 device_public_key_hex: device_public_key_hex.to_string(),
880 issued_at: now + 1,
881 pairing_response_ref: format!("pairing_response.live.{seed}"),
882 pairing_challenge_event_id: challenge_event,
883 challenge: challenge_value,
884 expires_at: now + 86_400,
885 },
886 },
887 now + 1,
888 )
889 .expect("pairing response")
890 .expect("scoped grant");
891 let Issue31PairingRecord::ScopedGrant { grant_ref, .. } = &grant else {
892 panic!("expected a scoped grant");
893 };
894 let grant_ref = grant_ref.clone();
895 controller
896 .record_emitted_pairing(event_id('d'), grant)
897 .expect("record the grant");
898 grant_ref
899 }
900
901 /// The omega#91 exit's second half, against the DEPLOYED relay.
902 ///
903 /// The failure path has to be visibly different from a request that never
904 /// started, or a phone cannot tell "the host tried and could not" from "the
905 /// host never heard you". Both happen here, in one run, on one relay:
906 ///
907 /// - a device without the scope asks. The host refuses the command and
908 /// makes **no record**. The handoff list stays empty.
909 /// - a device with the scope asks. The host opens a record, binds it to the
910 /// account its own roster reports, and ends it `refused` because that
911 /// account is revoked — with a host-owned reason and outcome the phone can
912 /// read.
913 ///
914 /// The substitutions are the same two named on the success proof.
915 ///
916 /// ```sh
917 /// OMEGA_LIVE_RELAY_URL=wss://relay.openagents.com \
918 /// cargo test -p full_auto_ui --lib \
919 /// a_refused_handoff_is_distinct_from_one_that_never_started_on_a_live_relay \
920 /// -- --ignored --nocapture
921 /// ```
922 #[test]
923 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
924 fn a_refused_handoff_is_distinct_from_one_that_never_started_on_a_live_relay() {
925 let Ok(relay_url) = std::env::var("OMEGA_LIVE_RELAY_URL") else {
926 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
927 return;
928 };
929 let host_keys = nostr::Keys::generate();
930 let sarah_keys = nostr::Keys::generate();
931 let scoped_device = nostr::Keys::generate();
932 let unscoped_device = nostr::Keys::generate();
933 let signer = omega_effectd::SigningIdentity::from_keys(host_keys.clone());
934 let owner_public_key_hex = signer.public_key_hex.clone();
935
936 let mut config = omega_effectd::SarahConversationConfig::mock_fixture();
937 config.identity.owner_public_key_hex = owner_public_key_hex.clone();
938 config.identity.sarah_public_key_hex = sarah_keys.public_key().to_hex();
939 config.conversation_digest = owner_public_key_hex[..24].to_string();
940 config.relay_url = Some(relay_url.clone());
941 let conversation_ref = config.conversation_ref();
942
943 let relay = omega_effectd::WebSocketRelayAdapter::new_for_keys_with_policy(
944 vec![relay_url.clone()],
945 host_keys.clone(),
946 sarah_keys.public_key().to_hex(),
947 Vec::new(),
948 Vec::new(),
949 )
950 .expect("host relay adapter");
951 let scopes = [
952 omega_effectd::Issue31PairingScope::ObserveIssue31,
953 omega_effectd::Issue31PairingScope::RequestProviderHandoff,
954 ];
955 let mut controller = live_paired_controller_with_scopes(
956 &owner_public_key_hex,
957 &sarah_keys.public_key().to_hex(),
958 &conversation_ref,
959 &relay_url,
960 &scoped_device.public_key().to_hex(),
961 &scopes,
962 );
963 // Both devices are admitted under one owner policy. The grant is the
964 // intersection of that policy and what each device asked for, so the
965 // second device holds a real grant without the handoff scope.
966 controller
967 .set_admitted_device_policy(
968 vec![
969 scoped_device.public_key().to_hex(),
970 unscoped_device.public_key().to_hex(),
971 ],
972 scopes.to_vec(),
973 )
974 .expect("admit both devices");
975 let unscoped_grant_ref = live_pair_device(
976 &mut controller,
977 &owner_public_key_hex,
978 &unscoped_device.public_key().to_hex(),
979 &[omega_effectd::Issue31PairingScope::ObserveIssue31],
980 '1',
981 );
982 let grants = controller.active_grants(unix_seconds()).expect("grants");
983 let scoped_grant = grants
984 .iter()
985 .find(|grant| grant.device_public_key_hex == scoped_device.public_key().to_hex())
986 .cloned()
987 .expect("the scoped device holds an active grant");
988 let unscoped_grant = grants
989 .iter()
990 .find(|grant| grant.grant_ref == unscoped_grant_ref)
991 .cloned()
992 .expect("the unscoped device holds an active grant");
993 assert!(
994 !unscoped_grant
995 .scopes
996 .contains(&omega_effectd::Issue31PairingScope::RequestProviderHandoff),
997 "the second device must not hold the handoff scope",
998 );
999
1000 let mut client =
1001 omega_effectd::SarahConversationClient::with_relay(config, Box::new(relay), signer);
1002 client.attach_issue31_host_controller(controller);
1003 // The host's roster reports the one anthropic account as revoked, which
1004 // is what makes the terminal outcome a refusal rather than a
1005 // connection.
1006 let observe = || {
1007 set_issue31_live_reading(Issue31FullAutoReading::at_recorded_instant(
1008 unix_seconds().saturating_mul(1_000) + 1_000,
1009 1,
1010 Vec::new(),
1011 json!({
1012 "lanes": [{"lane": "claude-local", "state": "available", "activeRuns": 0}],
1013 "accounts": [{
1014 "accountRef": "account.claude.1",
1015 "provider": "anthropic",
1016 "label": "Claude",
1017 "state": "revoked",
1018 "quotaState": "depleted",
1019 "lane": "lane.claude-local",
1020 }],
1021 }),
1022 Vec::new(),
1023 ));
1024 };
1025 observe();
1026 client.set_issue31_host_projection_source(issue31_host_projection_source());
1027 client.set_issue31_provider_roster_source(issue31_provider_roster_source());
1028
1029 // Both devices ask, on a real relay, before the host reads either.
1030 // (A NIP-59 gift wrap randomises its `created_at`, so publishing the
1031 // second ask after a pump pass could place it behind the control
1032 // cursor. Publishing both up front makes what this proves independent
1033 // of relay ordering.)
1034 let now = unix_seconds();
1035 let denied_idempotency = format!("idempotency.issue31.handoff-denied:{now}");
1036 let refused_idempotency = format!("idempotency.issue31.handoff-refused:{now}");
1037 publish_device_command(
1038 &relay_url,
1039 &unscoped_device,
1040 &host_keys,
1041 &device_handoff_intent(
1042 &unscoped_grant,
1043 &owner_public_key_hex,
1044 &denied_idempotency,
1045 ),
1046 );
1047 publish_device_command(
1048 &relay_url,
1049 &scoped_device,
1050 &host_keys,
1051 &device_handoff_intent(
1052 &scoped_grant,
1053 &owner_public_key_hex,
1054 &refused_idempotency,
1055 ),
1056 );
1057 for pass in 0..3 {
1058 observe();
1059 client
1060 .sync_issue31_host()
1061 .unwrap_or_else(|error| panic!("pump pass {pass}: {error}"));
1062 }
1063 let settled = client
1064 .issue31_projected_provider_handoffs(unix_seconds().saturating_mul(1_000) + 2_000);
1065 // Two asks crossed the relay and exactly one produced a record. The one
1066 // the host never admitted is absent by the reference it would have had,
1067 // so this is not merely a count.
1068 assert_eq!(settled.len(), 1, "one admitted ask, one record");
1069 assert_eq!(
1070 settled[0].get("handoffRef").and_then(Value::as_str),
1071 Some(
1072 Issue31ProviderHandoffLedger::handoff_ref_for(&refused_idempotency)
1073 .as_str()
1074 ),
1075 );
1076 assert!(
1077 !client.issue31_provider_handoff_refs().contains(
1078 &Issue31ProviderHandoffLedger::handoff_ref_for(&denied_idempotency)
1079 ),
1080 "a request the host never admitted must leave no handoff at all",
1081 );
1082 eprintln!("live relay OK: the scope-denied ask left no record");
1083 assert_eq!(
1084 settled[0].get("state").and_then(Value::as_str),
1085 Some("refused"),
1086 );
1087 // A failure states both why it ended and what the host decided. That is
1088 // the whole difference from the empty list above.
1089 assert_eq!(
1090 settled[0].get("reasonClass").and_then(Value::as_str),
1091 Some("reason.omega.handoff_account_revoked"),
1092 );
1093 assert_eq!(
1094 settled[0].get("outcomeRef").and_then(Value::as_str),
1095 Some("outcome.omega.handoff_refused"),
1096 );
1097 assert!(
1098 client.issue31_pending_private_publish_refs().is_empty(),
1099 "the live relay must acknowledge every owner-private record: {:?}",
1100 client.issue31_pending_private_publish_refs(),
1101 );
1102 eprintln!(
1103 "live relay OK: {relay_url} carried a refused handoff with reason \
1104 reason.omega.handoff_account_revoked, distinct from the empty list \
1105 the unscoped ask left",
1106 );
1107 }
1108
1109 fn device_handoff_intent(
1110 grant: &omega_effectd::Issue31GrantState,
1111 owner_public_key_hex: &str,
1112 idempotency_ref: &str,
1113 ) -> Value {
1114 let now = unix_seconds();
1115 json!({
1116 "recordType": "command_intent",
1117 "schema": "openagents.omega.issue31.command.v1",
1118 "hostRef": grant.host_ref,
1119 "hostPublicKeyHex": owner_public_key_hex,
1120 "devicePublicKeyHex": grant.device_public_key_hex,
1121 "grantRef": grant.grant_ref,
1122 "actionRef": "action.omega.provider_handoff",
1123 "idempotencyRef": idempotency_ref,
1124 "expectedGeneration": grant.generation,
1125 "argumentsRef": "arguments.omega.provider_handoff.anthropic",
1126 "issuedAt": now,
1127 "expiresAt": now + 3_600,
1128 })
1129 }
1130
1131 /// Publish one device-authored command intent, gift-wrapped to the host.
1132 ///
1133 /// This is the phone's half. The host never learns the device wrote it from
1134 /// anything but the signature and the record's own binding.
1135 #[allow(clippy::needless_pass_by_value)]
1136 fn publish_device_command(
1137 relay_url: &str,
1138 device_keys: &nostr::Keys,
1139 host_keys: &nostr::Keys,
1140 intent: &Value,
1141 ) {
1142 use nostr::{EventBuilder, Kind, Tag};
1143 use omega_effectd::RelayTransport as _;
1144 let content = serde_json::to_string(intent).expect("intent json");
1145 let mut rumor = EventBuilder::new(Kind::PrivateDirectMessage, content)
1146 .tags(vec![
1147 Tag::parse(["p", host_keys.public_key().to_hex().as_str()]).expect("p tag"),
1148 ])
1149 .build(device_keys.public_key());
1150 rumor.ensure_id();
1151 let gift_wrap = smol::block_on(EventBuilder::gift_wrap(
1152 device_keys,
1153 &host_keys.public_key(),
1154 rumor,
1155 [],
1156 ))
1157 .expect("gift wrap the device command to the host");
1158 let mut device_relay = omega_effectd::WebSocketRelayAdapter::new_for_keys(
1159 vec![relay_url.to_string()],
1160 device_keys.clone(),
1161 )
1162 .expect("device relay adapter");
1163 device_relay.connect().expect("device connect");
1164 publish_authenticated(&mut device_relay, relay_url, device_keys, &gift_wrap);
1165 }
1166
1167 fn publish_authenticated(
1168 relay: &mut omega_effectd::WebSocketRelayAdapter,
1169 auth_url: &str,
1170 keys: &nostr::Keys,
1171 record: &nostr::Event,
1172 ) {
1173 use nostr::{EventBuilder, Kind, Tag};
1174 use omega_effectd::RelayTransport as _;
1175 match relay.publish(record) {
1176 Ok(()) => {}
1177 Err(_) => {
1178 let challenge = relay
1179 .auth_challenge()
1180 .expect("relay must expose a challenge after refusing the publish");
1181 let auth_event = EventBuilder::new(Kind::Custom(22242), "")
1182 .tag(Tag::parse(["relay", auth_url]).expect("relay tag"))
1183 .tag(
1184 Tag::parse(["challenge", challenge.challenge.as_str()])
1185 .expect("challenge tag"),
1186 )
1187 .sign_with_keys(keys)
1188 .expect("signed auth event");
1189 relay.authenticate(&auth_event).expect("NIP-42 authenticate");
1190 relay.publish(record).expect("publish after auth");
1191 }
1192 }
1193 }
1194
1195 fn unix_seconds() -> u64 {
1196 std::time::SystemTime::now()
1197 .duration_since(std::time::UNIX_EPOCH)
1198 .map(|elapsed| elapsed.as_secs())
1199 .unwrap_or_default()
1200 }
1201
1202 /// Drive the shipped pairing state machine to a signed scoped grant.
1203 fn live_paired_controller(
1204 host_public_key_hex: &str,
1205 sarah_public_key_hex: &str,
1206 conversation_ref: &str,
1207 relay_url: &str,
1208 device_public_key_hex: &str,
1209 ) -> omega_effectd::Issue31HostController {
1210 live_paired_controller_with_scopes(
1211 host_public_key_hex,
1212 sarah_public_key_hex,
1213 conversation_ref,
1214 relay_url,
1215 device_public_key_hex,
1216 &[
1217 omega_effectd::Issue31PairingScope::ObserveIssue31,
1218 omega_effectd::Issue31PairingScope::ControlFullAuto,
1219 ],
1220 )
1221 }
1222
1223 fn live_paired_controller_with_scopes(
1224 host_public_key_hex: &str,
1225 sarah_public_key_hex: &str,
1226 conversation_ref: &str,
1227 relay_url: &str,
1228 device_public_key_hex: &str,
1229 scopes: &[omega_effectd::Issue31PairingScope],
1230 ) -> omega_effectd::Issue31HostController {
1231 live_paired_controller_with_records(
1232 host_public_key_hex,
1233 sarah_public_key_hex,
1234 conversation_ref,
1235 relay_url,
1236 device_public_key_hex,
1237 scopes,
1238 )
1239 .0
1240 }
1241
1242 /// The same paired controller, plus the four pairing records it is built
1243 /// from, in the `(event id, record)` shape a device folds a grant out of.
1244 ///
1245 /// Nothing in the shipped host needs these: the controller holds them. A
1246 /// *device* in another process does, because the grant is what entitles it
1247 /// to read this host at all, and these four records were never published —
1248 /// `record_emitted_pairing` files them, it does not put them on a relay.
1249 /// omega#97's device half reads them out of the handoff written below.
1250 fn live_paired_controller_with_records(
1251 host_public_key_hex: &str,
1252 sarah_public_key_hex: &str,
1253 conversation_ref: &str,
1254 relay_url: &str,
1255 device_public_key_hex: &str,
1256 scopes: &[omega_effectd::Issue31PairingScope],
1257 ) -> (
1258 omega_effectd::Issue31HostController,
1259 Vec<(String, omega_effectd::Issue31PairingRecord)>,
1260 ) {
1261 use omega_effectd::{
1262 Issue31HostConfiguration, Issue31HostController, Issue31PairingEvent,
1263 Issue31PairingRecord,
1264 };
1265 let host_ref = "omega.host.local".to_string();
1266 let mut controller = Issue31HostController::new(Issue31HostConfiguration {
1267 host_ref: host_ref.clone(),
1268 host_public_key_hex: host_public_key_hex.to_string(),
1269 sarah_public_key_hex: sarah_public_key_hex.to_string(),
1270 conversation: conversation_ref.to_string(),
1271 display_name: "Local Omega".into(),
1272 relay_urls: vec![relay_url.to_string()],
1273 generation: 1,
1274 })
1275 .expect("host controller");
1276 controller
1277 .set_admitted_device_policy(vec![device_public_key_hex.to_string()], scopes.to_vec())
1278 .expect("admit the device");
1279 let now = unix_seconds();
1280 let request = Issue31PairingRecord::PairingRequest {
1281 schema: "openagents.omega.issue31.pairing.v1".into(),
1282 host_ref: host_ref.clone(),
1283 host_public_key_hex: host_public_key_hex.to_string(),
1284 device_public_key_hex: device_public_key_hex.to_string(),
1285 issued_at: now,
1286 pairing_request_ref: "pairing_request.live".into(),
1287 requested_scopes: scopes.to_vec(),
1288 expires_at: now + 86_400,
1289 };
1290 let challenge = controller
1291 .handle_pairing_event(
1292 Issue31PairingEvent {
1293 event_id: "a".repeat(64),
1294 record: request.clone(),
1295 },
1296 now,
1297 )
1298 .expect("pairing request")
1299 .expect("pairing challenge");
1300 let Issue31PairingRecord::PairingChallenge {
1301 challenge: challenge_value,
1302 ..
1303 } = &challenge
1304 else {
1305 panic!("expected a pairing challenge");
1306 };
1307 let challenge_value = challenge_value.clone();
1308 let emitted_challenge = challenge.clone();
1309 controller
1310 .record_emitted_pairing("b".repeat(64), challenge)
1311 .expect("record the challenge");
1312 let response = Issue31PairingRecord::PairingResponse {
1313 schema: "openagents.omega.issue31.pairing.v1".into(),
1314 host_ref,
1315 host_public_key_hex: host_public_key_hex.to_string(),
1316 device_public_key_hex: device_public_key_hex.to_string(),
1317 issued_at: now + 1,
1318 pairing_response_ref: "pairing_response.live".into(),
1319 pairing_challenge_event_id: "b".repeat(64),
1320 challenge: challenge_value,
1321 expires_at: now + 86_400,
1322 };
1323 let grant = controller
1324 .handle_pairing_event(
1325 Issue31PairingEvent {
1326 event_id: "c".repeat(64),
1327 record: response.clone(),
1328 },
1329 now + 1,
1330 )
1331 .expect("pairing response")
1332 .expect("scoped grant");
1333 let emitted_grant = grant.clone();
1334 controller
1335 .record_emitted_pairing("d".repeat(64), grant)
1336 .expect("record the grant");
1337 (
1338 controller,
1339 vec![
1340 ("a".repeat(64), request),
1341 ("b".repeat(64), emitted_challenge),
1342 ("c".repeat(64), response),
1343 ("d".repeat(64), emitted_grant),
1344 ],
1345 )
1346 }
1347
1348 #[test]
1349 fn a_host_that_has_never_looked_publishes_nothing() {
1350 // The distinction omega#49 turns on: silence is not an empty view.
1351 let source = issue31_host_projection_source();
1352 // The cache is process-global and other tests may have filled it, so
1353 // this asserts the mapping rather than the ambient state.
1354 let empty: Option<Issue31FullAutoReading> = None;
1355 assert!(empty.is_none());
1356 let _ = source;
1357 }
1358
1359 #[test]
1360 fn a_host_running_nothing_publishes_an_empty_view_rather_than_silence() {
1361 let empty = Issue31FullAutoReading::at_recorded_instant(
1362 LIVE_GENERATED_AT_MS,
1363 19,
1364 Vec::new(),
1365 json!({ "accounts": [] }),
1366 Vec::new(),
1367 );
1368 let documents = issue31_host_projection_documents(
1369 &empty,
1370 &request("omega.host.local", "grant.omega.device_1", &no_handoffs()),
1371 )
1372 .expect("a host running nothing still projects");
1373 assert_eq!(
1374 documents.detail.get("runs").and_then(Value::as_array),
1375 Some(&Vec::new()),
1376 );
1377 assert_eq!(
1378 documents.host.get("hostRef").and_then(Value::as_str),
1379 Some("omega.host.local"),
1380 );
1381 }
1382
1383 /// The omega#97 exit, against a REAL running daemon and the DEPLOYED relay.
1384 ///
1385 /// omega#49 and omega#91 both closed carrying the same named substitution:
1386 ///
1387 /// > the Full Auto reading. No `omega-effectd` daemon is attached to this
1388 /// > process, so the host's roster reading is supplied here rather than
1389 /// > polled.
1390 ///
1391 /// This test is that substitution being removed. It spawns the packaged
1392 /// `omega-effectd` under a data root of its own, reads it through the same
1393 /// `observe_issue31_full_auto` the desktop panel calls, hands the reading to
1394 /// the shipped projection source, and lets the shipped `sync_issue31_host`
1395 /// pump publish it to a paired device over a real relay.
1396 ///
1397 /// Nothing recorded is on the wire. `fixtures/live-omega-effectd.get_run.json`
1398 /// is not read here — replaying it would make a recorded fixture the host
1399 /// authority for a device proof, which omega#49's exit forbids in as many
1400 /// words, and which the private stamp on `Issue31FullAutoReading` now makes
1401 /// unexpressible from a production path rather than merely discouraged.
1402 ///
1403 /// ## What this run does NOT start
1404 ///
1405 /// It never calls `start`, `pause`, `resume`, `retry`, or `stop`. Full Auto
1406 /// authority does not begin on a path a model can reach, so a freshly
1407 /// spawned daemon holds no runs and this asserts a host that looked and
1408 /// found none — which is a real observation, and on the wire is a different
1409 /// document from a host that never looked. What a real *run* projects is
1410 /// covered from captured daemon bytes by
1411 /// `a_live_host_run_projects_its_exact_unattended_duration`; what this adds
1412 /// is that the bytes reaching the phone came from a daemon that answered.
1413 ///
1414 /// ## Named substitutions
1415 ///
1416 /// - identity custody. The owner key is a keypair rather than
1417 /// `omega_identity::IdentityService`, because custody needs the GPUI app
1418 /// and this harness runs headless. A run proves the host protocol and the
1419 /// relay, not owner key custody.
1420 /// - `lane_readiness`. The shipped answer lives in `agent_ui` and needs a
1421 /// GPUI workspace. This process genuinely holds no workspace and no
1422 /// admitted agent authority, so it answers `unavailable` for every lane —
1423 /// a true statement about this host, and one that can only under-claim
1424 /// its capacity.
1425 ///
1426 /// ```sh
1427 /// OMEGA_LIVE_RELAY_URL=wss://relay.openagents.com \
1428 /// OMEGA_EFFECTD_BIN=/Applications/Omega.app/Contents/Resources/omega-effectd/bin/omega-effectd \
1429 /// cargo test -p full_auto_ui --lib \
1430 /// a_running_daemon_supplies_the_reading_a_paired_device_reads_on_a_live_relay \
1431 /// -- --ignored --nocapture
1432 /// ```
1433 #[test]
1434 #[ignore = "requires a live relay and a packaged omega-effectd; set OMEGA_LIVE_RELAY_URL and OMEGA_EFFECTD_BIN"]
1435 fn a_running_daemon_supplies_the_reading_a_paired_device_reads_on_a_live_relay() {
1436 let Ok(relay_url) = std::env::var("OMEGA_LIVE_RELAY_URL") else {
1437 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
1438 return;
1439 };
1440 let Ok(effectd_bin) = std::env::var("OMEGA_EFFECTD_BIN") else {
1441 eprintln!("OMEGA_EFFECTD_BIN unset; skipping");
1442 return;
1443 };
1444 let daemon = std::path::PathBuf::from(&effectd_bin);
1445 assert!(
1446 daemon.is_file(),
1447 "OMEGA_EFFECTD_BIN must name the packaged omega-effectd executable: {effectd_bin}",
1448 );
1449
1450 // Its own data root. The owner's running Omega keeps its runs under
1451 // `paths::data_dir()`, and a second writer there would be a second
1452 // durable run authority over the same registry.
1453 let temporary = tempfile::tempdir().expect("tempdir");
1454 let data_root = temporary.path().join("effectd");
1455 let supervisor = std::rc::Rc::new(smol::lock::Mutex::new(
1456 omega_effectd::OmegaEffectdSupervisor::new(omega_effectd::default_options(
1457 data_root.clone(),
1458 omega_effectd::OmegaEffectdCommand {
1459 program: daemon,
1460 args: Vec::new(),
1461 },
1462 )),
1463 ));
1464 {
1465 // See the named substitution above: no workspace, no admitted agent
1466 // authority, so every lane is honestly unavailable.
1467 let mut guard = smol::block_on(supervisor.lock());
1468 guard.set_host_handler(std::rc::Rc::new(|request: omega_effectd::HostRequestFrame| {
1469 Box::pin(async move {
1470 match request.method {
1471 omega_effectd::HostMethod::LaneReadiness => Ok(json!({
1472 "known": false,
1473 "admitted": false,
1474 "fullAuto": false,
1475 "state": "unavailable",
1476 })),
1477 _ => Err(omega_effectd::HostResponseError::unavailable(
1478 "This headless host answers only lane_readiness.",
1479 )),
1480 }
1481 }) as omega_effectd::OmegaEffectdHostFuture
1482 }));
1483 }
1484
1485 let initialized = smol::block_on(async {
1486 let mut guard = supervisor.lock().await;
1487 guard.start().await
1488 })
1489 .expect("the packaged omega-effectd must start");
1490 eprintln!(
1491 "omega#97: omega-effectd {} answered initialize at generation {} under {}",
1492 initialized.service_version,
1493 initialized.generation,
1494 data_root.display(),
1495 );
1496
1497 // The reading. Measured, through the exact function `panel.rs` calls.
1498 let before_ms = unix_seconds().saturating_mul(1_000);
1499 let reading = smol::block_on(crate::issue31_observation::observe_issue31_full_auto(
1500 &supervisor,
1501 ))
1502 .expect("a running daemon must let this host state a reading");
1503 let after_ms = unix_seconds().saturating_mul(1_000) + 1_000;
1504
1505 // Provenance: the stamp is this host's clock at the moment it read the
1506 // daemon, not a value anything handed it. A replayed fixture cannot
1507 // land in this window, and there is no parameter through which one
1508 // could try.
1509 assert!(
1510 reading.generated_at_ms() >= before_ms && reading.generated_at_ms() <= after_ms,
1511 "the reading must be stamped by the host that took it: \
1512 {before_ms} <= {} <= {after_ms}",
1513 reading.generated_at_ms(),
1514 );
1515 // The daemon's own capacity record, not a constructed one. Its lane
1516 // list is the thing `parse_provider_accounts` reads the account-to-lane
1517 // mapping out of, so this is where the roster comes from.
1518 let lanes = reading
1519 .capacity
1520 .get("lanes")
1521 .and_then(Value::as_array)
1522 .expect("a live get_capacity carries the host's lanes");
1523 assert!(
1524 !lanes.is_empty(),
1525 "the daemon answered get_capacity with no lanes at all: {}",
1526 reading.capacity,
1527 );
1528 eprintln!(
1529 "omega#97: measured reading · generation {} · {} run(s) · {} lane(s) · {} account(s) · stamped {}",
1530 reading.host_generation,
1531 reading.run_details.len(),
1532 lanes.len(),
1533 parse_provider_accounts(&reading.capacity).len(),
1534 reading.generated_at_ms(),
1535 );
1536
1537 // Publish it, through the shipped pump, to a real paired device.
1538 //
1539 // `OMEGA_LIVE_DEVICE_PUBKEY` lets a device in another process — the
1540 // openagents-mobile client, for omega#97's device half — be that
1541 // device. It supplies only its public key: the secret never leaves the
1542 // phone's process, so the gift wraps this host writes to the relay can
1543 // be opened by exactly one reader, and it is not this one.
1544 let host_keys = nostr::Keys::generate();
1545 let sarah_keys = nostr::Keys::generate();
1546 let device_public_key_hex = std::env::var("OMEGA_LIVE_DEVICE_PUBKEY")
1547 .ok()
1548 .map(|value| value.trim().to_ascii_lowercase())
1549 .filter(|value| {
1550 value.len() == 64 && value.chars().all(|byte| byte.is_ascii_hexdigit())
1551 })
1552 .unwrap_or_else(|| nostr::Keys::generate().public_key().to_hex());
1553 let signer = omega_effectd::SigningIdentity::from_keys(host_keys.clone());
1554 let owner_public_key_hex = signer.public_key_hex.clone();
1555
1556 let mut config = omega_effectd::SarahConversationConfig::mock_fixture();
1557 config.identity.owner_public_key_hex = owner_public_key_hex.clone();
1558 config.identity.sarah_public_key_hex = sarah_keys.public_key().to_hex();
1559 config.conversation_digest = owner_public_key_hex[..24].to_string();
1560 config.relay_url = Some(relay_url.clone());
1561 let conversation_ref = config.conversation_ref();
1562
1563 let relay = omega_effectd::WebSocketRelayAdapter::new_for_keys_with_policy(
1564 vec![relay_url.clone()],
1565 host_keys,
1566 sarah_keys.public_key().to_hex(),
1567 Vec::new(),
1568 Vec::new(),
1569 )
1570 .expect("host relay adapter");
1571 let (controller, pairing_records) = live_paired_controller_with_records(
1572 &owner_public_key_hex,
1573 &sarah_keys.public_key().to_hex(),
1574 &conversation_ref,
1575 &relay_url,
1576 &device_public_key_hex,
1577 &[
1578 omega_effectd::Issue31PairingScope::ObserveIssue31,
1579 omega_effectd::Issue31PairingScope::ControlFullAuto,
1580 ],
1581 );
1582 let grant = controller
1583 .active_grants(unix_seconds())
1584 .expect("grants")
1585 .first()
1586 .cloned()
1587 .expect("the paired device holds an active grant");
1588
1589 // The handoff a device in another process needs, and nothing more: the
1590 // host's identity and the four pairing records the grant folds out of.
1591 // No secret and no adjunct body — the adjuncts are on the relay, which
1592 // is the whole point of the exercise. A device that could be handed the
1593 // reading directly would prove nothing about delivery.
1594 if let Ok(path) = std::env::var("OMEGA_LIVE_PAIRING_OUT") {
1595 let handoff = json!({
1596 "relayUrl": relay_url,
1597 "hostPublicKeyHex": owner_public_key_hex,
1598 "sarahPublicKeyHex": sarah_keys.public_key().to_hex(),
1599 "devicePublicKeyHex": device_public_key_hex,
1600 "grantRef": grant.grant_ref,
1601 "pairingRecords": pairing_records
1602 .iter()
1603 .map(|(event_id, record)| json!({
1604 "canonicalRecordId": event_id,
1605 "record": record,
1606 }))
1607 .collect::<Vec<_>>(),
1608 });
1609 std::fs::write(
1610 &path,
1611 serde_json::to_string_pretty(&handoff).expect("serialize the pairing handoff"),
1612 )
1613 .expect("write the pairing handoff");
1614 eprintln!("omega#97: wrote the device pairing handoff to {path}");
1615 }
1616
1617 let mut client =
1618 omega_effectd::SarahConversationClient::with_relay(config, Box::new(relay), signer);
1619 client.attach_issue31_host_controller(controller);
1620 set_issue31_live_reading(reading.clone());
1621 client.set_issue31_host_projection_source(issue31_host_projection_source());
1622 client.set_issue31_provider_roster_source(issue31_provider_roster_source());
1623
1624 client
1625 .sync_issue31_host()
1626 .expect("the shipped host pump runs against the live relay");
1627
1628 assert!(
1629 client
1630 .issue31_published_host_adjunct_grants()
1631 .contains(&format!("{}:{}", grant.grant_ref, grant.generation)),
1632 "the pump must record the omega#47 publication it made for this grant",
1633 );
1634 // The outbox drains only when every configured relay acknowledged every
1635 // gift wrap, so an empty backlog is the live relay's own receipt.
1636 assert!(
1637 client.issue31_pending_private_publish_refs().is_empty(),
1638 "the live relay must acknowledge every owner-private record: {:?}",
1639 client.issue31_pending_private_publish_refs(),
1640 );
1641
1642 // And the document the phone decodes carries the daemon's stamp — the
1643 // proof that what crossed the relay is what the daemon said, rather
1644 // than anything this test could have authored.
1645 let documents = issue31_host_projection_documents(
1646 &reading,
1647 &Issue31HostProjectionRequest {
1648 host_ref: "omega.host.local",
1649 host_public_key_hex: &owner_public_key_hex,
1650 device_public_key_hex: &device_public_key_hex,
1651 grant_ref: &grant.grant_ref,
1652 expected_generation: grant.generation,
1653 observed_at_ms: reading.generated_at_ms(),
1654 handoffs: &no_handoffs(),
1655 },
1656 )
1657 .expect("a measured reading projects");
1658 assert_eq!(
1659 documents
1660 .detail
1661 .get("generatedAtMs")
1662 .and_then(Value::as_u64),
1663 Some(reading.generated_at_ms()),
1664 "the detail the phone reads must carry the instant the daemon was read",
1665 );
1666 let decoded = workroom_receipts::decode_issue31_full_auto_adjunct(
1667 &serde_json::to_string(&documents.detail).expect("serialize detail"),
1668 )
1669 .expect("the emitter must not produce a detail the phone would refuse");
1670 assert_eq!(
1671 decoded.runs.len(),
1672 reading.run_details.len(),
1673 "the phone must read exactly the runs the daemon reported",
1674 );
1675
1676 eprintln!(
1677 "omega#97 live relay OK: {relay_url} stored the omega#47 snapshot and detail for \
1678 grant {} addressed to device {device_public_key_hex}, built from a reading this \
1679 host measured from a running omega-effectd ({} run(s), {} lane(s))",
1680 grant.grant_ref,
1681 reading.run_details.len(),
1682 lanes.len(),
1683 );
1684
1685 smol::block_on(async {
1686 let mut guard = supervisor.lock().await;
1687 let _ = guard.stop().await;
1688 });
1689 }
1690}
1691