zc2 0.0.27

//! Publish-lock task distribution for P2P broker mesh.
//!
//! Instead of brokers calling remote workers directly, a broker that receives a
//! request it cannot serve locally **publishes** the task to its peers.  A peer
//! whose local worker matches the requirements and whose price aligns **locks**
//! the task, executes it on its own verified worker, and returns the result.
//!
//! This guarantees that only brokers with real, verified workers can execute
//! tasks — eliminating phantom-worker and phantom-owner problems by design.

use serde::{Deserialize, Serialize};

/// Task offer broadcast by the originating broker to peers.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskOffer {
    /// Unique task identifier (same as the original request_id)
    pub task_id: String,
    /// Serialised request body (opaque bytes, base64-encoded for JSON transport)
    pub payload_b64: String,
    /// Maximum price the requester is willing to pay (credits/hr).
    /// A peer will only accept if its worker's price_per_hour ≤ this value.
    pub max_price_per_hour: f64,
    /// Estimated duration in seconds (used for cost estimation)
    pub estimated_duration_secs: f64,
    /// Hard timeout — peer must abort execution after this many seconds.
    pub timeout_secs: f64,
    /// Minimum CPU cores required
    #[serde(default = "default_cpus")]
    pub cpus: f64,
    /// Minimum memory in bytes
    #[serde(default = "default_memory")]
    pub memory_bytes: u64,
    /// GPUs required
    #[serde(default)]
    pub gpus: u32,
    /// Required worker type (optional tag filter)
    #[serde(default)]
    pub worker_type: Option<String>,
    /// Required tags (all must match)
    #[serde(default)]
    pub tags: Vec<String>,
    /// The user_id being charged for this task
    pub requester_user_id: String,
    /// The originating broker's node name (for audit)
    pub source_broker: String,
    /// Shared peer secret (ZAKURO_PEER_KEY) for in-protocol QUIC auth.
    /// Empty when no key is configured; validated by the receiving broker.
    #[serde(default)]
    pub peer_key: String,
    /// Two-phase offer: when true the receiver RESERVES (does not execute) and
    /// returns a TaskAccept, awaiting a commit/cancel. False/absent = today's
    /// execute-on-offer. Set true only for two-phase-capable peers when the
    /// ZAKURO_TWO_PHASE_OFFERS flag is on.
    #[serde(default)]
    pub two_phase: bool,
    /// Dashboard-signed job voucher authorizing this task's spend (verbatim JSON
    /// bytes the dashboard signed). Empty/absent when the requester has no
    /// dashboard (local/standalone) or vouchers are not in use.
    #[serde(default)]
    pub voucher_signed_json: String,
    /// Ed25519 signature (b64) over `voucher_signed_json`.
    #[serde(default)]
    pub voucher_sig: String,
    /// Pin execution to this worker (bare name). A receiving peer accepts the
    /// offer only if it owns this worker. Empty = normal strategy routing.
    #[serde(default)]
    pub target_worker: String,
}

/// Normalize a zc:// target to its bare name (`zc://worker-x` → `worker-x`,
/// `zc://node-i9` → `node-i9`). Pass-through for a bare name.
pub fn norm_target(s: &str) -> String {
    s.strip_prefix("zc://").unwrap_or(s).to_string()
}

fn default_cpus() -> f64 {
    1.0
}
fn default_memory() -> u64 {
    1024 * 1024 * 1024
}

/// Result returned by the peer that locked and executed the task.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskResult {
    /// Echoed task identifier
    pub task_id: String,
    /// Response payload (base64-encoded)
    pub payload_b64: String,
    /// Actual wall-clock execution time (ms)
    pub duration_ms: f64,
    /// Actual cost charged (credits)
    pub actual_cost: f64,
    /// Name of the worker that executed the task
    pub worker_name: String,
    /// IP-free key-derived handle of the worker that executed the task
    /// (`zc://worker-<node_fp>-<slot>`, i.e. `Worker::zc_uri()`).
    /// Kept under its historical name — it has no `#[serde(default)]`, so an
    /// older peer would fail to deserialize a result that omitted it. The
    /// VALUE changed; the schema did not.
    pub worker_uri: String,
    /// Worker's price_per_hour (zkcr/hr)
    pub price_per_hour: f64,
    /// The peer broker's verified owner_user_id (the entity that earns credits)
    pub executor_owner: String,
    /// Worker process ID (from response header X-Zakuro-Pid, if present)
    #[serde(default)]
    pub worker_pid: Option<String>,
    /// Executing node's key-derived identity (`zc://node-<fp>`). Empty from a
    /// pre-identity peer; settlement then falls back to the locally-stamped
    /// `Worker::node_fp` (see `server.rs::executor_fp`).
    ///
    /// # UNSIGNED — attacker-controlled. Never a settlement input as-is.
    ///
    /// This field sits OUTSIDE the frozen [`receipt_bytes`], so `executor_sig`
    /// does not cover it: a malicious peer can put **any** node's identity here
    /// for free, at no cost and with no detectable tampering. It is a *claim*,
    /// not a proof. Treating it as an identity would let a peer redirect another
    /// node's earnings to itself.
    ///
    /// As of this commit nothing consumes it and settlement resolves via the
    /// locally-stamped `Worker::node_fp` (`server.rs::executor_fp`), so the
    /// exposure is latent, not live. The first consumer MUST apply both guards:
    ///
    /// 1. **Derive trust from the signature, not the claim.** Verify that this
    ///    value's fingerprint equals
    ///    `node_identity::fingerprint_of_pubkey_b64(executor_node_id)` — the key
    ///    that actually signed `executor_sig` (see [`verify_receipt`]). If they
    ///    disagree, the claim is forged: reject it, do not fall back to it.
    /// 2. **Resolving is now safe in either form.**
    ///    `PeerManager::get_url_for_fingerprint` normalizes its query with
    ///    `strip_node_arg`, so passing this field's canonical `zc://node-<fp>`
    ///    form resolves identically to bare hex. (It previously stripped only the
    ///    STORED side, so the canonical form returned a silent `None` that landed
    ///    settlement in the pay-nobody / free-work arm — a silent revenue loss,
    ///    not a loud error. Fixed; pinned by
    ///    `peer.rs::get_url_for_fingerprint_matches_cached_identity_not_host`.)
    ///    Guard 1 is still required, and is still on the consumer.
    ///
    /// Do NOT add this field to [`receipt_bytes`] to "fix" the signing gap: that
    /// would invalidate every existing peer's signature across the version
    /// boundary. The cross-check in guard 1 is the intended remedy.
    #[serde(default)]
    pub executor_node_uri: String,
    /// Executor's node identity public key (b64), used to verify `executor_sig`.
    /// None for peers that don't yet sign receipts (backward-compatible).
    #[serde(default)]
    pub executor_node_id: Option<String>,
    /// Ed25519 signature (b64) over `receipt_bytes(self)`, proving the executor
    /// node identified by `executor_node_id` produced this settlement data.
    #[serde(default)]
    pub executor_sig: Option<String>,
}

/// Canonical, stable byte encoding of the settlement-relevant fields of a
/// `TaskResult`, used as the message both signed by the executor and checked
/// by the verifier. Deliberately excludes `payload_b64` (big, irrelevant to
/// settlement) and `executor_sig`/`executor_node_id` themselves.
pub fn receipt_bytes(r: &TaskResult) -> Vec<u8> {
    format!(
        "zc-receipt-v1\n{}\n{}\n{}\n{}\n{}\n{}",
        r.task_id, r.duration_ms, r.actual_cost, r.price_per_hour, r.executor_owner, r.worker_name
    )
    .into_bytes()
}

/// Verify the proof-of-execution receipt on a `TaskResult`: the signature must
/// be valid over `receipt_bytes(r)` under `executor_node_id`, and that node's
/// key fingerprint must be authorized in `roster`. Never panics; any missing
/// field or verification failure returns `false`.
pub fn verify_receipt(r: &TaskResult, roster: &super::roster_cache::RosterCache) -> bool {
    let (node_id, sig) = match (&r.executor_node_id, &r.executor_sig) {
        (Some(n), Some(s)) => (n, s),
        _ => return false,
    };
    if !super::node_identity::verify_sig(node_id, &receipt_bytes(r), sig) {
        return false;
    }
    match super::node_identity::fingerprint_of_pubkey_b64(node_id) {
        Some(fp) => roster.fingerprint_authorized(&fp),
        None => false,
    }
}

/// Rejection returned when a peer cannot accept a task offer.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskReject {
    pub task_id: String,
    pub reason: String,
}

/// Two-phase: a peer accepts a reserve offer (reserved a worker slot, did NOT
/// execute) and awaits a commit or cancel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskAccept {
    pub task_id: String,
    pub estimated_cost: f64,
    pub worker_name: String,
    pub executor_owner: String,
    pub price_per_hour: f64,
}

/// Two-phase: tell the winning peer to execute its reserved offer.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskCommit {
    pub task_id: String,
}

/// Two-phase: tell a losing peer to release its reserved offer (no execution).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskCancel {
    pub task_id: String,
}

/// Control message: broker with workers subscribes to a peer to receive task offers (push).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubscribeMessage {
    pub action: String,
    /// Subscriber's own broker URL (so the publisher can identify and push offers).
    pub peer_url: String,
    /// Shared peer secret (ZAKURO_PEER_KEY) for in-protocol QUIC auth.
    #[serde(default)]
    pub peer_key: String,
}

/// Identity exchanged during peer handshake.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerIdentity {
    /// Broker's verified owner_user_id (validated against dashboard at startup)
    pub owner_user_id: String,
    /// Human-readable node name
    pub node_name: String,
    /// Whether this broker's owner was verified against the dashboard
    pub verified: bool,
    /// Summary of available workers and their pricing
    pub workers: Vec<WorkerSummary>,
    /// QUIC UDP port for task offers (0 = not available)
    #[serde(default)]
    pub quic_port: u16,
    /// Whether this broker supports the two-phase (reserve→commit/cancel) offer
    /// protocol. Old peers omit it → false → sender uses one-phase with them.
    #[serde(default)]
    pub supports_two_phase: bool,
}

/// Lightweight worker descriptor shared during handshake.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkerSummary {
    pub name: String,
    pub price_per_hour: f64,
    pub status: String,
    pub cpus: f64,
    pub memory_bytes: u64,
    pub gpus: u32,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn new_types_round_trip() {
        let a = TaskAccept {
            task_id: "t".into(),
            estimated_cost: 1.5,
            worker_name: "w".into(),
            executor_owner: "o".into(),
            price_per_hour: 3.6,
        };
        let s = serde_json::to_string(&a).unwrap();
        let back: TaskAccept = serde_json::from_str(&s).unwrap();
        assert_eq!(back.task_id, "t");
        assert_eq!(back.estimated_cost, 1.5);
        let c: TaskCommit = serde_json::from_str(r#"{"task_id":"t"}"#).unwrap();
        assert_eq!(c.task_id, "t");
        let x: TaskCancel = serde_json::from_str(r#"{"task_id":"t"}"#).unwrap();
        assert_eq!(x.task_id, "t");
    }

    #[test]
    fn task_offer_two_phase_defaults_false_for_old_senders() {
        // An offer JSON without `two_phase` (old sender) → false (execute-on-offer).
        let json = r#"{"task_id":"t","payload_b64":"","max_price_per_hour":1.0,
            "estimated_duration_secs":1.0,"timeout_secs":10.0,
            "requester_user_id":"u","source_broker":"n"}"#;
        let offer: TaskOffer = serde_json::from_str(json).unwrap();
        assert!(!offer.two_phase);
    }

    #[test]
    fn target_worker_defaults_empty_and_strips_scheme() {
        // An offer JSON without `target_worker` (old sender) → empty (no pin).
        let json = r#"{"task_id":"t","payload_b64":"","max_price_per_hour":1.0,
            "estimated_duration_secs":1.0,"timeout_secs":10.0,
            "requester_user_id":"u","source_broker":"n"}"#;
        let offer: TaskOffer = serde_json::from_str(json).unwrap();
        assert!(offer.target_worker.is_empty());
        // norm_target strips the zc:// scheme, passes bare names through.
        assert_eq!(norm_target("zc://worker-abc"), "worker-abc");
        assert_eq!(norm_target("zc://node-i9"), "node-i9");
        assert_eq!(norm_target("worker-abc"), "worker-abc");
    }

    #[test]
    fn peer_identity_supports_two_phase_defaults_false_for_old_peers() {
        let json = r#"{"owner_user_id":"o","node_name":"n","verified":true,"workers":[]}"#;
        let id: PeerIdentity = serde_json::from_str(json).unwrap();
        assert!(!id.supports_two_phase);
    }

    fn sample_result() -> TaskResult {
        TaskResult {
            task_id: "t1".into(),
            payload_b64: "irrelevant-to-settlement".into(),
            duration_ms: 1234.5,
            actual_cost: 0.789,
            worker_name: "worker-x".into(),
            worker_uri: "http://127.0.0.1:9000".into(),
            price_per_hour: 3.6,
            executor_owner: "owner-1".into(),
            worker_pid: None,
            executor_node_uri: String::new(),
            executor_node_id: None,
            executor_sig: None,
        }
    }

    /// The PRE-change shape of `TaskResult` as an un-upgraded peer sees it.
    /// This is a deliberate model of the *other side of the version boundary*
    /// (that code no longer exists in-tree), not a re-implementation of any
    /// production expression: field names and serde attributes are copied
    /// verbatim from the previous revision of `TaskResult`.
    #[derive(Debug, Deserialize)]
    #[allow(dead_code)]
    struct OldPeerTaskResult {
        task_id: String,
        payload_b64: String,
        duration_ms: f64,
        actual_cost: f64,
        worker_name: String,
        worker_uri: String,
        price_per_hour: f64,
        executor_owner: String,
        #[serde(default)]
        worker_pid: Option<String>,
        #[serde(default)]
        worker_ip: Option<String>,
        #[serde(default)]
        executor_node_id: Option<String>,
        #[serde(default)]
        executor_sig: Option<String>,
    }

    #[test]
    fn task_result_carries_identity_not_addresses() {
        let mut r = sample_result();
        r.worker_uri = "zc://worker-aaaaaaaaaaaaaaaa-3960".into();
        r.executor_node_uri = "zc://node-aaaaaaaaaaaaaaaa".into();
        let json = serde_json::to_string(&r).unwrap();
        assert!(
            !json.contains("worker_ip"),
            "worker_ip must be off the wire: {json}"
        );
        assert!(
            !json.contains("http://"),
            "no scheme+host on the wire: {json}"
        );
        assert!(
            !json.contains("10.13.13."),
            "no mesh address on the wire: {json}"
        );
        assert!(json.contains("zc://node-aaaaaaaaaaaaaaaa"));

        // Wire-compat, OLD → NEW (guaranteed): an old peer's result still has
        // worker_ip and no executor_node_uri. Unknown fields are ignored and the
        // new field defaults to empty.
        let old = r#"{"task_id":"t","payload_b64":"","duration_ms":1.0,"actual_cost":0.1,
            "worker_name":"w","worker_uri":"http://10.13.13.7:3960","price_per_hour":3.6,
            "executor_owner":"o","worker_ip":"10.13.13.7"}"#;
        let back: TaskResult = serde_json::from_str(old).unwrap();
        assert!(back.executor_node_uri.is_empty());

        // Wire-compat, NEW → OLD (guaranteed at the schema level): every field
        // the old shape requires is still emitted, and the two fields we changed
        // or added are tolerated (worker_ip has #[serde(default)]; the extra
        // executor_node_uri is an unknown field the old struct ignores).
        let old_view: OldPeerTaskResult = serde_json::from_str(&json)
            .expect("an un-upgraded peer must still deserialize a new-shape result");
        assert!(old_view.worker_ip.is_none());
        // The old peer reads worker_uri, but the VALUE is now an identity handle,
        // not a dialable address — semantic, not wire, incompatibility.
        assert_eq!(old_view.worker_uri, "zc://worker-aaaaaaaaaaaaaaaa-3960");
    }

    #[test]
    fn receipt_bytes_stable_and_sensitive_to_settlement_fields() {
        let a = sample_result();
        let b = sample_result();
        assert_eq!(receipt_bytes(&a), receipt_bytes(&b));

        let mut c = sample_result();
        c.actual_cost = 999.0;
        assert_ne!(receipt_bytes(&a), receipt_bytes(&c));

        // payload_b64 is excluded from the receipt.
        let mut d = sample_result();
        d.payload_b64 = "totally-different-payload".into();
        assert_eq!(receipt_bytes(&a), receipt_bytes(&d));
    }

    #[test]
    fn verify_receipt_true_for_rostered_signer() {
        use super::super::node_identity::NodeKey;
        use super::super::roster_cache::RosterCache;

        let key = NodeKey::generate();
        let mut r = sample_result();
        r.executor_node_id = Some(key.public_b64());
        r.executor_sig = Some(key.sign(&receipt_bytes(&r)));

        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);
        assert!(verify_receipt(&r, &roster));
    }

    #[test]
    fn verify_receipt_false_for_unrostered_signer() {
        use super::super::node_identity::NodeKey;
        use super::super::roster_cache::RosterCache;

        let key = NodeKey::generate();
        let other = NodeKey::generate();
        let mut r = sample_result();
        r.executor_node_id = Some(key.public_b64());
        r.executor_sig = Some(key.sign(&receipt_bytes(&r)));

        // Roster only authorizes a distinct key, not the signer.
        let roster = RosterCache::from_entries(vec![(other.public_b64(), false)]);
        assert!(!verify_receipt(&r, &roster));

        // Empty roster also rejects.
        let empty = RosterCache::from_entries(vec![]);
        assert!(!verify_receipt(&r, &empty));
    }

    #[test]
    fn verify_receipt_false_when_tampered_after_signing() {
        use super::super::node_identity::NodeKey;
        use super::super::roster_cache::RosterCache;

        let key = NodeKey::generate();
        let mut r = sample_result();
        r.executor_node_id = Some(key.public_b64());
        r.executor_sig = Some(key.sign(&receipt_bytes(&r)));
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);
        assert!(verify_receipt(&r, &roster));

        r.actual_cost += 1000.0; // tamper after signing
        assert!(!verify_receipt(&r, &roster));
    }

    #[test]
    fn verify_receipt_false_when_sig_or_node_id_missing() {
        use super::super::node_identity::NodeKey;
        use super::super::roster_cache::RosterCache;

        let key = NodeKey::generate();
        let roster = RosterCache::from_entries(vec![(key.public_b64(), false)]);

        let mut r = sample_result();
        assert!(!verify_receipt(&r, &roster)); // both missing

        r.executor_node_id = Some(key.public_b64());
        assert!(!verify_receipt(&r, &roster)); // sig missing

        r.executor_node_id = None;
        r.executor_sig = Some(key.sign(&receipt_bytes(&r)));
        assert!(!verify_receipt(&r, &roster)); // node_id missing
    }

    // NOTE: this file used to carry `mesh_facing_payloads_never_disclose_an_address`,
    // which built `TaskResult`/`PeerIdentity` by hand with `zc://…` already
    // substituted in and asserted on that literal. It never called a real
    // producer, so no production regression could fail it (proven: mutating
    // `server.rs`'s `run_offer_on` back to `worker_uri: worker.uri.clone()`
    // left it green 6/6). It has been removed in favor of tests that drive
    // the REAL producers and assert on what they actually emit:
    //   - `server.rs::task_result_producer_tests::producer_emits_no_address_in_task_result`
    //     drives `run_offer_on` against a stub worker for `TaskResult`.
    //   - `server.rs::task_result_producer_tests::producer_emits_no_address_in_peer_identity`
    //     drives `handle_peer_identity` against a registered worker for
    //     `PeerIdentity`/`WorkerSummary`.
    // See task-5-report.md in the sdd folder for the mutation evidence.
}