Operate an Agentic OS

pub async fn exercise(input: Value, run: Run, executable: PathBuf) -> Result<Value> {
    let exercise = input["exercise"]
        .as_str()
        .context("Select a recovery exercise")?;
    if ["lost-acknowledgement", "budget-exhaustion"].contains(&exercise) {
        let mut request = Fleet.sample();
        request["document"] = input["document"].clone();
        if exercise == "budget-exhaustion" {
            request["allowance"] = json!(30);
        }
        let child = Run::create(&run.directory()?.join("fleet-runs"), Fleet.name(), &request)?;
        let result = chio_worker_fleet::run_fleet(request, child.clone(), executable).await;
        child.finish(&result)?;
        let output = result?;
        run.emit(
            "operations.fleet",
            "operator",
            "Actual fleet restart followed by effect reconciliation",
            child.snapshot()?,
        )?;
        return Ok(json!({"exercise":exercise,"fleet":output}));
    }
    if exercise == "peer-removal" {
        let app = chio_personal_network::Personal;
        let mut request = app.sample();
        request["document"] = input["document"].clone();
        let child = Run::create(&run.directory()?.join("network-runs"), app.name(), &request)?;
        let result = chio_personal_network::run_network(request, child.clone(), executable).await;
        child.finish(&result)?;
        let output = result?;
        run.emit(
            "operations.network",
            "operator",
            "Live peer removal and restoration across node processes",
            child.snapshot()?,
        )?;
        return Ok(json!({"exercise":exercise,"network":output}));
    }
    anyhow::ensure!(
        [
            "worker-crash-before-dispatch",
            "host-crash-during-report",
            "cancel-running-report",
            "expired-authority"
        ]
        .contains(&exercise),
        "Choose a documented recovery exercise"
    );

async fn reconcile(
    State(service): State<Arc<Service>>,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> ApiResult {
    admin(&headers, &service)?;
    let mut stored = service.jobs.lock().await;
    let mut jobs = stored.clone();
    let job = jobs
        .iter_mut()
        .find(|job| job.operation_id.to_string() == id)
        .ok_or_else(|| error("Unknown operation"))?;
    if service.cancellations.lock().await.contains_key(&id) {
        return Err(error(
            "Worker is still executing; wait for its terminal state",
        ));
    }
    let file = service
        .config
        .directory
        .join("reports")
        .join(format!("{id}.json"));
    let resolution = if file.exists() {
        let artifact: Value =
            serde_json::from_slice(&std::fs::read(file).map_err(error)?).map_err(error)?;
        let expected =
            chio_core::sha256_hex(&chio_core::canonical_json_bytes(&job.input).map_err(error)?);
        if artifact["input_sha256"] != expected || artifact["operation_id"] != id {
            return Err(error(
                "Artifact does not match this operation; leave it unresolved",
            ));
        }
        let outcome = job.reconcile_published().map_err(error)?;
        json!({"state":"completed","reused_existing_effect":true,"artifact":artifact,"receipt_id":outcome["receipt_id"]})
    } else {
        json!({
            "state":job.label(),
            "reused_existing_effect":false,
            "recovery":"No published artifact proves completion. Inspect retained admission and partial output; this adapter does not blindly retry uncertain work."
        })
    };
    service.persist(&jobs).map_err(error)?;
    *stored = jobs;
    service
        .run
        .emit(
            "fleet.reconciled",
            "operator",
            "Reconciled the report's actual state",
            json!({"operation_id":id,"resolution":resolution}),
        )
        .map_err(error)?;
    Ok(Json(resolution))
}

async fn cancel(
    State(service): State<Arc<Service>>,
    headers: HeaderMap,
    Path(id): Path<String>,
) -> ApiResult {
    admin(&headers, &service)?;
    if let Some(flag) = service.cancellations.lock().await.get(&id) {
        flag.store(true, Ordering::SeqCst);
        return Ok(Json(json!({"cancellation_requested":true})));
    }
    let mut stored = service.jobs.lock().await;
    let mut jobs = stored.clone();
    let job = jobs
        .iter_mut()
        .find(|job| job.operation_id.to_string() == id)
        .ok_or_else(|| error("Unknown operation"))?;
    if matches!(job.state, JobState::Queued | JobState::Claimed { .. }) {
        job.cancel_before_dispatch().map_err(error)?;
        service.persist(&jobs).map_err(error)?;
        *stored = jobs;
        return Ok(Json(
            json!({"state":"cancelled_before_dispatch","receipt_id":null}),
        ));
    }
    Err(error(
        "No active worker; inspect the operation before attempting recovery",
    ))
}

    pub fn capture(root: &Path, id: &str) -> Result<Value> {
        // IDs are server-generated UUIDs. Validate before constructing paths.
        let id = uuid::Uuid::parse_str(id)?.to_string();
        let directory = root.join(id);
        let mut manifest: Value =
            serde_json::from_slice(&read_bounded(&directory.join("run.json"))?)?;
        let path = directory.join("events.ndjson");
        let mut events = Vec::new();
        if path.exists() {
            let bytes = read_bounded(&path)?;
            for line in std::str::from_utf8(&bytes)?.lines() {
                anyhow::ensure!(
                    events.len() < 4096,
                    "Retained event count exceeds its bound"
                );
                events.push(serde_json::from_str::<Event>(line)?);
            }
        }
        anyhow::ensure!(
            events
                .iter()
                .enumerate()
                .all(|(index, event)| event.sequence == index + 1),
            "Retained event sequence is incomplete"
        );
        if manifest["status"] != "running" {
            anyhow::ensure!(
                manifest["output_sha256"]
                    == chio_core::sha256_hex(&chio_core::canonical_json_bytes(
                        &manifest["output"]
                    )?),
                "Retained output digest mismatch"
            );
            anyhow::ensure!(
                manifest["events_sha256"]
                    == chio_core::sha256_hex(&chio_core::canonical_json_bytes(&events)?),
                "Retained event digest mismatch"
            );
        }
        manifest["events"] = serde_json::to_value(events)?;
        Ok(manifest)
    }
}