tenant.openagents/omega

1//! Read this host's own Full Auto state off a running `omega-effectd` (omega#97).
2//!
3//! omega#49 and omega#91 both shipped with the same named substitution:
4//!
5//! > the Full Auto reading. No `omega-effectd` daemon is attached to this
6//! > process, so the host's roster reading is supplied here rather than polled.
7//!
8//! That substitution is the whole of omega#97. A phone paired to a host that is
9//! genuinely running work still rendered `no host projection`, because nothing
10//! outside the desktop Full Auto panel ever handed the pump a reading, and the
11//! obvious way to fill the rows — replaying
12//! `fixtures/live-omega-effectd.get_run.json` — would make a recorded fixture
13//! the host authority for a device proof, which omega#49's exit forbids in as
14//! many words.
15//!
16//! So the reading is taken here instead, from a daemon that answered.
17//!
18//! ## One reading path, not two
19//!
20//! The polling this module does was previously inlined in `panel.rs`, reachable
21//! only from a GPUI view. That is the wrong owner twice over: a headless proof
22//! cannot reach it, and a host publishes its Full Auto state because it is a
23//! host, not because someone opened a panel. `panel.rs` now calls
24//! `observe_issue31_full_auto` like everyone else, so the desktop and the phone
25//! cannot end up looking at two different readings of one daemon.
26//!
27//! ## Refused rather than shortened
28//!
29//! If the daemon lists runs and then fails to describe one of them, this
30//! returns an error and no reading at all. A reading that dropped the run would
31//! be published as a shorter run list than the host actually has, and a run
32//! that vanishes from a snapshot reads on the phone as a run that ended. The
33//! host has no way to distinguish those after the fact, so the incomplete
34//! reading is never built.
35//!
36//! Evidence is the one exception, and deliberately: a run with no report or
37//! receipt yet is a run whose evidence chain has not been produced, which is an
38//! observation rather than a failure. It contributes no pair and the reading
39//! stands.
40//!
41//! ## What this module does not do
42//!
43//! It never starts, pauses, resumes, retries, or stops a run. Reading a run
44//! registry is not the same act as directing one, and Full Auto authority does
45//! not begin on a path a model can reach. Every method called here is one of
46//! `list_runs`, `get_run`, `get_capacity`, `get_report`, `get_receipt`.
47
48use omega_effectd::{SharedOmegaEffectdSupervisor, SupervisorError};
49use serde_json::Value;
50
51use crate::issue31_delivery::Issue31FullAutoReading;
52
53/// The omega#47 contract carries at most this many runs in one projection.
54pub const MAX_ISSUE31_PROJECTED_RUNS: usize = 16;
55
56/// Why this host could not state a reading of its own Full Auto surface.
57///
58/// Every variant means "no reading", never "an empty reading". The distinction
59/// omega#49 turns on is that a host which could not look is not a host that
60/// looked and found nothing: the first publishes silence and the phone says
61/// `no host projection`, the second publishes zero runs and the phone says the
62/// host is running nothing.
63#[derive(Debug)]
64pub enum Issue31ObservationError {
65    /// `list_runs` did not answer. The host does not know what it is running.
66    RunsUnavailable(String),
67    /// `get_capacity` did not answer. Without it there is no account-to-lane
68    /// mapping, and a snapshot carrying runs but no roster would report this
69    /// host as holding no provider accounts.
70    CapacityUnavailable(String),
71    /// A run the daemon listed could not be described. See the module note:
72    /// this refuses the whole reading rather than publishing a shorter one.
73    RunDetailUnavailable { run_ref: String, error: String },
74}
75
76impl std::fmt::Display for Issue31ObservationError {
77    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        match self {
79            Self::RunsUnavailable(error) => {
80                write!(formatter, "omega-effectd did not list its runs: {error}")
81            }
82            Self::CapacityUnavailable(error) => {
83                write!(formatter, "omega-effectd did not report capacity: {error}")
84            }
85            Self::RunDetailUnavailable { run_ref, error } => write!(
86                formatter,
87                "omega-effectd listed run {run_ref} and then could not describe it: {error}",
88            ),
89        }
90    }
91}
92
93impl std::error::Error for Issue31ObservationError {}
94
95impl Issue31ObservationError {
96    /// A stable token for logs and proof output, so a failed observation is
97    /// reported as the specific thing that failed rather than as a sentence.
98    pub fn token(&self) -> &'static str {
99        match self {
100            Self::RunsUnavailable(_) => "observation.omega.runs_unavailable",
101            Self::CapacityUnavailable(_) => "observation.omega.capacity_unavailable",
102            Self::RunDetailUnavailable { .. } => "observation.omega.run_detail_unavailable",
103        }
104    }
105}
106
107fn describe(error: SupervisorError) -> String {
108    error.to_string()
109}
110
111/// Take one complete reading of this host's Full Auto surface.
112///
113/// The supervisor is the same one the desktop drives, so this is the daemon
114/// the owner's Omega is actually running, answering the same five read
115/// methods. Nothing here is recorded, replayed, or defaulted: if the daemon
116/// does not answer, the host states no reading.
117pub async fn observe_issue31_full_auto(
118    supervisor: &SharedOmegaEffectdSupervisor,
119) -> Result<Issue31FullAutoReading, Issue31ObservationError> {
120    let listed = {
121        let mut guard = supervisor.lock().await;
122        guard
123            .list_runs()
124            .await
125            .map_err(|error| Issue31ObservationError::RunsUnavailable(describe(error)))?
126    };
127    let capacity = {
128        let mut guard = supervisor.lock().await;
129        guard
130            .get_capacity()
131            .await
132            .map_err(|error| Issue31ObservationError::CapacityUnavailable(describe(error)))?
133    };
134    let host_generation = {
135        let guard = supervisor.lock().await;
136        guard.generation()
137    };
138
139    let mut run_details: Vec<Value> = Vec::new();
140    let mut evidence: Vec<(Value, Value)> = Vec::new();
141    for run in listed.iter().take(MAX_ISSUE31_PROJECTED_RUNS) {
142        let mut guard = supervisor.lock().await;
143        let detail = guard.get_run(&run.run_ref).await.map_err(|error| {
144            Issue31ObservationError::RunDetailUnavailable {
145                run_ref: run.run_ref.clone(),
146                error: describe(error),
147            }
148        })?;
149        run_details.push(detail);
150        // A run whose evidence chain has not been produced yet contributes no
151        // pair. Both halves or neither: a report without its receipt is half a
152        // chain, and omega#43's chain is refused hop by hop rather than shown
153        // with a hop missing.
154        if let (Ok(report), Ok(receipt)) = (
155            guard.get_report(&run.run_ref).await,
156            guard.get_receipt(&run.run_ref).await,
157        ) {
158            evidence.push((report, receipt));
159        }
160    }
161
162    // Stamped inside the constructor, from one reading of this host's clock,
163    // with no parameter for a caller to supply. See `Issue31FullAutoReading`.
164    Ok(Issue31FullAutoReading::observed(
165        host_generation,
166        run_details,
167        capacity,
168        evidence,
169    ))
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175
176    use serde_json::json;
177
178    /// Stand up the observation stub against a data root of its own.
179    ///
180    /// The stub is a test double for this module's error handling, never host
181    /// authority — see the header of `fixtures/observation_stub_effectd.mjs`.
182    /// Nothing built here is published to a relay by any test.
183    fn stub(scenario: serde_json::Value) -> (tempfile::TempDir, SharedOmegaEffectdSupervisor) {
184        let temporary = tempfile::tempdir().expect("tempdir");
185        let data_root = temporary.path().join("effectd");
186        std::fs::create_dir_all(&data_root).expect("data root");
187        std::fs::write(
188            data_root.join("scenario.json"),
189            serde_json::to_string(&scenario).expect("scenario"),
190        )
191        .expect("write scenario");
192        let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
193            .join("fixtures")
194            .join("observation_stub_effectd.mjs");
195        let supervisor = std::rc::Rc::new(smol::lock::Mutex::new(
196            omega_effectd::OmegaEffectdSupervisor::new(omega_effectd::default_options(
197                data_root,
198                omega_effectd::fixture_command(&fixture),
199            )),
200        ));
201        smol::block_on(async {
202            let mut guard = supervisor.lock().await;
203            guard.start().await
204        })
205        .expect("the observation stub must start");
206        (temporary, supervisor)
207    }
208
209    fn one_run_scenario(refuse_get_run: bool) -> serde_json::Value {
210        json!({
211            "runs": [{
212                "runRef": "run.omega.1",
213                "threadRef": null,
214                "state": "running",
215                "title": "Observed run",
216                "updatedAt": "2026-07-26T12:00:00Z",
217                "startedAtMs": 1_785_000_000_000u64,
218            }],
219            "details": {
220                "run.omega.1": {
221                    "runRef": "run.omega.1",
222                    "state": "running",
223                    "title": "Observed run",
224                    "objective": "Prove the host measured this.",
225                    "doneCondition": "The phone reads it.",
226                    "lane": "codex-local",
227                    "turnCap": 40,
228                    "successfulAttempts": 1,
229                    "failedAttempts": 0,
230                    "recoveryAction": "none",
231                    "startedAtMs": 1_785_000_000_000u64,
232                    "updatedAt": "2026-07-26T12:00:00Z",
233                    "turns": [],
234                },
235            },
236            "capacity": {
237                "activeRunLimit": 8,
238                "activeRunCount": 1,
239                "lanes": [{ "lane": "codex-local", "state": "available", "activeRuns": 1 }],
240            },
241            "refuseGetRun": if refuse_get_run { vec!["run.omega.1"] } else { Vec::<&str>::new() },
242        })
243    }
244
245    /// The happy path, against a daemon that answered.
246    #[test]
247    fn a_daemon_that_answers_yields_the_runs_it_reported() {
248        let (_temporary, supervisor) = stub(one_run_scenario(false));
249        let reading =
250            smol::block_on(observe_issue31_full_auto(&supervisor)).expect("the daemon answered");
251        assert_eq!(reading.run_details.len(), 1);
252        assert_eq!(
253            reading.run_details[0].get("runRef").and_then(|v| v.as_str()),
254            Some("run.omega.1"),
255        );
256        assert_eq!(
257            reading.capacity.get("activeRunCount").and_then(|v| v.as_u64()),
258            Some(1),
259            "the capacity record must be the daemon's own, not a constructed one",
260        );
261        // No report or receipt exists yet, so no evidence pair is invented.
262        assert!(reading.evidence.is_empty());
263    }
264
265    /// A run the daemon listed and then would not describe refuses the whole
266    /// reading.
267    ///
268    /// Publishing the shorter list instead would drop a run the host is
269    /// actually holding, and a run that vanishes from a snapshot reads on the
270    /// phone as a run that ended. The host cannot tell those apart afterwards,
271    /// so the incomplete reading is never built.
272    #[test]
273    fn a_run_the_daemon_will_not_describe_refuses_the_whole_reading() {
274        let (_temporary, supervisor) = stub(one_run_scenario(true));
275        let error = smol::block_on(observe_issue31_full_auto(&supervisor))
276            .expect_err("an undescribable run must refuse the reading");
277        assert_eq!(error.token(), "observation.omega.run_detail_unavailable");
278        assert!(
279            matches!(
280                &error,
281                Issue31ObservationError::RunDetailUnavailable { run_ref, .. }
282                    if run_ref == "run.omega.1"
283            ),
284            "the refusal must name the run it could not describe: {error}",
285        );
286    }
287
288    /// A host that could not look is not a host that looked and found nothing.
289    #[test]
290    fn a_daemon_that_will_not_list_its_runs_yields_no_reading_rather_than_an_empty_one() {
291        let (_temporary, supervisor) = stub(json!({ "refuseListRuns": true }));
292        let error = smol::block_on(observe_issue31_full_auto(&supervisor))
293            .expect_err("a host that could not look states no reading");
294        assert_eq!(error.token(), "observation.omega.runs_unavailable");
295    }
296
297    /// Capacity is the account-to-lane mapping's only source. A reading that
298    /// carried runs and no capacity would report this host as holding no
299    /// provider accounts, which is a claim rather than an observation.
300    #[test]
301    fn a_daemon_that_will_not_report_capacity_yields_no_reading() {
302        let (_temporary, supervisor) = stub(json!({ "runs": [], "refuseCapacity": true }));
303        let error = smol::block_on(observe_issue31_full_auto(&supervisor))
304            .expect_err("no capacity means no reading");
305        assert_eq!(error.token(), "observation.omega.capacity_unavailable");
306    }
307
308    /// A daemon with nothing running is a real observation, and a different one
309    /// from a daemon that was never asked.
310    #[test]
311    fn a_daemon_running_nothing_yields_an_empty_reading_rather_than_a_refusal() {
312        let (_temporary, supervisor) = stub(json!({
313            "runs": [],
314            "capacity": { "activeRunLimit": 8, "activeRunCount": 0, "lanes": [] },
315        }));
316        let reading = smol::block_on(observe_issue31_full_auto(&supervisor))
317            .expect("a host that looked and found nothing still has a reading");
318        assert!(reading.run_details.is_empty());
319    }
320
321    /// The reading a host takes of a daemon it did not read is not an empty
322    /// reading — there is no way to express one.
323    ///
324    /// This is a compile-shaped assertion written as a runtime one: the only
325    /// constructor reachable from outside `issue31_delivery` is `observed`,
326    /// which takes no stamp. If someone adds a public setter or restores
327    /// `Default`, the omega#97 property is gone, and the tests that name it in
328    /// `issue31_delivery` go red.
329    #[test]
330    fn an_observed_reading_is_stamped_by_the_host_that_took_it() {
331        let before = std::time::SystemTime::now()
332            .duration_since(std::time::UNIX_EPOCH)
333            .expect("clock")
334            .as_millis() as u64;
335        let reading = Issue31FullAutoReading::observed(
336            7,
337            Vec::new(),
338            serde_json::json!({ "accounts": [] }),
339            Vec::new(),
340        );
341        let after = std::time::SystemTime::now()
342            .duration_since(std::time::UNIX_EPOCH)
343            .expect("clock")
344            .as_millis() as u64;
345        assert!(
346            reading.generated_at_ms() >= before && reading.generated_at_ms() <= after,
347            "the stamp must be this host's own clock reading, not a supplied value: \
348             {before} <= {} <= {after}",
349            reading.generated_at_ms(),
350        );
351        assert_eq!(reading.host_generation, 7);
352    }
353
354    #[test]
355    fn every_observation_failure_is_a_distinct_token() {
356        let tokens = [
357            Issue31ObservationError::RunsUnavailable("x".into()).token(),
358            Issue31ObservationError::CapacityUnavailable("x".into()).token(),
359            Issue31ObservationError::RunDetailUnavailable {
360                run_ref: "run.1".into(),
361                error: "x".into(),
362            }
363            .token(),
364        ];
365        let unique: std::collections::BTreeSet<&str> = tokens.iter().copied().collect();
366        assert_eq!(
367            unique.len(),
368            tokens.len(),
369            "a host that could not look must say which look failed",
370        );
371    }
372}
373