diff --git a/crates/mobee/src/agent_presets.rs b/crates/mobee-core/src/agent_presets.rs similarity index 99% rename from crates/mobee/src/agent_presets.rs rename to crates/mobee-core/src/agent_presets.rs index c998fb01..8a8414e2 100644 --- a/crates/mobee/src/agent_presets.rs +++ b/crates/mobee-core/src/agent_presets.rs @@ -7,7 +7,7 @@ use std::collections::BTreeMap; use std::path::{Path, PathBuf}; -use mobee_core::home::AgentPresetConfig; +use crate::home::AgentPresetConfig; /// Built-in preset names, in the order they are suggested/detected. pub const BUILTIN_PRESETS: [&str; 3] = ["claude", "cursor", "codex"]; diff --git a/crates/mobee-core/src/buyer/lifecycle.rs b/crates/mobee-core/src/buyer/lifecycle.rs index 549bc0bc..f08f2bf6 100644 --- a/crates/mobee-core/src/buyer/lifecycle.rs +++ b/crates/mobee-core/src/buyer/lifecycle.rs @@ -24,10 +24,9 @@ use super::reservations::{Converted, JobDisposition, ReservationState, ReserveRe use super::store::{BuyerStore, StoreError}; /// Hard filters an awardable claim must pass (issue #126). Grounded in the wire the offer/claim -/// actually carry: the offer's signed `amount_sats` is the fixed price, and the seller's claim -/// `creq` carries the payable terms + accepted mints. (`harness`/`model` targeting from #126 has -/// no offer/claim wire field yet, so it is deliberately not a filter here — it is added when the -/// wire carries it, rather than matched against a field that does not exist.) +/// actually carry: the offer's signed `amount_sats` is the fixed price, the seller's claim `creq` +/// carries the payable terms + accepted mints, and the claim's `mobee_agent` tag carries the +/// harnesses the seller can run. pub struct AwardFilters<'a> { /// The offer's signed amount — authority for the price. A claim whose `creq` quotes a /// different amount can never be accepted (the accept gate requires exact equality), so it @@ -40,6 +39,10 @@ pub struct AwardFilters<'a> { pub buyer_mint: &'a str, /// Whether real (non-testnut) mints are permitted; gates the mint-compat check. pub allow_real_mints: bool, + /// The harness the OFFER asked for, read back from the relay (never from award params — the + /// signed offer is the authority for what the job requested). `None` ⇒ no preference and every + /// claim passes this filter unchanged. + pub requested_agent: Option<&'a str>, } /// Select the claim to auto-award: the first LIVE claim whose seller-authored `creq` passes every @@ -51,10 +54,29 @@ pub fn select_awardable_claim(view: &JobView, filters: &AwardFilters) -> Option< } view.claims .iter() - .find(|claim| claim.live && claim_is_payable(&view.job_id, claim.creq.as_deref(), filters)) + .find(|claim| { + claim.live + && claim_serves_requested_agent(&claim.agents, filters.requested_agent) + && claim_is_payable(&view.job_id, claim.creq.as_deref(), filters) + }) .map(|claim| claim.claim_id.clone()) } +/// Whether a claim may be awarded a job that asked for a specific harness. +/// +/// No request ⇒ every claim passes. A request ⇒ the claim must ADVERTISE that harness. A claim +/// that advertises nothing does not pass: silence is not a capability, and awarding it would be +/// paying a seller to run the job on whatever it happens to prefer. Matching is on the +/// canonicalised name so wire casing/whitespace cannot smuggle a mismatch past the filter. +pub fn claim_serves_requested_agent(claim_agents: &[String], requested: Option<&str>) -> bool { + let Some(requested) = crate::seller_agents::normalize_request(requested) else { + return true; + }; + claim_agents + .iter() + .any(|advertised| advertised.trim().to_ascii_lowercase() == requested) +} + /// Why a specifically-named (manual) award was refused. The manual path names a `claim_id` instead /// of auto-selecting, so it must apply the SAME hard filters `select_awardable_claim` applies — /// otherwise `max_sats` and mint/price compatibility would be dead input on the manual path. @@ -69,6 +91,9 @@ pub enum NamedAwardRefused { /// The named claim cannot be paid (missing/malformed creq, price ≠ offer amount, wrong unit, or /// no mutually-payable mint) — awarding it would commit to something the buyer cannot settle. Unpayable { claim_id: String }, + /// The job asked for a harness the named claim does not advertise — awarding it would buy work + /// from a seller that never said it could do it this way. + AgentMismatch { claim_id: String, requested: String }, } impl std::fmt::Display for NamedAwardRefused { @@ -84,6 +109,10 @@ impl std::fmt::Display for NamedAwardRefused { formatter, "award refused: claim {claim_id} is not payable (price/mint/creq incompatible — the buyer could not settle it)" ), + Self::AgentMismatch { claim_id, requested } => write!( + formatter, + "award refused: job requested agent {requested:?}, which claim {claim_id} does not advertise" + ), } } } @@ -112,6 +141,12 @@ pub fn named_claim_awardable( if !claim.live { return Err(NamedAwardRefused::NotLive { claim_id: claim_id.to_owned() }); } + if !claim_serves_requested_agent(&claim.agents, filters.requested_agent) { + return Err(NamedAwardRefused::AgentMismatch { + claim_id: claim_id.to_owned(), + requested: filters.requested_agent.unwrap_or_default().to_owned(), + }); + } if !claim_is_payable(&view.job_id, claim.creq.as_deref(), filters) { return Err(NamedAwardRefused::Unpayable { claim_id: claim_id.to_owned() }); } @@ -384,6 +419,7 @@ mod tests { branch: None, job_class: None, contribution: None, + requested_agent: None, } } @@ -397,6 +433,7 @@ mod tests { status: "processing".into(), live, creq: Some(creq), + agents: Vec::new(), } } @@ -418,6 +455,7 @@ mod tests { max_sats, buyer_mint: DEFAULT_MINT_URL, allow_real_mints: false, + requested_agent: None, } } @@ -430,6 +468,74 @@ mod tests { assert_eq!(selected.as_deref(), Some("c".repeat(64).as_str())); } + // TOOTH (charter invariant 5, the strong one) — a job that asked for a harness is never + // awarded to a claim that does not advertise it, on EITHER award path. Everything else about + // these claims is payable: same price, same mint, live. Bite: drop the + // `claim_serves_requested_agent` arm from `select_awardable_claim` and the codex-only claim + // below wins a claude job; drop it from `named_claim_awardable` and the manual path pays it. + #[test] + fn a_job_requesting_a_harness_is_never_awarded_to_a_claim_without_it() { + let job = "a".repeat(64); + let mut codex_only = claim(&job, true, 10, &[DEFAULT_MINT_URL.into()]); + codex_only.agents = vec!["codex".to_owned()]; + let mut silent = claim(&job, true, 10, &[DEFAULT_MINT_URL.into()]); + silent.claim_id = "d".repeat(64); + let view = view_with(&job, 10, vec![codex_only, silent]); + + let mut wants_claude = filters(10, 100); + wants_claude.requested_agent = Some("claude"); + assert_eq!( + select_awardable_claim(&view, &wants_claude), + None, + "neither a wrong-harness claim nor a silent one may win a claude job" + ); + // …and the manual path refuses by NAME, with the reason on the error. + let refused = named_claim_awardable(&view, &"c".repeat(64), &wants_claude) + .expect_err("manual award must apply the same filter"); + assert!( + matches!(refused, NamedAwardRefused::AgentMismatch { .. }), + "unexpected refusal: {refused:?}" + ); + assert!(refused.to_string().contains("claude"), "{refused}"); + assert!( + matches!( + named_claim_awardable(&view, &"d".repeat(64), &wants_claude), + Err(NamedAwardRefused::AgentMismatch { .. }) + ), + "a claim advertising nothing does not satisfy a request either" + ); + + // The claim that DOES advertise it is awarded, so the filter selects rather than blocks. + let mut wants_codex = filters(10, 100); + wants_codex.requested_agent = Some("codex"); + assert_eq!( + select_awardable_claim(&view, &wants_codex).as_deref(), + Some("c".repeat(64).as_str()) + ); + assert!(named_claim_awardable(&view, &"c".repeat(64), &wants_codex).is_ok()); + } + + // Compat: with no harness requested the award path behaves exactly as before — a claim that + // advertises nothing is still awardable, so existing sellers keep winning existing jobs. + #[test] + fn no_harness_request_awards_exactly_as_before() { + let job = "a".repeat(64); + let view = view_with(&job, 10, vec![claim(&job, true, 10, &[DEFAULT_MINT_URL.into()])]); + let unfiltered = filters(10, 100); + assert!(unfiltered.requested_agent.is_none()); + assert_eq!( + select_awardable_claim(&view, &unfiltered).as_deref(), + Some("c".repeat(64).as_str()) + ); + // The explicit "any" is the same case as no request at all. + let mut any = filters(10, 100); + any.requested_agent = Some("any"); + assert_eq!( + select_awardable_claim(&view, &any).as_deref(), + Some("c".repeat(64).as_str()) + ); + } + // A non-live claim is never selected (nothing to award yet). #[test] fn select_skips_non_live_claim() { diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index a2d3f396..9bdd3a92 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -349,8 +349,9 @@ struct PostJobParams { /// never auto-awards a claim it cannot pay or priced above this. #[serde(default)] max_sats: Option, - /// Auto-award preferences recorded with the intent. Not yet hard filters — no offer/claim wire - /// field carries harness/model, so they are stored, not matched (added when the wire does). + /// Auto-award preferences recorded with the intent. `harness` is ALSO posted on the offer as + /// its requested agent, so it is a hard award filter: only a seller advertising that harness + /// can be awarded. `model` has no wire field yet and stays a recorded preference. #[serde(default)] harness: Option, #[serde(default)] @@ -410,6 +411,7 @@ async fn post_job(context: &Arc, id: Value, params: Value) -> Resp repo: params.repo, branch: params.branch, job, + requested_agent: harness.clone(), }; match job_lifecycle::post_job_async(&context.home, request).await { Ok(outcome) => { @@ -523,6 +525,7 @@ async fn award(context: &BuyerContext, id: Value, params: Value) -> Response { max_sats, buyer_mint: context.home.config.default_mint(), allow_real_mints: context.home.config.allow_real_mints, + requested_agent: offer.requested_agent.as_deref(), }; // Manual award names the claim but applies the SAME hard filters (max_sats, price, mint) as @@ -783,6 +786,7 @@ async fn drive_auto_award( max_sats, buyer_mint: context.home.config.default_mint(), allow_real_mints: context.home.config.allow_real_mints, + requested_agent: offer.requested_agent.as_deref(), }; if let Some(claim_id) = lifecycle::select_awardable_claim(&view, &filters) { return finalize_auto_award(context, job_id, offer.amount_sats, claim_id).await; @@ -1700,7 +1704,7 @@ mod tests { &seller_hex, ) .expect("creq"); - let claim_draft = crate::gateway::claim_draft(&job_id, &buyer_hex, &seller_hex, &creq); + let claim_draft = crate::gateway::claim_draft(&job_id, &buyer_hex, &seller_hex, &creq, &[]); let _ = publish(&seller, &claim_draft).await; let job_hash = job_lifecycle::job_hash_for_offer(&job_id, task, amount); diff --git a/crates/mobee-core/src/collect.rs b/crates/mobee-core/src/collect.rs index 93d18c28..9db19d56 100644 --- a/crates/mobee-core/src/collect.rs +++ b/crates/mobee-core/src/collect.rs @@ -530,7 +530,7 @@ mod tests { &seller_hex, ) .expect("creq"); - let claim_draft = crate::gateway::claim_draft(&offer_id, &buyer_hex, &seller_hex, &creq); + let claim_draft = crate::gateway::claim_draft(&offer_id, &buyer_hex, &seller_hex, &creq, &[]); let _ = publish(&seller, &claim_draft).await; let job_hash = crate::job_lifecycle::job_hash_for_offer(&offer_id, task, amount); diff --git a/crates/mobee-core/src/gateway.rs b/crates/mobee-core/src/gateway.rs index 088b443d..06f7cbb7 100644 --- a/crates/mobee-core/src/gateway.rs +++ b/crates/mobee-core/src/gateway.rs @@ -57,6 +57,9 @@ pub struct OfferDraft { pub amount_sats: u64, pub deadline_unix: u64, pub seller_pubkey: Option, + /// The harness this job asks for, as `["param", "agent", …]`. `None` (or `"any"`) ⇒ no + /// preference: any seller may claim and run it on whichever harness it prefers. + pub requested_agent: Option, } impl OfferDraft { @@ -73,6 +76,7 @@ impl OfferDraft { amount_sats, deadline_unix, seller_pubkey: Some(seller_pubkey.into()), + requested_agent: None, } } @@ -88,9 +92,17 @@ impl OfferDraft { amount_sats, deadline_unix, seller_pubkey: None, + requested_agent: None, } } + /// Request a specific harness for this job. A canonicalised-away value (`any`, blank) records + /// no request, so "no preference" has exactly one representation on the wire. + pub fn requesting_agent(mut self, requested_agent: Option<&str>) -> Self { + self.requested_agent = crate::seller_agents::normalize_request(requested_agent); + self + } + pub fn to_event_draft(&self) -> EventDraft { // The offer does not name a mint — the seller authors the accepted mint(s) in its claim // `creq`, so there is no `["mint", …]` tag here. @@ -100,6 +112,13 @@ impl OfferDraft { TagSpec::new(["amount", &self.amount_sats.to_string(), "sat"]), TagSpec::new(["param", "deadline", &self.deadline_unix.to_string()]), ]; + if let Some(requested_agent) = &self.requested_agent { + tags.push(TagSpec::new([ + "param", + crate::seller_agents::AGENT_PARAM, + requested_agent, + ])); + } if let Some(seller_pubkey) = &self.seller_pubkey { tags.push(TagSpec::new(["p", seller_pubkey])); } @@ -118,6 +137,9 @@ pub struct ParsedOffer { pub unit: String, pub deadline_unix: u64, pub seller_pubkey: Option, + /// The harness this job requested, canonicalised. `None` ⇒ no preference (the parameter was + /// absent, blank, or the explicit `any`). + pub requested_agent: Option, } impl ParsedOffer { @@ -305,9 +327,24 @@ pub fn parse_offer(event: &EventDraft) -> Result { unit: unit.clone(), deadline_unix, seller_pubkey: first_tag_value(&event.tags, "p").map(str::to_owned), + requested_agent: crate::seller_agents::normalize_request(param_value( + &event.tags, + crate::seller_agents::AGENT_PARAM, + )), }) } +/// Read a `["param", , ]` parameter off an event's tags. +fn param_value<'a>(tags: &'a [TagSpec], name: &str) -> Option<&'a str> { + tags.iter() + .find(|tag| { + tag.0.first().map(String::as_str) == Some("param") + && tag.0.get(1).map(String::as_str) == Some(name) + }) + .and_then(|tag| tag.0.get(2)) + .map(String::as_str) +} + /// Parses the buyer-visible git delivery fields carried by a result event. pub fn parse_git_result_delivery(event: &EventDraft) -> Result { if event.kind != JOB_RESULT_KIND { @@ -365,22 +402,27 @@ pub fn parse_bound_git_delivery( /// NUT-18 payment request as a `["creq", "creqA…"]` tag — the claim *is* /// the invoice. Build `creq` with [`creq::build_seller_creq`]; buyers read it back with /// [`creq::parse_creq`]. +/// +/// `agents` advertises the harnesses this seller can run (preference order) as +/// `["mobee_agent", …]`, so the buyer's award filter can hold the claim to the harness its job +/// asked for. Empty ⇒ the tag is omitted and the claim is byte-identical to a pre-registry claim. pub fn claim_draft( offer_id: &str, buyer_pubkey: &str, seller_pubkey: &str, creq: &str, + agents: &[String], ) -> EventDraft { - status_draft( - JOB_CLAIM_KIND, - "processing", - vec![ - TagSpec::new(["e", offer_id]), - TagSpec::new(["p", buyer_pubkey]), - TagSpec::new(["p", seller_pubkey]), - TagSpec::new(["creq", creq]), - ], - ) + let mut tags = vec![ + TagSpec::new(["e", offer_id]), + TagSpec::new(["p", buyer_pubkey]), + TagSpec::new(["p", seller_pubkey]), + TagSpec::new(["creq", creq]), + ]; + if let Some(tag) = crate::heartbeat::agent_tag(agents) { + tags.push(tag); + } + status_draft(JOB_CLAIM_KIND, "processing", tags) } /// Kind-award AWARD draft (`status=accepted`). Buyer-authored selection of a claim — e-tags the @@ -774,6 +816,64 @@ pub mod creq { #[cfg(test)] mod tests { + // TOOTH — an offer's harness request rides the existing `param` grammar and round-trips, and + // "no preference" has exactly ONE representation on the wire: no tag. `any` and blank + // canonicalise to that same absence, so a buyer stating indifference and one omitting it post + // byte-identical offers. + #[test] + fn offer_carries_a_requested_agent_or_nothing_at_all() { + let plain = OfferDraft::untargeted("t", "text/plain", 5, 1_800_000_001); + let asking = plain.clone().requesting_agent(Some("Codex")); + let draft = asking.to_event_draft(); + let param = draft + .tags + .iter() + .find(|tag| tag.first() == Some("param") && tag.0.get(1).map(String::as_str) == Some("agent")) + .expect("offer carries the agent param"); + assert_eq!(param.0, vec!["param", "agent", "codex"], "canonicalised on the way out"); + assert_eq!( + parse_offer(&draft).expect("parse").requested_agent.as_deref(), + Some("codex") + ); + + for indifferent in [None, Some("any"), Some(" "), Some("ANY")] { + let draft = plain.clone().requesting_agent(indifferent).to_event_draft(); + assert_eq!( + draft, + plain.to_event_draft(), + "{indifferent:?} must post the same offer as no request at all" + ); + assert_eq!(parse_offer(&draft).expect("parse").requested_agent, None); + } + } + + // TOOTH — a claim advertises the harnesses its seller can run, in order; a seller that states + // none emits a byte-identical pre-registry claim rather than an empty tag. + #[test] + fn claim_advertises_its_harnesses_in_order() { + let advertised = claim_draft( + "job-1", + "buyer", + "seller", + "creqAtest", + &["claude".to_owned(), "codex".to_owned()], + ); + let tag = advertised + .tags + .iter() + .find(|tag| tag.first() == Some("mobee_agent")) + .expect("claim advertises its harnesses"); + assert_eq!(tag.0, vec!["mobee_agent", "claude", "codex"]); + assert_eq!( + crate::heartbeat::agents_from_tags(&advertised.tags), + vec!["claude", "codex"] + ); + + let silent = claim_draft("job-1", "buyer", "seller", "creqAtest", &[]); + assert!(silent.tags.iter().all(|tag| tag.first() != Some("mobee_agent"))); + assert!(crate::heartbeat::agents_from_tags(&silent.tags).is_empty()); + } + use super::*; const BUYER: &str = "buyer"; @@ -846,6 +946,7 @@ mod tests { unit: "sat".into(), deadline_unix: 1_800_000_001, seller_pubkey: Some(SELLER.into()), + requested_agent: None, } ); } @@ -882,7 +983,7 @@ mod tests { // The claim (processing) is its own claim kind, and the buyer-authored award // is the award kind — each distinct from the seller's feedback kind. assert_eq!( - claim_draft("offer", BUYER, SELLER, "creqAtest"), + claim_draft("offer", BUYER, SELLER, "creqAtest", &[]), EventDraft::new( JOB_CLAIM_KIND, vec![ @@ -925,7 +1026,7 @@ mod tests { ); // A non-award event yields no selection. assert_eq!( - parse_award(&claim_draft("offer", BUYER, SELLER, "creqAtest")), + parse_award(&claim_draft("offer", BUYER, SELLER, "creqAtest", &[])), None ); } @@ -1165,7 +1266,7 @@ mod creq_tests { build_seller_creq("job-1", 21, "sat", &[MINT_A.to_string()], &seller).expect("build creq"); assert!(creq.starts_with("creqA"), "creq must start with creqA: {creq}"); - let draft = claim_draft("job-1", "buyer-pubkey", &seller, &creq); + let draft = claim_draft("job-1", "buyer-pubkey", &seller, &creq, &[]); let creq_tag = draft .tags .iter() @@ -1219,7 +1320,7 @@ mod creq_tests { let seller = seller_hex(); let creq = build_seller_creq("job-7", 5, "sat", &[MINT_A.to_string()], &seller).expect("build creq"); - let draft = claim_draft("job-7", "buyer", &seller, &creq); + let draft = claim_draft("job-7", "buyer", &seller, &creq, &[]); let tag: &TagSpec = draft .tags .iter() diff --git a/crates/mobee-core/src/heartbeat.rs b/crates/mobee-core/src/heartbeat.rs index 13100f4d..5e2a6030 100644 --- a/crates/mobee-core/src/heartbeat.rs +++ b/crates/mobee-core/src/heartbeat.rs @@ -13,6 +13,7 @@ use serde::Serialize; use crate::gateway::{EventDraft, MOBEE_TAG, PROTOCOL_VERSION, TagSpec}; +use crate::seller_agents::AGENT_TAG; /// Addressable kind for the seller heartbeat. MUST be in NIP-01's `30000..=39999` addressable /// range so the relay replaces it in place keyed by `(pubkey, d)` — hence `30340`, not a `34xx` @@ -46,6 +47,10 @@ pub struct HeartbeatDraft { pub rate_sats: u64, /// The mobee protocol versions this seller speaks. pub protocol_versions: Vec, + /// The agent harnesses this seller can run, in preference order. Empty ⇒ the seller states no + /// harness and the tag is omitted entirely (an unlabelled `agent_command` seller has no honest + /// name to publish). + pub agents: Vec, } impl HeartbeatDraft { @@ -60,6 +65,7 @@ impl HeartbeatDraft { queue_depth, rate_sats, protocol_versions, + agents: Vec::new(), } } @@ -73,6 +79,12 @@ impl HeartbeatDraft { ) } + /// Advertise `agents` (preference order) on this heartbeat. + pub fn with_agents(mut self, agents: Vec) -> Self { + self.agents = agents; + self + } + pub fn to_event_draft(&self) -> EventDraft { let accepting = if self.accepting { "y" } else { "n" }; let queue_depth = self.queue_depth.to_string(); @@ -82,7 +94,7 @@ impl HeartbeatDraft { let mut protocol_tag = vec!["protocol_versions".to_owned()]; protocol_tag.extend(self.protocol_versions.iter().cloned()); - let tags = vec![ + let mut tags = vec![ TagSpec::new(["d", SELLER_HEARTBEAT_D]), TagSpec::new(["t", MOBEE_TAG]), TagSpec::new(["accepting", accepting]), @@ -90,16 +102,41 @@ impl HeartbeatDraft { TagSpec::new(["rate", &rate]), TagSpec(protocol_tag), ]; + if let Some(tag) = agent_tag(&self.agents) { + tags.push(tag); + } EventDraft::new(SELLER_HEARTBEAT_KIND, tags, "") } } +/// The `["mobee_agent", …]` advertisement tag, or `None` for a seller that states no harness (the +/// tag is then omitted rather than emitted empty — absent means "unstated", never "none"). +pub fn agent_tag(agents: &[String]) -> Option { + if agents.is_empty() { + return None; + } + let mut tag = vec![AGENT_TAG.to_owned()]; + tag.extend(agents.iter().cloned()); + Some(TagSpec(tag)) +} + +/// Read a `["mobee_agent", …]` advertisement off any event's tags. Absent ⇒ empty. +pub fn agents_from_tags(tags: &[TagSpec]) -> Vec { + first_tag(tags, AGENT_TAG) + .map(|tag| tag.0[1..].to_vec()) + .unwrap_or_default() +} + /// Build the heartbeat for a seller's live state. `accepting` is `n` while a job holds the /// single-flight slot (a busy seller is not taking new work); `queue_depth` is that in-flight -/// count. This is the single mapping the daemon loop uses, factored out so the flip is unit- -/// testable without a live relay. -pub fn heartbeat_for_state(job_in_flight: bool, rate_sats: u64) -> HeartbeatDraft { - HeartbeatDraft::v1(!job_in_flight, u32::from(job_in_flight), rate_sats) +/// count; `agents` is what the resolved harness registry advertises. This is the single mapping +/// the daemon loop uses, factored out so the flip is unit-testable without a live relay. +pub fn heartbeat_for_state( + job_in_flight: bool, + rate_sats: u64, + agents: Vec, +) -> HeartbeatDraft { + HeartbeatDraft::v1(!job_in_flight, u32::from(job_in_flight), rate_sats).with_agents(agents) } /// A parsed heartbeat's payload. The author pubkey is NOT carried here — combine it with [`d`] @@ -113,6 +150,9 @@ pub struct ParsedHeartbeat { pub queue_depth: u32, pub rate_sats: u64, pub protocol_versions: Vec, + /// Advertised harnesses, preference order. Empty ⇒ the seller stated none (the tag was + /// absent) — NOT a claim that it can run nothing. + pub agents: Vec, } impl ParsedHeartbeat { @@ -217,6 +257,7 @@ pub fn parse_heartbeat(event: &EventDraft) -> Result, - /// Optional preset label (`claude` | `cursor` | `codex`) for rediscovery / status. + /// Optional preset label (`claude` | `cursor` | `codex`) for rediscovery / status. With an + /// `agents` list configured this names the harness the list's first entry resolved to; on its + /// own it labels the single `agent_command`. #[serde(default, skip_serializing_if = "Option::is_none")] pub agent: Option, + /// The harnesses this node enables, in preference order — the multi-harness registry + /// ([`crate::seller_agents`]). Each entry is a preset name (bare, or a `{ name, slots }` + /// table). Empty ⇒ the node serves with the single `agent_command` alone. + /// + /// The node advertises this list on its heartbeat and claims, and dispatches a job to the + /// harness its offer requested. Execution stays one job at a time across the whole list. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub agents: Vec, /// Opt-in to claim untargeted/open offers. Default **false** (targeted-only). #[serde(default)] pub claim_open_pool: bool, @@ -576,6 +586,110 @@ pub struct AgentPresetConfig { pub argv: Vec, } +/// One enabled harness in `[seller] agents` — a preset name plus the size of its pool. +/// +/// Written either as a bare name or as a table, in the same list: +/// +/// ```toml +/// agents = ["claude", { name = "codex", slots = 1 }] +/// ``` +/// +/// The bare form is the whole config today. The table form is why pool counts can arrive later +/// without reshaping anything an operator already wrote — a `slots` value only ever gets added to +/// an entry. `slots` above 1 is refused at boot while execution is serial (see +/// [`crate::seller_agents::RegistryError::ParallelismUnsupported`]). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AgentSlotConfig { + /// Preset name from the `[agents]` table or the built-ins (`claude|cursor|codex`). + pub name: String, + /// Concurrent jobs this harness may run. Always 1 today. + pub slots: u32, +} + +impl AgentSlotConfig { + /// A single-slot entry — the bare-name form. + pub fn named(name: impl Into) -> Self { + Self { + name: name.into(), + slots: default_agent_slots(), + } + } +} + +/// Serde default for [`AgentSlotConfig::slots`]. +fn default_agent_slots() -> u32 { + 1 +} + +impl Serialize for AgentSlotConfig { + /// Round-trips to the form it was written in: a single-slot entry serializes as the bare name, + /// so a config the CLI writes stays `agents = ["claude", "codex"]`. + fn serialize(&self, serializer: S) -> Result { + if self.slots == default_agent_slots() { + return serializer.serialize_str(&self.name); + } + use serde::ser::SerializeStruct; + let mut table = serializer.serialize_struct("AgentSlotConfig", 2)?; + table.serialize_field("name", &self.name)?; + table.serialize_field("slots", &self.slots)?; + table.end() + } +} + +impl<'de> Deserialize<'de> for AgentSlotConfig { + fn deserialize>(deserializer: D) -> Result { + use serde::de::{self, MapAccess, Visitor}; + use std::fmt; + + struct SlotVisitor; + + impl<'de> Visitor<'de> for SlotVisitor { + type Value = AgentSlotConfig; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + formatter.write_str("an agent preset name, or a { name, slots } table") + } + + fn visit_str(self, value: &str) -> Result { + if value.trim().is_empty() { + return Err(E::custom("agent name must be non-empty")); + } + Ok(AgentSlotConfig::named(value)) + } + + fn visit_string(self, value: String) -> Result { + self.visit_str(&value) + } + + fn visit_map>(self, mut map: A) -> Result { + let mut name: Option = None; + let mut slots: Option = None; + while let Some(key) = map.next_key::()? { + match key.as_str() { + "name" => name = Some(map.next_value()?), + "slots" => slots = Some(map.next_value()?), + other => { + return Err(de::Error::custom(format!( + "unknown agent entry field {other:?} (want name, slots)" + ))); + } + } + } + let name = name.ok_or_else(|| de::Error::missing_field("name"))?; + if name.trim().is_empty() { + return Err(de::Error::custom("agent name must be non-empty")); + } + Ok(AgentSlotConfig { + name, + slots: slots.unwrap_or_else(default_agent_slots), + }) + } + } + + deserializer.deserialize_any(SlotVisitor) + } +} + /// Buyer-facing packaged config (`~/.mobee/config.toml`). #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] diff --git a/crates/mobee-core/src/job_lifecycle.rs b/crates/mobee-core/src/job_lifecycle.rs index 0cf7c744..eef6c059 100644 --- a/crates/mobee-core/src/job_lifecycle.rs +++ b/crates/mobee-core/src/job_lifecycle.rs @@ -71,6 +71,10 @@ pub struct PostJobRequest { /// (byte-identical to a non-contribution offer); `Contribution` carries the required target + /// base pins. See [`JobKind`]. pub job: JobKind, + /// Ask for a specific agent harness (`claude`, `codex`, …). `None` or `"any"` ⇒ no + /// preference, and the offer is byte-identical to one posted before harness selection existed. + /// A requested harness narrows the market: only sellers advertising it may be awarded. + pub requested_agent: Option, } /// Job class of a posted offer. Making this an enum (rather than an all-or-nothing cluster of @@ -195,6 +199,11 @@ pub struct OfferView { /// contribution OR when a `contribution`-class offer's pins were malformed (fail-closed). #[serde(default, skip_serializing_if = "Option::is_none")] pub contribution: Option, + /// The harness this job requested (`["param", "agent", …]`), canonicalised. `None` ⇒ any. + /// The award filter reads it from HERE — the signed offer on the relay — never from award + /// parameters, so the request cannot be changed after the fact. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub requested_agent: Option, } /// Serializable view of a well-formed contribution offer's pins. @@ -232,6 +241,10 @@ pub struct ClaimView { /// accept-bind then records no `creq_hash` and binding behaves identically. #[serde(default, skip_serializing_if = "Option::is_none")] pub creq: Option, + /// Harnesses this seller advertised on the claim (`["mobee_agent", …]`), preference order. + /// Empty ⇒ the claim stated none, which never satisfies a job that asked for one. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub agents: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] @@ -537,7 +550,8 @@ fn build_offer_draft( ) })?, ) - }; + } + .requesting_agent(request.requested_agent.as_deref()); let mut draft = offer.to_event_draft(); if let (Some(repo), Some(branch)) = (&request.repo, &request.branch) { @@ -1758,6 +1772,7 @@ pub(crate) async fn fetch_job_view_async( job_class: first_tag_value(&draft.tags, crate::contribution::TAG_JOB_CLASS) .map(str::to_owned), contribution: contribution_offer_view(&draft.tags), + requested_agent: parsed.as_ref().and_then(|p| p.requested_agent.clone()), } }); @@ -1780,6 +1795,7 @@ pub(crate) async fn fetch_job_view_async( live: false, // Capture the seller-authored creq tag; absent on claims with no creq. creq: first_tag_value(&draft.tags, "creq").map(str::to_owned), + agents: crate::heartbeat::agents_from_tags(&draft.tags), }); } claims.sort_by_key(|c| std::cmp::Reverse(c.created_at)); @@ -2976,6 +2992,7 @@ mod tests { status: "processing".to_owned(), live: true, creq: None, + agents: Vec::new(), } } @@ -3063,6 +3080,7 @@ mod tests { status: status.to_owned(), live: false, creq: None, + agents: Vec::new(), } } @@ -3221,6 +3239,7 @@ mod tests { repo: None, branch: None, job: JobKind::FromScratch, + requested_agent: None, }, ) .expect_err("seller required"); @@ -3395,6 +3414,7 @@ mod tests { repo: None, branch: None, job: JobKind::FromScratch, + requested_agent: None, }, ) .await @@ -3480,6 +3500,7 @@ mod tests { repo: None, branch: None, job: JobKind::FromScratch, + requested_agent: None, }, ) .await @@ -3538,6 +3559,7 @@ mod tests { repo: None, branch: None, job: JobKind::FromScratch, + requested_agent: None, }, ) .expect_err("must refuse nested block_on"); @@ -3628,6 +3650,7 @@ mod tests { base_oid: base_oid.to_owned(), accepts: vec!["fork".into()], }), + requested_agent: None, } } @@ -3817,6 +3840,7 @@ mod tests { repo: None, branch: None, job: JobKind::Contribution(contribution_spec(owner, url, branch, oid, accepts)), + requested_agent: None, } } @@ -3882,6 +3906,7 @@ mod tests { repo: None, branch: None, job: JobKind::FromScratch, + requested_agent: None, }; let contribution: Option = match &request.job { JobKind::FromScratch => None, @@ -4043,6 +4068,7 @@ mod tests { repo: None, branch: None, job: JobKind::FromScratch, + requested_agent: None, }, ) .await diff --git a/crates/mobee-core/src/lib.rs b/crates/mobee-core/src/lib.rs index 67529c56..3dd85bd6 100644 --- a/crates/mobee-core/src/lib.rs +++ b/crates/mobee-core/src/lib.rs @@ -1,3 +1,4 @@ +pub mod agent_presets; pub mod announce; #[cfg(all(feature = "wallet", feature = "gateway"))] pub mod authorize_pay; @@ -54,6 +55,9 @@ pub mod receipt; pub mod relay_auth; pub mod runtime_guard; pub mod seller; +/// The seller's harness registry: the agent harnesses one node enables, what it advertises for +/// them, and which one a given job dispatches to. +pub mod seller_agents; /// Agent-run + delivery-shaping helpers for the seller node: run the awarded agent, compose its /// delivery prompt, derive the delivery kind, and shape the public exec-metadata block. Kept in its /// own module (with a neutral error) so the run loop stays focused on the relay surface. diff --git a/crates/mobee-core/src/payment_wallet.rs b/crates/mobee-core/src/payment_wallet.rs index f847e218..071c7932 100644 --- a/crates/mobee-core/src/payment_wallet.rs +++ b/crates/mobee-core/src/payment_wallet.rs @@ -3091,6 +3091,7 @@ mod tests { unit: "sat".into(), deadline_unix: 1, seller_pubkey: Some(seller.into()), + requested_agent: None, } } diff --git a/crates/mobee-core/src/seller.rs b/crates/mobee-core/src/seller.rs index c0664d4a..65ee468a 100644 --- a/crates/mobee-core/src/seller.rs +++ b/crates/mobee-core/src/seller.rs @@ -739,6 +739,7 @@ mod tests { unit: "sat".into(), deadline_unix: 2_000_000_000, seller_pubkey: seller.map(str::to_owned), + requested_agent: None, } } @@ -821,6 +822,67 @@ git_remote = "https://example.invalid/repo.git" ); } + // TOOTH (compat) — a seller config written before the registry existed parses unchanged, with + // an EMPTY agents list. It is the fallback path that then serves it, not a silently invented + // registry. + #[test] + fn a_config_without_an_agents_list_parses_with_none() { + let raw = r#" +relay_url = "wss://example.invalid" +accepted_mints = ["https://testnut.cashudevkit.org"] +per_job_budget_sats = 21 +total_budget_sats = 100 + +[seller] +agent_command = ["claude-agent-acp"] +agent = "claude" +rate_sats = 7 +git_remote = "https://example.invalid/repo.git" +"#; + let seller = toml::from_str::(raw).expect("parse").seller.expect("seller"); + assert!(seller.agents.is_empty()); + assert_eq!(seller.agent.as_deref(), Some("claude")); + } + + // TOOTH — the `agents` list takes bare names AND `{ name, slots }` tables in the SAME list, and + // a single-slot entry serializes back to the bare name. This is the growth seam: adding pool + // counts later means adding a field to an entry, never reshaping a config an operator wrote. + #[test] + fn the_agents_list_accepts_bare_names_and_tables_together() { + let raw = r#" +relay_url = "wss://example.invalid" +accepted_mints = ["https://testnut.cashudevkit.org"] +per_job_budget_sats = 21 +total_budget_sats = 100 + +[seller] +agent_command = ["claude-agent-acp"] +rate_sats = 7 +git_remote = "https://example.invalid/repo.git" +agents = ["claude", { name = "codex" }, { name = "cursor", slots = 3 }] +"#; + let seller = toml::from_str::(raw).expect("parse").seller.expect("seller"); + let listed: Vec<(&str, u32)> = seller + .agents + .iter() + .map(|slot| (slot.name.as_str(), slot.slots)) + .collect(); + assert_eq!( + listed, + vec![("claude", 1), ("codex", 1), ("cursor", 3)], + "order is preference order; an omitted slots defaults to 1" + ); + + // Single-slot entries round-trip as bare names, so the CLI keeps writing the simple form. + let rendered = toml::to_string(&seller).expect("serialize"); + assert!(rendered.contains("\"claude\""), "{rendered}"); + assert!(rendered.contains("slots = 3"), "{rendered}"); + + // A nameless or unknown-field entry is refused rather than half-understood. + assert!(toml::from_str::("slots = 2").is_err()); + assert!(toml::from_str::("name = \"x\"\nrate = 2").is_err()); + } + #[test] fn agent_command_argv_array_parses() { let raw = r#" @@ -887,6 +949,7 @@ offer_backfill_secs = {backfill} git_remote: "https://example.invalid/repo.git".into(), job_timeout_secs: None, agent: None, + agents: Vec::new(), claim_open_pool: false, offer_backfill_secs: 0, contribution_enabled: true, diff --git a/crates/mobee-core/src/seller_agents.rs b/crates/mobee-core/src/seller_agents.rs new file mode 100644 index 00000000..609a9875 --- /dev/null +++ b/crates/mobee-core/src/seller_agents.rs @@ -0,0 +1,493 @@ +//! The seller's harness registry: which agent harnesses this ONE node can run, and which one a +//! given job dispatches to. +//! +//! A node lists the harnesses it enables in `[seller] agents` (order = preference); each name is a +//! preset from [`crate::agent_presets`], so the same `claude|cursor|codex|` vocabulary the +//! CLI accepts is the vocabulary the wire advertises. The registry is resolved ONCE at boot — +//! every listed preset is checked for a launchable adapter and each gets its own PASS/FAIL verdict, +//! so a missing adapter is a named line rather than a runtime surprise mid-job. +//! +//! Two rules the rest of the node leans on: +//! +//! - **Advertise only what is dispatchable.** [`AgentRegistry::advertised`] is exactly the set +//! [`AgentRegistry::dispatch`] can serve, so a buyer reading the wire and the node choosing a +//! harness can never disagree. A raw `--agent-argv` seller carries no preset label, so it +//! advertises nothing — there is no honest harness name to publish. +//! - **A named request is exact or nothing.** A job requesting harness `X` dispatches to `X` or is +//! not served; there is no nearest-match fallback, because silently running a job on a harness +//! the buyer did not ask for is the failure this registry exists to prevent. +//! +//! Execution is serial regardless of size — one job at a time across the whole registry. Pool +//! counts (`slots`) are parsed and REFUSED above 1 (see [`RegistryError::ParallelismUnsupported`]): +//! the config shape is ready for parallel pools, the execution engine is deliberately not. + +use std::collections::BTreeMap; + +use crate::agent_presets::resolve_agent_preset; +use crate::home::{AgentPresetConfig, SellerConfig}; + +/// Wire tag naming the harnesses an event's author can run — `["mobee_agent", "claude", "codex"]`, +/// ordered by the seller's preference. Multi-value (the `protocol_versions` convention), so issue +/// #170's single-harness `["mobee_agent", "claude"]` is the one-entry case of the same tag. +pub const AGENT_TAG: &str = "mobee_agent"; + +/// The offer parameter naming a requested harness: `["param", "agent", "claude"]`, a sibling of +/// `["param", "deadline", …]`. The value is opaque to the wire — an exact harness name today, +/// leaving room for a tier vocabulary later without a grammar change. +pub const AGENT_PARAM: &str = "agent"; + +/// The reserved "no preference" request value. Explicitly equivalent to omitting the parameter, so +/// a buyer can state indifference rather than only imply it. +pub const AGENT_ANY: &str = "any"; + +/// One harness the node can actually launch. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RegisteredAgent { + /// The preset label — the public harness identity, advertised on the wire and matched against + /// a job's request. `None` for the raw `--agent-argv` hatch: an argv with no preset label has + /// no harness name we can honestly publish. + pub name: Option, + /// The launch argv for the ACP driver (no shell). + pub argv: Vec, +} + +/// One listed preset's boot verdict. A FAIL never names an argv — it names the reason, so the +/// degrade line tells the operator what to install. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AgentVerdict { + pub name: String, + /// `Ok` ⇒ resolved and launchable. `Err` ⇒ the resolver's reason (missing adapter, empty argv, + /// unknown preset name). + pub outcome: Result, String>, +} + +impl AgentVerdict { + /// `PASS argv0=` / `FAIL : ` — one line per listed preset. + pub fn line(&self) -> String { + match &self.outcome { + Ok(argv) => format!( + "PASS {} argv0={}", + self.name, + argv.first().map(String::as_str).unwrap_or("") + ), + Err(reason) => format!("FAIL {}: {reason}", self.name), + } + } + + pub fn passed(&self) -> bool { + self.outcome.is_ok() + } +} + +/// Why a registry could not be resolved at all. Both variants REFUSE the boot: a node that cannot +/// launch anything, or one whose config asks for capacity the engine does not have, must not serve. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RegistryError { + /// Every listed preset failed to resolve — there is nothing to run jobs with. + AllFailed(Vec), + /// The config declares a pool larger than one. Parallel execution is not implemented; running + /// such a config SERIALLY would quietly deliver a fraction of the declared capacity, so it is + /// refused rather than downgraded. + ParallelismUnsupported { name: String, slots: u32 }, + /// No harness at all: neither an `agents` list nor an `agent_command` to fall back to. + Empty, +} + +impl std::fmt::Display for RegistryError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::AllFailed(verdicts) => { + write!( + formatter, + "no usable agent harness: every configured preset failed to resolve [{}]", + verdicts + .iter() + .map(AgentVerdict::line) + .collect::>() + .join("; ") + ) + } + Self::ParallelismUnsupported { name, slots } => write!( + formatter, + "agent {name:?} declares slots={slots}: parallel execution is not supported \ + (this node runs one job at a time). Set slots = 1." + ), + Self::Empty => write!( + formatter, + "no agent harness configured: set [seller] agents = [\"claude\", …] or agent_command" + ), + } + } +} + +impl std::error::Error for RegistryError {} + +/// The harnesses a booted node can run, in preference order. Built by [`resolve`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AgentRegistry { + entries: Vec, +} + +impl AgentRegistry { + /// Build directly from resolved entries. Prefer [`resolve`]; this exists for tests and for + /// callers holding an already-resolved set. + pub fn new(entries: Vec) -> Self { + Self { entries } + } + + /// Every entry, in preference order. + pub fn entries(&self) -> &[RegisteredAgent] { + &self.entries + } + + /// The harness names to advertise on the wire, in preference order. Exactly the set + /// [`Self::dispatch`] can serve by name — unlabelled hatch entries are omitted (nothing honest + /// to publish), so an empty list means "this seller states no harness", never "it has none". + pub fn advertised(&self) -> Vec { + self.entries + .iter() + .filter_map(|entry| entry.name.clone()) + .collect() + } + + /// The harness that will run a job requesting `requested`. + /// + /// `None` or [`AGENT_ANY`] ⇒ the preferred (first) entry. A named request matches by exact + /// preset name and NOTHING else: an unmatched name returns `None` so the caller declines, + /// rather than running the job on a harness the buyer did not ask for. + pub fn dispatch(&self, requested: Option<&str>) -> Option<&RegisteredAgent> { + match normalize_request(requested) { + None => self.entries.first(), + Some(name) => self + .entries + .iter() + .find(|entry| entry.name.as_deref() == Some(name.as_str())), + } + } + + /// Whether a job requesting `requested` can be served at all — the claim-decision predicate. + pub fn serves(&self, requested: Option<&str>) -> bool { + self.dispatch(requested).is_some() + } +} + +/// Canonicalise a requested harness: trims, lowercases, and maps both absence and [`AGENT_ANY`] +/// (and an empty value) to "no preference", so `None`, `""`, `"any"` and `" ANY "` are one case. +pub fn normalize_request(requested: Option<&str>) -> Option { + let value = requested?.trim().to_ascii_lowercase(); + if value.is_empty() || value == AGENT_ANY { + return None; + } + Some(value) +} + +/// Resolved registry plus the per-preset verdicts that produced it. `verdicts` is empty when the +/// node fell back to the single `agent_command` (nothing was listed to verdict). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResolvedRegistry { + pub registry: AgentRegistry, + pub verdicts: Vec, +} + +impl ResolvedRegistry { + /// The listed presets that failed to resolve. Non-empty alongside a usable registry means the + /// node is DEGRADED: it serves with the remainder and advertises only those. + pub fn failures(&self) -> Vec<&AgentVerdict> { + self.verdicts.iter().filter(|v| !v.passed()).collect() + } + + /// The loud degrade line, or `None` when every listed preset resolved. + pub fn degrade_line(&self) -> Option { + let failures = self.failures(); + if failures.is_empty() { + return None; + } + Some(format!( + "seller node DEGRADED: {} of {} configured agents unavailable [{}]; serving with {:?}", + failures.len(), + self.verdicts.len(), + failures + .iter() + .map(|v| v.line()) + .collect::>() + .join("; "), + self.registry.advertised() + )) + } +} + +/// Resolve the node's harness registry from its seller config. +/// +/// `[seller] agents` is the registry when present: each name resolves through the preset table, +/// every listed name gets a verdict, and the node serves with whatever resolved. When the list is +/// absent the node falls back to the single configured harness — the stored `agent_command` under +/// its `agent` preset label — so a seller written before this existed resolves to the identical +/// one-entry registry and dispatches the identical argv. +pub fn resolve( + seller: &SellerConfig, + presets: &BTreeMap, +) -> Result { + if seller.agents.is_empty() { + return fallback_registry(seller); + } + + let mut verdicts = Vec::with_capacity(seller.agents.len()); + let mut entries = Vec::new(); + for slot in &seller.agents { + if slot.slots != 1 { + return Err(RegistryError::ParallelismUnsupported { + name: slot.name.clone(), + slots: slot.slots, + }); + } + match resolve_agent_preset(&slot.name, presets) { + Ok((label, argv)) => { + verdicts.push(AgentVerdict { + name: label.clone(), + outcome: Ok(argv.clone()), + }); + // A duplicate name would shadow itself on dispatch; keep the first (preference + // order) so the registry and its advertisement stay one-to-one. + if !entries + .iter() + .any(|e: &RegisteredAgent| e.name.as_deref() == Some(label.as_str())) + { + entries.push(RegisteredAgent { + name: Some(label), + argv, + }); + } + } + Err(reason) => verdicts.push(AgentVerdict { + name: slot.name.clone(), + outcome: Err(reason), + }), + } + } + + if entries.is_empty() { + return Err(RegistryError::AllFailed(verdicts)); + } + Ok(ResolvedRegistry { + registry: AgentRegistry::new(entries), + verdicts, + }) +} + +/// The single-harness registry: the stored `agent_command` labelled by the configured preset (or +/// unlabelled for the raw-argv hatch). The argv is taken from config as-is and never re-resolved, +/// so an existing seller launches the same binary it launched before. +fn fallback_registry(seller: &SellerConfig) -> Result { + if seller.agent_command.is_empty() { + return Err(RegistryError::Empty); + } + let name = seller + .agent + .as_ref() + .map(|label| label.trim().to_ascii_lowercase()) + .filter(|label| !label.is_empty()); + Ok(ResolvedRegistry { + registry: AgentRegistry::new(vec![RegisteredAgent { + name, + argv: seller.agent_command.clone(), + }]), + verdicts: Vec::new(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::home::AgentSlotConfig; + + fn seller_with(agents: Vec, label: Option<&str>) -> SellerConfig { + SellerConfig { + agent_command: vec!["fallback-bin".into()], + rate_sats: 5, + git_remote: "https://example.invalid/repo".into(), + job_timeout_secs: None, + agent: label.map(str::to_owned), + agents, + claim_open_pool: false, + offer_backfill_secs: 0, + contribution_enabled: true, + } + } + + fn presets(entries: &[(&str, &[&str])]) -> BTreeMap { + entries + .iter() + .map(|(name, argv)| { + ( + (*name).to_owned(), + AgentPresetConfig { + argv: argv.iter().map(|a| (*a).to_owned()).collect(), + }, + ) + }) + .collect() + } + + fn registry(names: &[&str]) -> AgentRegistry { + AgentRegistry::new( + names + .iter() + .map(|name| RegisteredAgent { + name: Some((*name).to_owned()), + argv: vec![format!("{name}-acp")], + }) + .collect(), + ) + } + + #[test] + fn a_named_request_dispatches_to_that_harness_and_nothing_else() { + // Invariant 2, the pure core: harness X runs X's argv, and a request the registry cannot + // serve dispatches to NOTHING rather than falling back to the preferred entry. + let registry = registry(&["claude", "codex"]); + assert_eq!( + registry.dispatch(Some("codex")).map(|a| a.argv.clone()), + Some(vec!["codex-acp".to_owned()]) + ); + assert_eq!( + registry.dispatch(Some("claude")).map(|a| a.argv.clone()), + Some(vec!["claude-acp".to_owned()]) + ); + assert_eq!(registry.dispatch(Some("goose")), None); + assert!(!registry.serves(Some("goose"))); + } + + #[test] + fn no_request_takes_the_preferred_entry_and_any_is_the_same_case() { + let registry = registry(&["codex", "claude"]); + let preferred = registry.dispatch(None).expect("preferred"); + assert_eq!(preferred.name.as_deref(), Some("codex")); + for indifferent in ["any", "ANY", " any ", ""] { + assert_eq!( + registry.dispatch(Some(indifferent)).map(|a| a.name.clone()), + Some(Some("codex".to_owned())), + "{indifferent:?} must mean no preference" + ); + } + // Case/whitespace on a NAMED request resolves too — the wire value is canonicalised. + assert_eq!( + registry.dispatch(Some(" Claude ")).map(|a| a.name.clone()), + Some(Some("claude".to_owned())) + ); + } + + #[test] + fn advertised_is_exactly_what_dispatch_serves() { + // The wire promise and the dispatch table are one set — a buyer can never read a harness + // the node would then refuse to run. + let registry = registry(&["claude", "codex"]); + let advertised = registry.advertised(); + assert_eq!(advertised, vec!["claude", "codex"]); + for name in &advertised { + assert!(registry.serves(Some(name)), "advertised {name} must dispatch"); + } + } + + #[test] + fn unlabelled_argv_hatch_advertises_nothing_but_still_runs_untargeted_jobs() { + let seller = seller_with(Vec::new(), None); + let resolved = resolve(&seller, &BTreeMap::new()).expect("hatch resolves"); + assert!( + resolved.registry.advertised().is_empty(), + "a raw argv seller has no honest harness name to publish" + ); + assert_eq!( + resolved.registry.dispatch(None).map(|a| a.argv.clone()), + Some(vec!["fallback-bin".to_owned()]) + ); + // …and it cannot serve a named request, because it cannot prove what it is. + assert!(!resolved.registry.serves(Some("claude"))); + } + + #[test] + fn single_preset_config_resolves_to_the_same_one_entry_registry_and_argv() { + // Compat: a seller written before `agents` existed keeps its stored argv verbatim (never + // re-resolved off PATH) and advertises exactly its one configured harness. + let seller = seller_with(Vec::new(), Some("claude")); + let resolved = resolve(&seller, &BTreeMap::new()).expect("single preset resolves"); + assert_eq!(resolved.registry.advertised(), vec!["claude"]); + assert_eq!( + resolved.registry.dispatch(None).map(|a| a.argv.clone()), + Some(vec!["fallback-bin".to_owned()]), + "the stored agent_command is the truth for an existing seller" + ); + assert!(resolved.verdicts.is_empty()); + assert!(resolved.degrade_line().is_none()); + } + + #[test] + fn partial_failure_degrades_loud_and_serves_with_the_remainder() { + let table = presets(&[("good", &["/bin/sh"])]); + let seller = seller_with( + vec![ + AgentSlotConfig::named("good"), + AgentSlotConfig::named("nope-not-a-preset"), + ], + None, + ); + let resolved = resolve(&seller, &table).expect("partial failure still serves"); + assert_eq!(resolved.registry.advertised(), vec!["good"]); + assert_eq!(resolved.failures().len(), 1); + let line = resolved.degrade_line().expect("degrade line"); + assert!(line.contains("DEGRADED"), "{line}"); + assert!(line.contains("nope-not-a-preset"), "{line}"); + // The reduced advertisement is part of the loud line, not just an internal state. + assert!(line.contains("good"), "{line}"); + // And the failed harness is not dispatchable. + assert!(!resolved.registry.serves(Some("nope-not-a-preset"))); + } + + #[test] + fn every_preset_failing_refuses_rather_than_serving_nothing() { + let seller = seller_with( + vec![ + AgentSlotConfig::named("nope-one"), + AgentSlotConfig::named("nope-two"), + ], + None, + ); + match resolve(&seller, &BTreeMap::new()) { + Err(RegistryError::AllFailed(verdicts)) => { + assert_eq!(verdicts.len(), 2); + assert!(verdicts.iter().all(|v| !v.passed())); + let message = RegistryError::AllFailed(verdicts).to_string(); + assert!(message.contains("nope-one") && message.contains("nope-two"), "{message}"); + } + other => panic!("all-fail must refuse the boot, got {other:?}"), + } + } + + #[test] + fn a_pool_larger_than_one_is_refused_never_silently_serialized() { + let table = presets(&[("good", &["/bin/sh"])]); + let seller = seller_with(vec![AgentSlotConfig { name: "good".into(), slots: 2 }], None); + assert_eq!( + resolve(&seller, &table), + Err(RegistryError::ParallelismUnsupported { name: "good".into(), slots: 2 }) + ); + } + + #[test] + fn a_duplicate_listing_keeps_one_entry_so_advertisement_matches_dispatch() { + let table = presets(&[("good", &["/bin/sh"])]); + let seller = seller_with( + vec![AgentSlotConfig::named("good"), AgentSlotConfig::named("good")], + None, + ); + let resolved = resolve(&seller, &table).expect("duplicates resolve"); + assert_eq!(resolved.registry.advertised(), vec!["good"]); + assert_eq!(resolved.registry.entries().len(), 1); + } + + #[test] + fn no_harness_at_all_refuses() { + let mut seller = seller_with(Vec::new(), None); + seller.agent_command = Vec::new(); + assert_eq!(resolve(&seller, &BTreeMap::new()), Err(RegistryError::Empty)); + } +} diff --git a/crates/mobee-core/src/seller_node/ingester.rs b/crates/mobee-core/src/seller_node/ingester.rs index 1b552a2d..4f14b346 100644 --- a/crates/mobee-core/src/seller_node/ingester.rs +++ b/crates/mobee-core/src/seller_node/ingester.rs @@ -119,6 +119,7 @@ mod tests { task: "task".to_owned(), deadline_unix: 9_999, targeted: true, + requested_agent: None, } } @@ -138,7 +139,7 @@ mod tests { let (store, path) = fresh_store("award"); let job = "j".repeat(64); let offer_id = "o".repeat(64); - let draft = crate::gateway::claim_draft(&offer_id, &"b".repeat(64), &"s".repeat(64), "creqA"); + let draft = crate::gateway::claim_draft(&offer_id, &"b".repeat(64), &"s".repeat(64), "creqA", &[]); store.claim_and_enqueue(&job, &offer_id, "creqA", &draft, 1, 9_999, 1).expect("claim"); let mut stream = VecStream( diff --git a/crates/mobee-core/src/seller_node/mod.rs b/crates/mobee-core/src/seller_node/mod.rs index 80162ab8..62141a6a 100644 --- a/crates/mobee-core/src/seller_node/mod.rs +++ b/crates/mobee-core/src/seller_node/mod.rs @@ -54,6 +54,9 @@ pub enum NodeError { Identity(HomeError), /// A live relay-surface failure (connect, NIP-42 auth, subscribe) raised by the run loop. Relay(String), + /// The configured agent harnesses could not be resolved into a registry the node can serve + /// with — boot refuses rather than advertise work it cannot run. + Agents(crate::seller_agents::RegistryError), } impl std::fmt::Display for NodeError { @@ -64,6 +67,7 @@ impl std::fmt::Display for NodeError { Self::Wallet(error) => write!(formatter, "seller node wallet error: {error}"), Self::Identity(error) => write!(formatter, "seller node identity error: {error}"), Self::Relay(message) => write!(formatter, "seller node relay error: {message}"), + Self::Agents(error) => write!(formatter, "seller node agent config error: {error}"), } } } @@ -249,7 +253,7 @@ mod tests { } fn claim_draft() -> crate::gateway::EventDraft { - crate::gateway::claim_draft(&"e".repeat(64), &"b".repeat(64), &"s".repeat(64), "creqA") + crate::gateway::claim_draft(&"e".repeat(64), &"b".repeat(64), &"s".repeat(64), "creqA", &[]) } struct FakePublisher { diff --git a/crates/mobee-core/src/seller_node/outbox.rs b/crates/mobee-core/src/seller_node/outbox.rs index 44842387..fd8f4066 100644 --- a/crates/mobee-core/src/seller_node/outbox.rs +++ b/crates/mobee-core/src/seller_node/outbox.rs @@ -79,7 +79,7 @@ mod tests { } fn claim_draft() -> crate::gateway::EventDraft { - crate::gateway::claim_draft(&"e".repeat(64), &"b".repeat(64), &"s".repeat(64), "creqA") + crate::gateway::claim_draft(&"e".repeat(64), &"b".repeat(64), &"s".repeat(64), "creqA", &[]) } /// Records every publish call; can be told to fail so the retry path is exercised. diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index b7420eff..04211dea 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -26,6 +26,7 @@ use crate::kinds::{JOB_AWARD_KIND, JOB_OFFER_KIND}; use crate::receipt::{ReceiptPreimage, EXEC_METADATA_COMMITMENT_EMPTY}; use crate::relay_auth::{self, AuthWait}; use crate::seller::rate_gate_allows; +use crate::seller_agents::AgentRegistry; use crate::seller_exec::{ compose_agent_prompt, delivery_message, job_workdir, run_agent_job, run_agent_with_retry, seller_delivery_kind, seller_exec_metadata, unified_job_timeout, @@ -74,6 +75,8 @@ enum SkipReason { Lapsed, /// Rate-gate refused: untargeted without open-pool opt-in, or below the seller's rate floor. RateGate, + /// The offer asked for a harness this node does not run. + AgentUnavailable, } impl SkipReason { @@ -82,6 +85,7 @@ impl SkipReason { match self { Self::Lapsed => "offer deadline already passed (lapsed; never resurrected)", Self::RateGate => "rate-gate refused (untargeted without opt-in / below rate)", + Self::AgentUnavailable => "requested agent harness not available on this node", } } } @@ -655,12 +659,39 @@ fn should_resume_execution(state: super::store::JobState) -> bool { ) } +/// The journaled offer facts for a parsed offer — the ONE place a wire offer becomes a stored row. +/// +/// Extracted from the claim path so this mapping is reachable by a test. Everything downstream +/// reads the ROW, not the event: the award arm authorizes against its buyer, the pay path takes its +/// amount/unit as the redeem terms, and execution (possibly a restart later) takes its +/// `requested_agent` as the harness to dispatch. A field dropped here is a field that silently does +/// not exist for the rest of the job's life, which is why the mapping is a named function with its +/// own tooth rather than a struct literal inlined at the only call site. +fn offer_row(job_id: &str, buyer_pubkey: &str, offer: &ParsedOffer) -> super::store::Offer { + super::store::Offer { + offer_id: job_id.to_owned(), + buyer_pubkey: buyer_pubkey.to_owned(), + amount_sats: offer.amount, + unit: offer.unit.clone(), + task: offer.task.clone(), + deadline_unix: offer.deadline_unix as i64, + targeted: offer.is_targeted(), + requested_agent: offer.requested_agent.clone(), + } +} + /// Decide whether to claim `offer`, applying the always-on money-safety gates in the legacy order: /// a lapsed offer is refused BEFORE its deadline is re-derived (never resurrect a stale offer with a -/// fresh `now + timeout`), then the targeting/rate gate. Pure over (offer, config, now). +/// fresh `now + timeout`), then the targeting/rate gate, then the harness the offer asked for. +/// Pure over (offer, config, registry, now). +/// +/// The harness gate is a CLAIM-time decision, not a delivery-time one: a node that cannot run the +/// requested harness never parks a claim at all, so the buyer's offer stays visible to a seller +/// that can, instead of being answered by one that would fail later. fn classify_offer( offer: &ParsedOffer, seller: &crate::home::SellerConfig, + agents: &AgentRegistry, seller_pubkey: &str, now_unix: u64, ) -> ClaimDecision { @@ -672,11 +703,44 @@ fn classify_offer( if rate_gate_allows(offer, seller_pubkey, seller.rate_sats, seller.claim_open_pool).is_err() { return ClaimDecision::Skip(SkipReason::RateGate); } + if !agents.serves(offer.requested_agent.as_deref()) { + return ClaimDecision::Skip(SkipReason::AgentUnavailable); + } ClaimDecision::Claim { deadline_unix: crate::seller::job_deadline_unix(offer, seller, now_unix), } } +/// Resolve + report the harness registry at boot: one PASS/FAIL line per configured preset, then +/// either a loud degrade line (some resolved) or a refusal (none did). +/// +/// The three outcomes are deliberately distinct. ALL configured presets failing REFUSES the boot — +/// a node with no launchable harness that still claimed work would take jobs it must then fail. +/// SOME failing DEGRADES loudly and serves with the remainder, advertising only those, because a +/// two-harness seller that loses one is still a working one-harness seller. A node with no `agents` +/// list at all resolves to its single `agent_command` and prints nothing new. +fn boot_agent_registry(home: &MobeeHome) -> Result { + let Some(seller) = home.config.seller.as_ref() else { + // No `[seller]` section: nothing serves offers, and the run loop already no-ops. An empty + // registry keeps that path unchanged rather than turning it into a boot failure. + return Ok(AgentRegistry::new(Vec::new())); + }; + let resolved = + crate::seller_agents::resolve(seller, &home.config.agents).map_err(NodeError::Agents)?; + for verdict in &resolved.verdicts { + eprintln!("seller node agent {}", verdict.line()); + } + if let Some(degraded) = resolved.degrade_line() { + eprintln!("{degraded}"); + } else if !resolved.registry.advertised().is_empty() { + eprintln!( + "seller node agents ready: {:?} (serial execution — one job at a time)", + resolved.registry.advertised() + ); + } + Ok(resolved.registry) +} + /// How long boot waits for the relay connection and the NIP-42 challenge. const CONNECT_WAIT: Duration = Duration::from_secs(20); /// Cadence of the outbox drain / housekeeping tick. @@ -692,6 +756,10 @@ pub struct SellerNodeRunner { /// Outcome of the boot NIP-42 handshake, which seeds the run loop's view of whether the current /// socket is authenticated. `NoChallenge` is not authentication. boot_auth: AuthWait, + /// The harnesses this node can run, resolved once at boot. Every claim decision, every + /// advertisement, and every dispatch reads THIS — never the config — so what the node + /// advertises is what it verified it can launch. + agents: AgentRegistry, } impl SellerNodeRunner { @@ -714,6 +782,10 @@ impl SellerNodeRunner { let node = SellerNode::open(home).await?; + // Resolve the harness registry BEFORE anything goes on the wire: a node that cannot launch + // a single harness must refuse to boot rather than claim work it can never run. + let agents = boot_agent_registry(node.home())?; + // Reconcile durable state before serving anything live: expire stale outbox rows, report the // non-terminal jobs that resume. Reconcile must NOT release parked claims (invariant 5). match node.reconcile_on_start(now_unix()) { @@ -776,6 +848,7 @@ impl SellerNodeRunner { relay_url, seller_pubkey, boot_auth, + agents, }) } @@ -1436,8 +1509,12 @@ impl SellerNodeRunner { return; }; let job_in_flight = self.node.store().health().map(|h| h.jobs > 0).unwrap_or(false); - let draft = - crate::heartbeat::heartbeat_for_state(job_in_flight, seller.rate_sats).to_event_draft(); + let draft = crate::heartbeat::heartbeat_for_state( + job_in_flight, + seller.rate_sats, + self.agents.advertised(), + ) + .to_event_draft(); match self.node.signer().sign(draft, now_unix()).await { Ok(Ok(signed)) => { use nostr_sdk::JsonUtil as _; @@ -1584,7 +1661,8 @@ impl SellerNodeRunner { let seller_pubkey = self.seller_pubkey.to_hex(); let now = now_unix(); - let deadline_unix = match classify_offer(&offer, &seller, &seller_pubkey, now as u64) { + let deadline_unix = match classify_offer(&offer, &seller, &self.agents, &seller_pubkey, now as u64) + { ClaimDecision::Claim { deadline_unix } => deadline_unix, ClaimDecision::Skip(skip) => { eprintln!("seller node offer skip id={}: {}", event.id, skip.reason()); @@ -1624,18 +1702,11 @@ impl SellerNodeRunner { // Journal the offer facts BEFORE claiming: the award arm reads the buyer to authorize an // award (author MUST be the offer's buyer), and the pay path reads amount/unit as the redeem // terms. Idempotent — a re-seen offer is a no-op. - if let Err(error) = self.node.store().record_offer( - &super::store::Offer { - offer_id: job_id.clone(), - buyer_pubkey: buyer_pubkey.clone(), - amount_sats: offer.amount, - unit: offer.unit.clone(), - task: offer.task.clone(), - deadline_unix: offer.deadline_unix as i64, - targeted: offer.is_targeted(), - }, - now, - ) { + if let Err(error) = self + .node + .store() + .record_offer(&offer_row(&job_id, &buyer_pubkey, &offer), now) + { eprintln!("seller node offer skip id={job_id}: record offer failed ({error})"); return; } @@ -1653,7 +1724,15 @@ impl SellerNodeRunner { return; } }; - let claim = claim_draft(&job_id, &buyer_pubkey, &seller_pubkey, &creq); + // The claim advertises what this node can run, so the buyer's award filter can hold it to + // the harness its job asked for. + let claim = claim_draft( + &job_id, + &buyer_pubkey, + &seller_pubkey, + &creq, + &self.agents.advertised(), + ); match self.node.store().claim_and_enqueue( &job_id, &job_id, @@ -1820,6 +1899,33 @@ impl SellerNodeRunner { _ => now_unix(), }; + // Which harness runs this job. Read from the STORED offer (not live config), so a job + // resumed after a restart still dispatches to the harness its buyer asked for. A request + // this node cannot serve fails the job rather than substituting another harness — the + // claim gate should already have refused it, and quietly running the wrong agent is the + // one outcome the registry exists to prevent. + let requested_agent = offer.requested_agent.clone(); + let Some(agent) = self.agents.dispatch(requested_agent.as_deref()) else { + eprintln!( + "seller node execute fail job_id={job_id}: requested agent {:?} is not available on \ + this node (never substituted)", + requested_agent.as_deref().unwrap_or("") + ); + self.fail_job_with_feedback(job_id, &offer.buyer_pubkey, EXEC_FAILURE_FEEDBACK).await; + return; + }; + let agent_command = agent.argv.clone(); + let agent_label = agent.name.clone(); + // Journal WHICH harness ran it before the run starts, so the row exists even if the job + // then fails — the journal answers "what ran this", not only "what finished it". + if let Some(label) = agent_label.as_deref() + && let Err(error) = self.node.store().assign_agent(job_id, label) + { + eprintln!( + "seller node execute job_id={job_id}: agent journal write failed (continuing): {error}" + ); + } + // Move awarded -> executing (idempotent). A failed mark is logged, never fatal. if let Err(error) = self.node.store().mark_executing(job_id, now_unix()) { eprintln!("seller node execute job_id={job_id}: mark_executing failed (continuing): {error}"); @@ -1850,7 +1956,7 @@ impl SellerNodeRunner { || now_unix() as u64, |_attempt| { let job_timeout = unified_job_timeout(deadline, now_unix() as u64); - run_agent_job(&seller.agent_command, &prompt, &workdir, &identity, job_timeout) + run_agent_job(&agent_command, &prompt, &workdir, &identity, job_timeout) }, ) .await; @@ -1957,8 +2063,8 @@ impl SellerNodeRunner { // Harness-generic PUBLIC seller-claimed usage block (opportunistic; absent fields stay // absent). `usage` carries what the ACP driver surfaced this run — `None` when it exposed none. let exec_metadata = seller_exec_metadata( - &seller.agent_command, - seller.agent.as_deref(), + &agent_command, + agent_label.as_deref(), wall_time_ms, usage.as_ref(), ); @@ -2117,6 +2223,8 @@ impl SellerNodeRunner { unit: offer.unit.clone(), deadline_unix: offer.deadline_unix.max(0) as u64, seller_pubkey: offer.targeted.then(|| seller_pubkey.clone()), + // The pay path is harness-blind: which harness ran the job never changes the terms. + requested_agent: None, }; let accepted_mints: std::collections::HashSet = request.mints.iter().cloned().collect(); @@ -2265,12 +2373,21 @@ mod tests { git_remote: "https://example.invalid/repo.git".to_owned(), job_timeout_secs: None, agent: Some("claude".to_owned()), + agents: Vec::new(), claim_open_pool, offer_backfill_secs: 0, contribution_enabled: true, } } + /// The registry an existing single-preset (`agent = "claude"`) seller resolves to. + fn claude_only() -> AgentRegistry { + AgentRegistry::new(vec![crate::seller_agents::RegisteredAgent { + name: Some("claude".to_owned()), + argv: vec!["claude-agent-acp".to_owned()], + }]) + } + fn offer(amount: u64, targeted_to: Option<&str>, deadline_unix: u64) -> ParsedOffer { ParsedOffer { task: "do the thing".to_owned(), @@ -2279,13 +2396,14 @@ mod tests { unit: "sat".to_owned(), deadline_unix, seller_pubkey: targeted_to.map(str::to_owned), + requested_agent: None, } } // A fresh, in-rate, targeted offer is claimed and carries the resolved deadline. #[test] fn claims_fresh_targeted_offer_at_rate() { - let decision = classify_offer(&offer(5, Some(SELLER), NOW + 600), &seller_cfg(2, false), SELLER, NOW); + let decision = classify_offer(&offer(5, Some(SELLER), NOW + 600), &seller_cfg(2, false), &claude_only(), SELLER, NOW); assert_eq!(decision, ClaimDecision::Claim { deadline_unix: NOW + 600 }); } @@ -2293,14 +2411,14 @@ mod tests { // it is never resurrected with a fresh window, even though it clears the rate floor. #[test] fn refuses_lapsed_offer_before_rate() { - let decision = classify_offer(&offer(100, Some(SELLER), NOW), &seller_cfg(2, false), SELLER, NOW); + let decision = classify_offer(&offer(100, Some(SELLER), NOW), &seller_cfg(2, false), &claude_only(), SELLER, NOW); assert_eq!(decision, ClaimDecision::Skip(SkipReason::Lapsed)); } // Below the rate floor ⇒ skip (never claim work priced under the seller's floor). #[test] fn refuses_below_rate() { - let decision = classify_offer(&offer(1, Some(SELLER), NOW + 600), &seller_cfg(5, false), SELLER, NOW); + let decision = classify_offer(&offer(1, Some(SELLER), NOW + 600), &seller_cfg(5, false), &claude_only(), SELLER, NOW); assert_eq!(decision, ClaimDecision::Skip(SkipReason::RateGate)); } @@ -2360,21 +2478,174 @@ mod tests { #[test] fn untargeted_needs_open_pool_opt_in() { assert_eq!( - classify_offer(&offer(5, None, NOW + 600), &seller_cfg(2, false), SELLER, NOW), + classify_offer(&offer(5, None, NOW + 600), &seller_cfg(2, false), &claude_only(), SELLER, NOW), ClaimDecision::Skip(SkipReason::RateGate) ); assert_eq!( - classify_offer(&offer(5, None, NOW + 600), &seller_cfg(2, true), SELLER, NOW), + classify_offer(&offer(5, None, NOW + 600), &seller_cfg(2, true), &claude_only(), SELLER, NOW), + ClaimDecision::Claim { deadline_unix: NOW + 600 } + ); + } + + // TOOTH (charter invariant 3) — a node that cannot run the requested harness never CLAIMS. + // The refusal is a decision over the offer, not an outcome discovered at delivery: the offer + // stays available to a seller that can serve it, instead of being answered by one that would + // then fail. Bite: drop the `agents.serves(...)` arm from `classify_offer` and the codex offer + // below is claimed by a claude-only node. + #[test] + fn a_node_without_the_requested_harness_never_claims() { + let mut wants_codex = offer(5, Some(SELLER), NOW + 600); + wants_codex.requested_agent = Some("codex".to_owned()); + assert_eq!( + classify_offer(&wants_codex, &seller_cfg(2, false), &claude_only(), SELLER, NOW), + ClaimDecision::Skip(SkipReason::AgentUnavailable) + ); + + // The same offer at a node that DOES run codex is claimed — the gate is the harness, not + // the presence of a request. + let both = AgentRegistry::new(vec![ + crate::seller_agents::RegisteredAgent { + name: Some("claude".to_owned()), + argv: vec!["claude-agent-acp".to_owned()], + }, + crate::seller_agents::RegisteredAgent { + name: Some("codex".to_owned()), + argv: vec!["codex-acp".to_owned()], + }, + ]); + assert_eq!( + classify_offer(&wants_codex, &seller_cfg(2, false), &both, SELLER, NOW), + ClaimDecision::Claim { deadline_unix: NOW + 600 } + ); + + // And an offer asking for nothing is claimed by the claude-only node exactly as before. + assert_eq!( + classify_offer(&offer(5, Some(SELLER), NOW + 600), &seller_cfg(2, false), &claude_only(), SELLER, NOW), ClaimDecision::Claim { deadline_unix: NOW + 600 } ); } + // TOOTH (the seam my other teeth do not look at) — the harness request survives the trip from + // WIRE EVENT to STORED ROW. + // + // Every other tooth here either builds the `Offer` row by hand or reads one back, so all of + // them stay green if this mapping silently drops the field — invariant 2 would then be built, + // green, and dead the moment execution happened after a restart. This one starts from an + // offer draft, parses it the way the claim path does, and asserts the row carries the request. + // Bite (measured): replace `requested_agent` in `offer_row` with `None` — before this tooth + // existed the whole suite stayed green; with it, this test and only this test goes red. + #[test] + fn the_harness_request_survives_the_wire_to_row_mapping() { + let asked = gateway::OfferDraft::new("do a task", "text/plain", 5, NOW + 600, "a".repeat(64)) + .requesting_agent(Some("codex")) + .to_event_draft(); + let parsed = parse_offer(&asked).expect("parse offer"); + let row = offer_row("job-1", "buyer-1", &parsed); + assert_eq!( + row.requested_agent.as_deref(), + Some("codex"), + "the request must reach the row — everything downstream reads the ROW, not the event" + ); + // The rest of the mapping is asserted alongside it, so a field dropped here is caught too. + assert_eq!(row.amount_sats, 5); + assert_eq!(row.unit, "sat"); + assert_eq!(row.task, "do a task"); + assert_eq!(row.deadline_unix, (NOW + 600) as i64); + assert!(row.targeted); + + // An offer that asked for nothing stores nothing — absence is carried, not invented. + let plain = gateway::OfferDraft::new("do a task", "text/plain", 5, NOW + 600, "a".repeat(64)) + .to_event_draft(); + let parsed = parse_offer(&plain).expect("parse offer"); + assert_eq!(offer_row("job-2", "buyer-1", &parsed).requested_agent, None); + } + + // TOOTH (charter invariant 2, RESTART form — the strong one) — a job requesting harness X is + // dispatched to X even when the process that claimed it is gone. The request is journaled with + // the offer facts, so the resumed execute path reads it from the STORE; the registry below + // deliberately PREFERS claude, so a regression that dispatches the preferred harness (or that + // re-reads live config) runs claude and goes red. + #[test] + fn a_resumed_job_still_dispatches_to_the_harness_it_requested() { + let job = "a".repeat(64); + let buyer = "b".repeat(64); + let seller = nostr_sdk::prelude::Keys::generate().public_key().to_hex(); + let creq = gateway::creq::build_seller_creq( + &job, + 21, + "sat", + &["https://testnut.cashudevkit.org".to_owned()], + &seller, + ) + .expect("creq"); + let root = temp_dir("restart-dispatch"); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).expect("mk root"); + let db = root.join("seller.sqlite"); + + // Claim time: the offer asks for codex, and its facts are journaled with the claim. + { + let store = SellerStore::open(&db).expect("open store"); + store + .record_offer( + &Offer { + offer_id: job.clone(), + buyer_pubkey: buyer.clone(), + amount_sats: 21, + unit: "sat".to_owned(), + task: "build a widget".to_owned(), + deadline_unix: 2_000_000_000, + targeted: true, + requested_agent: Some("codex".to_owned()), + }, + 1, + ) + .expect("record offer"); + let draft = claim_draft(&job, &buyer, &seller, &creq, &["codex".to_owned()]); + store + .claim_and_enqueue(&job, &job, &creq, &draft, 1, 9_999_999_999, 1) + .expect("claim"); + } + + // …the process dies here. A fresh store handle is all the resumed node has. + let store = SellerStore::open(&db).expect("reopen store"); + let resumed = store.offer_row(&job).expect("offer row").expect("offer survives"); + assert_eq!(resumed.requested_agent.as_deref(), Some("codex")); + + let registry = AgentRegistry::new(vec![ + crate::seller_agents::RegisteredAgent { + name: Some("claude".to_owned()), + argv: vec!["claude-agent-acp".to_owned()], + }, + crate::seller_agents::RegisteredAgent { + name: Some("codex".to_owned()), + argv: vec!["codex-acp".to_owned()], + }, + ]); + let dispatched = registry + .dispatch(resumed.requested_agent.as_deref()) + .expect("the requested harness is available"); + assert_eq!(dispatched.name.as_deref(), Some("codex")); + assert_eq!(dispatched.argv, vec!["codex-acp"], "the RUN command is codex's, not the preferred harness's"); + + // And the journal names what ran it. + store + .record_award(&"w".repeat(64), &job, &buyer, 4242) + .expect("award"); + store + .assign_agent(&job, dispatched.name.as_deref().expect("label")) + .expect("journal the harness"); + assert_eq!(store.job_agent(&job).expect("job agent"), Some("codex".to_owned())); + + let _ = std::fs::remove_dir_all(&root); + } + // TOOTH (#146 / #117 refusal taxonomy) — a cross-version offer is a DISTINCT refusal, not the // generic "unparseable" bucket. Build a well-formed offer, then swap ONLY its `v` tag so the sole // parse failure is version skew; the node's on_offer routes that to the unsupported-version skip. #[test] fn unsupported_version_offer_is_a_distinct_parse_refusal() { - let offer = gateway::OfferDraft::new("do a task", "text/plain", 5, NOW + 600, &"a".repeat(64)); + let offer = gateway::OfferDraft::new("do a task", "text/plain", 5, NOW + 600, "a".repeat(64)); let mut draft = offer.to_event_draft(); for tag in &mut draft.tags { if tag.0.first().map(String::as_str) == Some("v") { @@ -2907,7 +3178,7 @@ mod tests { "nothing delivered yet" ); - let draft = claim_draft(&job, &buyer, &seller, &creq); + let draft = claim_draft(&job, &buyer, &seller, &creq, &[]); let delivered_at = 6_000; assert!(store .deliver_and_enqueue( @@ -2989,7 +3260,7 @@ mod tests { let job = "a".repeat(64); let buyer = "b".repeat(64); let (store, root) = store_with_awarded_job(&creq, &job, &buyer, 4242); - let draft = claim_draft(&job, &buyer, &seller, &creq); + let draft = claim_draft(&job, &buyer, &seller, &creq, &[]); // Deliver ⇒ state Delivered ⇒ NOT re-execute-eligible (the guard early-returns). assert!(store @@ -3067,11 +3338,12 @@ mod tests { task: "build a widget".to_owned(), deadline_unix: 2_000_000_000, targeted: true, + requested_agent: None, }, 1, ) .expect("record offer"); - let draft = claim_draft(job, buyer, &"s".repeat(64), creq); + let draft = claim_draft(job, buyer, &"s".repeat(64), creq, &[]); store .claim_and_enqueue(job, job, creq, &draft, 1, 9_999_999_999, 1) .expect("claim"); @@ -3177,7 +3449,7 @@ mod tests { ) .expect("creq"); let (store, root) = store_with_awarded_job(&creq, &job, &buyer, author_date); - let draft = claim_draft(&job, &buyer, &seller, &creq); + let draft = claim_draft(&job, &buyer, &seller, &creq, &[]); let now = 5000; assert!( store @@ -3361,7 +3633,7 @@ mod tests { ); // Re-execution delivers exactly once: deliver_and_enqueue is idempotent on the job. - let draft = claim_draft(&job, &buyer, &seller, &creq); + let draft = claim_draft(&job, &buyer, &seller, &creq, &[]); let now = 5000; assert!( store diff --git a/crates/mobee-core/src/seller_node/signer.rs b/crates/mobee-core/src/seller_node/signer.rs index ebbcc1d7..64171c8f 100644 --- a/crates/mobee-core/src/seller_node/signer.rs +++ b/crates/mobee-core/src/seller_node/signer.rs @@ -453,7 +453,7 @@ mod tests { } fn claim_draft() -> EventDraft { - gateway::claim_draft(&"e".repeat(64), &"b".repeat(64), &"s".repeat(64), "creqA-test") + gateway::claim_draft(&"e".repeat(64), &"b".repeat(64), &"s".repeat(64), "creqA-test", &[]) } // The fixed created_at makes the signed event id deterministic — the property the outbox relies diff --git a/crates/mobee-core/src/seller_node/store.rs b/crates/mobee-core/src/seller_node/store.rs index c8d6b733..0cfe718c 100644 --- a/crates/mobee-core/src/seller_node/store.rs +++ b/crates/mobee-core/src/seller_node/store.rs @@ -23,7 +23,7 @@ use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; use crate::gateway::EventDraft; /// Current on-disk schema version. -pub const SCHEMA_VERSION: i64 = 1; +pub const SCHEMA_VERSION: i64 = 2; /// A cloneable handle to the node-owned SQLite state. #[derive(Clone)] @@ -59,6 +59,11 @@ pub struct Offer { pub task: String, pub deadline_unix: i64, pub targeted: bool, + /// The harness the offer asked for (`["param", "agent", …]`), canonicalised; `None` ⇒ no + /// preference. Journaled with the other offer facts because execution can be a RESTART away + /// from the claim: a resumed job reads its requested harness from here, so it dispatches to + /// the harness the buyer asked for and not to whichever one happens to be preferred now. + pub requested_agent: Option, } /// The lifecycle state of a job (execution side of a claim that was awarded). @@ -172,7 +177,10 @@ impl SellerStore { task TEXT NOT NULL, deadline_unix INTEGER NOT NULL, targeted INTEGER NOT NULL, - created_at_unix INTEGER NOT NULL + created_at_unix INTEGER NOT NULL, + -- The harness the offer requested. NULL ⇒ no preference, which is also what an + -- offer recorded before this column existed reads as. + requested_agent TEXT ); -- Claims the node parked. `state` is the claim's own lifecycle; `awarded` marks the -- one the buyer selected, `released` the ones it stepped back from. @@ -198,8 +206,9 @@ impl SellerStore { buyer_pubkey TEXT NOT NULL, created_at_unix INTEGER NOT NULL ); - -- Jobs the node is executing (one per awarded claim). `agent_name` is the roster - -- agent selected to run it (never published to buyers). + -- Jobs the node is executing (one per awarded claim). `agent_name` is the harness that + -- actually ran it — the journal row naming which agent did the job, and the evidence + -- that a harness-requesting job was served by the harness it asked for. CREATE TABLE IF NOT EXISTS jobs ( job_id TEXT PRIMARY KEY, offer_id TEXT NOT NULL, @@ -254,6 +263,7 @@ impl SellerStore { updated_at_unix INTEGER NOT NULL );", )?; + Self::migrate(conn)?; conn.execute( "INSERT INTO seller_meta (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value @@ -263,6 +273,30 @@ impl SellerStore { Ok(()) } + /// Bring a store created by an older binary up to [`SCHEMA_VERSION`]. `CREATE TABLE IF NOT + /// EXISTS` never alters a table that already exists, so a column added to the schema above + /// reaches existing stores only through here. + /// + /// Every step is ADDITIVE and idempotent — a nullable column whose absence reads the same as + /// its default. Nothing here rewrites or drops a row: this store holds live trade state. + fn migrate(conn: &Connection) -> Result<(), StoreError> { + if !Self::column_exists(conn, "offers", "requested_agent")? { + conn.execute_batch("ALTER TABLE offers ADD COLUMN requested_agent TEXT;")?; + } + Ok(()) + } + + fn column_exists(conn: &Connection, table: &str, column: &str) -> Result { + let mut statement = conn.prepare(&format!("PRAGMA table_info({table})"))?; + let mut rows = statement.query([])?; + while let Some(row) = rows.next()? { + if row.get::<_, String>(1)? == column { + return Ok(true); + } + } + Ok(false) + } + /// Record (idempotently overwrite) the node's most recent start time. pub fn record_start(&self, now_unix: i64) -> Result<(), StoreError> { let conn = self.lock()?; @@ -288,8 +322,9 @@ impl SellerStore { let conn = self.lock()?; let changed = conn.execute( "INSERT OR IGNORE INTO offers - (offer_id, buyer_pubkey, amount_sats, unit, task, deadline_unix, targeted, created_at_unix) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + (offer_id, buyer_pubkey, amount_sats, unit, task, deadline_unix, targeted, created_at_unix, + requested_agent) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![ offer.offer_id, offer.buyer_pubkey, @@ -299,6 +334,7 @@ impl SellerStore { offer.deadline_unix, offer.targeted as i64, now_unix, + offer.requested_agent, ], )?; Ok(changed == 1) @@ -332,7 +368,8 @@ impl SellerStore { let conn = self.lock()?; let row = conn .query_row( - "SELECT offer_id, buyer_pubkey, amount_sats, unit, task, deadline_unix, targeted + "SELECT offer_id, buyer_pubkey, amount_sats, unit, task, deadline_unix, targeted, + requested_agent FROM offers WHERE offer_id = ?1", [offer_id], |row| { @@ -344,6 +381,7 @@ impl SellerStore { task: row.get(4)?, deadline_unix: row.get(5)?, targeted: row.get::<_, i64>(6)? != 0, + requested_agent: row.get(7)?, }) }, ) @@ -459,7 +497,7 @@ impl SellerStore { // ---- Job execution -------------------------------------------------------------------------- - /// Record which roster agent was selected to run a job. Idempotent (last write wins). + /// Record which harness ran a job. Idempotent (last write wins). pub fn assign_agent(&self, job_id: &str, agent_name: &str) -> Result<(), StoreError> { let conn = self.lock()?; conn.execute( @@ -930,6 +968,7 @@ mod tests { task: "do the thing".to_owned(), deadline_unix: 10_000, targeted: true, + requested_agent: None, } } @@ -971,6 +1010,71 @@ mod tests { let _ = std::fs::remove_file(&path); } + // TOOTH — the harness an offer requested is journaled with its other facts and READS BACK + // across a reopen. Execution can be a restart away from the claim, so a request that lived only + // in memory would let a resumed job run on whatever harness the node prefers now. + #[test] + fn requested_agent_survives_a_reopen() { + let path = temp_db("requested-agent"); + let _ = std::fs::remove_file(&path); + { + let store = SellerStore::open(&path).expect("open"); + let mut offer = sample_offer("o1"); + offer.requested_agent = Some("codex".to_owned()); + store.record_offer(&offer, 1).expect("record"); + // An offer with no preference stays None — absence is a value here, not a default. + store.record_offer(&sample_offer("o2"), 1).expect("record"); + } + let store = SellerStore::open(&path).expect("reopen"); + assert_eq!( + store.offer_row("o1").expect("row").expect("o1").requested_agent.as_deref(), + Some("codex") + ); + assert_eq!( + store.offer_row("o2").expect("row").expect("o2").requested_agent, + None + ); + let _ = std::fs::remove_file(&path); + } + + // TOOTH — a store written by a binary from before this column opens, MIGRATES, and reads its + // existing rows as "no preference". `CREATE TABLE IF NOT EXISTS` silently skips an existing + // table, so without the ALTER an upgraded node would fail every offer read on a live store. + #[test] + fn a_store_from_before_the_column_migrates_and_reads_no_preference() { + let path = temp_db("pre-agent-schema"); + let _ = std::fs::remove_file(&path); + // The offers table exactly as the previous schema had it, holding a live row. + { + let conn = Connection::open(&path).expect("create old store"); + conn.execute_batch( + "CREATE TABLE offers ( + offer_id TEXT PRIMARY KEY, + buyer_pubkey TEXT NOT NULL, + amount_sats INTEGER NOT NULL CHECK (amount_sats >= 0), + unit TEXT NOT NULL, + task TEXT NOT NULL, + deadline_unix INTEGER NOT NULL, + targeted INTEGER NOT NULL, + created_at_unix INTEGER NOT NULL + ); + INSERT INTO offers VALUES ('old', 'buyer', 21, 'sat', 'task', 10000, 1, 1);", + ) + .expect("old schema"); + } + + let store = SellerStore::open(&path).expect("open migrates"); + let row = store.offer_row("old").expect("read").expect("the pre-existing row survives"); + assert_eq!(row.amount_sats, 21, "the row is migrated, not replaced"); + assert_eq!(row.requested_agent, None, "an offer from before the column asked for no harness"); + // Migration is idempotent: opening again neither errors nor double-adds. + drop(store); + let store = SellerStore::open(&path).expect("second open"); + assert_eq!(store.health().expect("health").schema_version, SCHEMA_VERSION); + assert!(store.offer_row("old").expect("read").is_some()); + let _ = std::fs::remove_file(&path); + } + #[test] fn record_offer_is_idempotent() { let (store, path) = fresh_store("offer"); diff --git a/crates/mobee/src/doctor.rs b/crates/mobee/src/doctor.rs index 977129f1..1c01f962 100644 --- a/crates/mobee/src/doctor.rs +++ b/crates/mobee/src/doctor.rs @@ -99,7 +99,7 @@ mod checks { use mobee_core::seller_git; use super::Check; - use crate::agent_presets; + use mobee_core::agent_presets; const RELAY_TIMEOUT: Duration = Duration::from_secs(15); const MINT_TIMEOUT: Duration = Duration::from_secs(10); diff --git a/crates/mobee/src/main.rs b/crates/mobee/src/main.rs index caef8224..93a077e7 100644 --- a/crates/mobee/src/main.rs +++ b/crates/mobee/src/main.rs @@ -1,5 +1,4 @@ mod accept_cli; -mod agent_presets; mod buyer; mod cli; mod collect_cli; diff --git a/crates/mobee/src/sell.rs b/crates/mobee/src/sell.rs index c2c88449..1446aaa8 100644 --- a/crates/mobee/src/sell.rs +++ b/crates/mobee/src/sell.rs @@ -17,7 +17,7 @@ use mobee_core::delivery_transport::is_relay_git_locator; use mobee_core::home::{self, MobeeHome, SellerConfig, DEFAULT_MINT_URL, DEFAULT_RELAY_URL}; use mobee_core::profile::{self, SetProfileRequest}; -use crate::agent_presets; +use mobee_core::agent_presets; const SUCCESS: i32 = 0; const USAGE_ERROR: i32 = 1; @@ -38,8 +38,11 @@ fn readiness_decision(gate: Option>) -> Result<(), i32> { struct SellOptions { /// Force fail-closed naming of missing fields (no TTY prompts). non_interactive: bool, - /// Named preset: claude | cursor | codex. + /// The preferred named preset — the first `--agent`. agent: Option, + /// Every `--agent` in the order given: the harness registry, preference first. One entry is + /// the single-harness case and writes the config a single-harness seller has always had. + agents: Vec, /// Power-user escape hatch (repeatable). agent_argv: Vec, rate_sats: Option, @@ -394,6 +397,42 @@ fn ensure_seller_config( .or_else(|| options.agent.clone()) .or(existing_agent); + // The harness registry. Every named preset is resolved here so a typo or a missing adapter is + // a config-time refusal, not a boot-time degrade. A single `--agent` writes no list: it is the + // one-harness case the fallback already serves identically, and leaving the list out keeps a + // single-harness config exactly the shape it has always been. + let agents = if options.agents.len() > 1 { + let mut resolved = Vec::with_capacity(options.agents.len()); + for name in &options.agents { + let (label, _argv) = agent_presets::resolve_agent_preset(name, &custom_agents) + .map_err(|message| { + let _ = writeln!(err, "{message}"); + USAGE_ERROR + })?; + if !resolved + .iter() + .any(|slot: &home::AgentSlotConfig| slot.name == label) + { + resolved.push(home::AgentSlotConfig::named(label)); + } + } + let _ = writeln!( + err, + "agent registry: {}", + resolved + .iter() + .map(|slot| slot.name.as_str()) + .collect::>() + .join(", ") + ); + resolved + } else { + existing + .as_ref() + .map(|seller| seller.agents.clone()) + .unwrap_or_default() + }; + let seller = SellerConfig { agent_command, rate_sats: rate_sats.ok_or_else(|| { @@ -406,6 +445,7 @@ fn ensure_seller_config( })?, job_timeout_secs, agent, + agents, claim_open_pool, offer_backfill_secs, // Contribution (freelance-PR fork) support is ON by default for CLI-configured @@ -554,7 +594,9 @@ impl SellOptions { let name = args .get(index) .ok_or_else(|| "missing value for --agent".to_owned())?; - options.agent = Some(name.clone()); + // Repeatable: each occurrence enables one more harness, first = preferred. + options.agents.push(name.clone()); + options.agent.get_or_insert_with(|| name.clone()); } "--agent-argv" => { index += 1;