tenant.openagents/omega

1//! The filesystem half of harness maintenance. omega#81, `OMEGA-DELTA-0025`.
2//!
3//! [`omega_harness`] holds every decision and every contract; this module is
4//! the only thing that touches a disk, and it makes no judgements of its own.
5//! The split matters because the decision layer is a leaf that a test can run
6//! in microseconds, while this layer is bound to `Fs`, `paths`, and the async
7//! agent-launch path.
8//!
9//! The enforcement point is [`authorize_installed_harness`], and it sits
10//! between "the harness tree is on disk" and "Omega spawns it". A registry
11//! agent runs with the tool permissions of the thread that started it, so the
12//! question it answers is not *may we install this* but *may these exact bytes
13//! run* — which is why the digest is measured from the installed tree on every
14//! launch rather than trusted from the install that produced it.
15
16use std::{
17    path::{Path, PathBuf},
18    sync::Arc,
19};
20
21use anyhow::Result;
22use fs::Fs;
23use futures::StreamExt as _;
24use omega_harness::{
25    HARNESS_MAINTENANCE_LOG_FILE_NAME, HARNESS_PIN_LEDGER_FILE_NAME, HarnessDistribution,
26    HarnessFrontDoorState, HarnessMaintenanceReceipt, HarnessPinLedger, HarnessPinLedgerError,
27    InstallationProvenance, LoadedPinLedger, MaintenanceAction, MaintenanceDecision,
28    MaintenanceRefusal, MeasuredDigest, admits_package_manager_launch, admits_version,
29    decide_maintenance, encode_harness_pin_ledger, harness_front_door_state, latest_record_for,
30    receipt_for_decision, receipt_log_line,
31};
32use util::ResultExt as _;
33
34/// The host reference every receipt written on this machine carries.
35///
36/// A constant rather than a device identity on purpose. The log is local, so
37/// the host is whichever machine the file is on; minting a stable device id
38/// here would put a correlatable identifier into a file that has no reason to
39/// carry one, and `omega#75` is explicit that the first-party agent does not
40/// sign with its own principal.
41pub const LOCAL_HOST_REF: &str = "host.omega.local";
42
43/// The most files one installed harness tree may hold and still be measured.
44///
45/// A bound rather than an ambition. Exceeding it yields no measurement, which
46/// [`decide_maintenance`] refuses — a harness too large to attest does not run
47/// unattested.
48pub const MAX_MEASURED_TREE_FILES: usize = 4096;
49
50/// The largest single file that will be read into memory to measure it.
51pub const MAX_MEASURED_FILE_BYTES: u64 = 512 * 1024 * 1024;
52
53/// Where the owner's pins live.
54pub fn pin_ledger_path() -> PathBuf {
55    paths::external_agents_dir().join(HARNESS_PIN_LEDGER_FILE_NAME)
56}
57
58/// Where maintenance receipts are appended.
59pub fn receipt_log_path() -> PathBuf {
60    paths::external_agents_dir().join(HARNESS_MAINTENANCE_LOG_FILE_NAME)
61}
62
63/// Read the pin ledger.
64///
65/// A file that is not there is [`LoadedPinLedger::Absent`]. A file that is
66/// there and cannot be read — bad JSON, or an IO error — is
67/// [`LoadedPinLedger::Unreadable`], which refuses. Collapsing those two would
68/// make deleting one byte of the ledger a way to unfreeze every harness.
69pub async fn load_pin_ledger(fs: &dyn Fs, path: &Path) -> LoadedPinLedger {
70    if !fs.is_file(path).await {
71        return LoadedPinLedger::Absent;
72    }
73    match fs.load(path).await {
74        Ok(contents) => LoadedPinLedger::read(Some(&contents)),
75        Err(_) => LoadedPinLedger::Unreadable(HarnessPinLedgerError::InvalidJson),
76    }
77}
78
79/// Measure every regular file in an installed harness tree.
80///
81/// `None` is "this host could not read what would run", which the decision
82/// layer treats as a refusal rather than as an absence of objection. Every
83/// failure path here — an unreadable file, a tree above the bound, a directory
84/// that vanished mid-walk — collapses to `None` deliberately: a partial
85/// measurement is a wrong measurement, and there is no way to express one.
86pub async fn measure_installed_tree(fs: &dyn Fs, dir: &Path) -> Option<MeasuredDigest> {
87    let mut files: Vec<(String, MeasuredDigest)> = Vec::new();
88    let mut pending: Vec<PathBuf> = vec![dir.to_path_buf()];
89
90    while let Some(current) = pending.pop() {
91        let mut entries = fs.read_dir(&current).await.ok()?;
92        while let Some(entry) = entries.next().await {
93            let entry = entry.ok()?;
94            let metadata = fs.metadata(&entry).await.ok()??;
95            if metadata.is_dir {
96                pending.push(entry);
97                continue;
98            }
99            // A symlink's target is not part of this tree, and following one
100            // would let a link planted in the install directory attest bytes
101            // from anywhere on the machine. Its presence still changes the
102            // tree digest, because the path is recorded.
103            let relative = entry.strip_prefix(dir).ok()?.to_str()?.to_string();
104            if metadata.is_symlink {
105                files.push((relative, MeasuredDigest::measure(b"")));
106            } else {
107                if metadata.len > MAX_MEASURED_FILE_BYTES {
108                    return None;
109                }
110                files.push((relative, MeasuredDigest::measure(&fs.load_bytes(&entry).await.ok()?)));
111            }
112            if files.len() > MAX_MEASURED_TREE_FILES {
113                return None;
114            }
115        }
116    }
117
118    if files.is_empty() {
119        return None;
120    }
121    Some(MeasuredDigest::measure_tree(&mut files))
122}
123
124/// Append one receipt to the log.
125///
126/// Best-effort: a log that cannot be written must not stop a *refusal* from
127/// taking effect, because the refusal is the safety property and the record is
128/// the evidence for it. The caller therefore refuses first and records second.
129pub async fn append_receipt(fs: &dyn Fs, path: &Path, receipt: &HarnessMaintenanceReceipt) {
130    let Some(parent) = path.parent() else {
131        return;
132    };
133    if fs.create_dir(parent).await.log_err().is_none() {
134        return;
135    }
136    let mut existing = fs.load(path).await.unwrap_or_default();
137    existing.push_str(&receipt_log_line(receipt));
138    fs.write(path, existing.as_bytes()).await.log_err();
139}
140
141/// The gate that runs *before* a harness version is fetched.
142///
143/// Returns the sentence to show if the fetch must not happen. See
144/// [`admits_version`] for why this is a prefilter and not the authority.
145pub async fn authorize_version_fetch(
146    fs: Arc<dyn Fs>,
147    harness_id: &str,
148    candidate_version: &str,
149    now_ms: u64,
150) -> Result<()> {
151    let ledger = load_pin_ledger(fs.as_ref(), &pin_ledger_path()).await;
152    let Some(refusal) = admits_version(ledger.pin_state(harness_id), candidate_version) else {
153        return Ok(());
154    };
155    let decision = MaintenanceDecision::Refused(refusal.clone());
156    if let Some(receipt) = receipt_for_decision(
157        LOCAL_HOST_REF,
158        harness_id,
159        now_ms,
160        MaintenanceAction::Update,
161        &decision,
162    )
163    .log_err()
164    {
165        append_receipt(fs.as_ref(), &receipt_log_path(), &receipt).await;
166    }
167    anyhow::bail!("{}", refusal.reason())
168}
169
170/// The gate that runs *after* the tree is on disk and before Omega spawns it.
171///
172/// This is the enforcement point the omega#81 falsifier names: a binary that
173/// will run with full tool permissions does not run unless this host hashed it
174/// and the owner's pins admit the hash. Every call writes a receipt, permitted
175/// or refused.
176pub async fn authorize_installed_harness(
177    fs: Arc<dyn Fs>,
178    harness_id: &str,
179    version: &str,
180    installed_dir: &Path,
181    action: MaintenanceAction,
182    now_ms: u64,
183) -> Result<MeasuredDigest> {
184    let ledger = load_pin_ledger(fs.as_ref(), &pin_ledger_path()).await;
185    let measured = measure_installed_tree(fs.as_ref(), installed_dir).await;
186    let candidate = match measured.as_ref() {
187        Some(digest) => omega_harness::CandidateArtifact::Measured { version, digest },
188        None => omega_harness::CandidateArtifact::Unmeasured { version },
189    };
190
191    let decision = decide_maintenance(ledger.pin_state(harness_id), candidate);
192    if let Some(receipt) =
193        receipt_for_decision(LOCAL_HOST_REF, harness_id, now_ms, action, &decision).log_err()
194    {
195        append_receipt(fs.as_ref(), &receipt_log_path(), &receipt).await;
196    }
197
198    match decision {
199        MaintenanceDecision::Permitted { digest, .. } => Ok(digest.clone()),
200        MaintenanceDecision::Refused(refusal) => anyhow::bail!("{}", refusal.reason()),
201    }
202}
203
204/// The resolver name recorded for npm-distributed harnesses.
205pub const NPX_RESOLVER: &str = "npx";
206
207/// The gate for a harness whose bytes no directory of Omega's holds.
208///
209/// `LocalRegistryNpxAgent` resolves its package inside the node runtime's own
210/// cache, at exec time, against a version *range* rather than an exact version
211/// (see `bounded_npm_package_spec`). There is no tree to measure, so
212/// [`authorize_installed_harness`] cannot be the gate here — routing this path
213/// through it would refuse every npx harness on every machine, which is a
214/// larger change than omega#81 asks for and would be dishonest about what the
215/// measurement proved.
216///
217/// What this closes instead is the state where an owner froze such a harness
218/// and Omega ran whatever `npm` resolved anyway. A pin that cannot be enforced
219/// refuses. Every call writes a receipt, permitted or refused, so the log
220/// records that this launch was *not* attested rather than leaving a silence a
221/// reader would mistake for an attested one.
222pub async fn authorize_package_manager_launch(
223    fs: Arc<dyn Fs>,
224    harness_id: &str,
225    resolver: &str,
226    now_ms: u64,
227) -> Result<()> {
228    let ledger = load_pin_ledger(fs.as_ref(), &pin_ledger_path()).await;
229    let Some(refusal) = admits_package_manager_launch(ledger.pin_state(harness_id), resolver)
230    else {
231        return Ok(());
232    };
233    record_refusal(
234        fs.as_ref(),
235        harness_id,
236        now_ms,
237        MaintenanceAction::Verify,
238        &refusal,
239    )
240    .await;
241    anyhow::bail!("{}", refusal.reason())
242}
243
244/// Decide whether a version the registry channel now names may be offered.
245///
246/// Separate from the launch gates, and it runs when nothing is about to launch:
247/// the registry document refreshes in the background and the store decides
248/// whether to tell the owner an update is available. A harness frozen at
249/// another version must not be offered one — the offer would lead to a refusal
250/// on the next launch, which is the front door promising something the gate
251/// then takes back.
252///
253/// Returns the refusal when the channel's answer is not offerable. The refusal
254/// is recorded under [`MaintenanceAction::ResolveChannel`], because an update
255/// that is never started leaves no other trace that it was considered.
256pub async fn resolve_channel(
257    fs: Arc<dyn Fs>,
258    harness_id: &str,
259    resolved_version: &str,
260    now_ms: u64,
261) -> Option<MaintenanceRefusal> {
262    let ledger = load_pin_ledger(fs.as_ref(), &pin_ledger_path()).await;
263    let refusal = admits_version(ledger.pin_state(harness_id), resolved_version)?;
264    record_refusal(
265        fs.as_ref(),
266        harness_id,
267        now_ms,
268        MaintenanceAction::ResolveChannel,
269        &refusal,
270    )
271    .await;
272    Some(refusal)
273}
274
275/// Re-establish what an installed tree is, on the owner's request.
276///
277/// The same measurement the launch path takes, under its own action, with
278/// nothing about to run. It is what the front door's verify control does and
279/// what a caller runs after an update rather than carrying the prior
280/// measurement forward.
281pub async fn reprobe_installed_harness(
282    fs: Arc<dyn Fs>,
283    harness_id: &str,
284    version: &str,
285    installed_dir: &Path,
286    now_ms: u64,
287) -> Result<MeasuredDigest> {
288    authorize_installed_harness(
289        fs,
290        harness_id,
291        version,
292        installed_dir,
293        MaintenanceAction::ReprobeCapability,
294        now_ms,
295    )
296    .await
297}
298
299/// Write one refusal to the log.
300///
301/// Goes through [`receipt_for_decision`] like every other write in this module,
302/// so there is still exactly one receipt writer and it still takes a decision
303/// rather than a caller's belief about one.
304async fn record_refusal(
305    fs: &dyn Fs,
306    harness_id: &str,
307    now_ms: u64,
308    action: MaintenanceAction,
309    refusal: &MaintenanceRefusal,
310) {
311    let decision = MaintenanceDecision::Refused(refusal.clone());
312    if let Some(receipt) =
313        receipt_for_decision(LOCAL_HOST_REF, harness_id, now_ms, action, &decision).log_err()
314    {
315        append_receipt(fs, &receipt_log_path(), &receipt).await;
316    }
317}
318
319/// Persist a pin ledger.
320///
321/// [`encode_harness_pin_ledger`] routes through its own reader before returning,
322/// so this cannot leave a file the next launch would read as
323/// [`LoadedPinLedger::Unreadable`] — which would freeze every harness on the
324/// machine rather than the one the owner meant.
325pub async fn write_pin_ledger(fs: &dyn Fs, ledger: &HarnessPinLedger) -> Result<()> {
326    let path = pin_ledger_path();
327    let contents = encode_harness_pin_ledger(ledger)?;
328    if let Some(parent) = path.parent() {
329        fs.create_dir(parent).await?;
330    }
331    fs.write(&path, contents.as_bytes()).await?;
332    Ok(())
333}
334
335/// Freeze a harness at the bytes this host measures now. The front door's
336/// "Pin" control.
337///
338/// The measurement is taken through [`authorize_installed_harness`], not beside
339/// it. That means taking a pin runs the real gate and writes a real receipt,
340/// and a tree the gate refuses cannot be pinned — a pin recorded against bytes
341/// the host would not run is a pin that means nothing.
342///
343/// It refuses to overwrite an existing pin. Moving a pin from one release to
344/// another is remove-then-pin, two deliberate actions, because one click that
345/// silently re-pointed a freeze at whatever is currently installed would undo
346/// the freeze in the exact case it exists for.
347pub async fn pin_installed_harness(
348    fs: Arc<dyn Fs>,
349    harness_id: &str,
350    version: &str,
351    installed_dir: &Path,
352    now_ms: u64,
353) -> Result<()> {
354    let loaded = load_pin_ledger(fs.as_ref(), &pin_ledger_path()).await;
355    let mut ledger = match loaded {
356        LoadedPinLedger::Absent => HarnessPinLedger::empty(),
357        LoadedPinLedger::Loaded(ledger) => ledger,
358        LoadedPinLedger::Unreadable(error) => {
359            anyhow::bail!(
360                "Omega cannot read the pin ledger ({error}), so it will not rewrite it. \
361                 Writing would drop pins it cannot see."
362            )
363        }
364    };
365    anyhow::ensure!(
366        ledger.pin(harness_id).is_none(),
367        "{harness_id} is already pinned. Remove the pin before taking a new one."
368    );
369
370    let digest = authorize_installed_harness(
371        fs.clone(),
372        harness_id,
373        version,
374        installed_dir,
375        MaintenanceAction::ReprobeCapability,
376        now_ms,
377    )
378    .await?;
379
380    ledger.set_pin(harness_id, version, &digest);
381    write_pin_ledger(fs.as_ref(), &ledger).await
382}
383
384/// Unfreeze a harness. The front door's "Remove pin" control.
385///
386/// An unreadable ledger refuses here too. "Remove the pin" on a file whose pins
387/// cannot be read would mean rewriting it from the subset this build could
388/// parse, which is not removal — it is deletion of everything unreadable.
389pub async fn unpin_harness(fs: Arc<dyn Fs>, harness_id: &str) -> Result<()> {
390    let loaded = load_pin_ledger(fs.as_ref(), &pin_ledger_path()).await;
391    let mut ledger = match loaded {
392        LoadedPinLedger::Absent => return Ok(()),
393        LoadedPinLedger::Loaded(ledger) => ledger,
394        LoadedPinLedger::Unreadable(error) => {
395            anyhow::bail!(
396                "Omega cannot read the pin ledger ({error}), so it will not rewrite it. \
397                 Writing would drop pins it cannot see."
398            )
399        }
400    };
401    if !ledger.remove_pin(harness_id) {
402        return Ok(());
403    }
404    write_pin_ledger(fs.as_ref(), &ledger).await
405}
406
407/// Read everything the front door shows for one harness.
408///
409/// Reads only. Nothing here writes a receipt, because looking at a settings
410/// page is not a maintenance action and a log that recorded every render would
411/// bury the actions that mattered.
412pub async fn read_front_door_state(
413    fs: &dyn Fs,
414    harness_id: &str,
415    version: &str,
416    distribution: HarnessDistribution,
417    installed_dir: Option<&Path>,
418) -> HarnessFrontDoorState {
419    let ledger = load_pin_ledger(fs, &pin_ledger_path()).await;
420    let measured = match installed_dir {
421        Some(dir) if fs.is_dir(dir).await => measure_installed_tree(fs, dir).await,
422        _ => None,
423    };
424    let log = fs.load(&receipt_log_path()).await.unwrap_or_default();
425    let provenance = if log.is_empty() {
426        InstallationProvenance::Unattested
427    } else {
428        latest_record_for(&log, harness_id)
429    };
430
431    harness_front_door_state(
432        harness_id,
433        version,
434        distribution,
435        ledger.pin_state(harness_id),
436        measured.as_ref(),
437        &provenance,
438    )
439}
440
441/// The host's clock, read once.
442///
443/// Every receipt this module writes is stamped from one call to this function,
444/// made at the moment the action happened. Nothing that reaches the receipt
445/// writer carries a time of its own, so there is no path by which a registry
446/// document or a settings file can supply one.
447pub fn now_ms() -> u64 {
448    std::time::SystemTime::now()
449        .duration_since(std::time::UNIX_EPOCH)
450        .map(|elapsed| elapsed.as_millis() as u64)
451        .unwrap_or_default()
452}
453