diff --git a/crates/maxplayer-core/Cargo.toml b/crates/maxplayer-core/Cargo.toml index 5cee4ccd3..84dee5ac0 100644 --- a/crates/maxplayer-core/Cargo.toml +++ b/crates/maxplayer-core/Cargo.toml @@ -165,3 +165,10 @@ tokio = { version = "1.52.0", features = ["macros", "net", "rt", "rt-multi-threa tokio-tungstenite = { version = "0.26", default-features = false, features = ["handshake"] } url = "2.5.8" uuid = "1.23.5" + +# F3 (advisor verdict on the discovery foundation): the discovery read driven over a SCRIPTED relay, +# so an EOSE, a timeout, a drop and a CLOSED each land where they must. Needs the buyer identity and +# the relay leg, both `wallet`-gated, and tokio-tungstenite for the fixture's raw NIP-01 socket. +[[test]] +name = "discovery_relay_behavior" +required-features = ["wallet"] diff --git a/crates/maxplayer-core/src/buyer/lifecycle.rs b/crates/maxplayer-core/src/buyer/lifecycle.rs index 344752c20..60bbbee04 100644 --- a/crates/maxplayer-core/src/buyer/lifecycle.rs +++ b/crates/maxplayer-core/src/buyer/lifecycle.rs @@ -1745,6 +1745,7 @@ mod tests { capabilities: vec!["python".to_owned()], harness_variant: None, hardware: None, + specialty: None, }; // The seat serves the `codex` PRESET as well as the family. A model axis now has to name a // preset, because dispatch reads nothing else — so without this the model case would be @@ -3677,6 +3678,7 @@ mod tests { capabilities: capabilities.iter().map(|c| (*c).to_owned()).collect(), harness_variant: None, hardware: None, + specialty: None, } } diff --git a/crates/maxplayer-core/src/buyer/mod.rs b/crates/maxplayer-core/src/buyer/mod.rs index 93b0cbb87..c7b58b91b 100644 --- a/crates/maxplayer-core/src/buyer/mod.rs +++ b/crates/maxplayer-core/src/buyer/mod.rs @@ -381,6 +381,7 @@ async fn dispatch(context: &Arc, request: Request) -> Response { "status" | "health" => status(context, id).await, "post_job" => post_job(context, id, request.params).await, "get_job" => get_job(context, id, request.params).await, + "discover_sellers" => discover_sellers(context, id, request.params).await, "award" => award(context, id, request.params).await, "collect" => collect(context, id, request.params).await, "accept_claim" | "authorize_pay" => Response::err( @@ -526,19 +527,33 @@ fn post_job_kind(params: &PostJobParams) -> Result { /// CLI/MCP use), record its auto-award intent, and spawn the background auto-award task — the /// daemon-drives-the-award half of the 2-call trade loop (post_job → collect). No reservation is /// taken at post — funds are reserved at award. -async fn post_job(context: &Arc, id: Value, params: Value) -> Response { - let params: PostJobParams = match serde_json::from_value(params) { - Ok(params) => params, - Err(error) => return Response::err(id, CODE_METHOD_NOT_FOUND, format!("post_job params: {error}")), - }; - let job = match post_job_kind(¶ms) { - Ok(job) => job, - Err(message) => return Response::err(id, CODE_METHOD_NOT_FOUND, message), - }; - let payment_mode = match post_job_payment_mode(params.payment.as_deref(), params.amount_sats) { - Ok(mode) => mode, - Err(message) => return Response::err(id, CODE_METHOD_NOT_FOUND, message), - }; +/// The `post_job` RPC body, mapped — the request the lifecycle will be handed, plus the three +/// values the daemon keeps for its auto-award intent. +/// +/// A named result rather than a tuple because [`map_post_job_params`] is the boundary a caller +/// hands a discovered pubkey to, and "which field did the seller end up in" is the whole question +/// at that boundary. +pub struct PostJobMapping { + pub request: PostJobRequest, + pub max_sats: u64, + pub harness: Option, + pub model: Option, +} + +/// Map a `post_job` RPC body to a [`PostJobRequest`], with NO daemon, relay, wallet or money. +/// +/// Lifted out of the RPC handler so this boundary can be exercised on its own. It is the mapping a +/// buyer's pubkey actually travels through: the MCP `post_job` tool routes here, so a value that +/// `OfferDraft` would accept but this rejects is a value no user can post with. Discovery ends at a +/// pubkey precisely so it can be handed in here, which is why the handoff is asserted through this +/// function rather than around it. +/// +/// Errors are the RPC's own strings, unchanged, so the handler's replies read exactly as before. +pub fn map_post_job_params(params: Value) -> Result { + let params: PostJobParams = + serde_json::from_value(params).map_err(|error| format!("post_job params: {error}"))?; + let job = post_job_kind(¶ms)?; + let payment_mode = post_job_payment_mode(params.payment.as_deref(), params.amount_sats)?; let max_sats = params.max_sats.unwrap_or(params.amount_sats); let harness = params.harness.clone(); let model = params.model.clone(); @@ -568,6 +583,24 @@ async fn post_job(context: &Arc, id: Value, params: Value) -> Resp // free bind — uncollectable. payment_mode, }; + Ok(PostJobMapping { + request, + max_sats, + harness, + model, + }) +} + +async fn post_job(context: &Arc, id: Value, params: Value) -> Response { + let PostJobMapping { + request, + max_sats, + harness, + model, + } = match map_post_job_params(params) { + Ok(mapping) => mapping, + Err(message) => return Response::err(id, CODE_METHOD_NOT_FOUND, message), + }; match job_lifecycle::post_job_async(&context.home, request).await { Ok(outcome) => { // Record the intent BEFORE spawning so a crash right after post still re-arms on restart. @@ -721,6 +754,116 @@ async fn get_job(context: &BuyerContext, id: Value, params: Value) -> Response { } } +/// Params for the `discover_sellers` RPC. Every field is optional and every default is the shipped +/// discovery rule, so `{}` is the whole call. +/// +/// ⛔ **THERE IS NO QUERY FIELD, AND ITS ABSENCE IS THE DESIGN.** No `specialty_contains`, no +/// keyword, no required-skill list: a server-side text predicate over seller-declared prose is the +/// string-match gate this slice is explicitly ordered not to build, and shipping it as a *filter* +/// would make it a de-facto matcher the moment a caller trusted the shortlist it returned. The +/// three parameters here bound the READ (how many, how fresh, how long to wait); the choosing is +/// the caller's, over rows it can see. +#[derive(Debug, Deserialize)] +struct DiscoverSellersParams { + /// Cap on announcements requested from the relay. Defaults to + /// [`crate::discovery::DEFAULT_DIRECTORY_LIMIT`]. + #[serde(default)] + limit: Option, + /// Recency window in seconds. Defaults to [`crate::discovery::DEFAULT_MAX_AGE_SECS`]. + #[serde(default)] + max_age_secs: Option, + /// Total budget for the read in seconds. Defaults to + /// [`crate::discovery::DEFAULT_DISCOVERY_TIMEOUT_SECS`], capped at + /// [`crate::discovery::MAX_DISCOVERY_TIMEOUT_SECS`]. + #[serde(default)] + timeout_secs: Option, +} + +/// Refuse an out-of-range `discover_sellers` bound at the RPC boundary, before any relay work. +/// +/// Zero is refused on all three rather than read as "no limit": an unbounded read has no place on a +/// surface whose caller holds a deadline, and a zero recency window admits nothing, so a caller +/// that sent it wants an answer this call cannot give. Refusing beats a silent substitution the +/// caller would then mistake for evidence. +fn discover_sellers_bounds_error(params: &DiscoverSellersParams) -> Option { + if params.limit == Some(0) { + return Some( + "limit=0 is refused (it would mean an unbounded relay read on a deadline-bound call); \ + omit limit for the default or pass a positive value" + .to_owned(), + ); + } + if let Some(limit) = params.limit { + if limit > crate::discovery::DEFAULT_DIRECTORY_LIMIT { + return Some(format!( + "limit={limit} exceeds the directory read cap of {}; omit limit for the default", + crate::discovery::DEFAULT_DIRECTORY_LIMIT + )); + } + } + if params.max_age_secs == Some(0) { + return Some( + "max_age_secs=0 is refused: no beat can be zero seconds old, so it would report an \ + empty market as a fact; omit it for the default window" + .to_owned(), + ); + } + match params.timeout_secs { + Some(0) => Some( + "timeout_secs=0 is refused: a read with no budget cannot confirm anything; omit it \ + for the default" + .to_owned(), + ), + Some(secs) if secs > crate::discovery::MAX_DISCOVERY_TIMEOUT_SECS => Some(format!( + "timeout_secs={secs} exceeds the discovery cap of {}s (bounded under the MCP tool \ + deadline); omit timeout_secs for the default or pass a value <= {}", + crate::discovery::MAX_DISCOVERY_TIMEOUT_SECS, + crate::discovery::MAX_DISCOVERY_TIMEOUT_SECS, + )), + _ => None, + } +} + +/// Read the public seat directory — **a READ, and the only money-free RPC on this surface besides +/// `status`**. It takes no `money_lock`, opens no wallet, touches no reservation ledger and +/// publishes no event; the whole of it is [`crate::discovery::fetch_directory_async`] plus the +/// clock. Relay failure surfaces as an error and an unanswered read as `read_confirmed: false`, so +/// no caller can read either as "the market is empty" (see [`crate::discovery::DiscoveryError`]). +async fn discover_sellers(context: &BuyerContext, id: Value, params: Value) -> Response { + let params: DiscoverSellersParams = match serde_json::from_value(params) { + Ok(params) => params, + Err(error) => { + return Response::err( + id, + CODE_METHOD_NOT_FOUND, + format!("discover_sellers params: {error}"), + ); + } + }; + if let Some(error) = discover_sellers_bounds_error(¶ms) { + return Response::err(id, CODE_METHOD_NOT_FOUND, error); + } + // The reader's clock, taken once here so every row's `age_secs` is measured against the same + // instant the recency rule used. + let now_unix = u64::try_from(now_unix()).unwrap_or(0); + let mut policy = crate::discovery::DirectoryPolicy::at(now_unix); + if let Some(max_age_secs) = params.max_age_secs { + policy.max_age_secs = max_age_secs; + } + let limit = params + .limit + .unwrap_or(crate::discovery::DEFAULT_DIRECTORY_LIMIT); + let budget = Duration::from_secs( + params + .timeout_secs + .unwrap_or(crate::discovery::DEFAULT_DISCOVERY_TIMEOUT_SECS), + ); + match crate::discovery::fetch_directory_async(&context.home, policy, limit, budget).await { + Ok(directory) => Response::ok(id, json!(directory)), + Err(error) => Response::err(id, CODE_INTERNAL, format!("discover_sellers: {error}")), + } +} + /// Params for the `award` RPC. `claim_id` present ⇒ MANUAL award of that claim (the fine-grain /// flag from #126); absent ⇒ AUTO-award the first claim passing the hard filters. `max_sats` /// caps the price the buyer will commit to (defaults to the offer amount). @@ -6503,4 +6646,112 @@ mod tests { ); assert_eq!(paid["amount_sats"], 21); } + + // ── discover_sellers: the read-only directory RPC ──────────────────────────────────────── + // + // The relay leg is covered offline in `crate::discovery`; what belongs HERE is the RPC + // boundary — the params it accepts, the bounds it refuses before any relay work, and the shape + // its answer serialises to. + + /// The params a caller sends, through the REAL deserializer. + fn discover_params(body: Value) -> Result { + serde_json::from_value(body).map_err(|error| error.to_string()) + } + + // `{}` is a complete call: every bound has a shipped default, so a caller that wants the + // shipped rules sends nothing. Unknown/absent fields must not force a caller to state them. + #[test] + fn an_empty_discover_sellers_body_takes_every_shipped_default() { + let params = discover_params(json!({})).expect("empty body is valid"); + assert_eq!(params.limit, None); + assert_eq!(params.max_age_secs, None); + assert_eq!(params.timeout_secs, None); + assert_eq!(discover_sellers_bounds_error(¶ms), None); + } + + // Every bound is refused OUT OF RANGE rather than clamped, and the refusal happens before a + // socket is opened. A silently-clamped 60s budget would hand the caller a 10s empty answer it + // would read as sixty seconds of evidence — the one failure mode this whole module exists to + // prevent. + #[test] + fn an_out_of_range_discover_sellers_bound_is_refused_not_clamped() { + let over_budget = discover_params(json!({ "timeout_secs": 60 })).expect("parses"); + let error = discover_sellers_bounds_error(&over_budget).expect("refused"); + assert!( + error.contains(&format!( + "exceeds the discovery cap of {}s", + crate::discovery::MAX_DISCOVERY_TIMEOUT_SECS + )), + "the refusal must name the cap: {error}" + ); + + let over_limit = discover_params(json!({ + "limit": crate::discovery::DEFAULT_DIRECTORY_LIMIT + 1 + })) + .expect("parses"); + assert!( + discover_sellers_bounds_error(&over_limit) + .expect("refused") + .contains("exceeds the directory read cap"), + ); + + // Zero on any bound is a request this call cannot honour, so it is named rather than + // reinterpreted as "no limit" / "no window" / "no wait". + for body in [ + json!({ "limit": 0 }), + json!({ "max_age_secs": 0 }), + json!({ "timeout_secs": 0 }), + ] { + let params = discover_params(body.clone()).expect("parses"); + assert!( + discover_sellers_bounds_error(¶ms).is_some(), + "zero must be refused: {body}" + ); + } + + // At the caps, and inside them, nothing is refused. + let at_caps = discover_params(json!({ + "limit": crate::discovery::DEFAULT_DIRECTORY_LIMIT, + "max_age_secs": 60, + "timeout_secs": crate::discovery::MAX_DISCOVERY_TIMEOUT_SECS + })) + .expect("parses"); + assert_eq!(discover_sellers_bounds_error(&at_caps), None); + } + + // ⛔ NO MATCH PREDICATE ON THE RPC EITHER. The MCP schema refuses unknown inputs, but the + // daemon is reachable directly over the socket, so the absence has to hold here too: a + // `specialty_contains` a caller could send would be the string-match gate, and serde would + // accept the field silently if this ever grew one. + #[test] + fn the_discover_sellers_rpc_offers_no_specialty_predicate() { + let params = discover_params(json!({ + "specialty_contains": "rust", + "required_skills": ["rust"], + "query": "rust" + })) + .expect("unknown fields are ignored, not honoured"); + // Every bound stayed default: nothing in that body reached a filter. + assert_eq!(params.limit, None); + assert_eq!(params.max_age_secs, None); + assert_eq!(params.timeout_secs, None); + assert_eq!(discover_sellers_bounds_error(¶ms), None); + } + + // The wire shape of an answer: the caller must be able to tell an answered-empty market from an + // unanswered read, and `read_confirmed` is the only field that says so. + #[test] + fn a_directory_answer_serialises_its_confirmation_flag() { + let confirmed = json!(crate::discovery::SellerDirectory::empty_confirmed()); + assert_eq!(confirmed["read_confirmed"], json!(true)); + assert_eq!(confirmed["sellers"], json!([])); + assert_eq!(confirmed["events_read"], json!(0)); + + let unverified = json!(crate::discovery::SellerDirectory::unverified()); + assert_eq!(unverified["read_confirmed"], json!(false)); + assert_eq!( + unverified["sellers"], confirmed["sellers"], + "the two differ ONLY in the flag — which is why the flag has to be read" + ); + } } diff --git a/crates/maxplayer-core/src/discovery.rs b/crates/maxplayer-core/src/discovery.rs new file mode 100644 index 000000000..925f61e5f --- /dev/null +++ b/crates/maxplayer-core/src/discovery.rs @@ -0,0 +1,1262 @@ +//! Buyer-side **specialist discovery**: read the public seat directory off kind-30340 +//! announcements so a buyer can find a seat it has never met, read what that seat says it is for, +//! and then target it with the posting path that already exists. +//! +//! ## What this is, and the three things it is NOT +//! +//! It is ONE read. A buyer asks the relay for seat announcements, this module reduces them to the +//! latest live beat per seat, and the caller gets rows to look at. Nothing here posts, awards, pays, +//! or reserves — see [`SellerDirectory`] and the `discovery_never_writes` test for the pinned form +//! of that claim. +//! +//! - **NOT automatic matching.** There is no scoring, no ranking, no keyword predicate. A buyer (or +//! its agent, or its human) reads [`DiscoveredSeller::specialty`] and decides. The rows come back +//! in a deliberately merit-free order — see [`SellerDirectory::sellers`]. +//! - **NOT a competence assertion.** `specialty` is text the seat's operator typed. It is unverified +//! by construction and rides the announcement alone, never a claim; see +//! [`crate::heartbeat::SPECIALTY_TAG`] for why that placement is load-bearing rather than +//! incidental. +//! - **NOT a capacity or eligibility signal.** ⛔ A FRESH BEAT PROVES NEITHER. `accepting=y` says the +//! seat is alive and serving, not that it has a free execution slot (`slots` defaults to 3 and the +//! gate is `SlotGate::try_reserve` at claim time), and the admission fields say who the seat +//! *advertises* it admits, which is intent and not a guarantee. The authoritative signal that a +//! seat will take a job remains that the seat CLAIMS one. A buyer that treats a row here as +//! "will serve me" has read it wrong. +//! +//! ## Shape: a pure reducer plus a thin transport +//! +//! [`reduce_directory`] holds every rule — recency, future-dating, retraction, latest-per-address — +//! and touches no relay, so all of it is testable offline against drafts built by the same +//! [`crate::heartbeat`] emitters a real seat publishes through. [`fetch_directory_async`] is the +//! only part that needs a socket. The split is why the acceptance tests need no live relay and no +//! sats. + +use std::collections::HashMap; + +use crate::gateway::EventDraft; +use crate::heartbeat::{HeartbeatKey, ParsedHeartbeat, parse_heartbeat}; + +/// Wire spelling for an admission field the seat did NOT state. +/// +/// ⛔ **UNSTATED IS NOT `closed`.** A seat published before the §4.2 admission tags existed states +/// neither half, and rendering that as "closed" would tell a buyer that every seat running today +/// refuses it. It is a third value because it is a third fact: the seat did not say. +pub const ADMISSION_UNSTATED: &str = "unstated"; + +/// How old a beat may be and still count as a LIVE seat, in seconds. +/// +/// Derived from the shipped cadence rather than picked: [`crate::home`]'s heartbeat defaults are a +/// 300 s interval and 3 missed intervals before the seat itself calls a publish stalled, so 900 s is +/// the same patience the seller side already applies to its own beat. A buyer using a different +/// window passes its own through [`DirectoryPolicy::max_age_secs`]. +/// +/// ⚠ THE WINDOW IS THE ONLY COVER FOR AN UNGRACEFUL EXIT, and that is why it exists at all. +/// kind-30340 is addressable: the relay holds exactly one announcement per `(pubkey, d)`, and a seat +/// killed by SIGKILL, an OOM or a power cut leaves its last `accepting=y` standing as its permanent +/// public answer with no later event to correct it. Waiting produces nothing. Recency filtering is +/// therefore REQUIRED of every consumer and is not a tuning nicety — see +/// [`crate::heartbeat::retraction_for_state`], which covers the graceful case and explicitly does +/// not cover this one. +pub const DEFAULT_MAX_AGE_SECS: u64 = 900; + +/// How far into the future a beat's `created_at` may sit before it is discarded, in seconds. +/// +/// Relay and seat clocks disagree by seconds in normal operation, so a small tolerance keeps honest +/// seats visible. Beyond it the timestamp is not usable: an addressable event is superseded by +/// `created_at` order, so a far-future beat would outrank every genuine later one and pin a stale +/// row in place until real time caught up. Discarding it costs one seat's visibility; keeping it +/// costs the correctness of the whole ordering rule. +pub const DEFAULT_MAX_CLOCK_SKEW_SECS: u64 = 300; + +/// Default cap on how many announcements one directory read asks the relay for. +pub const DEFAULT_DIRECTORY_LIMIT: usize = 500; + +/// How long one directory read waits on the relay, in seconds. +/// +/// Finite by construction — there is deliberately no unbounded variant, because an MCP tool that +/// never returns is indistinguishable to its caller from a hung buyer. Declared here rather than in +/// the relay leg so a build with no relay features can still state the tool's contract. +pub const DEFAULT_DISCOVERY_TIMEOUT_SECS: u64 = 8; + +/// The largest directory-read budget a caller may ask for, in seconds. +/// +/// ⚠ THE CEILING EXISTS BECAUSE THE CALLER THAT MATTERS HAS ITS OWN DEADLINE. `maxplayer mcp` +/// caps a `tools/call` at 15 s (`mcp::TOOL_DEADLINE_SECS`) and reports a cap-hit as a tool error +/// with no directory in it, so a budget at or above that ceiling converts every slow relay into +/// "the tool broke" instead of the honest "the relay did not answer" this module goes to some +/// trouble to be able to say. 10 s leaves the daemon room to reply inside the client's window. +/// +/// It is REFUSED at the RPC boundary rather than silently clamped, the same choice +/// [`crate::job_lifecycle::WAIT_FOR_CAP_SECS`] makes for the long poll: a caller that asked for +/// 60 s and got 10 would read the empty answer as sixty seconds of evidence. +pub const MAX_DISCOVERY_TIMEOUT_SECS: u64 = 10; + +/// One seat as the public directory describes it, at the moment of the read. +/// +/// Every field comes off ONE announcement — the latest live beat for this seat's `(pubkey, d)` +/// address — so the row is internally consistent rather than assembled from several events. +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] +pub struct DiscoveredSeller { + /// The seat's pubkey, 64-hex lowercase. **This is the discovery output that matters**: it is the + /// value a buyer hands to the existing targeted-post parameter (`seller_pubkey`) to hire this + /// seat. Nothing else here is an identifier. + pub pubkey: String, + /// What the seat says it specialises in, or `None` when it stated nothing. + /// + /// ⛔ SELLER-DECLARED, NEVER VERIFIED. Read it as an advertisement, not a credential. `None` is + /// unstated and NOT a claim to be a generalist — a seat published before the field existed and a + /// seat whose operator declined to describe it are the same value here, and both stay listed. + pub specialty: Option, + /// The announcement's `created_at` (unix seconds), as the relay served it. + pub announced_at: u64, + /// How long ago that was, at the moment of this read. Carried alongside the timestamp rather + /// than left to the caller so two callers cannot compute "age" against two different clocks. + pub age_secs: u64, + /// The seat's advertised rate floor in sats — the LOWEST it accepts, per §4.2, not a quote. + pub rate_sats: u64, + /// The seat states it takes NO payment at all (§4.1). ⚠ Do not substitute `rate_sats == 0`: + /// that means "any amount ≥ 0", which a buyer holding zero sats cannot act on. + pub takes_no_payment: bool, + /// Every mint this seat can be paid on. Never empty — a seat naming none does not parse. + pub accepted_mints: Vec, + /// The harnesses the seat advertises, in its preference order. Empty ⇒ stated none, which is not + /// a claim that it can run nothing (the unlabelled `--agent-argv` hatch has no name to publish). + pub agents: Vec, + /// The enum-bound harness families the seat serves. Empty ⇒ unstated. + pub harness_families: Vec, + /// Untargeted (open-pool) admission: `open`, `closed`, or [`ADMISSION_UNSTATED`]. + pub admits_pool: String, + /// Targeted admission: `open`, `named`, `closed`, or [`ADMISSION_UNSTATED`]. + /// + /// `named` discloses only that an allowlist EXISTS, never who is on it — so a buyer reading it + /// learns that targeting this seat may be refused, which is exactly the fact a boolean would + /// have hidden. + pub admits_targeted: String, +} + +/// Why a returned announcement did not become a row. Counts only — a directory read is a diagnostic +/// surface, and naming the pubkeys it dropped would publish a list of dead seats to no purpose. +/// +/// It exists so an empty directory can be EXPLAINED. "The relay answered and held 40 beats, all of +/// them stale" and "the relay answered and held nothing" are different facts about the market, and +/// without this they are the same empty list. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize)] +pub struct DirectorySkips { + /// Events that are not parseable maxplayer seat announcements at all (wrong kind, missing the + /// `t=maxplayer` guard, a protocol major this build does not speak, no payable mint). + pub unparseable: u32, + /// Seats whose latest beat is older than [`DirectoryPolicy::max_age_secs`]. + pub stale: u32, + /// Seats whose latest beat is dated further ahead than + /// [`DirectoryPolicy::max_clock_skew_secs`]. + pub future_dated: u32, + /// Seats whose latest beat says `accepting=n` — the seat has left the market or is closed. This + /// is the retraction being HONOURED: the terminal beat superseded the seat's old `accepting=y` + /// at the same address, and this read resolves the address, so the newer word wins. + pub retracted: u32, +} + +/// The rules one directory read applies. Taken as a value rather than read from globals so every +/// rule is exercisable offline with a fixed clock. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DirectoryPolicy { + /// The reader's "now", unix seconds. Supplied by the caller so a test can pin it. + pub now_unix: u64, + /// Recency window — see [`DEFAULT_MAX_AGE_SECS`]. + pub max_age_secs: u64, + /// Future-dating tolerance — see [`DEFAULT_MAX_CLOCK_SKEW_SECS`]. + pub max_clock_skew_secs: u64, +} + +impl DirectoryPolicy { + /// The shipped rules against a caller-supplied clock. + pub fn at(now_unix: u64) -> Self { + Self { + now_unix, + max_age_secs: DEFAULT_MAX_AGE_SECS, + max_clock_skew_secs: DEFAULT_MAX_CLOCK_SKEW_SECS, + } + } +} + +/// The result of one directory read. +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] +pub struct SellerDirectory { + /// The live seats, ordered by `pubkey` ascending. + /// + /// ⛔ **THE ORDER IS DELIBERATELY MERIT-FREE, AND SORTING BY FRESHNESS WOULD NOT BE.** Any order + /// this function chooses is the order a caller reads first, so freshness-descending would ship a + /// ranking policy — "the seat that beat most recently is the best one to hire" — which is a + /// claim discovery has no standing to make and which rewards beating often. Sorting on the + /// pubkey is stable, total, and says nothing. `announced_at`/`age_secs` are on every row for a + /// caller that wants to order by them and owns that decision. + pub sellers: Vec, + /// **Whether the relay ANSWERED this read.** `fetch_events` resolves `Ok(empty)` on timeout, so + /// an empty `sellers` cannot by itself tell "no specialists are advertising" from "we stopped + /// waiting" — the same bytes, and the discriminator has to be asked for. `true` ⇒ the emptiness + /// is a fact about the market. `false` ⇒ it is a fact about our patience. + /// + /// It is the same discipline [`crate::job_lifecycle::JobView::read_confirmed`] applies to offer + /// reads (#291/#322), and it is `false` on any directory not built from a confirmed read, so the + /// misleading direction is the one a caller has to opt into. A hard relay failure is an + /// [`DiscoveryError::Relay`] instead — that is a THIRD outcome, not this flag. + pub read_confirmed: bool, + /// What the read saw and dropped. See [`DirectorySkips`]. + pub skipped: DirectorySkips, + /// How many announcements the relay returned, before any rule was applied. The denominator for + /// everything above. + pub events_read: u32, +} + +impl SellerDirectory { + /// An answered read of a market with nothing in it. + /// + /// Public rather than crate-private because the DISTINCTION it makes with [`Self::unverified`] + /// is the module's contract, not an implementation detail: a caller assembling a directory from + /// its own transport has to be able to state which of the two it got, and a private + /// constructor would leave it building the struct field-by-field and choosing + /// `read_confirmed` by hand. + pub fn empty_confirmed() -> Self { + Self { + sellers: Vec::new(), + read_confirmed: true, + skipped: DirectorySkips::default(), + events_read: 0, + } + } + + /// A read the relay never answered. Empty AND unconfirmed — see [`Self::read_confirmed`]. + pub fn unverified() -> Self { + Self { + sellers: Vec::new(), + read_confirmed: false, + skipped: DirectorySkips::default(), + events_read: 0, + } + } +} + +/// One announcement as the relay served it: the author, the timestamp, and the event's tag content. +/// +/// A plain struct rather than a `nostr_sdk::Event` so [`reduce_directory`] compiles and tests +/// without the gateway feature, and so a test can build one from +/// [`crate::heartbeat::HeartbeatDraft::to_event_draft`] — the very emitter a real seat publishes +/// through, rather than a hand-written tag set made to agree with it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AnnouncedSeat { + /// The event's author pubkey, 64-hex. + pub author_pubkey: String, + /// The event's `created_at`, unix seconds. + pub created_at: u64, + /// The event's SIGNED id, 64-hex lowercase — the NIP-01 tie-breaker, carried across the + /// transport-to-reducer seam because `(author, created_at)` is NOT a total order. Two signed + /// beats for one address CAN share a timestamp, and if they disagree about `accepting` then + /// whichever the reducer keeps decides whether the seat is live or retracted. Without the id + /// that decision falls to iteration luck. + pub event_id: String, + /// The event's kind/tags/content. + pub event: EventDraft, +} + +/// Whether the relay actually FINISHED answering the directory request. +/// +/// This is the whole of [`SellerDirectory::read_confirmed`], and it is a transport fact the reducer +/// cannot derive: rows alone cannot tell a completed answer from a stream that stopped early. +/// `fetch_events`-style helpers end on either an `EOSE` or a spent deadline WITHOUT distinguishing +/// them, so a caller that infers completion from "we got here with some events" certifies a partial +/// read. Only the directory REQ's own `EOSE` may set [`Self::ConfirmedByEose`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ReadCompletion { + /// The directory subscription's own `EOSE` arrived: the relay served everything it holds for + /// this filter, so an empty row set is a genuinely empty market. + ConfirmedByEose, + /// The read ended without its `EOSE` — deadline spent, socket dropped, or stream closed. Rows + /// already received are kept and reported HONESTLY as unconfirmed; absence proves nothing. + Unconfirmed, +} + +impl ReadCompletion { + /// `true` only for [`Self::ConfirmedByEose`]. Written once, here, so no call site can decide + /// what "confirmed" means for itself. + pub fn is_confirmed(self) -> bool { + matches!(self, Self::ConfirmedByEose) + } +} + +/// Reduce raw announcements to the live seat directory — **every discovery rule, and no I/O**. +/// +/// In order: +/// +/// 1. **Parse.** Anything [`parse_heartbeat`] refuses is counted and dropped. A junk event squatting +/// the kind must not become a row a buyer might target. +/// 2. **Resolve the address.** Rows are keyed by `(pubkey, d)` via [`ParsedHeartbeat::key`], NEVER +/// by event id: kind-30340 is addressable and superseded IN PLACE, so an id-keyed reduce would +/// keep a seat's dead announcements alongside its live one. Newest `created_at` wins; an exact +/// timestamp tie is broken by the LOWEST lexical event id, per NIP-01's retention rule for +/// addressable events. Both together are a total order over signed input, so the result does not +/// depend on the order the relay handed the events over. +/// 3. **Discard a future-dated beat**, past the skew tolerance — it would outrank every genuine +/// later beat for this address. +/// 4. **Discard a stale beat**, past the recency window. +/// 5. **Honour a retraction.** A resolved beat with `accepting=n` is the seat's own last word that +/// it is not taking work; it leaves the directory. +/// +/// Steps 3–5 apply to the RESOLVED beat, after step 2, and that ordering is the point: a seat's +/// newer retraction must not be filtered out on its own merits and leave the seat's older +/// `accepting=y` standing as the survivor. Resolve the address first, judge the winner second. +pub fn reduce_directory( + announcements: impl IntoIterator, + policy: DirectoryPolicy, + completion: ReadCompletion, +) -> SellerDirectory { + let mut skipped = DirectorySkips::default(); + let mut events_read: u32 = 0; + // (pubkey, d) -> the winning beat for that address so far: its timestamp, its signed id (the + // tie-breaker), and the parse. + let mut newest: HashMap = HashMap::new(); + + for announcement in announcements { + events_read = events_read.saturating_add(1); + let Ok(parsed) = parse_heartbeat(&announcement.event) else { + skipped.unparseable = skipped.unparseable.saturating_add(1); + continue; + }; + let pubkey = announcement.author_pubkey.to_ascii_lowercase(); + let key = parsed.key(&pubkey); + let created = announcement.created_at; + let id = announcement.event_id.to_ascii_lowercase(); + // NEWER WINS, and on an exact tie the LOWER id wins — NIP-01's own rule for retaining an + // addressable event. A conforming relay may never serve the conflicting pair, but this + // reducer is reusable and is handed whatever arrives: two same-address beats sharing a + // timestamp and disagreeing about `accepting` must resolve the SAME way whichever order + // they come in, or a seat is live or retracted by luck. + let supersedes = match newest.get(&key) { + None => true, + Some((previous_created, previous_id, _)) => match created.cmp(previous_created) { + std::cmp::Ordering::Greater => true, + std::cmp::Ordering::Less => false, + std::cmp::Ordering::Equal => id < *previous_id, + }, + }; + if supersedes { + newest.insert(key, (created, id, parsed)); + } + } + + let mut sellers: Vec = Vec::with_capacity(newest.len()); + for (key, (created_at, _id, parsed)) in newest { + if created_at > policy.now_unix.saturating_add(policy.max_clock_skew_secs) { + skipped.future_dated = skipped.future_dated.saturating_add(1); + continue; + } + // Saturating, so a beat inside the skew tolerance but still ahead of our clock reads as age + // zero rather than wrapping to a colossal age and being called stale. + let age_secs = policy.now_unix.saturating_sub(created_at); + if age_secs > policy.max_age_secs { + skipped.stale = skipped.stale.saturating_add(1); + continue; + } + if !parsed.accepting { + skipped.retracted = skipped.retracted.saturating_add(1); + continue; + } + sellers.push(row(key.pubkey, created_at, age_secs, parsed)); + } + sellers.sort_by(|left, right| left.pubkey.cmp(&right.pubkey)); + + SellerDirectory { + sellers, + // The TRANSPORT's word, never this function's guess. A reducer handed a partial stream + // sees perfectly well-formed rows and has no way to know the relay stopped early. + read_confirmed: completion.is_confirmed(), + skipped, + events_read, + } +} + +/// Project one resolved beat into a directory row. Reads the capability off the ALREADY-PARSED +/// [`ParsedHeartbeat`] rather than re-reading tags, so this shares the one reader +/// ([`crate::heartbeat::SeatCapability::from_tags`]) with the claim path and cannot spell a field +/// differently from it. +fn row( + pubkey: String, + announced_at: u64, + age_secs: u64, + parsed: ParsedHeartbeat, +) -> DiscoveredSeller { + let (admits_pool, admits_targeted) = match parsed.admission { + Some(admission) => ( + if admission.pool { + crate::home::ADMISSION_OPEN.to_owned() + } else { + crate::home::ADMISSION_CLOSED.to_owned() + }, + admission.targeted.as_str().to_owned(), + ), + // Unstated on BOTH halves, together: a seat that predates the tags published neither, and + // guessing one of them would be inventing a policy the seat never advertised. + None => (ADMISSION_UNSTATED.to_owned(), ADMISSION_UNSTATED.to_owned()), + }; + DiscoveredSeller { + pubkey, + specialty: parsed.capability.specialty, + announced_at, + age_secs, + rate_sats: parsed.rate_sats, + takes_no_payment: parsed.takes_no_payment, + accepted_mints: parsed.accepted_mints, + agents: parsed.agents, + harness_families: parsed.capability.harness_families, + admits_pool, + admits_targeted, + } +} + +/// Why a directory read could not be performed at all. +/// +/// ⚠ **A RELAY FAILURE IS NOT AN EMPTY MARKET, AND THAT IS THIS TYPE'S ONLY JOB.** Collapsing the +/// two would tell a buyer "no specialists are advertising" whenever its own network is down — the +/// single most misleading answer discovery can give, because it looks exactly like a true one. The +/// three outcomes are: `Err(_)` (the read failed), `Ok` with `read_confirmed == false` (the relay +/// did not answer in time), and `Ok` with `read_confirmed == true` (whatever `sellers` holds, +/// including nothing, is what the market has). +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum DiscoveryError { + /// The relay could not be added, reached, or served the read. + Relay(String), + /// The home has no usable identity to read with. + Identity(String), + /// Called from inside a Tokio runtime through the sync entry point. + Runtime(String), +} + +impl std::fmt::Display for DiscoveryError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Relay(detail) => write!(f, "relay: {detail}"), + Self::Identity(detail) => write!(f, "identity: {detail}"), + Self::Runtime(detail) => write!(f, "runtime: {detail}"), + } + } +} + +impl std::error::Error for DiscoveryError {} + +#[cfg(all(feature = "wallet", feature = "gateway"))] +pub use transport::{fetch_directory, fetch_directory_async}; + +/// The relay leg. Gated with `job_lifecycle`/`profile` because it needs the buyer identity and the +/// relay client; every RULE lives in [`reduce_directory`], which is ungated. +#[cfg(all(feature = "wallet", feature = "gateway"))] +mod transport { + use super::{ + AnnouncedSeat, DEFAULT_DIRECTORY_LIMIT, DirectoryPolicy, DiscoveryError, ReadCompletion, + SellerDirectory, reduce_directory, + }; + use crate::gateway::{EventDraft, TagSpec}; + use crate::home::MaxplayerHome; + use std::time::{Duration, Instant}; + + /// Our own REQ's subscription id. The completion evidence is scoped to THIS id: an `EOSE` for + /// anything else — a liveness probe, the buyer's job subscription, another read sharing the + /// socket — says nothing about whether the directory request finished. + const DIRECTORY_SUB_ID: &str = "maxplayer-discovery-directory"; + + /// Ceiling on the connect leg alone. `connect()` only SPAWNS the connection, so a fetch racing + /// the handshake burns its whole window and comes back empty — which this module would then + /// have to report as an unconfirmed read of an apparently dead market. It is a CEILING, not a + /// floor: the wait is whatever is left of the caller's budget, capped here. + const RELAY_CONNECT_WAIT: Duration = Duration::from_secs(20); + + /// What is left of the caller's budget. Zero once it is spent — never a negative wrap. + fn remaining(deadline: Instant) -> Duration { + deadline.saturating_duration_since(Instant::now()) + } + + /// Read the live seat directory off the home's relay. **Read-only**: it subscribes and reads, + /// and publishes no event of any kind. + /// + /// Completion is read off OUR OWN REQ and nothing else. The subscription is opened with a known + /// id, the notification receiver is established BEFORE the REQ goes out (an `EOSE` that lands + /// first would otherwise be missed), and only that id's `EOSE` sets + /// [`ReadCompletion::ConfirmedByEose`]. A deadline that expires, a socket that drops, or a + /// stream that ends leaves the read UNCONFIRMED however many rows arrived — a partial answer is + /// still returned, honestly flagged, because rows already in hand are useful and pretending they + /// are the whole market is not. + /// + /// This is deliberately NOT `fetch_events`. That helper ends on either an `EOSE` or a spent + /// timeout and returns the same `Ok(events)` for both (`job_lifecycle.rs:202` documents the + /// same trap), so no caller of it can honestly certify completion. Nor can a preceding liveness + /// probe stand in: a probe is a DIFFERENT subscription with a different filter, and its `EOSE` + /// is evidence about the probe. + /// + /// A relay that REJECTS our NIP-42 authentication is an error, not silence. That rejection + /// arrives on this relay's own notification channel and is never forwarded to the pool's + /// (`relay/inner.rs:417-419`), and a refused relay owes the subscription neither an `EOSE` nor + /// a `CLOSED` — so a reader watching only the pool waits out its deadline and reports an + /// unanswered read, throwing away a rejection that has a reason and a fix. + /// + /// The REQ goes out through the SINGLE RELAY, whose result is `Result<(), Error>`. The + /// pool-level call returns `Result>` and folds per-relay send failures into + /// `output.failed`, returning `Ok(output)` even when nothing succeeded (`pool/mod.rs:955-973`), + /// so its outer `Err` alone cannot tell a sent REQ from one that reached nobody. + /// + /// `budget` bounds the read loop rather than each leg, because a caller (an MCP tool with a + /// client read-timeout) can only honour a promise about the total. Running out mid-read yields + /// an UNCONFIRMED directory — an unanswered read, which is what it is — never an empty market + /// and never an error. Connect and cleanup are bounded separately and are not inside that one + /// timeout, so this is not a strict wall-clock cap on the whole call. + pub async fn fetch_directory_async( + home: &MaxplayerHome, + policy: DirectoryPolicy, + limit: usize, + budget: Duration, + ) -> Result { + use nostr_sdk::RelayMessage; + use nostr_sdk::pool::relay::RelayNotification; + use nostr_sdk::prelude::{Client, Filter, Kind, SubscribeOptions, SubscriptionId}; + + let deadline = Instant::now() + budget; + let secret = crate::home::read_secret_key_hex(home) + .map_err(|error| DiscoveryError::Identity(error.to_string()))?; + let keys = nostr_sdk::Keys::parse(&secret) + .map_err(|error| DiscoveryError::Identity(format!("key parse: {error}")))?; + + let client = Client::new(keys.clone()); + // Same discipline as every other read on this relay: auto-auth on, and WAIT for the socket. + client.automatic_authentication(true); + client + .add_relay(&home.config.relay_url) + .await + .map_err(|error| DiscoveryError::Relay(format!("add relay: {error}")))?; + let relay = client + .relay(&home.config.relay_url) + .await + .map_err(|error| DiscoveryError::Relay(format!("relay handle: {error}")))?; + + // THIS RELAY'S OWN notifications, and opened BEFORE `connect()` — before any authentication + // activity exists to observe. Two reasons, both load-bearing: + // + // 1. `AuthenticationFailed` is emitted on the RELAY's channel and is NOT forwarded to the + // pool's (`relay/inner.rs:417-419`). A reader watching only pool notifications sees a + // relay that rejected its AUTH as a relay that simply never answered: the rejection has a + // reason and a fix, and reporting it as an unanswered read throws both away. + // 2. The challenge and the negative OK can both land during `connect()`. A receiver opened + // afterwards would miss them, which is the same ordering trap the EOSE receiver avoids. + let mut notifications = relay.notifications(); + + client.connect().await; + relay + .wait_for_connection(remaining(deadline).min(RELAY_CONNECT_WAIT)) + .await; + + // Scoped to the seat address: the kind, the `#t=maxplayer` namespace guard so a foreign + // event squatting the kind is never delivered, and the `d` identifier so only seat + // announcements match. + let filter = Filter::new() + .kind(Kind::Custom(crate::heartbeat::SELLER_HEARTBEAT_KIND)) + .hashtag(crate::gateway::MAXPLAYER_TAG) + .identifier(crate::heartbeat::SELLER_HEARTBEAT_D) + .limit(if limit == 0 { + DEFAULT_DIRECTORY_LIMIT + } else { + limit + }); + + let sub_id = SubscriptionId::new(DIRECTORY_SUB_ID); + + // Subscribed through the SINGLE RELAY, whose result is `Result<(), Error>` — the failure of + // THIS relay's REQ, propagated. `Client::subscribe_with_id` returns `Result>`, + // and the pool collects per-relay failures into `output.failed` and returns `Ok(output)` + // even when NOTHING succeeded (`pool/mod.rs:955-973`); checking only the outer `Err` there + // reads a REQ that reached nobody as a REQ that was sent. One relay is the whole market + // here, so its disposition is the read's disposition. + if let Err(error) = relay + .subscribe_with_id(sub_id.clone(), filter, SubscribeOptions::default()) + .await + { + client.disconnect().await; + return Err(DiscoveryError::Relay(format!( + "subscribe seat directory: {error}" + ))); + } + + let mut events: Vec = Vec::new(); + let mut refusal: Option = None; + // A relay that rejected our AUTH. Distinct from `refusal` because the two are different + // facts with different fixes — "this relay will not serve you" versus "this relay will not + // serve this subscription" — and a reader that collapses them cannot tell an identity + // problem from a policy one. + let mut auth_failed = false; + // Pessimistic until this subscription's own EOSE says otherwise. Every early exit below + // leaves it as it is, so "we fell out of the loop somehow" can only ever mean unconfirmed. + let mut completion = ReadCompletion::Unconfirmed; + + let _ = tokio::time::timeout(remaining(deadline), async { + loop { + match notifications.recv().await { + Ok(RelayNotification::Event { + subscription_id, + event, + }) if subscription_id == sub_id => events.push((*event).clone()), + Ok(RelayNotification::Message { + message: RelayMessage::EndOfStoredEvents(id), + }) if *id == sub_id => { + completion = ReadCompletion::ConfirmedByEose; + return; + } + // A CLOSED naming our subscription is the relay REFUSING this read — a policy + // rejection with a reason. Reporting it as "no sellers" would be the worst + // possible lie about it. + Ok(RelayNotification::Message { + message: + RelayMessage::Closed { + subscription_id, + message, + }, + }) if *subscription_id == sub_id => { + refusal = Some(message.to_string()); + return; + } + // The relay rejected the AUTH we signed for it. Nothing else is coming: a relay + // that refuses the identity need send neither EOSE nor CLOSED, and waiting out + // the deadline would convert an explicit, actionable rejection into an + // unanswered read — the exact downgrade this arm exists to stop. + Ok(RelayNotification::AuthenticationFailed) => { + auth_failed = true; + return; + } + Ok(RelayNotification::Shutdown) => return, + Ok(_) => continue, + // The notification stream ending is a lost socket, never a finished read. + Err(_) => return, + } + } + }) + .await; + + // Cleanup on EVERY path, including the timeout: drop our REQ before the socket goes, so a + // relay is not left streaming into a subscription nobody is reading. + client.unsubscribe(&sub_id).await; + client.disconnect().await; + + if auth_failed { + return Err(DiscoveryError::Relay(format!( + "relay {} rejected our authentication; the seat directory was not read", + home.config.relay_url + ))); + } + + if let Some(reason) = refusal { + return Err(DiscoveryError::Relay(format!( + "relay refused the seat-directory subscription: {reason}" + ))); + } + + let announcements = events.into_iter().map(|event| AnnouncedSeat { + author_pubkey: event.pubkey.to_hex().to_ascii_lowercase(), + created_at: event.created_at.as_secs(), + event_id: event.id.to_hex().to_ascii_lowercase(), + event: EventDraft::new( + u16::try_from(event.kind.as_u16()).unwrap_or(event.kind.as_u16()), + event + .tags + .iter() + .map(|tag| TagSpec(tag.clone().to_vec())) + .collect(), + event.content.clone(), + ), + }); + Ok(reduce_directory(announcements, policy, completion)) + } + + /// Sync entry point for callers not already on a runtime. `budget` bounds the whole read, as in + /// [`fetch_directory_async`]. + pub fn fetch_directory( + home: &MaxplayerHome, + policy: DirectoryPolicy, + limit: usize, + budget: Duration, + ) -> Result { + crate::runtime_guard::refuse_nested_block_on("discovery::fetch_directory") + .map_err(DiscoveryError::Runtime)?; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| DiscoveryError::Runtime(error.to_string()))?; + runtime.block_on(fetch_directory_async(home, policy, limit, budget)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::heartbeat::{ + HeartbeatDraft, SeatCapability, heartbeat_for_state, retraction_for_state, + }; + use crate::home::{AdmissionPolicy, TargetedAdmission}; + + const NOW: u64 = 1_800_000_000; + const MINT: &str = "https://mint.example/Bitcoin"; + const SEAT_A: &str = "aa11111111111111111111111111111111111111111111111111111111111111"; + const SEAT_B: &str = "bb22222222222222222222222222222222222222222222222222222222222222"; + + fn open_policy() -> AdmissionPolicy { + AdmissionPolicy { + pool: true, + targeted: TargetedAdmission::Open, + } + } + + /// A beat built through the PRODUCTION emitter, so a discovery test can never pass against a + /// tag set hand-written to agree with the reader. `specialty` rides the same + /// `SeatCapability` a real seat's roster read fills. + fn beat(specialty: Option<&str>, admission: Option) -> HeartbeatDraft { + let capability = SeatCapability { + harness_families: vec!["claude-code".to_owned()], + specialty: specialty.map(str::to_owned), + ..SeatCapability::default() + }; + match admission { + Some(admission) => heartbeat_for_state( + 0, + true, + 12, + false, + vec![MINT.to_owned()], + vec!["claude".to_owned()], + capability, + admission, + ), + // The pre-§4.2 shape: a seat that states no admission policy at all. + None => HeartbeatDraft::new(true, 0, 12, vec![MINT.to_owned()]) + .with_agents(vec!["claude".to_owned()]) + .with_capability(capability), + } + } + + /// The reducer under a COMPLETED read. Completion is a transport fact, and these tests are + /// about the rules; the confirmed/unconfirmed distinction has its own tests below and a + /// behavioral one against a scripted relay in `tests/discovery_relay_behavior.rs`. + fn reduced( + announcements: impl IntoIterator, + policy: DirectoryPolicy, + ) -> SellerDirectory { + reduce_directory(announcements, policy, ReadCompletion::ConfirmedByEose) + } + + /// A distinct id per (pubkey, created_at), so the ordinary tests carry signed-shaped ids + /// without caring what they are. Tie-break tests state their ids explicitly instead. + fn announced(pubkey: &str, created_at: u64, draft: &HeartbeatDraft) -> AnnouncedSeat { + let id = format!("{:0>64}", format!("{}{created_at}", &pubkey[..4])); + announced_with_id(pubkey, created_at, &id, draft) + } + + fn announced_with_id( + pubkey: &str, + created_at: u64, + event_id: &str, + draft: &HeartbeatDraft, + ) -> AnnouncedSeat { + AnnouncedSeat { + event_id: event_id.to_owned(), + ..announced_inner(pubkey, created_at, draft) + } + } + + fn announced_inner(pubkey: &str, created_at: u64, draft: &HeartbeatDraft) -> AnnouncedSeat { + AnnouncedSeat { + author_pubkey: pubkey.to_owned(), + created_at, + event_id: String::new(), + event: draft.to_event_draft(), + } + } + + #[test] + fn an_equal_created_at_tie_resolves_to_the_lowest_id_in_either_input_order() { + // F2. Two SIGNED beats for one address, same timestamp, opposite `accepting`: one says the + // seat is live, the other retracts it. NIP-01 retains the lowest id, so the retraction here + // (id "11…") must win both times. Before the id crossed the transport seam this resolved by + // whichever event the relay happened to hand over last — a seat live or dead by luck. + let live = beat(Some("Rust"), Some(open_policy())); + let terminal = retraction_for_state( + 0, + 12, + false, + vec![MINT.to_owned()], + vec!["claude".to_owned()], + SeatCapability { + harness_families: vec!["claude-code".to_owned()], + specialty: Some("Rust".to_owned()), + ..SeatCapability::default() + }, + open_policy(), + ); + let low = format!("{:1>64}", ""); + let high = format!("{:f>64}", ""); + + for (first, second) in [(&live, &terminal), (&terminal, &live)] { + let first_id = if std::ptr::eq(first, &live) { + &high + } else { + &low + }; + let second_id = if std::ptr::eq(second, &live) { + &high + } else { + &low + }; + let directory = reduced( + [ + announced_with_id(SEAT_A, NOW - 30, first_id, first), + announced_with_id(SEAT_A, NOW - 30, second_id, second), + ], + DirectoryPolicy::at(NOW), + ); + assert!( + directory.sellers.is_empty(), + "the lowest-id event is the retraction, so the seat must be retracted whichever \ + order it arrives in: {directory:?}" + ); + assert_eq!(directory.skipped.retracted, 1); + } + + // And the mirror image: when the LIVE beat holds the lowest id, the seat stays listed in + // both orders. A tie-break that always dropped the seat would pass the half above. + for (first, second) in [(&live, &terminal), (&terminal, &live)] { + let first_id = if std::ptr::eq(first, &live) { + &low + } else { + &high + }; + let second_id = if std::ptr::eq(second, &live) { + &low + } else { + &high + }; + let directory = reduced( + [ + announced_with_id(SEAT_A, NOW - 30, first_id, first), + announced_with_id(SEAT_A, NOW - 30, second_id, second), + ], + DirectoryPolicy::at(NOW), + ); + assert_eq!( + directory.sellers.len(), + 1, + "the lowest-id event is the live beat, so the seat must be listed whichever order \ + it arrives in: {directory:?}" + ); + assert_eq!(directory.skipped.retracted, 0); + } + } + + #[test] + fn a_newer_timestamp_still_outranks_a_lower_id() { + // The tie-break must be SECONDARY. An id-first order would let a stale low-id beat outrank + // the seat's newer word about itself. + let live = beat(Some("Rust"), Some(open_policy())); + let terminal = retraction_for_state( + 0, + 12, + false, + vec![MINT.to_owned()], + vec!["claude".to_owned()], + SeatCapability { + harness_families: vec!["claude-code".to_owned()], + specialty: Some("Rust".to_owned()), + ..SeatCapability::default() + }, + open_policy(), + ); + let low = format!("{:1>64}", ""); + let high = format!("{:f>64}", ""); + + // Older retraction with the LOW id; newer live beat with the HIGH id. Newer wins. + let directory = reduced( + [ + announced_with_id(SEAT_A, NOW - 300, &low, &terminal), + announced_with_id(SEAT_A, NOW - 30, &high, &live), + ], + DirectoryPolicy::at(NOW), + ); + assert_eq!(directory.sellers.len(), 1, "{directory:?}"); + + // And the reverse: newer retraction with the high id beats an older live beat with the low. + let directory = reduced( + [ + announced_with_id(SEAT_A, NOW - 300, &low, &live), + announced_with_id(SEAT_A, NOW - 30, &high, &terminal), + ], + DirectoryPolicy::at(NOW), + ); + assert!(directory.sellers.is_empty(), "{directory:?}"); + } + + #[test] + fn an_unconfirmed_read_never_reports_itself_as_confirmed_however_many_rows_it_holds() { + // F1, at the reducer seam. A partial stream carries perfectly well-formed rows; the reducer + // cannot tell it from a completed one, so completion is passed IN and never inferred. + let live = beat(Some("Rust"), Some(open_policy())); + + let partial = reduce_directory( + [announced(SEAT_A, NOW - 30, &live)], + DirectoryPolicy::at(NOW), + ReadCompletion::Unconfirmed, + ); + assert_eq!(partial.sellers.len(), 1, "partial rows are KEPT"); + assert!( + !partial.read_confirmed, + "rows in hand are not evidence the relay finished answering" + ); + + // The nastiest shape of the same bug: one stale beat arrives, the stream dies, and the + // filtered-out row leaves an EMPTY seller list. Confirmed here would read as "the market is + // empty" on the strength of an answer that never came. + let stale_partial = reduce_directory( + [announced(SEAT_A, NOW - DEFAULT_MAX_AGE_SECS - 1, &live)], + DirectoryPolicy::at(NOW), + ReadCompletion::Unconfirmed, + ); + assert!(stale_partial.sellers.is_empty()); + assert!( + !stale_partial.read_confirmed, + "an empty list from an unfinished read is not an empty market" + ); + assert_eq!( + stale_partial.events_read, 1, + "and it still says what it saw" + ); + + assert!(ReadCompletion::ConfirmedByEose.is_confirmed()); + assert!(!ReadCompletion::Unconfirmed.is_confirmed()); + } + + #[test] + fn a_declared_specialty_reaches_the_discovery_output_with_its_pubkey() { + // The chain scope 1 started, finished: config -> beat -> tag -> parse -> a row a buyer can + // act on. `pubkey` is the load-bearing field — it is what the targeted post takes. + let directory = reduced( + [announced( + SEAT_A, + NOW - 30, + &beat( + Some("Rust async runtimes and tokio internals"), + Some(open_policy()), + ), + )], + DirectoryPolicy::at(NOW), + ); + assert_eq!(directory.sellers.len(), 1, "{directory:?}"); + let seat = &directory.sellers[0]; + assert_eq!(seat.pubkey, SEAT_A); + assert_eq!( + seat.specialty.as_deref(), + Some("Rust async runtimes and tokio internals") + ); + assert_eq!(seat.announced_at, NOW - 30); + assert_eq!(seat.age_secs, 30); + assert_eq!(seat.rate_sats, 12); + assert_eq!(seat.accepted_mints, vec![MINT]); + assert_eq!(seat.agents, vec!["claude"]); + assert_eq!(seat.harness_families, vec!["claude-code"]); + assert_eq!(seat.admits_pool, crate::home::ADMISSION_OPEN); + assert_eq!(seat.admits_targeted, crate::home::ADMISSION_OPEN); + assert!(directory.read_confirmed); + assert_eq!(directory.events_read, 1); + assert_eq!(directory.skipped, DirectorySkips::default()); + } + + #[test] + fn a_discovered_pubkey_is_accepted_by_the_unchanged_targeted_post_path() { + // The handoff, end to end and OFFLINE: the pubkey a discovery row carries goes into the + // EXISTING targeted-post parameter and comes back out of the parsed offer as the seat the + // offer addresses. No relay, no post, no payment — only the two ends of the flow the order + // names, joined by nothing but a 64-hex string. + // + // This test exists to catch a whole class of quiet breakage: a row field that renders fine + // and is not a valid target (padded, truncated, npub-encoded, upper-cased). Discovery's + // whole purpose is to end at a value `post_job` accepts, so the value is asserted through + // the real `OfferDraft` -> `to_event_draft` -> `parse_offer` path rather than eyeballed. + use crate::gateway::{OfferDraft, assert_seller_matches, is_targeted, parse_offer}; + + let directory = reduced( + [announced( + SEAT_A, + NOW - 30, + &beat(Some("Rust async runtimes"), Some(open_policy())), + )], + DirectoryPolicy::at(NOW), + ); + let discovered = directory.sellers[0].pubkey.clone(); + + let draft = OfferDraft::new( + "port a crate to tokio", + "text/plain", + 1, + NOW + 600, + &discovered, + ) + .to_event_draft(); + let offer = + parse_offer(&draft).expect("an offer targeted at a discovered pubkey must parse"); + + assert!( + is_targeted(&offer), + "a discovered pubkey must produce a TARGETED offer, not an open-pool one" + ); + assert!(offer.seller_matches(&discovered)); + assert_seller_matches(&offer, &discovered) + .expect("the discovered seat must be the seat the offer addresses"); + assert!( + !offer.seller_matches(SEAT_B), + "targeting one discovered seat must not address another" + ); + + // And the row's own specialty is nowhere on the offer. Discovery informed the CHOICE; it + // did not become a term of the deal. + assert!( + !draft + .tags + .iter() + .any(|tag| tag.0.iter().any(|value| value.contains("Rust async"))), + "specialty text must not ride the offer: {:?}", + draft.tags + ); + } + + #[test] + fn a_seat_with_no_specialty_is_still_discoverable() { + // The migration property the order names: old sellers without a description must not vanish + // from the directory. Unstated is a missing FIELD, never a missing SEAT. + let directory = reduced( + [announced( + SEAT_A, + NOW - 10, + &beat(None, Some(open_policy())), + )], + DirectoryPolicy::at(NOW), + ); + assert_eq!(directory.sellers.len(), 1); + assert_eq!(directory.sellers[0].specialty, None); + assert_eq!(directory.sellers[0].pubkey, SEAT_A); + } + + #[test] + fn a_legacy_beat_that_states_no_admission_reads_as_unstated_never_closed() { + // Rendering unstated as `closed` would tell a buyer that every seat older than the §4.2 + // tags refuses it. The seat did not say; the directory must not say either. + let directory = reduced( + [announced(SEAT_A, NOW - 10, &beat(Some("Rust"), None))], + DirectoryPolicy::at(NOW), + ); + assert_eq!(directory.sellers.len(), 1, "a legacy seat stays listed"); + assert_eq!(directory.sellers[0].admits_pool, ADMISSION_UNSTATED); + assert_eq!(directory.sellers[0].admits_targeted, ADMISSION_UNSTATED); + assert_ne!( + directory.sellers[0].admits_pool, + crate::home::ADMISSION_CLOSED + ); + } + + #[test] + fn a_newer_retraction_removes_the_seat_even_though_its_older_beat_was_live() { + // The ordering rule that makes retraction work. The address is resolved FIRST, then the + // winner is judged: resolve-after-filter would drop the `accepting=n` beat on its own + // merits and leave the seat's older `accepting=y` standing as the survivor — the seat would + // stay advertised by the very event that retracted it. + let live = beat(Some("Rust"), Some(open_policy())); + let terminal = retraction_for_state( + 0, + 12, + false, + vec![MINT.to_owned()], + vec!["claude".to_owned()], + SeatCapability { + harness_families: vec!["claude-code".to_owned()], + specialty: Some("Rust".to_owned()), + ..SeatCapability::default() + }, + open_policy(), + ); + let directory = reduced( + [ + announced(SEAT_A, NOW - 600, &live), + announced(SEAT_A, NOW - 60, &terminal), + ], + DirectoryPolicy::at(NOW), + ); + assert!( + directory.sellers.is_empty(), + "a retracted seat must not appear as a live seller: {directory:?}" + ); + assert_eq!(directory.skipped.retracted, 1); + assert_eq!(directory.events_read, 2); + + // AND THE OTHER DIRECTION, or the assertion above is satisfied by any rule that drops + // `accepting=n`: an OLDER retraction must NOT bury a newer live beat. + let recovered = reduced( + [ + announced(SEAT_A, NOW - 600, &terminal), + announced(SEAT_A, NOW - 60, &live), + ], + DirectoryPolicy::at(NOW), + ); + assert_eq!( + recovered.sellers.len(), + 1, + "a seat that came back is live again: {recovered:?}" + ); + assert_eq!(recovered.skipped.retracted, 0); + } + + #[test] + fn a_stale_or_future_dated_beat_is_not_a_live_seller() { + let live = beat(Some("Rust"), Some(open_policy())); + let stale = reduced( + [announced(SEAT_A, NOW - DEFAULT_MAX_AGE_SECS - 1, &live)], + DirectoryPolicy::at(NOW), + ); + assert!(stale.sellers.is_empty(), "{stale:?}"); + assert_eq!(stale.skipped.stale, 1); + + // Exactly AT the window is still live — the bound is inclusive, so a seat is not dropped by + // one second of arithmetic it cannot observe. + let edge = reduced( + [announced(SEAT_A, NOW - DEFAULT_MAX_AGE_SECS, &live)], + DirectoryPolicy::at(NOW), + ); + assert_eq!(edge.sellers.len(), 1, "{edge:?}"); + + let future = reduced( + [announced( + SEAT_A, + NOW + DEFAULT_MAX_CLOCK_SKEW_SECS + 1, + &live, + )], + DirectoryPolicy::at(NOW), + ); + assert!(future.sellers.is_empty(), "{future:?}"); + assert_eq!(future.skipped.future_dated, 1); + + // Inside the skew tolerance a seat stays visible, at age zero rather than a wrapped age. + let skewed = reduced( + [announced(SEAT_A, NOW + 10, &live)], + DirectoryPolicy::at(NOW), + ); + assert_eq!(skewed.sellers.len(), 1, "{skewed:?}"); + assert_eq!(skewed.sellers[0].age_secs, 0); + } + + #[test] + fn an_unparseable_event_is_counted_and_never_becomes_a_row() { + // A junk event squatting the kind must not become a seat a buyer might target. The count is + // what lets an empty directory be EXPLAINED rather than merely reported. + let junk = AnnouncedSeat { + author_pubkey: SEAT_B.to_owned(), + created_at: NOW - 10, + event_id: format!("{:0>64}", "junk"), + event: EventDraft::new( + crate::heartbeat::SELLER_HEARTBEAT_KIND, + vec![crate::gateway::TagSpec::new(["d", "maxplayer-seller"])], + "", + ), + }; + let directory = reduced( + [ + announced(SEAT_A, NOW - 10, &beat(Some("Rust"), Some(open_policy()))), + junk, + ], + DirectoryPolicy::at(NOW), + ); + assert_eq!(directory.sellers.len(), 1); + assert_eq!(directory.sellers[0].pubkey, SEAT_A); + assert_eq!(directory.skipped.unparseable, 1); + assert_eq!(directory.events_read, 2); + } + + #[test] + fn the_latest_beat_per_address_wins_and_the_order_carries_no_ranking() { + // Addressable events supersede IN PLACE, so a seat's older beats must not survive alongside + // its newest — an id-keyed reduce would list the same seat twice at two rates. + let old = beat(Some("Rust, old text"), Some(open_policy())); + let new = beat(Some("Rust, current text"), Some(open_policy())); + let directory = reduced( + [ + announced(SEAT_B, NOW - 20, &new), + announced(SEAT_A, NOW - 300, &old), + announced(SEAT_A, NOW - 5, &new), + ], + DirectoryPolicy::at(NOW), + ); + assert_eq!( + directory.sellers.len(), + 2, + "one row per seat: {directory:?}" + ); + assert_eq!( + directory.sellers[0].specialty.as_deref(), + Some("Rust, current text"), + "the superseded text must not be what a buyer reads" + ); + // Pubkey order, NOT freshness order: SEAT_B beat more recently than SEAT_A's resolved beat + // would in a freshness sort, and it still comes second. Freshness-descending would be a + // ranking policy this slice deliberately does not ship. + assert_eq!( + directory + .sellers + .iter() + .map(|seat| seat.pubkey.as_str()) + .collect::>(), + vec![SEAT_A, SEAT_B] + ); + } + + #[test] + fn an_answered_empty_market_is_not_the_same_value_as_an_unanswered_read() { + // The distinction the order demands, asserted on the two constructors the transport returns. + // Both hold zero sellers; only one of them is a statement about the market. + let answered = SellerDirectory::empty_confirmed(); + let unanswered = SellerDirectory::unverified(); + assert!(answered.sellers.is_empty() && unanswered.sellers.is_empty()); + assert!( + answered.read_confirmed, + "an answered empty read is a fact about the market" + ); + assert!( + !unanswered.read_confirmed, + "an unanswered read is a fact about our patience, and must not read as an empty market" + ); + assert_ne!(answered, unanswered); + + // And a REDUCED directory is always a confirmed read: the reducer only ever runs on events + // the relay actually served, so the unconfirmed value cannot be produced by this path. + let reduced = reduced([], DirectoryPolicy::at(NOW)); + assert!(reduced.read_confirmed); + assert_eq!(reduced, answered); + } + + #[test] + fn discovery_never_writes() { + // A structural check, not a behavioural one: the discovery module must contain no publish, + // no award, no payment. Asserted against the SOURCE because the property is "this code + // cannot spend", and a runtime test can only show that one path did not. + let source = include_str!("discovery.rs"); + // Split the needles so this test's own text does not match them. + for forbidden in [ + concat!("send_", "event"), + concat!("send_", "event_to"), + concat!("publish_", "signed"), + concat!("EventBuilder", "::"), + concat!("award_", "claim"), + concat!("reserve_", "for_award"), + concat!("pay_", "invoice"), + ] { + assert!( + !source.contains(forbidden), + "discovery is read-only and must not reference `{forbidden}`" + ); + } + } +} diff --git a/crates/maxplayer-core/src/heartbeat.rs b/crates/maxplayer-core/src/heartbeat.rs index 332570c7c..b169ea6b1 100644 --- a/crates/maxplayer-core/src/heartbeat.rs +++ b/crates/maxplayer-core/src/heartbeat.rs @@ -297,6 +297,45 @@ pub const ADMITS_TARGETED_TAG: &str = "admits_targeted"; /// allowed to be arbitrary text at all, and a test names the filter surface to keep it true. pub const HARDWARE_TAG: &str = "hardware"; +/// Wire tag carrying the seat's own description of what it SPECIALISES in — free text, single +/// value, e.g. `"Rust async runtimes and tokio internals"`. This is what lets a buyer who has +/// never met a seat read what it says it is for. +/// +/// ⛔ **SELLER-DECLARED, NOT VERIFIED — AND THEREFORE NEVER FILTERED.** It joins +/// [`HARDWARE_TAG`] and [`HARNESS_VARIANT_TAG`] in the display-only half of [`SeatCapability`], +/// which is not a convenience: the provenance rule there is *filterable ⟺ machine-sourced*, and +/// nothing the daemon can run measures whether a seat is good at Rust. A buyer commits sats at +/// award, so a field the operator typed must not be able to gate that award — which is why this +/// rides the kind-30340 beat ALONE and is absent from [`SeatCapability::filterable_tags`], hence +/// absent from every kind-3402 claim. +/// +/// A buyer READS this and decides for itself. Discovery hands the text to a human or an agent; +/// the choice of who to hire stays a buyer act, and it targets the discovered pubkey through the +/// posting path that already existed. There is deliberately no string-match gate anywhere: a +/// substring test against operator-typed text would be a competence assertion the protocol +/// cannot back, and it would reward keyword-stuffing over the one signal that costs something +/// (a seat that claims, delivers, and gets paid). +/// +/// Absent means UNSTATED, never "generalist" — a seat that predates this tag, and a seat whose +/// operator declined to describe it, are the same state on the wire and both stay discoverable. +/// +/// Bounded at [`SPECIALTY_MAX_BYTES`]; see that constant for what happens to a longer value. +pub const SPECIALTY_TAG: &str = "specialty"; + +/// Upper bound (bytes, UTF-8) on a [`SPECIALTY_TAG`] value. One bounded text field, not a token +/// taxonomy: a controlled vocabulary would need a registry, a versioning story, and an answer for +/// every specialty nobody thought to enumerate, and it would still be operator-typed. +/// +/// **A longer value is TRUNCATED on a char boundary, never refused.** Both directions of that +/// choice matter. Refusing at config would brick a seat's boot over cosmetic text; refusing at +/// parse would drop the whole beat and make an over-talkative seat invisible — a seat that is +/// perfectly able to work would vanish from the market for a description that is too long. So the +/// FIELD degrades and the SEAT survives, and the same bound is applied by the emitter and the +/// reader alike (see [`bounded_specialty`]) so a reader can never render more than a seat could +/// have published. Truncating on a char boundary rather than a byte one is what keeps the result +/// valid UTF-8 for a multi-byte description. +pub const SPECIALTY_MAX_BYTES: usize = 1024; + /// The capability a seat advertises (#784), as ONE object rather than five loose fields. /// /// It exists to make the split structural instead of remembered. #784 has two kinds of field: @@ -305,9 +344,9 @@ pub const HARDWARE_TAG: &str = "hardware"; /// reads these off the CLAIM, so they appear on the kind-3402 claim as well as the kind-30340 /// beat, and must be spelled identically on both. [`Self::filterable_tags`] is that single /// spelling. -/// - **Display-only** — `harness_variant`, `hardware`. Colour for a human or a seat directory. They -/// go on the beat alone, because the award decision never reads them, and putting them on every -/// claim would be weight with no reader. +/// - **Display-only** — `harness_variant`, `hardware`, `specialty`. Colour for a human or a seat +/// directory. They go on the beat alone, because the award decision never reads them, and putting +/// them on every claim would be weight with no reader. /// /// ## The line between them is PROVENANCE /// @@ -315,7 +354,7 @@ pub const HARDWARE_TAG: &str = "hardware"; /// buyer commits sats at award and an operator-typed claim has nothing to contradict it. Each /// filterable field earns its place by being measured: `harness_family` from the dispatchable /// roster, `harness_model` from the harness handshake, `capabilities` from a probe of the job -/// execution environment. The display-only two are operator-declared, which is exactly why they are +/// execution environment. The display-only ones are operator-declared, which is exactly why they are /// harmless — nothing pays out on them. /// /// Enum-binding is NOT what buys this. Enum-binding solves canonicalisation — that `rust` and `Rust` @@ -344,6 +383,10 @@ pub struct SeatCapability { pub harness_variant: Option, /// Free-text machine colour. Never filtered — see [`HARDWARE_TAG`]. pub hardware: Option, + /// The seat's own description of what it specialises in, bounded at + /// [`SPECIALTY_MAX_BYTES`]. Seller-declared and never verified, therefore never filtered — + /// see [`SPECIALTY_TAG`]. `None` ⇒ unstated, which is not a claim to be a generalist. + pub specialty: Option, } impl SeatCapability { @@ -419,7 +462,7 @@ impl SeatCapability { /// prevent, and the read side is worth no less. Every consumer — the seat directory, the buyer's /// claim parse, the award filter — goes through here. /// - /// Reading all five off ANY event is deliberate, including a claim, which carries no display + /// Reading every field off ANY event is deliberate, including a claim, which carries no display /// fields: those simply come back `None`. Absent means unstated, so there is nothing to /// special-case per event kind, and no place for a per-kind rule to be applied inconsistently. pub fn from_tags(tags: &[TagSpec]) -> Self { @@ -429,6 +472,7 @@ impl SeatCapability { capabilities: capabilities_from_tags(tags), harness_variant: harness_variant_from_tags(tags), hardware: hardware_from_tags(tags), + specialty: specialty_from_tags(tags), } } @@ -451,6 +495,7 @@ impl SeatCapability { [ harness_variant_tag(self.harness_variant.as_deref()), hardware_tag(self.hardware.as_deref()), + specialty_tag(self.specialty.as_deref()), ] .into_iter() .flatten() @@ -756,6 +801,51 @@ pub fn hardware_from_tags(tags: &[TagSpec]) -> Option { first_tag_value(tags, HARDWARE_TAG).and_then(stated) } +/// One specialty value, normalized to the "stated or absent" contract AND bounded to +/// [`SPECIALTY_MAX_BYTES`]. `None` when nothing survives trimming. +/// +/// ⚠ **THE EMITTER AND THE READER BOTH CALL THIS, and that is the point.** A bound applied only at +/// emit would leave a reader rendering whatever arbitrary length a foreign seat published; a bound +/// applied only at read would let this build publish a value its own reader then silently shortens. +/// Applying it in one function on both sides means what a reader shows is always something a seat +/// could have published. +/// +/// Truncation walks BACK to a char boundary, so a description ending mid-codepoint loses that +/// codepoint rather than yielding invalid UTF-8. The result is therefore at most +/// [`SPECIALTY_MAX_BYTES`] bytes and may be shorter by up to three. +pub fn bounded_specialty(value: Option<&str>) -> Option { + let declared = stated(value?)?; + if declared.len() <= SPECIALTY_MAX_BYTES { + return Some(declared); + } + let mut end = SPECIALTY_MAX_BYTES; + while end > 0 && !declared.is_char_boundary(end) { + end -= 1; + } + // Trim again: the cut can expose trailing whitespace that was interior before it. + let truncated = declared[..end].trim_end(); + (!truncated.is_empty()).then(|| truncated.to_owned()) +} + +/// The `["specialty", text]` tag, or `None` for a seat that describes itself as nothing. +/// +/// Beat-only and never filtered — see [`SPECIALTY_TAG`]. Bounded by [`bounded_specialty`], so an +/// operator who pastes an essay into `[seat] specialty` publishes a valid beat carrying its +/// leading 1024 bytes rather than a beat a relay might refuse. +pub fn specialty_tag(specialty: Option<&str>) -> Option { + bounded_specialty(specialty).map(|value| TagSpec::new([SPECIALTY_TAG, &value])) +} + +/// Read the `["specialty", text]` value off a seat announcement's tags. Absent ⇒ `None`, which is +/// UNSTATED: a seat too old to carry the tag and a seat whose operator wrote no description are +/// the same state here, and both remain discoverable. +/// +/// Bounded on the way in by [`bounded_specialty`] — a foreign seat's oversized value is shortened +/// by the reader, never a reason to reject the beat (see [`SPECIALTY_MAX_BYTES`]). +pub fn specialty_from_tags(tags: &[TagSpec]) -> Option { + bounded_specialty(first_tag_value(tags, SPECIALTY_TAG)) +} + /// The `["admits_pool", …]` and `["admits_targeted", …]` tags for a stated admission policy. /// /// BEAT ONLY — never on a kind-3402 claim. A claim already proves admission (the seat claimed), so @@ -1513,7 +1603,7 @@ mod tests { /// The **PRESENT** row: the exact tag set of a beat that states EVERY #784 field. /// /// This is the one that goes red when a tag is added, renamed or dropped, because its input - /// states all five names. Its sibling above states none, so between them a tag cannot appear + /// states every name. Its sibling above states none, so between them a tag cannot appear /// without being declared here nor vanish without failing there. **One row alone cannot tell a /// working emitter from a fixture that populates nothing** — the same argument /// `capability::probe_capabilities`'s own `a_stock_image_with_no_toolchain_advertises_nothing` @@ -1547,14 +1637,16 @@ mod tests { "harness_variant", "queue_depth", "rate", + "specialty", "t", "v", ] ); // The denominator, stated rather than left to be counted off the list above: 8 pre-#784 tags - // (7 plus `agents`) and 6 capability tags, because `full_capability` carries two models. - assert_eq!(event.tags.len(), 14, "14 tags, 13 distinct names: {:?}", tag_names(&event)); + // (7 plus `agents`) and 7 capability tags — 6 from #784 (`full_capability` carries two + // models) plus the display-only `specialty`. + assert_eq!(event.tags.len(), 15, "15 tags, 14 distinct names: {:?}", tag_names(&event)); assert_eq!( tag_names(&event).iter().filter(|name| **name == HARNESS_MODEL_TAG).count(), 2, @@ -2108,6 +2200,7 @@ mod tests { capabilities: vec!["rust".to_owned(), "node".to_owned()], harness_variant: Some("my-fork".to_owned()), hardware: Some("mac studio, 64GB".to_owned()), + specialty: Some("Rust async runtimes and tokio internals".to_owned()), } } @@ -2376,7 +2469,7 @@ mod tests { let written = full_capability(); let mut tags = written.filterable_tags(); tags.extend(written.display_tags()); - assert_eq!(tags.len(), 6, "denominator: 4 filterable + 2 display: {tags:?}"); + assert_eq!(tags.len(), 7, "denominator: 4 filterable + 3 display: {tags:?}"); assert_eq!(SeatCapability::from_tags(&tags), written); } @@ -2463,6 +2556,115 @@ mod tests { "hardware is display colour; the award decision never reads it, so it must not ride every claim" ); assert_eq!(harness_variant_from_tags(&claim.tags), None); + assert_eq!( + specialty_from_tags(&beat.tags).as_deref(), + Some("Rust async runtimes and tokio internals") + ); + assert_eq!( + specialty_from_tags(&claim.tags), + None, + "a specialty is seller-declared and unverified, so it must not reach the event the \ + award filter reads — that is the whole reason it is allowed to be free text" + ); + } + + #[test] + fn a_specialty_is_unreachable_from_the_filterable_surface() { + // The structural guarantee behind "no string-match claim gate": there is no path from a + // declared specialty to anything a buyer's award predicate consults. Asserted against the + // ONE function that DEFINES the filterable surface, so it survives new fields. + let capability = SeatCapability { + specialty: Some("Rust async runtimes".to_owned()), + ..SeatCapability::default() + }; + assert!( + capability.filterable_tags().is_empty(), + "an operator-typed specialty must expose nothing filterable" + ); + assert_eq!( + capability.display_tags().len(), + 1, + "it must still ride the beat — otherwise no buyer can discover the seat by reading it" + ); + } + + #[test] + fn an_oversized_specialty_is_truncated_on_a_char_boundary_and_the_beat_survives() { + // A multi-byte description whose cut lands mid-codepoint. Truncation must walk BACK to a + // boundary: slicing on the byte index would panic, and refusing the value outright would + // take a working seat off the market over cosmetic text. + let essay = "é".repeat(SPECIALTY_MAX_BYTES); // 2 bytes each ⇒ 2× the bound + let capability = SeatCapability { + specialty: Some(essay), + ..SeatCapability::default() + }; + let beat = draft(true, 0, 5) + .with_capability(capability) + .to_event_draft(); + let parsed = + parse_heartbeat(&beat).expect("an over-long description must not lose the beat"); + let carried = parsed + .capability + .specialty + .expect("the description is shortened, not dropped"); + assert!( + carried.len() <= SPECIALTY_MAX_BYTES, + "published {} bytes, bound is {SPECIALTY_MAX_BYTES}", + carried.len() + ); + assert_eq!( + carried.len(), + SPECIALTY_MAX_BYTES, + "the cut must take the largest whole-codepoint prefix that fits, not a smaller one" + ); + assert!( + carried.chars().all(|c| c == 'é'), + "the result must stay valid UTF-8 é's" + ); + } + + #[test] + fn a_blank_or_absent_specialty_is_unstated_and_the_seat_stays_discoverable() { + // Three ways to say nothing, one state. The legacy case is the third: a beat published by a + // seat that predates the tag must parse, and must not read as a claim to be a generalist. + assert_eq!(bounded_specialty(None), None); + assert_eq!(bounded_specialty(Some(" \t ")), None); + let legacy = draft(true, 0, 5) + .with_agents(vec!["claude".to_owned()]) + .to_event_draft(); + assert!( + !legacy + .tags + .iter() + .any(|tag| tag.0.first().map(String::as_str) == Some(SPECIALTY_TAG)), + "a seat that states no specialty must emit no tag at all" + ); + let parsed = parse_heartbeat(&legacy).expect("a legacy announcement stays parseable"); + assert_eq!(parsed.capability.specialty, None); + assert!( + parsed.accepting && !parsed.accepted_mints.is_empty(), + "and it stays a usable, discoverable seat: {parsed:?}" + ); + } + + #[test] + fn a_specialty_read_off_a_foreign_beat_is_trimmed_and_bounded_by_the_reader() { + // The reader applies the same bound as the emitter, because a beat can be written by + // anything. Padding a value that no operator typed would render as present-and-blank. + let padded = [TagSpec::new([SPECIALTY_TAG, " Rust async runtimes "])]; + assert_eq!( + specialty_from_tags(&padded).as_deref(), + Some("Rust async runtimes") + ); + let oversized = [TagSpec::new([ + SPECIALTY_TAG, + &"x".repeat(SPECIALTY_MAX_BYTES + 500), + ])]; + assert_eq!( + specialty_from_tags(&oversized).map(|value| value.len()), + Some(SPECIALTY_MAX_BYTES), + "a reader must never render more than a seat could have published" + ); } #[test] diff --git a/crates/maxplayer-core/src/home.rs b/crates/maxplayer-core/src/home.rs index b88b19040..63a5c5bfd 100644 --- a/crates/maxplayer-core/src/home.rs +++ b/crates/maxplayer-core/src/home.rs @@ -1493,15 +1493,15 @@ pub struct MaxplayerConfig { /// `[seat]` — the operator-declared half of the seat's advertisement (#784). /// /// These are the DISPLAY-ONLY fields of [`crate::heartbeat::SeatCapability`], and they are declared -/// here precisely because they are the fields no probe can answer. A fork name and a machine -/// description are facts about the operator's intent and hardware; nothing the daemon can run -/// measures them. +/// here precisely because they are the fields no probe can answer. A fork name, a machine +/// description and a statement of what the seat is FOR are facts about the operator's intent and +/// hardware; nothing the daemon can run measures them. /// /// ## Why a config key is safe here and forbidden for `capabilities` /// /// The provenance rule is [`crate::heartbeat::SeatCapability`]'s: **filterable ⟺ machine-sourced**, /// because a buyer commits sats at award and an operator-typed claim has nothing to contradict it. -/// These two are never filtered — they ride the kind-30340 beat alone and no award decision reads +/// These are never filtered — they ride the kind-30340 beat alone and no award decision reads /// them — so an operator may state them freely. That is the same rule read from its other end, not /// an exception to it. Adding a FILTERABLE field to this struct would break it; see /// [`crate::seller_roster::Advertisement::capability`] for the seam that keeps the two halves apart. @@ -1522,6 +1522,28 @@ pub struct SeatConfig { /// [`crate::heartbeat::HARDWARE_TAG`] documents why that is acceptable: nothing filters on it. #[serde(default, skip_serializing_if = "Option::is_none")] pub hardware: Option, + /// Free-text description of what this seat SPECIALISES in — e.g. + /// `specialty = "Rust async runtimes, tokio internals, and tracing instrumentation"`. Absent ⇒ + /// the tag is omitted and the seat states nothing, which is NOT a claim to be a generalist. + /// + /// This is the field a buyer that has never met this seat reads to decide whether to target + /// it. It belongs in this struct and not in `[seller]` for the reason the other two do: + /// **EXPLICITLY UNVERIFIED.** No probe can measure whether an operator is good at Rust, so by + /// [`crate::heartbeat::SeatCapability`]'s provenance rule the field can be operator-typed only + /// because nothing filters on it — it rides the kind-30340 beat alone and no award decision + /// reads it. Adding a match gate on it later would move it to the filterable half and break + /// that rule; see [`crate::heartbeat::SPECIALTY_TAG`]. + /// + /// Bounded at [`crate::heartbeat::SPECIALTY_MAX_BYTES`] bytes of UTF-8. A longer value is + /// truncated at publish, never refused — the seat stays on the market and only the description + /// is shortened. + /// + /// ⛔ It is a DESCRIPTION the operator wrote, and nothing else. The daemon never fills it from + /// local memory, a client list, an episode log, or any file on the box: everything on a beat is + /// public to every relay reader, and a field that auto-published who a seat has worked for + /// would leak the operator's business as a side effect of being discoverable. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub specialty: Option, } impl SeatConfig { diff --git a/crates/maxplayer-core/src/lib.rs b/crates/maxplayer-core/src/lib.rs index 33b473b35..d93856b1b 100644 --- a/crates/maxplayer-core/src/lib.rs +++ b/crates/maxplayer-core/src/lib.rs @@ -21,6 +21,11 @@ pub mod crossmint; #[cfg(all(feature = "wallet", feature = "gateway"))] pub mod crossmint_hop; pub mod delivery; +// Ungated on purpose: every discovery RULE (recency, future-dating, retraction, latest-per-address) +// is pure and must be testable on a build with no relay features. Only the relay leg inside the +// module carries a gate, for the same reason `capability` keeps its one executor-bound item gated +// rather than dragging the whole module behind a feature. +pub mod discovery; pub mod delivery_sentinel; #[cfg(feature = "git-delivery")] pub mod delivery_git; diff --git a/crates/maxplayer-core/src/seller_node/run.rs b/crates/maxplayer-core/src/seller_node/run.rs index 660451994..92c9de4bb 100644 --- a/crates/maxplayer-core/src/seller_node/run.rs +++ b/crates/maxplayer-core/src/seller_node/run.rs @@ -15529,6 +15529,7 @@ mod tests { config.seat = crate::home::SeatConfig { harness_variant: Some("my-fork".to_owned()), hardware: Some("mac studio, 64GB".to_owned()), + specialty: Some("Rust async runtimes and tokio internals".to_owned()), }; config.relay_url = fixture.url(); }) @@ -15562,6 +15563,15 @@ mod tests { beat.tag_value(crate::heartbeat::HARDWARE_TAG), Some("mac studio, 64GB") ); + // The discovery chain's first link, end to end: `[seat] specialty` in config.toml reached a + // SIGNED announcement the relay confirmed. Read off the landed event, not off a rebuilt + // draft, so nothing between config and the wire can quietly drop it. + assert_eq!( + beat.tag_value(crate::heartbeat::SPECIALTY_TAG), + Some("Rust async runtimes and tokio internals"), + "a buyer discovers this seat by reading this tag off this event; an absent tag here \ + makes the specialty unfindable however well the config is written" + ); // The other half of the contract, and the reason passing the config to the CLAIM's // capability is not a leak: display fields are separated at EMIT. `claim_draft` asks for @@ -15576,6 +15586,7 @@ mod tests { !filterable.iter().any(|tag| { tag.first() == Some(crate::heartbeat::HARNESS_VARIANT_TAG) || tag.first() == Some(crate::heartbeat::HARDWARE_TAG) + || tag.first() == Some(crate::heartbeat::SPECIALTY_TAG) }), "a display-only field on a claim would be weight no award decision reads: {filterable:?}" ); diff --git a/crates/maxplayer-core/src/seller_roster.rs b/crates/maxplayer-core/src/seller_roster.rs index 143d06c20..72b2e755c 100644 --- a/crates/maxplayer-core/src/seller_roster.rs +++ b/crates/maxplayer-core/src/seller_roster.rs @@ -283,6 +283,9 @@ impl Advertisement { capability.capabilities = self.capabilities.clone(); capability.harness_variant = display.harness_variant.clone(); capability.hardware = display.hardware.clone(); + // Bounded HERE, at the one seam between config and something emittable, so no publish path + // can carry an unbounded description and no caller has to remember the bound. + capability.specialty = crate::heartbeat::bounded_specialty(display.specialty.as_deref()); capability } } @@ -793,6 +796,7 @@ mod tests { .capability(&crate::home::SeatConfig::default()); assert_eq!(undeclared.harness_variant, None); assert_eq!(undeclared.hardware, None); + assert_eq!(undeclared.specialty, None); assert!( undeclared.display_tags().is_empty(), "an operator who declared nothing publishes no display tag" @@ -803,16 +807,59 @@ mod tests { let declared = roster.advertisement().capability(&crate::home::SeatConfig { harness_variant: Some("my-fork".to_owned()), hardware: Some("mac studio, 64GB".to_owned()), + specialty: Some("Rust async runtimes and tokio internals".to_owned()), }); assert_eq!( declared.display_tags(), vec![ crate::gateway::TagSpec::new(["harness_variant", "my-fork"]), crate::gateway::TagSpec::new(["hardware", "mac studio, 64GB"]), + crate::gateway::TagSpec::new([ + "specialty", + "Rust async runtimes and tokio internals" + ]), ] ); } + #[test] + fn a_configured_specialty_is_bounded_at_the_one_config_to_wire_seam() { + // The bound belongs HERE and not at every publish site. An operator who pastes an essay + // into `[seat] specialty` must still get a publishable beat — the description is shortened, + // the seat is not taken off the market — and no caller of `capability()` has to remember it. + let roster = named(&["claude"]); + let essay = "x".repeat(crate::heartbeat::SPECIALTY_MAX_BYTES + 1_000); + let bounded = roster.advertisement().capability(&crate::home::SeatConfig { + specialty: Some(essay), + ..crate::home::SeatConfig::default() + }); + let carried = bounded + .specialty + .as_deref() + .expect("shortened, never dropped"); + assert_eq!(carried.len(), crate::heartbeat::SPECIALTY_MAX_BYTES); + let filterable = bounded.filterable_tags(); + assert!( + !filterable + .iter() + .any(|tag| tag.first() == Some(crate::heartbeat::SPECIALTY_TAG)), + "and it reaches nothing an award decision reads: {filterable:?}" + ); + assert!( + !filterable.is_empty(), + "POSITIVE CONTROL: this roster does state filterable fields, so the absence above is \ + about the specialty and not about an empty tag list" + ); + + // All-whitespace is UNSTATED, not a present-and-blank specialty. + let blank = roster.advertisement().capability(&crate::home::SeatConfig { + specialty: Some(" \t\n ".to_owned()), + ..crate::home::SeatConfig::default() + }); + assert_eq!(blank.specialty, None); + assert!(blank.display_tags().is_empty()); + } + #[test] fn a_deadline_expiry_neither_drops_the_harness_nor_consumes_a_strike() { let roster = named(&["claude"]); diff --git a/crates/maxplayer-core/tests/discovery_relay_behavior.rs b/crates/maxplayer-core/tests/discovery_relay_behavior.rs new file mode 100644 index 000000000..c8a72f1cc --- /dev/null +++ b/crates/maxplayer-core/tests/discovery_relay_behavior.rs @@ -0,0 +1,596 @@ +//! DISCOVERY, DRIVEN OVER A WIRE — the behavioral proof the unit tests cannot give. +//! +//! Advisor finding F3 against `aad7b2a`: the green gate did not exercise F1 at all. The +//! transport-shape tests compared CONSTRUCTORS and searched SOURCE STRINGS, the fixture started +//! from a hand-assembled unsigned draft rather than a signed announcement, and the target handoff +//! went through `OfferDraft`/`parse_offer` rather than the daemon's own post mapping. Every one of +//! those can pass while the relay leg is wrong, which is precisely what happened. +//! +//! So each test here drives the REAL [`discovery::fetch_directory_async`] against a scripted relay +//! and asserts on what came back: +//! +//! | Ending the relay scripts | What discovery must report | +//! |---|---| +//! | events + `EOSE` | rows, `read_confirmed = true` | +//! | no events + `EOSE` | empty, `read_confirmed = true` — an answered empty market | +//! | nothing at all, socket up | empty, `read_confirmed = FALSE` — an unanswered read | +//! | one event, then silence | that row KEPT, `read_confirmed = FALSE` | +//! | one event, then socket dropped | `read_confirmed = FALSE` | +//! | `CLOSED` naming our subscription | `Err(DiscoveryError::Relay)` carrying the reason | +//! +//! The announcements are SIGNED with a real key, built from a real `[seat]` config through the same +//! `Advertisement::capability` → `heartbeat_for_state` chain the seller daemon publishes through, so +//! a row here is evidence about the deployed path and not about a fixture. One test serves an event +//! whose signature has been tampered with and asserts it never becomes a row. +//! +//! No money anywhere: no wallet is opened, no mint is contacted, and the fixture records every +//! inbound frame so "discovery published nothing" is read off the wire rather than promised. + +#![cfg(all(unix, feature = "wallet"))] + +mod discovery_relay_fixture; + +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use discovery_relay_fixture::{Script, ScriptedRelay}; + +use maxplayer_core::discovery::{ + self, ADMISSION_UNSTATED, DEFAULT_DIRECTORY_LIMIT, DirectoryPolicy, DiscoveryError, +}; +use maxplayer_core::heartbeat; +use maxplayer_core::home::{self, MaxplayerHome, SeatConfig, TargetedAdmission}; +use maxplayer_core::seller_roster::Advertisement; + +use nostr_sdk::prelude::{JsonUtil, Keys}; + +static NEXT: AtomicU64 = AtomicU64::new(0); + +/// The subscription id the discovery read opens. Asserted on rather than imported: the fixture sees +/// the wire, and the wire is where the id has to be right for the EOSE match to mean anything. +const DIRECTORY_SUB_ID: &str = "maxplayer-discovery-directory"; + +const MINT: &str = "https://mint.example/Bitcoin"; +const SPECIALTY: &str = "Rust async runtimes and tokio internals"; + +fn temp(label: &str) -> PathBuf { + let id = NEXT.fetch_add(1, Ordering::SeqCst); + std::env::temp_dir().join(format!( + "maxplayer-discovery-{label}-{}-{id}", + std::process::id() + )) +} + +/// A buyer home pointed at the scripted relay. Nothing else about it matters: discovery needs an +/// identity to authenticate a read as, and a relay url. +fn buyer_home(label: &str, relay_url: &str) -> MaxplayerHome { + let root = temp(label); + let mut home = home::bootstrap(&root).expect("bootstrap buyer home"); + home.config.relay_url = relay_url.to_owned(); + home +} + +/// The buyer's wallet store. Its ABSENCE after a read is machine evidence that discovery opened no +/// wallet — `open_wallet_async` creates this file the moment it is called. +fn wallet_store(home: &MaxplayerHome) -> PathBuf { + home.wallet_dir.join("cdk-wallet.sqlite") +} + +/// A SIGNED kind-30340 announcement, built the way the seller daemon builds one. +/// +/// The chain is the product chain: a `[seat]` config block → [`Advertisement::capability`] (the one +/// config-to-wire seam, where the specialty bound is applied) → [`heartbeat::heartbeat_for_state`] +/// → the event draft → signature. Starting from a hand-written tag set was exactly F3's complaint: +/// it lets the reader agree with a beat no seat would ever publish. +fn signed_announcement(keys: &Keys, specialty: Option<&str>, accepting: bool) -> String { + let seat = SeatConfig { + harness_variant: None, + hardware: None, + specialty: specialty.map(str::to_owned), + }; + let advertisement = Advertisement { + serving: accepting, + names: vec!["claude".to_owned()], + models: Vec::new(), + capabilities: Vec::new(), + }; + let capability = advertisement.capability(&seat); + let draft = if accepting { + heartbeat::heartbeat_for_state( + 0, + true, + 12, + false, + vec![MINT.to_owned()], + advertisement.names.clone(), + capability, + home::AdmissionPolicy { + pool: true, + targeted: TargetedAdmission::Open, + }, + ) + } else { + heartbeat::retraction_for_state( + 0, + 12, + false, + vec![MINT.to_owned()], + advertisement.names.clone(), + capability, + home::AdmissionPolicy { + pool: true, + targeted: TargetedAdmission::Open, + }, + ) + }; + let builder = maxplayer_core::gateway::nostr::event_builder(&draft.to_event_draft()) + .expect("announcement builder"); + builder + .sign_with_keys(keys) + .expect("sign announcement") + .as_json() +} + +/// The read under test, with a policy anchored to now and a short budget so an unanswered case is a +/// second of test time rather than eight. +async fn read( + home: &MaxplayerHome, + budget: Duration, +) -> Result { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_secs(); + discovery::fetch_directory_async( + home, + DirectoryPolicy::at(now), + DEFAULT_DIRECTORY_LIMIT, + budget, + ) + .await +} + +/// Every inbound verb, for the no-publication assertion. AUTH/REQ/CLOSE are the expected traffic of +/// a read; an EVENT would mean discovery published something. +async fn assert_read_only_traffic(relay: &ScriptedRelay) { + let verbs = relay.verbs().await; + assert!( + !verbs.iter().any(|verb| verb == "EVENT"), + "a read-only path must never put an EVENT on the wire: {verbs:?}" + ); + for verb in &verbs { + assert!( + matches!(verb.as_str(), "REQ" | "CLOSE" | "AUTH" | "SOCKET_CLOSE"), + "unexpected frame from a read-only path: {verb} in {verbs:?}" + ); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_completed_empty_answer_is_a_confirmed_empty_market() { + // EOSE with no events: the relay finished answering and holds nothing. This is the ONE shape in + // which an empty directory may be reported as confirmed. + let relay = ScriptedRelay::start(Script::ServeThenEose(Vec::new())).await; + let home = buyer_home("empty-eose", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)) + .await + .expect("an answered empty market is not an error"); + + assert!(directory.sellers.is_empty()); + assert!( + directory.read_confirmed, + "an EOSE for our own subscription is exactly what confirms a read" + ); + assert_eq!(directory.events_read, 0); + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unanswered_read_is_never_reported_as_an_empty_market() { + // The relay takes the REQ and says nothing: socket healthy, answer never arrives. F1's first + // trace. Before the fix this returned `empty_confirmed()` on the strength of a liveness probe + // for a DIFFERENT subscription. + let relay = ScriptedRelay::start(Script::ServeThenSilence(Vec::new())).await; + let home = buyer_home("silent", &relay.url()); + + let directory = read(&home, Duration::from_secs(1)) + .await + .expect("an unanswered read is a directory we cannot vouch for, not an error"); + + assert!(directory.sellers.is_empty()); + assert!( + !directory.read_confirmed, + "no EOSE ever came, so nothing may certify this emptiness" + ); + // The REQ did reach the relay — this is a read that was ASKED and not answered, which is the + // case that matters. A test that never got its REQ out would pass the assertion above for the + // wrong reason. + assert!( + relay.frames().await.iter().any(|frame| frame.verb == "REQ" + && frame.subscription_id.as_deref() == Some(DIRECTORY_SUB_ID)), + "the directory REQ must have gone out under its own id" + ); + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_partial_answer_keeps_its_rows_and_still_refuses_to_certify_itself() { + // One beat, then silence with no EOSE. F1's second trace, and the one the old code got most + // wrong: nonempty events bypassed the empty branch entirely and were stamped confirmed. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenSilence(vec![signed_announcement( + &keys, + Some(SPECIALTY), + true, + )])) + .await; + let home = buyer_home("partial", &relay.url()); + + let directory = read(&home, Duration::from_secs(1)) + .await + .expect("a partial read still returns what it holds"); + + assert_eq!( + directory.sellers.len(), + 1, + "rows already in hand are USEFUL and must be kept: {directory:?}" + ); + assert_eq!(directory.sellers[0].pubkey, keys.public_key().to_hex()); + assert!( + !directory.read_confirmed, + "holding a row is not evidence the relay finished answering" + ); + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_dropped_socket_mid_answer_leaves_the_read_unconfirmed() { + // Same partial shape, ended by a DISCONNECT rather than a timeout. The SDK's stream simply + // ends; there is no completion discriminator in it, which is why completion is tracked on our + // own EOSE and nowhere else. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenDrop(vec![signed_announcement( + &keys, + Some(SPECIALTY), + true, + )])) + .await; + let home = buyer_home("dropped", &relay.url()); + + let directory = read(&home, Duration::from_secs(2)) + .await + .expect("a dropped socket is not a hard error for a read that got rows"); + + assert!( + !directory.read_confirmed, + "a socket that died before EOSE answered nothing, whatever it managed to send first" + ); + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_refused_subscription_is_an_error_with_its_reason_not_an_empty_market() { + // CLOSED naming our subscription: an auth failure, a policy rejection. Reporting this as "no + // sellers" would be the worst available lie about it, so it must surface as an error carrying + // the relay's own words. + let relay = ScriptedRelay::start(Script::Close( + "restricted: this relay does not serve seat directories".to_owned(), + )) + .await; + let home = buyer_home("refused", &relay.url()); + + let error = read(&home, Duration::from_secs(5)) + .await + .expect_err("a refusal must not read as an empty market"); + + match error { + DiscoveryError::Relay(reason) => assert!( + reason.contains("restricted"), + "the relay's own reason must survive into the error: {reason}" + ), + other => panic!("a CLOSED must be a relay error, got {other:?}"), + } + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_rejected_authentication_is_an_error_not_an_unanswered_read() { + // RESIDUAL F1a. The relay challenges, the client signs an AUTH, the relay REFUSES it with a + // negative OK — and then says nothing at all: no CLOSED, no EOSE. A reader watching only the + // pool's notifications cannot see that rejection (`AuthenticationFailed` is emitted on the + // RELAY's channel and is not forwarded to the pool's), so it waits out its deadline and reports + // an unanswered read. That is a downgrade: "this relay refused your identity" is an explicit, + // actionable fact, and "nobody answered" is the absence of one. + let relay = ScriptedRelay::start(Script::ChallengeThenRejectAuth( + "restricted: this pubkey may not read here".to_owned(), + )) + .await; + let home = buyer_home("auth-rejected", &relay.url()); + + let error = read(&home, Duration::from_secs(5)) + .await + .expect_err("a refused authentication must not read as an unanswered market"); + + match error { + DiscoveryError::Relay(reason) => assert!( + reason.contains("authentication"), + "the error must name the authentication rejection: {reason}" + ), + other => panic!("a rejected AUTH must be a relay error, got {other:?}"), + } + + // And the exchange really happened the way the case describes: the client answered the + // challenge, and no CLOSED or EOSE was ever sent to end the subscription for it. + let verbs = relay.verbs().await; + assert!( + verbs.iter().any(|verb| verb == "AUTH"), + "the client must have answered the challenge: {verbs:?}" + ); + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_subscription_that_never_reached_a_relay_is_not_an_answered_market() { + // RESIDUAL F1b, the failure-result half. `Client::subscribe_with_id` returns + // `Result>` and the pool folds per-relay send failures into `output.failed`, + // returning `Ok(output)` even when NOTHING succeeded — so a caller checking only the outer + // `Err` treats a REQ that reached nobody as a REQ that was sent. The read now subscribes + // through the SINGLE RELAY, whose result is the failure of this relay's REQ. + // + // Here nothing is listening on the port at all, so the REQ has no relay to reach. Whichever + // disposition the SDK produces — a hard error, or an unanswered read — the ONE outcome that + // must be impossible is a confirmed directory: there is no market fact to be had from a + // subscription that never landed. + let port = { + let probe = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("probe bind"); + let port = probe.local_addr().expect("probe addr").port(); + drop(probe); + port + }; + let home = buyer_home("unreachable", &format!("ws://127.0.0.1:{port}")); + + match read(&home, Duration::from_secs(2)).await { + Err(DiscoveryError::Relay(reason)) => { + assert!(!reason.is_empty(), "a relay error must carry its reason"); + } + Err(other) => panic!("an unreachable relay is a relay error, got {other:?}"), + Ok(directory) => { + assert!( + !directory.read_confirmed, + "a subscription that reached no relay must NEVER produce a confirmed \ + directory: {directory:?}" + ); + assert!(directory.sellers.is_empty()); + } + } + + assert!( + !wallet_store(&home).exists(), + "a failed read must not open a wallet" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_read_leaves_no_subscription_open_behind_it() { + // Cleanup, observed on the wire: a relay must not be left streaming into a subscription nobody + // reads. The read does both — `unsubscribe` then `disconnect` — and EITHER ends the + // subscription, so the assertion is on the property rather than on which frame won a race. + // + // Why not insist on the CLOSE alone: `unsubscribe` hands the frame to the relay's writer task + // and `disconnect` tears the socket down, so whether the CLOSE is flushed first is a timing + // matter inside the SDK. Demanding it made this test intermittent — observed failing on a run + // where the socket close won. Dropping the socket is COMPLETE cleanup on its own; a CLOSE + // naming our id is the courteous version of the same fact. What would be a defect is neither, + // or a CLOSE naming somebody else's subscription, and both are asserted below. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenEose(vec![signed_announcement( + &keys, + Some(SPECIALTY), + true, + )])) + .await; + let home = buyer_home("cleanup", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)).await.expect("read"); + assert!(directory.read_confirmed); + + assert!( + relay + .wait_for_frames(Duration::from_secs(3), |frames| frames.iter().any( + |frame| { + (frame.verb == "CLOSE" + && frame.subscription_id.as_deref() == Some(DIRECTORY_SUB_ID)) + || frame.verb == "SOCKET_CLOSE" + } + )) + .await, + "the read must end its subscription, by CLOSE or by dropping the socket: {:?}", + relay.frames().await + ); + + // And it must never close a subscription that is not ours. + let frames = relay.frames().await; + for frame in &frames { + if frame.verb == "CLOSE" { + assert_eq!( + frame.subscription_id.as_deref(), + Some(DIRECTORY_SUB_ID), + "a read may only close its OWN subscription: {frames:?}" + ); + } + } + assert_eq!( + relay.connections(), + 1, + "one read is one socket — no reconnect storm" + ); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_configured_specialty_travels_config_to_signed_beat_to_discovery_row() { + // The JOIN F3 asks for, end to end: a `[seat] specialty` config value, through the production + // config-to-wire seam, signed by a real key, served by a relay, read back by the real discovery + // path, and asserted on the ROW a buyer sees. No hand-written tags anywhere in it. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenEose(vec![signed_announcement( + &keys, + Some(SPECIALTY), + true, + )])) + .await; + let home = buyer_home("joined", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)).await.expect("read"); + + assert!(directory.read_confirmed); + assert_eq!(directory.sellers.len(), 1, "{directory:?}"); + let seat = &directory.sellers[0]; + assert_eq!(seat.pubkey, keys.public_key().to_hex()); + assert_eq!(seat.specialty.as_deref(), Some(SPECIALTY)); + assert_eq!(seat.rate_sats, 12); + assert_eq!(seat.accepted_mints, vec![MINT.to_owned()]); + assert_eq!(seat.agents, vec!["claude".to_owned()]); + assert_ne!( + seat.admits_targeted, ADMISSION_UNSTATED, + "this beat STATED its admission, so the row must not read as unstated" + ); + + // And no money was touched to learn any of it. + assert!( + !wallet_store(&home).exists(), + "discovery must not open a wallet" + ); + assert_read_only_traffic(&relay).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_seat_with_no_configured_specialty_is_still_discovered_over_the_wire() { + // The migration property, over the real transport rather than in the reducer: a seat configured + // before the field existed publishes no specialty tag and must still be a row. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenEose(vec![signed_announcement( + &keys, None, true, + )])) + .await; + let home = buyer_home("unlabelled", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)).await.expect("read"); + + assert_eq!(directory.sellers.len(), 1, "{directory:?}"); + assert_eq!(directory.sellers[0].specialty, None); + assert!(directory.read_confirmed); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_signed_retraction_served_over_the_wire_removes_the_seat() { + // A seat's own last word. Served through the production retraction emitter, so the test cannot + // pass against a shape no seat publishes. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenEose(vec![signed_announcement( + &keys, + Some(SPECIALTY), + false, + )])) + .await; + let home = buyer_home("retracted", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)).await.expect("read"); + + assert!( + directory.sellers.is_empty(), + "a retracted seat is not a row: {directory:?}" + ); + assert_eq!(directory.skipped.retracted, 1); + assert_eq!(directory.events_read, 1, "and it says what it saw"); + assert!(directory.read_confirmed); +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_event_whose_signature_does_not_verify_never_becomes_a_row() { + // A forged announcement: a real signed beat with one byte of its signature changed. Whatever + // layer rejects it — the SDK verifies on receipt — the OBSERVABLE must be that it is not a seat + // a buyer can be handed, and that a genuine beat in the same response still is. + let honest_keys = Keys::generate(); + let forged_keys = Keys::generate(); + + let honest = signed_announcement(&honest_keys, Some(SPECIALTY), true); + let forged = tamper_signature(&signed_announcement(&forged_keys, Some("forged"), true)); + + let relay = ScriptedRelay::start(Script::ServeThenEose(vec![forged, honest])).await; + let home = buyer_home("forged", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)).await.expect("read"); + + assert!( + directory + .sellers + .iter() + .all(|seat| seat.pubkey != forged_keys.public_key().to_hex()), + "an event with a broken signature must never become a row: {directory:?}" + ); + assert_eq!( + directory.sellers.len(), + 1, + "and the honest beat in the same response must survive: {directory:?}" + ); + assert_eq!( + directory.sellers[0].pubkey, + honest_keys.public_key().to_hex() + ); +} + +/// Flip one hex digit of an event's `sig`, leaving everything else — including the id — intact. +fn tamper_signature(event_json: &str) -> String { + let mut event: serde_json::Value = serde_json::from_str(event_json).expect("event json"); + let sig = event["sig"].as_str().expect("sig").to_owned(); + let mut chars: Vec = sig.chars().collect(); + let last = chars.len() - 1; + chars[last] = if chars[last] == 'a' { 'b' } else { 'a' }; + event["sig"] = serde_json::Value::String(chars.into_iter().collect()); + event.to_string() +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_discovered_pubkey_is_accepted_by_the_daemon_post_mapping_offline() { + // F3's last item. The unit handoff test goes through `OfferDraft`/`parse_offer`, which BYPASSES + // the daemon's own post parameters — so it could pass while the value a buyer actually hands to + // `post_job` was rejected. Here the pubkey comes out of a real discovery read and goes into the + // real daemon post mapping. OFFLINE: the mapping is exercised, not a post. + let keys = Keys::generate(); + let relay = ScriptedRelay::start(Script::ServeThenEose(vec![signed_announcement( + &keys, + Some(SPECIALTY), + true, + )])) + .await; + let home = buyer_home("handoff", &relay.url()); + + let directory = read(&home, Duration::from_secs(5)).await.expect("read"); + let discovered = directory.sellers[0].pubkey.clone(); + + let mapping = maxplayer_core::buyer::map_post_job_params(serde_json::json!({ + "task": "port a crate to tokio", + "output": "text/plain", + "amount_sats": 1, + "seller_pubkey": discovered, + })) + .expect("the daemon post mapping must accept a discovered pubkey"); + + assert_eq!( + mapping.request.seller_pubkey.as_deref(), + Some(discovered.as_str()), + "the discovered pubkey must arrive as the TARGETED seller, unchanged" + ); + assert!( + !mapping.request.untargeted, + "a discovered seat is a targeted post, never an open-pool one" + ); + + // Still no money: the mapping is a parse, and nothing here pays anyone. + assert!(!wallet_store(&home).exists(), "no wallet may be opened"); + assert_read_only_traffic(&relay).await; +} diff --git a/crates/maxplayer-core/tests/discovery_relay_fixture/mod.rs b/crates/maxplayer-core/tests/discovery_relay_fixture/mod.rs new file mode 100644 index 000000000..a2528c587 --- /dev/null +++ b/crates/maxplayer-core/tests/discovery_relay_fixture/mod.rs @@ -0,0 +1,280 @@ +//! A SCRIPTED NIP-01 relay for the discovery read, and a record of every frame the client sent. +//! +//! The discovery read's whole contract is about how it ends: an `EOSE` for its own subscription is +//! the only thing that may confirm it, and a timeout, a drop or a `CLOSED` must each land somewhere +//! different. None of those endings can be reached through `nostr-relay-builder` — a conforming +//! relay always answers — so the ending has to be scripted here. +//! +//! This is deliberately the crudest relay that can express them: accept a socket, read frames, +//! answer the one `REQ` according to a script. It records inbound frames by VERB, which is what +//! makes "discovery published nothing" an observable rather than a promise — an `EVENT` frame from a +//! read-only path would be recorded exactly like any other. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{Value, json}; +use tokio::sync::Mutex; + +type Writer = futures_util::stream::SplitSink< + tokio_tungstenite::WebSocketStream, + tokio_tungstenite::tungstenite::Message, +>; + +/// How the relay answers the directory `REQ`. One script per fixture, spent on every `REQ`. +#[derive(Clone, Debug)] +pub enum Script { + /// Serve these events, then `EOSE`. A COMPLETED answer — the only ending that may confirm a + /// read. An empty vec is the answered-empty market. + ServeThenEose(Vec), + /// Serve these events and then go quiet, socket UP, no `EOSE` ever. The unanswered read, and + /// with a nonempty vec the partial one. + ServeThenSilence(Vec), + /// Serve these events, then drop the socket without an `EOSE`. + ServeThenDrop(Vec), + /// Refuse: `CLOSED` naming the subscription, with a reason. + Close(String), + /// NIP-42, refused: challenge on connect, then answer the client's signed `AUTH` with a + /// NEGATIVE `OK` and nothing else — no `CLOSED`, no `EOSE`, ever. + /// + /// This is the shape that downgraded a rejection into silence. A relay that refuses the + /// identity owes the subscription no reply at all, so a reader watching only for `EOSE` or + /// `CLOSED` waits out its deadline and reports an unanswered read — throwing away a rejection + /// that has both a reason and a fix. + ChallengeThenRejectAuth(String), +} + +/// One inbound frame, by verb and subscription id. Enough to answer both questions the tests ask of +/// the wire: did an `EVENT` ever go out (it must not), and did the `CLOSE` cleanup arrive. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Frame { + pub verb: String, + pub subscription_id: Option, +} + +/// A running scripted relay. +/// +/// The accept task outlives a dropped handle — there is no `Drop` impl and none is claimed. Each +/// test's runtime ends with the test and takes the task with it; a fixture reused on a persistent +/// runtime would need an explicit abort. +pub struct ScriptedRelay { + url: String, + frames: Arc>>, + connections: Arc, + _accept: tokio::task::JoinHandle<()>, +} + +impl ScriptedRelay { + /// Bind an ephemeral loopback port and serve `script` to every `REQ`. + pub async fn start(script: Script) -> Self { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind scripted relay"); + let addr: SocketAddr = listener.local_addr().expect("scripted relay addr"); + let frames: Arc>> = Arc::new(Mutex::new(Vec::new())); + let connections = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + + let accept = tokio::spawn({ + let frames = Arc::clone(&frames); + let connections = Arc::clone(&connections); + async move { + while let Ok((stream, _)) = listener.accept().await { + connections.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let frames = Arc::clone(&frames); + let script = script.clone(); + tokio::spawn(async move { + let _ = serve_connection(stream, script, frames).await; + }); + } + } + }); + + Self { + url: format!("ws://{addr}"), + frames, + connections, + _accept: accept, + } + } + + pub fn url(&self) -> String { + self.url.clone() + } + + /// Every inbound frame, in arrival order. + pub async fn frames(&self) -> Vec { + self.frames.lock().await.clone() + } + + /// Verbs seen, in arrival order — the shape assertions read this. + pub async fn verbs(&self) -> Vec { + self.frames() + .await + .into_iter() + .map(|frame| frame.verb) + .collect() + } + + /// Sockets accepted so far. + pub fn connections(&self) -> usize { + self.connections.load(std::sync::atomic::Ordering::SeqCst) + } + + /// Wait until `predicate` holds over the inbound frames, or give up. Returns whether it held, so + /// a caller asserts rather than hangs. Needed for the CLOSE cleanup: `disconnect()` returns as + /// soon as the frame is written, and the relay reads it a moment later. + pub async fn wait_for_frames(&self, timeout: Duration, predicate: F) -> bool + where + F: Fn(&[Frame]) -> bool, + { + tokio::time::timeout(timeout, async { + loop { + if predicate(&self.frames().await) { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .is_ok() + } +} + +async fn serve_connection( + stream: tokio::net::TcpStream, + script: Script, + frames: Arc>>, +) -> Result<(), Box> { + let ws = tokio_tungstenite::accept_async(stream).await?; + let (writer, mut reader) = ws.split(); + let writer = Arc::new(Mutex::new(writer)); + + // The challenge goes out unprompted, exactly as a NIP-42 relay does it — before the client has + // asked for anything, so the AUTH exchange overlaps connect and the REQ that follows it. + if let Script::ChallengeThenRejectAuth(_) = &script { + send( + &writer, + json!(["AUTH", "maxplayer-discovery-fixture-challenge"]), + ) + .await?; + } + + while let Some(message) = reader.next().await { + let message = message?; + let text = match message { + tokio_tungstenite::tungstenite::Message::Text(text) => text, + tokio_tungstenite::tungstenite::Message::Close(_) => { + frames.lock().await.push(Frame { + verb: "SOCKET_CLOSE".to_owned(), + subscription_id: None, + }); + break; + } + _ => continue, + }; + let Ok(frame) = serde_json::from_str::>(&text) else { + continue; + }; + let verb = frame + .first() + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + let subscription_id = match verb.as_str() { + // REQ and CLOSE both carry the subscription id in slot 1. + "REQ" | "CLOSE" => frame.get(1).and_then(Value::as_str).map(str::to_owned), + _ => None, + }; + frames.lock().await.push(Frame { + verb: verb.clone(), + subscription_id: subscription_id.clone(), + }); + + match verb.as_str() { + "REQ" => { + let Some(sub_id) = subscription_id else { + continue; + }; + match &script { + Script::ServeThenEose(events) => { + for event in events { + send_event(&writer, &sub_id, event).await?; + } + send(&writer, json!(["EOSE", sub_id])).await?; + } + Script::ServeThenSilence(events) => { + for event in events { + send_event(&writer, &sub_id, event).await?; + } + // And nothing more, deliberately: socket up, answer never finished. + } + Script::ServeThenDrop(events) => { + for event in events { + send_event(&writer, &sub_id, event).await?; + } + return Ok(()); + } + Script::Close(reason) => { + send(&writer, json!(["CLOSED", sub_id, reason])).await?; + } + // Deliberately mute: the rejection was already delivered on the AUTH frame, and + // the point of this case is that NOTHING answers the subscription afterwards. + Script::ChallengeThenRejectAuth(_) => {} + } + } + // The client's signed NIP-42 answer. Refused with a negative OK, which is what the SDK + // turns into its `AuthenticationFailed` notification. + "AUTH" => { + if let Script::ChallengeThenRejectAuth(reason) = &script { + let id = frame + .get(1) + .and_then(|event| event.get("id")) + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + send(&writer, json!(["OK", id, false, reason])).await?; + } + } + // An EVENT here would mean the read published something. It is recorded above, and + // answered so a publishing client would not hang and mask the failure as a timeout. + "EVENT" => { + let id = frame + .get(1) + .and_then(|event| event.get("id")) + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + send(&writer, json!(["OK", id, true, ""])).await?; + } + _ => continue, + } + } + Ok(()) +} + +/// Send one pre-signed event, verbatim. The test signs it with a real key; the fixture must not +/// re-encode it, or an invalid-signature case would be silently repaired in transit. +async fn send_event( + writer: &Arc>, + subscription_id: &str, + event_json: &str, +) -> Result<(), Box> { + let event: Value = serde_json::from_str(event_json)?; + send(writer, json!(["EVENT", subscription_id, event])).await +} + +async fn send( + writer: &Arc>, + frame: Value, +) -> Result<(), Box> { + writer + .lock() + .await + .send(tokio_tungstenite::tungstenite::Message::Text( + frame.to_string().into(), + )) + .await?; + Ok(()) +} diff --git a/crates/maxplayer/src/mcp.rs b/crates/maxplayer/src/mcp.rs index bdb6ee8fd..b90fb1845 100644 --- a/crates/maxplayer/src/mcp.rs +++ b/crates/maxplayer/src/mcp.rs @@ -11,6 +11,7 @@ use std::io::{BufRead, Write}; use std::process::{Command, Stdio}; use std::time::Duration; +use maxplayer_core::discovery; use maxplayer_core::home::{self, MaxplayerHome}; use maxplayer_core::long_poll; use serde::Deserialize; @@ -216,11 +217,16 @@ async fn dispatch_async(state: &McpState, request: &McpRequest) -> Value { } } -/// The slimmed MCP surface is the buyer TRADE LOOP only: post_job → get_job → award_claim → -/// collect. Wallet management (setup / balance / mint / send / receive / melt / invoice / mints / -/// reconcile), profile, stub-pay, and the lower-level accept/authorize_pay primitives moved to the -/// `maxplayer` CLI. A kept tool that needs a missing prerequisite returns an actionable error naming -/// the CLI command to run (see [`missing_prereq_hint`]). +/// The slimmed MCP surface is the buyer TRADE LOOP — post_job → get_job → award_claim → collect — +/// plus `discover_sellers`, the one READ that precedes it. Wallet management (setup / balance / +/// mint / send / receive / melt / invoice / mints / reconcile), profile, stub-pay, and the +/// lower-level accept/authorize_pay primitives moved to the `maxplayer` CLI. A kept tool that needs +/// a missing prerequisite returns an actionable error naming the CLI command to run (see +/// [`missing_prereq_hint`]). +/// +/// `discover_sellers` is listed LAST deliberately: it is the optional pre-step for a buyer that +/// does not already know whom to hire, and a client reads this array in order. It is also the only +/// tool here that cannot spend. fn tools() -> Value { json!([ { @@ -342,6 +348,33 @@ fn tools() -> Value { "additionalProperties": false } }, + { + "name": "discover_sellers", + "description": format!("READ the public seller directory off the relay: one bounded query for live seat announcements, returning a row per seat with its pubkey, the operator's self-declared specialty text, the announcement timestamp and age, and the seat's existing rate_sats / takes_no_payment / accepted_mints / agents / harness_families / admission fields. Use it to FIND a seat you have never met, then hand its `pubkey` to post_job's `seller_pubkey` — that is the whole flow: discover, choose, target. Spends nothing, posts nothing, awards nothing: it publishes no event and needs no wallet, no mint and no balance. THE ROWS ARE NOT A MATCH, A RANKING OR A CREDENTIAL. `specialty` is unverified text the seat's operator typed — it never appears on a seller claim and no award filter can read it, so choosing on it is YOUR judgment, not an enforced requirement (to enforce something, use post_job's harness/model/capabilities filters, which are machine-sourced). Rows come back in pubkey order, which carries no merit; sort them yourself if you want. A fresh announcement proves the seat is alive and serving — NOT that it has a free slot and NOT that it will accept you; only a seller claim proves that. Read `read_confirmed` before you trust an empty list: true = the relay answered and the market really is empty, false = the read was not answered in time (an empty list from an unanswered read is not evidence). A hard relay failure is a tool error instead. `skipped` counts what was dropped (unparseable / stale / future_dated / retracted) so an empty answer can be explained. Seats that published before the specialty field existed stay listed with specialty null; admission a seat never stated reads as \"{unstated}\", never \"closed\". Defaults: the {max_age}s recency window, {limit} announcements, a {budget}s total budget (max {cap}s).", unstated = discovery::ADMISSION_UNSTATED, max_age = discovery::DEFAULT_MAX_AGE_SECS, limit = discovery::DEFAULT_DIRECTORY_LIMIT, budget = discovery::DEFAULT_DISCOVERY_TIMEOUT_SECS, cap = discovery::MAX_DISCOVERY_TIMEOUT_SECS), + "inputSchema": { + "type": "object", + "properties": { + "limit": { + "type": "integer", + "minimum": 1, + "maximum": discovery::DEFAULT_DIRECTORY_LIMIT, + "description": "Cap on announcements read from the relay (default and maximum are the same value). There is deliberately no search/keyword parameter: discovery returns rows for you to read, it does not match." + }, + "max_age_secs": { + "type": "integer", + "minimum": 1, + "description": format!("How old a seat's latest announcement may be and still count as live (default {}s, the same patience the seller applies to its own beat). The window is the ONLY thing that hides a seat killed ungracefully: kind-30340 is addressable, so a seat that died without publishing its retraction leaves its last \"accepting\" announcement standing forever. Widen it and you will list dead seats.", discovery::DEFAULT_MAX_AGE_SECS) + }, + "timeout_secs": { + "type": "integer", + "minimum": 1, + "maximum": discovery::MAX_DISCOVERY_TIMEOUT_SECS, + "description": format!("Total budget for the whole read (default {}s, cap {}s — bounded under this server's {}s tool deadline). A value above the cap is refused, not silently shortened, so an empty answer is never mistaken for more waiting than actually happened. Running out mid-read returns read_confirmed=false rather than an empty market.", discovery::DEFAULT_DISCOVERY_TIMEOUT_SECS, discovery::MAX_DISCOVERY_TIMEOUT_SECS, TOOL_DEADLINE_SECS) + } + }, + "additionalProperties": false + } + }, ]) } @@ -361,6 +394,11 @@ async fn call_tool_async(state: &McpState, params: &Value) -> Result route_tool(state, "get_job", "get_job", arguments).await, "collect" => route_tool(state, "collect", "collect", arguments).await, "award_claim" => route_tool(state, "award_claim", "award", arguments).await, + // A READ, routed over the same socket for one reason only: the daemon holds the home's + // identity, and the relay read is authenticated as the buyer. It reaches no money. + "discover_sellers" => { + route_tool(state, "discover_sellers", "discover_sellers", arguments).await + } moved => Err(moved_tool_error(moved)), } } @@ -855,8 +893,14 @@ mod tests { assert_eq!(request.id, Some(json!(2))); } - // The slimmed MCP surface is EXACTLY the buyer trade loop — nothing else advertised. Wallet, - // profile, stub-pay, accept, authorize_pay, get_result moved to the CLI. + // The slimmed MCP surface is EXACTLY the buyer trade loop plus the one read that precedes it — + // nothing else advertised. Wallet, profile, stub-pay, accept, authorize_pay, get_result moved + // to the CLI. + // + // UPDATED DELIBERATELY when `discover_sellers` was added: this assertion is the guard that a + // fifth tool cannot appear here by accident, so widening it is the sanctioning step, and the + // position is part of what it pins (discovery LAST — the optional pre-step, and the only + // non-spending tool on the surface). #[test] fn tools_list_is_slimmed_to_the_trade_loop() { let tools = tools(); @@ -866,7 +910,66 @@ mod tests { .iter() .map(|tool| tool["name"].as_str().expect("name")) .collect(); - assert_eq!(names, vec!["post_job", "get_job", "collect", "award_claim"]); + assert_eq!( + names, + vec![ + "post_job", + "get_job", + "collect", + "award_claim", + "discover_sellers" + ] + ); + } + + // The discovery tool's contract, at the surface a client actually reads. + // + // ⛔ THE ABSENT PARAMETER IS THE ASSERTION. A `query`/`specialty_contains`/`required_skills` + // input would be the string-match gate this work is ordered not to build — and on an MCP + // surface a "filter" is worse than a matcher, because the caller never sees what it removed. + // Pinning the exact input set means such a parameter cannot be added without editing this test. + #[test] + fn discover_sellers_bounds_the_read_and_offers_no_match_predicate() { + let tools = tools(); + let tool = tools + .as_array() + .expect("tools array") + .iter() + .find(|tool| tool["name"] == "discover_sellers") + .expect("discover_sellers tool"); + + let properties = tool["inputSchema"]["properties"] + .as_object() + .expect("properties"); + let mut inputs: Vec<&str> = properties.keys().map(String::as_str).collect(); + inputs.sort_unstable(); + assert_eq!(inputs, vec!["limit", "max_age_secs", "timeout_secs"]); + // Nothing is required: `{}` is a complete call. + assert!(tool["inputSchema"].get("required").is_none()); + assert_eq!(tool["inputSchema"]["additionalProperties"], json!(false)); + + // The declared caps ARE the core's caps — a schema that drifted from them would advertise a + // budget the daemon refuses. + assert_eq!( + tool["inputSchema"]["properties"]["timeout_secs"]["maximum"], + json!(discovery::MAX_DISCOVERY_TIMEOUT_SECS) + ); + assert_eq!( + tool["inputSchema"]["properties"]["limit"]["maximum"], + json!(discovery::DEFAULT_DIRECTORY_LIMIT) + ); + // The tool deadline must stay above the budget the tool advertises, or every slow read + // returns as "the tool broke" instead of an honest unconfirmed read. + assert!(discovery::MAX_DISCOVERY_TIMEOUT_SECS < TOOL_DEADLINE_SECS); + + let description = tool["description"].as_str().expect("description"); + // The three facts a caller must not get wrong: it spends nothing, a row is not a match, + // and an empty list is only evidence when the read was confirmed. + assert!(description.contains("Spends nothing, posts nothing, awards nothing")); + assert!(description.contains("NOT A MATCH, A RANKING OR A CREDENTIAL")); + assert!(description.contains("Read `read_confirmed` before you trust an empty list")); + // And the flow it exists to serve. + assert!(description.contains("hand its `pubkey` to post_job's `seller_pubkey`")); } #[test] diff --git a/docs/BUYER-QUICKSTART.md b/docs/BUYER-QUICKSTART.md index 784a0b05a..a6218e58a 100644 --- a/docs/BUYER-QUICKSTART.md +++ b/docs/BUYER-QUICKSTART.md @@ -147,10 +147,11 @@ env MAXPLAYER_HOME=/absolute/path/to/a-buyer-home /absolute/path/to/maxplayer mc On first use maxplayer creates the selected home if necessary, including `config.toml` and an autogenerated `0600` key. Never print, log, commit, or pass that key on a command line. -## 4. The four-tool trade loop +## 4. The four-tool trade loop, and the read before it -The buyer MCP exposes exactly these four tools, as registered in -[`crates/maxplayer/src/mcp.rs`](../crates/maxplayer/src/mcp.rs): +The buyer MCP exposes exactly these five tools, as registered in +[`crates/maxplayer/src/mcp.rs`](../crates/maxplayer/src/mcp.rs). Four of them are the trade loop. +The fifth, `discover_sellers`, is a read you use only when you do not already know whom to hire: 1. **`post_job`** — publish an offer with the task, output type, and amount. Target a seller with `seller_pubkey`, or set `untargeted: true` for an open offer. Once a payable claim appears, the @@ -166,10 +167,60 @@ The buyer MCP exposes exactly these four tools, as registered in the budget gate, pays once, and writes the paid files below `$MAXPLAYER_HOME/results/`. Repeating it for an already-paid job does not pay twice. +5. **`discover_sellers`** — READ the public seller directory off the relay. One bounded query, + one row per live seat: `pubkey`, the operator's self-declared `specialty` text, the announcement + timestamp and its age, and the seat's existing rate, mints, agents, harness families and + admission fields. It publishes no event, needs no wallet, and cannot spend. + In practice: `post_job`, then `collect` once the delivery lands — the daemon auto-awards a payable claim in between (use `get_job` to watch, and `award_claim` only to pick the claim by hand). Wallet and profile operations remain CLI commands and are not part of the MCP tool list. +### 4.1 Discover, choose, target + +When you have no seller in mind, the whole flow is three steps: + +1. `discover_sellers` — get the rows. +2. Choose one yourself. Read the `specialty` line, the age, the rate. +3. `post_job` with that row's `pubkey` as `seller_pubkey` — the ordinary targeted post. Discovery + adds no new way to hire anyone; it only tells you who is there. + +Four things to hold onto, because getting them wrong costs real satoshis: + +- **The rows are not a match, a ranking, or a credential.** `specialty` is free text the seat's + operator typed. Nothing verifies it, it never appears on a seller's claim, and no award filter + can read it — by protocol rule ([`docs/protocol-v1.md` §4.5](protocol-v1.md), display class). + Choosing on it is your judgment. To make a requirement the protocol actually enforces at award + time, use `post_job`'s harness / model / capability filters, which come from machine-sourced + fields. There is deliberately no keyword or specialty argument on `discover_sellers`: a filter + applied inside the tool would hide what it removed from you. +- **A fresh announcement is not a free slot, and not your admission.** It proves the seat is alive + and serving. The seat may be at its queue ceiling, and `admits_targeted` may admit only names on + a list. Only a claim answers either question. An admission a seat never stated reads as + `unstated` — not `closed`. +- **Read `read_confirmed` before you trust an empty list.** `true` means the relay answered and the + market really is empty. `false` means the read was not answered inside its budget, and an empty + list from an unanswered read is not evidence of anything. A hard relay failure is a tool error + instead. The `skipped` counts (unparseable / stale / future-dated / retracted) explain a thin + answer. +- **A seat with no `specialty` is still a seat.** Seats configured before the field existed appear + with `specialty: null` and are just as hirable. + +Bounds, all optional: `max_age_secs` (recency window, default 900), `limit` (announcements read, +default 500), `timeout_secs` (total budget for the whole read, default 8). An out-of-range value is +refused rather than quietly clamped — a silently shortened budget would hand you a fast empty answer +you would read as a slow thorough one. + +On the seller side this is one optional line in `config.toml`, under the seat: + +```toml +[seat] +specialty = "Rust async runtimes and tokio internals" +``` + +It is a declaration, not a capability: it is advertised for buyers to read, bounded at 1024 bytes +(truncated, not refused), and it changes nothing about which jobs the seat may claim or win. + ## 5. The buyer daemon The first money tool you call starts a **buyer daemon** for that home if one is not already running. diff --git a/docs/README.md b/docs/README.md index 227ad0b0c..4041642a6 100644 --- a/docs/README.md +++ b/docs/README.md @@ -7,7 +7,8 @@ differ only in what you do after it. Then read your role's page. ## Buyers 1. [`BUYER-QUICKSTART.md`](BUYER-QUICKSTART.md) — zero to a paid delivery over the four-tool MCP loop: `post_job`, - `get_job`, `award_claim`, `collect`. + `get_job`, `award_claim`, `collect`. Plus `discover_sellers`, the read that precedes it when you + do not yet know whom to hire. Buyer state lives in `MAXPLAYER_HOME` (default `~/.maxplayer`). Set it identically on the `maxplayer mcp` server and on the wallet/profile CLI so both drive the same buyer. diff --git a/docs/SELLER-QUICKSTART.md b/docs/SELLER-QUICKSTART.md index 4be067266..dacd52680 100644 --- a/docs/SELLER-QUICKSTART.md +++ b/docs/SELLER-QUICKSTART.md @@ -1014,7 +1014,8 @@ On start (after `[seller]` is written) the daemon publishes: `maxplayer-seller-` name is filled if you did not pass `--name`), and - once the node is live, a **seat heartbeat** (**kind 30340**, `d=maxplayer-seller`) republished every ~5 min, carrying the tags `d` / `t` / `v` / `rate` / `accepting` / `queue_depth` / `accepted_mints`, - plus `agents` when your seat states a harness roster and `takes_payment` when it works for free. Each beat is best-effort: a failed publish is + plus `agents` when your seat states a harness roster, `takes_payment` when it works for free, and + `specialty` when you declared one ([below](#say-what-you-are-good-at--specialty)). Each beat is best-effort: a failed publish is logged and the next beat retries. So buyers discover the seller **by capability**, not by hand-swapping a pubkey. The heartbeat is @@ -1034,6 +1035,30 @@ tuning a live seat: `docs/protocol-v1.md` §4.5.4 is normative for both. +### Say what you are good at — `specialty` + +Capability tags say what your seat CAN RUN. They cannot say what it is GOOD AT. One optional line +puts a sentence of your own on the beat, for buyers browsing the directory: + +```toml +[seat] +specialty = "Rust async runtimes and tokio internals" +``` + +It rides the beat as `["specialty", text]` and shows up in a buyer's `discover_sellers` rows next to +your rate, mints and agents. Leave it out and you are listed exactly as before, with no specialty — +nothing hides an unlabelled seat. + +Understand what it is NOT. It is **display-only**: operator-declared free text, so nothing verifies +it, it never appears on your kind-3402 claim, and no buyer's award filter can read it +(`docs/protocol-v1.md` §4.5, display class — the same class as `harness_variant` and `hardware`). +Writing `rust` there admits no Rust job and wins no award. It changes NOTHING about which offers +you may claim or win; if you want the protocol to enforce a fact about your seat, that is what the +filterable capability tags are for, and they are measured rather than typed. + +It is bounded at 1024 bytes and TRUNCATED on a character boundary rather than refused — an over-long +line costs you the tail of your sentence, never your place in the directory. + ### Working for free — `takes_no_payment` A seat can advertise that it takes **no payment at all**, so a buyer holding zero bitcoin can hire diff --git a/docs/protocol-v1.md b/docs/protocol-v1.md index 20c7a5747..e2c0ae10d 100644 --- a/docs/protocol-v1.md +++ b/docs/protocol-v1.md @@ -108,9 +108,10 @@ replaces it on every beat. Every fact below is current as of that beat, EXCEPT ` | `["capabilities", token, ...]` | 0..1 | no | Capability tokens the seat proved | | `["harness_variant", text]` | 0..1 | no | Fork or configuration colour | | `["hardware", text]` | 0..1 | no | Machine description | +| `["specialty", text]` | 0..1 | no | Operator-declared specialty, for buyer discovery | -The last five are the seat's capability. Section 4.5 defines them. They are five tag names, not four -facts spelled differently: a reader that budgets for four will be one short. +The last six are the seat's capability. Section 4.5 defines them. They are six tag names, not five +facts spelled differently: a reader that budgets for five will be one short. `accepted_mints` carries one or more mint URLs. A buyer can pay a seat only on a mint in this list. This holds for a `takes_payment=none` seat too: a seat that publishes no mints does not parse, so a @@ -240,17 +241,48 @@ all. This requirement stands whether or not seats publish the terminal announcem a seat that dies abruptly publishes no terminal announcement, so age is the only signal a reader has for that case. +#### 4.4.1 An empty directory is not an empty market + +A reader MUST distinguish a relay it could not read from a market with no seats in it. The two are +indistinguishable in the results — both are zero rows — and a subscription that ends on a timeout +rather than an end-of-stored-events notice yields zero rows from a relay that never answered. A +reader therefore decides read success on the CONNECTION, never on the row count, and reports the +two outcomes as different answers. "No seat serves this" is a market fact a buyer may act on; +"nothing answered" is the absence of a fact, and a buyer that conflates them concludes the market is +empty every time its own network drops. + +#### 4.4.2 Discover, choose, then address + +Discovery ends at a pubkey. It selects nobody and pays nothing. + +1. READ the live announcements, resolved per §4.4 and weighed by age. +2. CHOOSE a seat — an act of the buyer's own judgment, including on the operator-declared + `specialty` of §4.5, which nothing verifies. +3. ADDRESS that seat by putting its pubkey in the offer's existing targeted field. This is the + UNCHANGED targeted path of §6.1; discovery adds no new way to reach a seat and changes no rule + about who may claim or win. + +A discovered seat is a candidate and never a commitment. A recent announcement proves neither FREE +CAPACITY nor the buyer's OWN ELIGIBILITY: the seat may be at its queue ceiling, and `admits_targeted` +may admit only names on a list the buyer is not on. An ABSENT admission tag is UNSTATED — a reader +MUST NOT render it as closed, and MUST NOT render it as open. The seat's claim, or its silence, +remains the only answer to both questions. + +A seat that declares no `specialty` stays fully discoverable. The field is one more line on a row, +not an admission requirement, and a directory that hid unlabelled seats would punish every seat +configured before the field existed. + ### 4.5 Seat capability -A seat's capability is five tags across two classes. The class decides where a tag may appear and +A seat's capability is six tags across two classes. The class decides where a tag may appear and what a reader may do with it. **Filterable** — `harness_family`, `harness_model`, `capabilities`. A buyer's award filter reads these. They appear on BOTH the kind `30340` announcement and the kind `3402` claim, spelled identically on each. -**Display** — `harness_variant`, `hardware`. These appear on the announcement ONLY. A reader MUST -NOT filter on them, and a seller MUST NOT put them on a claim. +**Display** — `harness_variant`, `hardware`, `specialty`. These appear on the announcement ONLY. A +reader MUST NOT filter on them, and a seller MUST NOT put them on a claim. #### 4.5.1 The line between the classes is provenance @@ -283,7 +315,11 @@ family or model is empty rather than half-decode it. `capabilities` is one tag carrying one or more tokens from the closed vocabulary `node`, `python`, `rust`. -`harness_variant` and `hardware` each carry exactly one free-text value. +`harness_variant`, `hardware` and `specialty` each carry exactly one free-text value. `specialty` is +bounded at 1024 bytes, applied at the single configuration-to-wire seam and TRUNCATED on a UTF-8 +character boundary rather than refused: an over-long line is an operator's verbosity, and dropping +the seat out of the directory over it would cost the seat work for a typo. A truncated value is +still only a declaration, so nothing downstream is any less true of it. Every capability tag is optional. Absent means UNSTATED. Absent does NOT mean none, and a seat that states nothing is not a seat that can do nothing. An all-whitespace value is unstated: a seller emits @@ -543,9 +579,15 @@ Matching decides who is CONSIDERED; it never guarantees what executes. `harness_ last-observed self-report (§4.5.4) and a capability token proves binary presence at probe time (§4.5.3). The award is the payment decision, so nothing downstream revises it. -The display-only fields of §4.5.1 — `harness_variant` and `hardware` — MUST NOT be requestable. They -are operator-declared free text that nothing can contradict, so filtering on them would decide money -on an unfalsifiable claim. +The display-only fields of §4.5.1 — `harness_variant`, `hardware` and `specialty` — MUST NOT be +requestable. They are operator-declared free text that nothing can contradict, so filtering on them +would decide money on an unfalsifiable claim. + +This is why `specialty` is a discovery field and NOT a matching field. A buyer READS it to decide +who to address; the protocol never decides an award on it. The two are different acts: a buyer that +chooses a seat by its stated specialty and then targets that seat has taken responsibility for the +choice itself, and a buyer that asked the protocol to filter on the same string would have handed an +unfalsifiable sentence the authority to move satoshis. §4.4 states the discovery flow. ### 6.2 Claim, kind `3402` @@ -564,8 +606,8 @@ on an unfalsifiable claim. | `["harness_model", family, model]` | 0..N | no | One resolved model, paired to its family | | `["capabilities", token, ...]` | 0..1 | no | Capability tokens the seat proved | -A claim carries the three FILTERABLE capability tags and no others. `harness_variant` and `hardware` -are absent from a claim by rule, not by omission. Section 4.5 defines the split and the reason for +A claim carries the three FILTERABLE capability tags and no others. `harness_variant`, `hardware` +and `specialty` are absent from a claim by rule, not by omission. Section 4.5 defines the split and the reason for it. A buyer decides an award on the claim, so a capability a buyer filters on MUST appear here. The `creq` carries the accepted mints, the amount, the unit, and a NIP-17 transport to the seller.