From a705a10b6672647ffc244ba6706b795660849f01 Mon Sep 17 00:00:00 2001 From: mobee-multiharness Date: Mon, 27 Jul 2026 13:34:11 -0700 Subject: [PATCH 1/6] seller: multi-harness registry, wire advertisement, and harness-aware awarding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A seller node can enable several agent harnesses at once. It resolves them once at boot, advertises them on its heartbeat and claims, claims only jobs whose requested harness it can run, and dispatches each job to the harness its buyer asked for. Execution stays serial: one job at a time. - seller_agents: the registry — resolve, advertise, dispatch. A named request is exact or nothing; advertised is exactly what dispatch can serve. - config: [seller] agents = ["claude", "codex"], entries bare or {name, slots}, so pool counts arrive later without reshaping the list. slots > 1 refused. - wire: mobee_agent multi-value on heartbeat 30340 + claim 3402 (issue #170's single-harness tag, generalized); offers request one via param/agent. - store: offers.requested_agent (additive migration) so a job resumed after a restart still dispatches to the harness it was posted for. - buyer: the award filter holds a claim to the harness the offer asked for. Co-Authored-By: Claude Opus 5 --- .buildenv | 4 + .../src/agent_presets.rs | 2 +- crates/mobee-core/src/buyer/lifecycle.rs | 48 +- crates/mobee-core/src/buyer/mod.rs | 10 +- crates/mobee-core/src/collect.rs | 2 +- crates/mobee-core/src/gateway.rs | 71 ++- crates/mobee-core/src/heartbeat.rs | 80 ++- crates/mobee-core/src/home.rs | 116 ++++- crates/mobee-core/src/job_lifecycle.rs | 28 +- crates/mobee-core/src/lib.rs | 4 + crates/mobee-core/src/payment_wallet.rs | 1 + crates/mobee-core/src/seller.rs | 2 + crates/mobee-core/src/seller_agents.rs | 493 ++++++++++++++++++ crates/mobee-core/src/seller_node/ingester.rs | 3 +- crates/mobee-core/src/seller_node/mod.rs | 6 +- crates/mobee-core/src/seller_node/outbox.rs | 2 +- crates/mobee-core/src/seller_node/run.rs | 142 ++++- crates/mobee-core/src/seller_node/signer.rs | 2 +- crates/mobee-core/src/seller_node/store.rs | 55 +- crates/mobee/src/doctor.rs | 2 +- crates/mobee/src/main.rs | 1 - crates/mobee/src/sell.rs | 48 +- 22 files changed, 1054 insertions(+), 68 deletions(-) create mode 100644 .buildenv rename crates/{mobee => mobee-core}/src/agent_presets.rs (99%) create mode 100644 crates/mobee-core/src/seller_agents.rs diff --git a/.buildenv b/.buildenv new file mode 100644 index 00000000..74b9afbf --- /dev/null +++ b/.buildenv @@ -0,0 +1,4 @@ +export PATH="/nix/store/lrap4wy75llyp86axair0f1v2n87r99m-rust-default-1.94.0/bin:/nix/store/a245z3cvf9x9sn0xlk6k8j9xhxbhda1z-gcc-wrapper-15.2.0/bin:/etc/profiles/per-user/gudnuf/bin:$PATH" +export CARGO_HOME="/srv/forge/workspaces/mobee-devsoak/.cargo-home" +export CARGO_TARGET_DIR="/srv/forge/workspaces/.multiharness-target" +export GIT_CONFIG_GLOBAL=/dev/null 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..e809e32b 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, } } 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..cc0bd619 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 @@ -846,6 +888,7 @@ mod tests { unit: "sat".into(), deadline_unix: 1_800_000_001, seller_pubkey: Some(SELLER.into()), + requested_agent: None, } ); } @@ -882,7 +925,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 +968,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 +1208,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 +1262,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..9fd0b122 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, } } @@ -887,6 +888,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..c42025b6 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", } } } @@ -657,10 +661,16 @@ fn should_resume_execution(state: super::store::JobState) -> bool { /// 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 +682,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 +735,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 +761,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 +827,7 @@ impl SellerNodeRunner { relay_url, seller_pubkey, boot_auth, + agents, }) } @@ -1436,8 +1488,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 +1640,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()); @@ -1633,6 +1690,7 @@ impl SellerNodeRunner { task: offer.task.clone(), deadline_unix: offer.deadline_unix as i64, targeted: offer.is_targeted(), + requested_agent: offer.requested_agent.clone(), }, now, ) { @@ -1653,7 +1711,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 +1886,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() { + if 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 +1943,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 +2050,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 +2210,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 +2360,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 +2383,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 +2398,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,11 +2465,11 @@ 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 } ); } @@ -2907,7 +3012,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 +3094,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 +3172,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 +3283,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 +3467,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..33b550e1 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, } } 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; From d7eaa4bcc9cff0d4e8972b57187421443b409f0d Mon Sep 17 00:00:00 2001 From: mobee-multiharness Date: Mon, 27 Jul 2026 13:49:53 -0700 Subject: [PATCH 2/6] seller: tooth the harness registry, its wire, and harness-aware awarding Co-Authored-By: Claude Opus 5 --- crates/mobee-core/src/buyer/lifecycle.rs | 68 ++++++++++++ crates/mobee-core/src/gateway.rs | 58 ++++++++++ crates/mobee-core/src/seller.rs | 61 +++++++++++ crates/mobee-core/src/seller_node/run.rs | 118 +++++++++++++++++++++ crates/mobee-core/src/seller_node/store.rs | 65 ++++++++++++ 5 files changed, 370 insertions(+) diff --git a/crates/mobee-core/src/buyer/lifecycle.rs b/crates/mobee-core/src/buyer/lifecycle.rs index e809e32b..f08f2bf6 100644 --- a/crates/mobee-core/src/buyer/lifecycle.rs +++ b/crates/mobee-core/src/buyer/lifecycle.rs @@ -468,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/gateway.rs b/crates/mobee-core/src/gateway.rs index cc0bd619..06f7cbb7 100644 --- a/crates/mobee-core/src/gateway.rs +++ b/crates/mobee-core/src/gateway.rs @@ -816,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"; diff --git a/crates/mobee-core/src/seller.rs b/crates/mobee-core/src/seller.rs index 9fd0b122..65ee468a 100644 --- a/crates/mobee-core/src/seller.rs +++ b/crates/mobee-core/src/seller.rs @@ -822,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#" diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index c42025b6..866953e3 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -2474,6 +2474,124 @@ mod tests { ); } + // 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 (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. diff --git a/crates/mobee-core/src/seller_node/store.rs b/crates/mobee-core/src/seller_node/store.rs index 33b550e1..0cfe718c 100644 --- a/crates/mobee-core/src/seller_node/store.rs +++ b/crates/mobee-core/src/seller_node/store.rs @@ -1010,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"); From 5f6c06413afcc72b569de6a416aff7dbcf74cf1a Mon Sep 17 00:00:00 2001 From: mobee-multiharness Date: Mon, 27 Jul 2026 13:56:40 -0700 Subject: [PATCH 3/6] seller: collapse the agent-journal guard (clippy) Co-Authored-By: Claude Opus 5 --- crates/mobee-core/src/seller_node/run.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 866953e3..6f1042c5 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -1905,12 +1905,12 @@ impl SellerNodeRunner { 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() { - if 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}" - ); - } + 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. From 5c115f9a7c6cc6aa950227519b5e6fa6d34fed87 Mon Sep 17 00:00:00 2001 From: mobee-multiharness Date: Mon, 27 Jul 2026 14:07:24 -0700 Subject: [PATCH 4/6] drop the workspace-local build env from the branch Co-Authored-By: Claude Opus 5 --- .buildenv | 4 ---- 1 file changed, 4 deletions(-) delete mode 100644 .buildenv diff --git a/.buildenv b/.buildenv deleted file mode 100644 index 74b9afbf..00000000 --- a/.buildenv +++ /dev/null @@ -1,4 +0,0 @@ -export PATH="/nix/store/lrap4wy75llyp86axair0f1v2n87r99m-rust-default-1.94.0/bin:/nix/store/a245z3cvf9x9sn0xlk6k8j9xhxbhda1z-gcc-wrapper-15.2.0/bin:/etc/profiles/per-user/gudnuf/bin:$PATH" -export CARGO_HOME="/srv/forge/workspaces/mobee-devsoak/.cargo-home" -export CARGO_TARGET_DIR="/srv/forge/workspaces/.multiharness-target" -export GIT_CONFIG_GLOBAL=/dev/null From 0a901ac059d91b291b399de7970ba4d695c2d77e Mon Sep 17 00:00:00 2001 From: mobee-multiharness Date: Mon, 27 Jul 2026 15:21:10 -0700 Subject: [PATCH 5/6] seller: tooth the wire-to-row mapping of the harness request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every other tooth either builds the offer row by hand or reads one back, so all of them stayed green if the claim path dropped the request on its way into the store — invariant 2 would hold in one process and be dead the moment execution happened after a restart. Give the mapping a name and a test that starts from a wire event. Co-Authored-By: Claude Opus 5 --- crates/mobee-core/src/seller_node/run.rs | 74 +++++++++++++++++++----- 1 file changed, 61 insertions(+), 13 deletions(-) diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 6f1042c5..3d4abb2e 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -659,6 +659,27 @@ 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, then the harness the offer asked for. @@ -1681,19 +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(), - requested_agent: offer.requested_agent.clone(), - }, - 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; } @@ -2512,6 +2525,41 @@ mod tests { ); } + // 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 From 0a4061c02ec30d89da3a7608ca09016fae3368fe Mon Sep 17 00:00:00 2001 From: mobee-multiharness Date: Mon, 27 Jul 2026 16:10:40 -0700 Subject: [PATCH 6/6] seller: drop needless borrows in the offer-draft test helpers (clippy) Co-Authored-By: Claude Opus 5 --- crates/mobee-core/src/seller_node/run.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/mobee-core/src/seller_node/run.rs b/crates/mobee-core/src/seller_node/run.rs index 3d4abb2e..04211dea 100644 --- a/crates/mobee-core/src/seller_node/run.rs +++ b/crates/mobee-core/src/seller_node/run.rs @@ -2536,7 +2536,7 @@ mod tests { // 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)) + 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"); @@ -2554,7 +2554,7 @@ mod tests { 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)) + 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); @@ -2645,7 +2645,7 @@ mod tests { // 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") {