Build an agent cooperative

pub fn allow(
    policy: &Policy,
    action: &Action,
    local: &str,
    other: &str,
    gate: &DirectoryGate,
) -> Result<()> {
    anyhow::ensure!(
        policy.accept,
        "This member declined the proposal under its local policy"
    );
    anyhow::ensure!(
        action.schema == "chio.cooperative.computation.v1",
        "Unsupported joint action"
    );
    uuid::Uuid::parse_str(&action.nonce)?;
    let now = now_ms();
    anyhow::ensure!(
        action.issued_at_ms <= now
            && now < action.expires_at_ms
            && action.expires_at_ms - action.issued_at_ms <= 300_000,
        "Proposal expired or has an excessive validity window"
    );
    anyhow::ensure!(
        action.participants.len() == 2
            && action.participants.contains(&local.to_string())
            && action.participants.contains(&other.to_string())
            && local != other,
        "Proposal does not bind these two independent members"
    );
    anyhow::ensure!(
        action.resource_id == "shared-research-compute"
            && !action.values.is_empty()
            && action.values.len() <= policy.max_values
            && action.values.iter().all(|v| v.is_finite())
            && action.work_units == action.values.len() as u64
            && action.work_units <= policy.max_work_units,
        "Computation exceeds this member's resource policy"
    );
    let directory = gate.directory();
    anyhow::ensure!(
        directory.expires_at_unix_ms() > now
            && action
                .participants
                .iter()
                .all(|id| directory.resolve_passport_key(id).is_some()),
        "A proposal member is no longer admitted"
    );
    Ok(())
}

async fn complete(
    State(node): State<Arc<Node>>,
    h: HeaderMap,
    Json(receipts): Json<Vec<FrostDkgPackageReceipt>>,
) -> Api {
    admin(&node, &h).map_err(error)?;
    let mut guard = node.ceremony.lock().map_err(error)?;
    let state = guard.as_mut().ok_or_else(|| error("No ceremony"))?;
    let complete = complete_frost_ceremony_scoped(
        &state.config,
        state
            .secret
            .take()
            .ok_or_else(|| error("Round two already consumed"))?,
        &state.round1,
        &state.inbound,
        &receipts,
    )
    .map_err(error)?;
    private_bytes(
        &node.config.root.join("key-package.custody"),
        complete.key_package.custody_bytes(),
    )
    .map_err(error)?;
    let value = json!({
        "group_public_key":complete.group_public_key,
        "verification_shares":complete.verification_shares,
        "transcript_digest":complete.transcript_digest,
        "received_private_packages":state.inbound.len()
    });
    state.key = Some(complete.key_package);
    state.public = Some(value.clone());
    state.inbound.clear();
    Ok(Json(value))
}

    fn execute(&self, input: Value) -> Result<Value> {
        let settlement: Settlement = serde_json::from_value(input["settlement"].clone())?;
        settlement.validate()?;
        let proof: FrostAuthorizationV1 = serde_json::from_value(input["proof"].clone())?;
        let action = action(&settlement)?;
        let digest = action.action_digest()?;
        let registration = frost_action_registration(FrostAuthorizationDomain::SettleCommitment)
            .context("Settlement is not registered")?;
        let contract = registration.ladder_contract_digest()?;
        let expected = ExpectedFrostAuthorization {
            domain: FrostAuthorizationDomain::SettleCommitment,
            ladder_action_class: registration.ladder_action_class,
            ladder_contract_digest: &contract,
            scope_id: super::SCOPE,
            resource_id: &settlement.operation_id,
            resource_version: 1,
            resource_fence: 1,
            action_digest: &digest,
        };
        verify_for_execution(
            &proof,
            &expected,
            &self.0.active()?,
            &*self.0,
            &*self.0,
            &self.0.trust()?,
            now(),
        )?;
        // The local credit transfer and its useful deterministic computation commit together.
        let mut db = self
            .0
            .db
            .lock()
            .map_err(|_| anyhow::anyhow!("Settlement store lock failed"))?;
        let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let count: i64 = tx.query_row(
            "SELECT COUNT(*) FROM settlements WHERE id=?1",
            [&settlement.operation_id],
            |r| r.get(0),
        )?;
        anyhow::ensure!(
            count == 0,
            "This operation already settled; inspect the retained result instead of spending again"
        );
        let raw: String = tx.query_row("SELECT body FROM records WHERE id='credits'", [], |r| {
            r.get(0)
        })?;
        let mut credits: Value = serde_json::from_str(&raw)?;
        let available = credits["research"]
            .as_u64()
            .context("Invalid payer balance")?;
        anyhow::ensure!(
            available >= settlement.units,
            "The research account has insufficient local credits"
        );
        let mean = settlement.values.iter().sum::<f64>() / (settlement.values.len() as f64);
        let output = json!({
            "mean":mean,
            "samples":settlement.values.len(),
            "operation_id":settlement.operation_id,
            "local_credits_transferred":settlement.units,
            "action_digest":digest,
            "authorization_id":proof.body.authorization_id
        });
        credits["research"] = json!(available - settlement.units);
        credits["compute"] = json!(
            credits["compute"]
                .as_u64()
                .context("Invalid recipient balance")?
                + settlement.units
        );
        tx.execute(
            "UPDATE records SET body=?1 WHERE id='credits'",
            [serde_json::to_string(&credits)?],
        )?;
        tx.execute(
            "INSERT INTO settlements(id,action_digest,result) VALUES(?1,?2,?3)",
            rusqlite::params![
                settlement.operation_id,
                digest,
                serde_json::to_string(&output)?
            ],
        )?;
        tx.commit()?;
        Ok(output)
    }
}

async fn execute(State(node): State<Arc<Node>>, h: HeaderMap, Json(input): Json<Value>) -> Api {
    admin(&node, &h).map_err(error)?;
    let result = node
        .host
        .call(
            &node.run,
            "research-operator",
            &node.cap,
            "cooperative",
            "settle_and_compute",
            input,
        )
        .await
        .map_err(error)?;
    Ok(Json(json!({
        "allowed":result.allowed,
        "output":result.output,
        "reason":result.reason,
        "receipt_id":result.receipt_id,
        "run":node.run.snapshot().map_err(error)?
    })))
}