From d81180ddf3c43f1b097e8c878a84d24062418d32 Mon Sep 17 00:00:00 2001 From: orveth Date: Thu, 23 Jul 2026 21:35:04 -0700 Subject: [PATCH 1/3] =?UTF-8?q?buyer:=20MCP/CLI=20connect-or-spawn=20?= =?UTF-8?q?=E2=80=94=20route=20money=20ops=20through=20the=20daemon=20sock?= =?UTF-8?q?et?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The MCP is now a thin client of the buyer daemon: post_job/get_job/collect/ award_claim forward over $MOBEE_HOME/buyer.sock (connect-or-spawn) instead of opening the wallet/budget/relay in-process. `mobee collect` (CLI) routes the same way, closing the in-process money bypass so the daemon is the single money owner (daemon-owns-home enforced, not convention). - crates/mobee/src/daemon.rs: connect-or-spawn client. Connects if a daemon is serving; else spawns `mobee buyer serve` detached (own process group, MOBEE_HOME pinned to the caller's home) and polls until it binds. A double-spawn is safe — the loser fails closed on the exclusive home lock, both callers reach the winner. - mcp.rs: thin router; McpState drops the in-process BudgetGate. post_job gains max_sats/harness/model inputs (forwarded); award_claim gains max_sats. - collect_cli.rs: routes collect over the daemon. - daemon post_job: contribution offers (all-or-nothing pins) — no silent from-scratch fallback now that the MCP forwards contribution mode; status carries pid. - tests/mcp_daemon.rs: connect-or-spawn (spawn + reuse + second-serve fails closed) and money-op-served-by-daemon-never-in-process (a failed collect burns zero). Co-Authored-By: Claude Opus 4.8 --- crates/mobee-core/src/buyer/mod.rs | 55 +++- crates/mobee/src/collect_cli.rs | 43 +-- crates/mobee/src/daemon.rs | 99 +++++++ crates/mobee/src/main.rs | 2 + crates/mobee/src/mcp.rs | 449 +++++++---------------------- crates/mobee/tests/mcp_daemon.rs | 175 +++++++++++ 6 files changed, 435 insertions(+), 388 deletions(-) create mode 100644 crates/mobee/src/daemon.rs create mode 100644 crates/mobee/tests/mcp_daemon.rs diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index 73181903..867d07f5 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -45,7 +45,7 @@ use crate::buyer_fund::{self, FundError}; use crate::collect::{self, CollectRequest}; use crate::home::{self, HomeError, MobeeHome}; use crate::job_lifecycle::{ - self, AwardClaimRequest, GetJobRequest, JobKind, PostJobRequest, WaitFor, + self, AwardClaimRequest, ContributionSpec, GetJobRequest, JobKind, PostJobRequest, WaitFor, }; use crate::payment::{PaymentMachine, PaymentRecord, PaymentState}; use lifecycle::{AwardError, AwardFilters, PaymentProgress, SettleError}; @@ -261,8 +261,9 @@ async fn dispatch(context: &BuyerContext, request: Request) -> Response { } } -/// Params for the `post_job` RPC — a from-scratch offer's fields (the daemon's default flow; -/// contribution offers stay on the CLI/MCP path). `job_id` returned is the offer event id. +/// Params for the `post_job` RPC. From-scratch by default; the four contribution pins +/// (`target_repo_owner`/`target_repo_url`/`base_branch`/`base_oid`) are ALL-OR-NOTHING — all four +/// select a contribution offer, a partial set is refused. `job_id` returned is the offer event id. #[derive(Debug, Deserialize)] struct PostJobParams { task: String, @@ -278,15 +279,56 @@ struct PostJobParams { repo: Option, #[serde(default)] branch: Option, + #[serde(default)] + target_repo_owner: Option, + #[serde(default)] + target_repo_url: Option, + #[serde(default)] + base_branch: Option, + #[serde(default)] + base_oid: Option, + #[serde(default)] + accepts: Option>, } -/// Publish a from-scratch offer (reuses [`job_lifecycle::post_job_async`], the same money-checked -/// post path the CLI/MCP use). No reservation is taken at post — funds are reserved at award. +/// Resolve the offer kind from the contribution pins: all four present ⇒ contribution; none ⇒ +/// from-scratch; a partial set is refused so the core never sees a half-specified contribution. +fn post_job_kind(params: &PostJobParams) -> Result { + match ( + ¶ms.target_repo_owner, + ¶ms.target_repo_url, + ¶ms.base_branch, + ¶ms.base_oid, + ) { + (None, None, None, None) => Ok(JobKind::FromScratch), + (Some(owner), Some(url), Some(branch), Some(oid)) => { + Ok(JobKind::Contribution(ContributionSpec { + target_repo_owner: owner.clone(), + target_repo_url: url.clone(), + base_branch: branch.clone(), + base_oid: oid.clone(), + accepts: params.accepts.clone(), + })) + } + _ => Err( + "post_job contribution mode requires ALL of target_repo_owner, target_repo_url, \ + base_branch, base_oid (a partial set is refused)" + .to_owned(), + ), + } +} + +/// Publish an offer (reuses [`job_lifecycle::post_job_async`], the same money-checked post path the +/// CLI/MCP use). No reservation is taken at post — funds are reserved at award. async fn post_job(context: &BuyerContext, 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 request = PostJobRequest { task: params.task, output: params.output, @@ -296,7 +338,7 @@ async fn post_job(context: &BuyerContext, id: Value, params: Value) -> Response deadline_unix: params.deadline_unix, repo: params.repo, branch: params.branch, - job: JobKind::FromScratch, + job, }; match job_lifecycle::post_job_async(&context.home, request).await { Ok(outcome) => Response::ok(id, json!(outcome)), @@ -668,6 +710,7 @@ async fn status(context: &BuyerContext, id: Value) -> Response { "version": crate::version(), "home": context.home.root.display().to_string(), "socket": context.home.root.join(SOCKET_FILE).display().to_string(), + "pid": std::process::id(), "pubkey": context.signer.public_key_hex(), "started_at_unix": context.started_at_unix, "wallet": wallet, diff --git a/crates/mobee/src/collect_cli.rs b/crates/mobee/src/collect_cli.rs index 87713c19..8775c28f 100644 --- a/crates/mobee/src/collect_cli.rs +++ b/crates/mobee/src/collect_cli.rs @@ -60,10 +60,13 @@ fn usage(err: &mut dyn Write) { } /// Entry from `cli::run` for `mobee collect ...`. +/// +/// Money ops are owned by the buyer daemon (exclusive home lock, single wallet + budget + reservation +/// ledger). This command routes `collect` over the daemon socket (connect-or-spawn) rather than +/// opening the wallet itself — so a CLI collect can never pay outside the daemon path or race a +/// concurrent MCP/daemon melt. #[cfg(feature = "wallet")] pub fn run(args: &[String], out: &mut dyn Write, err: &mut dyn Write) -> i32 { - use mobee_core::budget::BudgetGate; - use mobee_core::collect::{self, CollectRequest}; use mobee_core::home; let opts = match parse(args) { @@ -92,42 +95,18 @@ pub fn run(args: &[String], out: &mut dyn Write, err: &mut dyn Write) -> i32 { return RUNTIME_ERROR; } }; - let mut gate = match BudgetGate::from_home(&home) { - Ok(gate) => gate, - Err(error) => { - let _ = writeln!(err, "{error}"); - return RUNTIME_ERROR; - } - }; - let outcome = match collect::collect_blocking( - &home, - &mut gate, - CollectRequest { - job_id: opts.job_id, - out: opts.out, - }, - ) { - Ok(outcome) => outcome, - Err(error) => { - let _ = writeln!(err, "{error}"); + let params = serde_json::json!({ "job_id": opts.job_id, "out": opts.out }); + let body = match crate::daemon::ensure_then_call(&home, "collect", params) { + Ok(body) => body, + Err(message) => { + let _ = writeln!(err, "{message}"); return RUNTIME_ERROR; } }; - let body = serde_json::json!({ - "ok": true, - "amount_sats": outcome.pay.amount_sats, - "attempt_id": outcome.pay.attempt_id, - "spent_total_sats": outcome.pay.spent_total_sats, - "remaining_sats": outcome.pay.remaining_sats, - "state": outcome.pay.state, - "commit": outcome.commit_oid, - "path": outcome.path, - "files": outcome.files, - }); let rendered = body.to_string(); - // Defense in depth: never let the secret key appear on stdout. + // Defense in depth: never let the secret key appear on stdout (the daemon never returns it). if let Ok(secret) = home::read_secret_key_hex(&home) { if !secret.is_empty() && rendered.contains(&secret) { let _ = writeln!(err, "collect refused: response would echo secret key"); diff --git a/crates/mobee/src/daemon.rs b/crates/mobee/src/daemon.rs new file mode 100644 index 00000000..94c98eda --- /dev/null +++ b/crates/mobee/src/daemon.rs @@ -0,0 +1,99 @@ +//! Connect-or-spawn client to the per-home buyer daemon. +//! +//! The MCP and the CLI money commands never own the wallet, key, or budget ledger — the buyer +//! daemon does, guarded by the exclusive home lock. A caller that needs the daemon calls +//! [`ensure`], which connects to `$MOBEE_HOME/buyer.sock` if a daemon is already serving, or spawns +//! one (this same binary, `mobee buyer serve`, detached) and waits for it to come up. A concurrent +//! double-spawn is safe: the loser fails closed at the exclusive home lock and exits, the winner +//! binds the socket, and both callers connect to the winner. No manual `mobee buyer` command is +//! ever needed. + +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use mobee_core::buyer::{client, SOCKET_FILE}; +use mobee_core::home::MobeeHome; +use serde_json::Value; + +/// How long to wait for a freshly spawned daemon to bind its socket before giving up. Kept under the +/// MCP tool deadline so a cold start surfaces a clear error rather than a deadline timeout. +const SPAWN_READY_TIMEOUT: Duration = Duration::from_secs(10); +/// Poll cadence while waiting for the socket to answer. +const SPAWN_POLL_INTERVAL: Duration = Duration::from_millis(100); + +/// The daemon socket path for a home. +pub fn socket_path(home: &MobeeHome) -> PathBuf { + home.root.join(SOCKET_FILE) +} + +/// Connect to the home's buyer daemon, spawning it if none is serving. Returns the socket path a +/// thin client can call. Never runs a money op itself — it only guarantees a daemon owns the home. +pub fn ensure(home: &MobeeHome) -> Result { + let sock = socket_path(home); + if client::status(&sock).is_ok() { + return Ok(sock); // a daemon is already serving this home. + } + spawn_detached(&home.root)?; + // Poll until our daemon — or a racing winner's — answers, or time out. + let deadline = Instant::now() + SPAWN_READY_TIMEOUT; + loop { + if client::status(&sock).is_ok() { + return Ok(sock); + } + if Instant::now() >= deadline { + return Err(format!( + "buyer daemon did not come up on {} within {}s", + sock.display(), + SPAWN_READY_TIMEOUT.as_secs() + )); + } + std::thread::sleep(SPAWN_POLL_INTERVAL); + } +} + +/// Spawn this binary as a detached `mobee buyer serve`. Detached into its own process group so it +/// outlives the spawning session (a later session connects to the same daemon). A double-spawn is +/// safe — the loser fails closed at the exclusive home lock and exits. +fn spawn_detached(home_root: &Path) -> Result<(), String> { + let exe = std::env::current_exe() + .map_err(|error| format!("cannot resolve the mobee binary to spawn the buyer daemon: {error}"))?; + let mut command = Command::new(exe); + command + .arg("buyer") + .arg("serve") + // Pin the daemon to exactly this home so it serves the caller's home (default, MOBEE_HOME, + // or a --home path) rather than whatever the ambient env resolves to. + .env("MOBEE_HOME", home_root) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + // New process group: the daemon is not killed when the spawning session exits. + command.process_group(0); + } + command + .spawn() + .map(|_child| ()) + .map_err(|error| format!("failed to spawn the buyer daemon (`mobee buyer serve`): {error}")) +} + +/// Call a daemon method over the socket, returning its `result` value or a flattened error message. +/// The daemon owns the money authority; this is a pure transport. +pub fn call(sock: &Path, method: &str, params: Value) -> Result { + let response = client::call(sock, method, params).map_err(|error| error.to_string())?; + if let Some(error) = response.error { + return Err(error.message); + } + response + .result + .ok_or_else(|| format!("buyer daemon returned neither result nor error for {method}")) +} + +/// Ensure a daemon is serving `home`, then call `method` — the one-shot the MCP and CLI use. +pub fn ensure_then_call(home: &MobeeHome, method: &str, params: Value) -> Result { + let sock = ensure(home)?; + call(&sock, method, params) +} diff --git a/crates/mobee/src/main.rs b/crates/mobee/src/main.rs index 4be9c9b6..caef8224 100644 --- a/crates/mobee/src/main.rs +++ b/crates/mobee/src/main.rs @@ -3,6 +3,8 @@ mod agent_presets; mod buyer; mod cli; mod collect_cli; +#[cfg(feature = "wallet")] +mod daemon; mod doctor; mod exec; mod mcp; diff --git a/crates/mobee/src/mcp.rs b/crates/mobee/src/mcp.rs index 55044ffb..bf5c4757 100644 --- a/crates/mobee/src/mcp.rs +++ b/crates/mobee/src/mcp.rs @@ -8,20 +8,15 @@ //! returns a graceful tool-error — the server never exits. use std::io::{BufRead, Write}; -use std::sync::Mutex; use std::time::Duration; -use mobee_core::budget::BudgetGate; -#[cfg(feature = "wallet")] -use mobee_core::collect; use mobee_core::home::{self, MobeeHome}; -#[cfg(feature = "wallet")] -use mobee_core::job_lifecycle::{ - self, AwardClaimRequest, ContributionSpec, GetJobRequest, JobKind, PostJobRequest, WaitFor, -}; use serde::Deserialize; use serde_json::{Value, json}; +#[cfg(feature = "wallet")] +use crate::daemon; + const SUCCESS: i32 = 0; const RUNTIME_ERROR: i32 = 2; @@ -37,9 +32,10 @@ struct McpRequest { params: Value, } +/// The MCP owns no money authority — the buyer daemon does. The MCP only needs the home to resolve +/// the daemon socket (connect-or-spawn) and to run the never-echo-secret guard over daemon replies. struct McpState { home: MobeeHome, - gate: Mutex, } /// Run the MCP server on the provided stdio handles until stdin EOF. @@ -106,8 +102,7 @@ pub fn run(out: &mut dyn Write, err: &mut dyn Write) -> i32 { fn bootstrap_state() -> Result { let root = home::default_home_dir().map_err(|error| error.to_string())?; let home = home::bootstrap(root).map_err(|error| error.to_string())?; - let gate = Mutex::new(BudgetGate::from_home(&home).map_err(|error| error.to_string())?); - Ok(McpState { home, gate }) + Ok(McpState { home }) } #[cfg(test)] @@ -166,13 +161,26 @@ fn tools() -> Value { json!([ { "name": "post_job", - "description": "Publish a real mobee job offer (OFFER kind) to the configured mobee relay. Targeted seller p-tag is the documented default (pass seller_pubkey); set untargeted=true for an open offer. Optional repo+branch attach git delivery tags. CONTRIBUTION (freelance-PR) mode: supply target_repo_owner + target_repo_url + base_branch + base_oid to post a job-class=contribution offer against a repo you own (seller forks it and delivers a PR); these four are ALL-OR-NOTHING (a partial set is refused). Omit all four ⇒ from-scratch job (unchanged). Never echoes secrets.", + "description": "Publish a real mobee job offer (OFFER kind) to the configured mobee relay, then let the buyer daemon drive the award: once a payable seller claim appears the daemon auto-awards it under the hood, so the normal flow is just post_job then collect (two calls). max_sats caps what the daemon will commit to (defaults to amount_sats); it never auto-awards a claim it cannot pay. harness/model are recorded auto-award preferences. Targeted seller p-tag is the documented default (pass seller_pubkey); set untargeted=true for an open offer. Optional repo+branch attach git delivery tags. CONTRIBUTION (freelance-PR) mode: supply target_repo_owner + target_repo_url + base_branch + base_oid to post a job-class=contribution offer against a repo you own (seller forks it and delivers a PR); these four are ALL-OR-NOTHING (a partial set is refused). Omit all four ⇒ from-scratch job. Never echoes secrets.", "inputSchema": { "type": "object", "properties": { "task": { "type": "string" }, "output": { "type": "string", "description": "MIME / output type (e.g. text/plain)" }, "amount_sats": { "type": "integer", "minimum": 0 }, + "max_sats": { + "type": "integer", + "minimum": 0, + "description": "Per-job spend ceiling for the daemon's auto-award (defaults to amount_sats). A claim priced above it, or that the buyer cannot pay, is never auto-awarded." + }, + "harness": { + "type": "string", + "description": "Preferred seller harness (e.g. claude|cursor|codex). Recorded as an auto-award preference; not yet a hard filter (no claim wire field carries it)." + }, + "model": { + "type": "string", + "description": "Preferred seller model. Recorded as an auto-award preference; not yet a hard filter." + }, "seller_pubkey": { "type": "string", "description": "Targeted seller hex pubkey (documented default)" @@ -240,12 +248,17 @@ fn tools() -> Value { }, { "name": "award_claim", - "description": "Award a seller claim BEFORE work: publish the buyer AWARD (kind-3405, status=accepted) selecting one claim so that seller executes and every other claimant releases without spending compute. Verifies the claim is present, still processing, and (for a targeted offer) authored by the targeted seller. Returns quoted_mints — the mints the claim's creq will be paid at — so an incompatible award is visible before you commit. No pay-bind — settle after delivery with collect. Never echoes secrets.", + "description": "Manually award a specific seller claim (the fine-grain override of the daemon's auto-award): reserve the funds and publish the buyer AWARD (kind-3405, status=accepted) selecting that claim so the seller executes and every other claimant releases without spending compute. The daemon refuses to award a claim it cannot pay or whose price exceeds max_sats (defaults to the offer amount). No pay-bind — settle after delivery with collect. Never echoes secrets.", "inputSchema": { "type": "object", "properties": { "job_id": { "type": "string" }, - "claim_id": { "type": "string" } + "claim_id": { "type": "string" }, + "max_sats": { + "type": "integer", + "minimum": 0, + "description": "Spend ceiling for this manual award (defaults to the offer amount). A claim priced above it is refused." + } }, "required": ["job_id", "claim_id"], "additionalProperties": false @@ -260,43 +273,77 @@ async fn call_tool_async(state: &McpState, params: &Value) -> Result { - #[cfg(feature = "wallet")] - { - post_job_tool_async(state, &arguments).await - } - #[cfg(not(feature = "wallet"))] - { - post_job_tool(state, &arguments) - } - } - "get_job" => { - #[cfg(feature = "wallet")] - { - get_job_tool_async(state, &arguments).await - } - #[cfg(not(feature = "wallet"))] - { - get_job_tool(state, &arguments) - } + "post_job" => route_tool(state, "post_job", "post_job", arguments).await, + "get_job" => 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, + moved => Err(moved_tool_error(moved)), + } +} + +/// Route a trade-loop tool over the buyer daemon socket (connect-or-spawn). `tool` is the MCP tool +/// name (for errors/hints); `method` is the daemon RPC. The tool arguments are forwarded verbatim as +/// RPC params — the daemon is the sole authority for validation and money, so the MCP adds no +/// second-guessing. The daemon never returns the secret key; the never-echo guard is defense in +/// depth over its reply. +#[cfg(feature = "wallet")] +async fn route_tool( + state: &McpState, + tool: &str, + method: &str, + arguments: Value, +) -> Result { + let home = state.home.clone(); + let rpc = method.to_owned(); + // The daemon client is synchronous (a plain UnixStream) and may spawn+poll a daemon on cold + // start, so run it off the async reactor under the outer tool deadline. + let body = tokio::task::spawn_blocking(move || daemon::ensure_then_call(&home, &rpc, arguments)) + .await + .map_err(|error| format!("buyer daemon call task failed: {error}"))? + .map_err(|error| with_prereq_hint(tool, error))?; + guard_never_echo(state, tool, &body)?; + Ok(tool_ok(with_ok(body))) +} + +#[cfg(not(feature = "wallet"))] +async fn route_tool( + _state: &McpState, + tool: &str, + _method: &str, + _arguments: Value, +) -> Result { + Err(format!( + "{tool} requires the wallet feature (rebuild with --features wallet, on by default)" + )) +} + +/// Tag a daemon result object with `ok: true` for the MCP surface (a non-object result is wrapped). +fn with_ok(body: Value) -> Value { + match body { + Value::Object(mut map) => { + map.insert("ok".into(), json!(true)); + Value::Object(map) } - "collect" => { - #[cfg(feature = "wallet")] - { - collect_tool_async(state, &arguments).await - } - #[cfg(not(feature = "wallet"))] - { - let _ = arguments; - Err("collect requires the wallet feature".into()) - } + other => json!({ "ok": true, "result": other }), + } +} + +/// Defense in depth: refuse a daemon reply that would echo the buyer secret key. The daemon never +/// includes it, so this only ever fires on a bug. +#[cfg(feature = "wallet")] +fn guard_never_echo(state: &McpState, tool: &str, body: &Value) -> Result<(), String> { + if let Ok(secret) = home::read_secret_key_hex(&state.home) { + if !secret.is_empty() && body.to_string().contains(&secret) { + return Err(format!("{tool} refused: response would echo secret key")); } - "award_claim" => award_claim_tool_async(state, &arguments).await, - moved => Err(moved_tool_error(moved)), } + Ok(()) } /// Actionable error for a tool that moved off the MCP surface to the `mobee` CLI, or an unknown @@ -348,236 +395,6 @@ fn with_prereq_hint(tool: &str, error: String) -> String { } } -#[cfg(feature = "wallet")] -async fn collect_tool_async(state: &McpState, arguments: &Value) -> Result { - let job_id = arguments - .get("job_id") - .and_then(Value::as_str) - .ok_or_else(|| "collect requires job_id".to_owned())? - .to_owned(); - let out = arguments - .get("out") - .and_then(Value::as_str) - .map(str::to_owned); - - let mut gate = state - .gate - .lock() - .map_err(|_| "budget gate lock poisoned".to_owned())?; - let outcome = collect::collect_async( - &state.home, - &mut gate, - collect::CollectRequest { job_id, out }, - ) - .await - .map_err(|error| with_prereq_hint("collect", error.to_string()))?; - - let body = json!({ - "ok": true, - "amount_sats": outcome.pay.amount_sats, - "attempt_id": outcome.pay.attempt_id, - "spent_total_sats": outcome.pay.spent_total_sats, - "remaining_sats": outcome.pay.remaining_sats, - "per_job_cap_sats": gate.per_job_cap(), - "total_cap_sats": gate.total_cap(), - "state": outcome.pay.state, - "commit": outcome.commit_oid, - "path": outcome.path, - "files": outcome.files, - }); - Ok(tool_ok(body)) -} - -#[cfg(feature = "wallet")] -async fn post_job_tool_async(state: &McpState, arguments: &Value) -> Result { - let require_str = |key: &str| -> Result { - arguments - .get(key) - .and_then(Value::as_str) - .map(str::to_owned) - .ok_or_else(|| format!("post_job requires {key} (string)")) - }; - let amount_sats = arguments - .get("amount_sats") - .and_then(Value::as_u64) - .ok_or_else(|| "post_job requires amount_sats (integer)".to_owned())?; - let untargeted = arguments - .get("untargeted") - .and_then(Value::as_bool) - .unwrap_or(false); - let seller_pubkey = arguments - .get("seller_pubkey") - .and_then(Value::as_str) - .map(str::to_owned); - let deadline_unix = match arguments.get("deadline_unix") { - Some(value) => Some( - value - .as_u64() - .ok_or_else(|| "post_job requires deadline_unix (integer >= 0)".to_owned())?, - ), - None => None, - }; - let repo = arguments - .get("repo") - .and_then(Value::as_str) - .map(str::to_owned); - let branch = arguments - .get("branch") - .and_then(Value::as_str) - .map(str::to_owned); - let opt_str = |key: &str| -> Option { - arguments.get(key).and_then(Value::as_str).map(str::to_owned) - }; - let accepts = arguments.get("accepts").and_then(Value::as_array).map(|values| { - values - .iter() - .filter_map(|value| value.as_str().map(str::to_owned)) - .collect::>() - }); - - // The four target/base pins are ALL-OR-NOTHING: all four ⇒ a contribution offer; none ⇒ - // from-scratch; a partial set is refused so the core never sees a half-specified contribution. - let job = match ( - opt_str("target_repo_owner"), - opt_str("target_repo_url"), - opt_str("base_branch"), - opt_str("base_oid"), - ) { - (None, None, None, None) => JobKind::FromScratch, - (Some(target_repo_owner), Some(target_repo_url), Some(base_branch), Some(base_oid)) => { - JobKind::Contribution(ContributionSpec { - target_repo_owner, - target_repo_url, - base_branch, - base_oid, - accepts, - }) - } - _ => { - return Err( - "post_job contribution mode requires ALL of target_repo_owner, target_repo_url, \ - base_branch, base_oid (a partial set is refused)" - .into(), - ) - } - }; - - let outcome = job_lifecycle::post_job_async( - &state.home, - PostJobRequest { - task: require_str("task")?, - output: require_str("output")?, - amount_sats, - seller_pubkey, - untargeted, - deadline_unix, - repo, - branch, - job, - }, - ) - .await - .map_err(|error| with_prereq_hint("post_job", error.to_string()))?; - - let body = json!({ - "ok": true, - "job_id": outcome.job_id, - "job_hash": outcome.job_hash, - "offer_kind": outcome.offer_kind, - "targeted": outcome.targeted, - "seller_pubkey": outcome.seller_pubkey, - "amount_sats": outcome.amount_sats, - "relay_url": outcome.relay_url, - "task": outcome.task, - "output": outcome.output, - }); - // never-echo: secret key must not appear - let rendered = body.to_string(); - if let Ok(secret) = home::read_secret_key_hex(&state.home) { - if rendered.contains(&secret) { - return Err("post_job refused: response would echo secret key".into()); - } - } - Ok(tool_ok(body)) -} - -#[cfg(not(feature = "wallet"))] -fn post_job_tool(_state: &McpState, _arguments: &Value) -> Result { - Err("post_job requires the wallet feature".into()) -} - -#[cfg(feature = "wallet")] -async fn get_job_tool_async(state: &McpState, arguments: &Value) -> Result { - let job_id = arguments - .get("job_id") - .and_then(Value::as_str) - .ok_or_else(|| "get_job requires job_id (string)".to_owned())? - .to_owned(); - let wait_for = match arguments.get("wait_for").and_then(Value::as_str) { - Some(raw) => Some(WaitFor::parse(raw).map_err(|error| error.to_string())?), - None => None, - }; - let timeout_secs = arguments.get("timeout_secs").and_then(Value::as_u64); - let include_display_names = arguments - .get("include_display_names") - .and_then(Value::as_bool) - .unwrap_or(false); - let view = job_lifecycle::get_job_async( - &state.home, - GetJobRequest { - job_id, - wait_for, - timeout_secs, - include_display_names, - }, - ) - .await - .map_err(|error| error.to_string())?; - let body = serde_json::to_value(&view).map_err(|error| error.to_string())?; - let mut body = body; - if let Some(obj) = body.as_object_mut() { - obj.insert("ok".into(), json!(true)); - obj.insert("source".into(), json!("relay")); - if view.pending { - obj.insert("status".into(), json!("pending")); - } - } - Ok(tool_ok(body)) -} - -#[cfg(not(feature = "wallet"))] -fn get_job_tool(_state: &McpState, _arguments: &Value) -> Result { - Err("get_job requires the wallet feature".into()) -} - -#[cfg(feature = "wallet")] -async fn award_claim_tool_async(state: &McpState, arguments: &Value) -> Result { - let require_str = |key: &str| -> Result { - arguments - .get(key) - .and_then(Value::as_str) - .map(str::to_owned) - .ok_or_else(|| format!("award_claim requires {key} (string)")) - }; - let outcome = job_lifecycle::award_claim_async( - &state.home, - AwardClaimRequest { - job_id: require_str("job_id")?, - claim_id: require_str("claim_id")?, - }, - ) - .await - .map_err(|error| error.to_string())?; - Ok(tool_ok(json!({ - "ok": true, - "award_event_id": outcome.award_event_id, - "job_id": outcome.job_id, - "claim_id": outcome.claim_id, - "seller_pubkey": outcome.seller_pubkey, - "quoted_mints": outcome.quoted_mints, - }))) -} - fn tool_ok(body: Value) -> Value { json!({ "content": [{ "type": "text", "text": body.to_string() }], @@ -675,8 +492,7 @@ mod tests { fn state_at(root: &std::path::Path) -> McpState { let home = home::bootstrap(root).expect("bootstrap"); - let gate = Mutex::new(BudgetGate::from_home(&home).expect("gate")); - McpState { home, gate } + McpState { home } } // #98: get_result materializes a PAID delivery's files to /results/ and returns @@ -785,75 +601,8 @@ mod tests { assert_eq!(response["result"]["isError"], true); } - #[cfg(feature = "wallet")] - #[test] - fn post_job_error_path_never_echoes_secret() { - let root = temp_home("post-echo"); - let _ = std::fs::remove_dir_all(&root); - let state = state_at(&root); - let secret = home::read_secret_key_hex(&state.home).expect("secret"); - let response = dispatch( - &state, - &McpRequest { - id: Some(json!(41)), - method: "tools/call".into(), - params: json!({ - "name": "post_job", - "arguments": { - "task": "gate-e", - "output": "text/plain", - "amount_sats": 1 - } - }), - }, - ); - let rendered = response.to_string(); - assert!(!rendered.contains(&secret), "secret leaked on post_job error"); - assert_eq!(response["result"]["isError"], true); - let message = response["result"]["content"][0]["text"] - .as_str() - .unwrap_or("") - .to_ascii_lowercase(); - assert!( - message.contains("seller_pubkey") || message.contains("untargeted"), - "message={message}" - ); - } - - #[cfg(feature = "wallet")] - #[test] - fn post_job_mcp_refuses_zero_deadline_before_publish() { - let root = temp_home("post-zero-deadline"); - let _ = std::fs::remove_dir_all(&root); - let state = state_at(&root); - let secret = home::read_secret_key_hex(&state.home).expect("secret"); - let response = dispatch( - &state, - &McpRequest { - id: Some(json!(42)), - method: "tools/call".into(), - params: json!({ - "name": "post_job", - "arguments": { - "task": "deadline-gate", - "output": "text/plain", - "amount_sats": 1, - "seller_pubkey": "aa".repeat(32), - "deadline_unix": 0 - } - }), - }, - ); - let rendered = response.to_string(); - assert!(!rendered.contains(&secret), "secret leaked on post_job error"); - assert_eq!(response["result"]["isError"], true); - let message = response["result"]["content"][0]["text"] - .as_str() - .unwrap_or(""); - assert!(message.contains("deadline_unix"), "message={message}"); - assert!(message.contains("given=0"), "message={message}"); - assert!(message.contains("current="), "message={message}"); - let _ = std::fs::remove_dir_all(&root); - } - + // post_job/collect/award business validation (targeting, zero/past deadline, contribution pins, + // over-budget, integrity+pay) now lives in the buyer daemon and mobee-core, tested there + // (job_lifecycle::post_job_* , buyer::* , collect_integrity). The MCP is a thin router; its own + // tests cover the transport surface (framing, tool list, moved-tool pointers, never-echo). } diff --git a/crates/mobee/tests/mcp_daemon.rs b/crates/mobee/tests/mcp_daemon.rs new file mode 100644 index 00000000..51020149 --- /dev/null +++ b/crates/mobee/tests/mcp_daemon.rs @@ -0,0 +1,175 @@ +//! Connect-or-spawn + daemon-owned money path (the #126/#127/#131 acceptance). +//! +//! A fresh home with ONLY the MCP/CLI configured — zero manual daemon commands — must transparently +//! start the buyer daemon on first use, reuse the same daemon for later sessions, and serve every +//! money op through it (never in-process). These drive the real `mobee` binary. +//! +//! Relay is pinned to a dead loopback (`MOBEE_RELAY_URL`) so a routed trade op fails fast and the +//! test stays hermetic — the point under test is the daemon boundary, not a live trade. +#![cfg(all(unix, feature = "wallet"))] + +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use serde_json::Value; + +static NEXT: AtomicU64 = AtomicU64::new(0); + +fn temp_home(label: &str) -> PathBuf { + let id = NEXT.fetch_add(1, Ordering::SeqCst); + std::env::temp_dir().join(format!("mobee-mcp-daemon-{label}-{}-{id}", std::process::id())) +} + +/// A `mobee` command pinned to `home` with a dead relay (fast, network-free relay failures). +fn mobee(home: &Path) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_mobee")); + command + .env("MOBEE_HOME", home) + .env("MOBEE_RELAY_URL", "ws://127.0.0.1:1"); + command +} + +/// Run `mobee ` to completion; returns (exit code, stdout, stderr). +fn run(home: &Path, args: &[&str]) -> (i32, String, String) { + let output = mobee(home).args(args).output().expect("spawn mobee"); + ( + output.status.code().unwrap_or(-1), + String::from_utf8_lossy(&output.stdout).into_owned(), + String::from_utf8_lossy(&output.stderr).into_owned(), + ) +} + +/// The daemon's `status` result object, or `None` while no daemon answers. +fn status(home: &Path) -> Option { + let (code, stdout, _stderr) = run(home, &["buyer", "status"]); + if code != 0 { + return None; + } + let response: Value = serde_json::from_str(stdout.trim()).ok()?; + response.get("result").cloned() +} + +/// Poll `buyer status` until a daemon answers or the deadline passes. +fn wait_for_daemon(home: &Path) -> Option { + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if let Some(result) = status(home) { + return Some(result); + } + if Instant::now() >= deadline { + return None; + } + std::thread::sleep(Duration::from_millis(100)); + } +} + +/// Terminate a daemon by pid (test teardown; exact-pid, never a broad pattern kill). +fn kill(pid: u64) { + let _ = Command::new("kill").arg("-TERM").arg(pid.to_string()).status(); +} + +/// No daemon → a routed CLI money op connect-or-spawns one → a later session reuses the SAME daemon, +/// and a direct second `buyer serve` fails closed on the exclusive home lock. +#[test] +fn connect_or_spawn_starts_then_reuses_one_daemon() { + let home = temp_home("spawn"); + let _ = std::fs::remove_dir_all(&home); + + // Nothing running yet: the thin-client status has no daemon to talk to. + assert!(status(&home).is_none(), "no daemon should be running before first use"); + + // A routed money op (collect) must start the daemon under the hood — zero manual commands. The + // collect itself fails (bogus job, dead relay), but the daemon must come up to serve it. + let (collect_code, _out, _err) = run(&home, &["collect", &"a".repeat(64)]); + assert_ne!(collect_code, 0, "collect on a bogus job must fail"); + + let first = wait_for_daemon(&home).expect("connect-or-spawn must start the daemon"); + let pid = first["pid"].as_u64().expect("status carries a pid"); + let pubkey = first["pubkey"].as_str().expect("status carries a pubkey").to_owned(); + assert!(home.join("buyer.sock").exists(), "daemon bound its socket"); + + // A second session finds the daemon already serving and reuses it — same pid, same identity. + let second = wait_for_daemon(&home).expect("daemon still serving"); + assert_eq!(second["pid"].as_u64().unwrap(), pid, "second session reuses the same daemon"); + assert_eq!(second["pubkey"].as_str().unwrap(), pubkey, "same daemon identity"); + + // The exclusive home lock holds: a direct second `buyer serve` fails closed rather than becoming + // a second money owner. + let mut second_serve = mobee(&home) + .args(["buyer", "serve"]) + .spawn() + .expect("spawn second buyer serve"); + let exit = wait_child(&mut second_serve, Duration::from_secs(8)); + assert!( + matches!(exit, Some(status) if !status.success()), + "a second `buyer serve` on the same home must fail closed on the lock" + ); + + kill(pid); + let _ = std::fs::remove_dir_all(&home); +} + +/// A money op is served by the daemon, never in-process: after a routed collect the daemon owns the +/// home lock + socket, and a collect with no delivery burns nothing (no payment journal / results). +#[test] +fn money_op_is_served_by_the_daemon_never_in_process() { + let home = temp_home("served"); + let _ = std::fs::remove_dir_all(&home); + + // Before: no daemon artifacts at all. + assert!(!home.join("buyer.lock").exists(), "no lock before first use"); + assert!(!home.join("buyer.sock").exists(), "no socket before first use"); + + // A CLI collect on an undelivered job routes to the daemon and fails (nothing to pay). + let job = "b".repeat(64); + let (code, stdout, _stderr) = run(&home, &["collect", &job]); + assert_ne!(code, 0, "collect with no delivery must fail"); + + // It was SERVED BY THE DAEMON, not run in-process: a daemon now owns the home lock + socket. Had + // collect opened the wallet in-process, no daemon (and no held lock) would exist afterward. + let result = wait_for_daemon(&home).expect("the money op brought up a daemon to serve it"); + let pid = result["pid"].as_u64().expect("pid"); + assert!(home.join("buyer.lock").exists(), "daemon holds the exclusive home lock"); + assert!(home.join("buyer.sock").exists(), "daemon owns the socket"); + + // The failed collect burned nothing: no payment journal, no materialized results. + assert!( + !home.join("payment-journal").exists(), + "a failed collect must write no payment journal (zero spend)" + ); + assert!( + !home.join("results").join(&job).exists(), + "a failed collect must materialize nothing" + ); + + // Never-echo: the buyer secret must not appear on the CLI output. + let secret_path = home.join("key"); + if let Ok(secret) = std::fs::read_to_string(&secret_path) { + let secret = secret.trim(); + if !secret.is_empty() { + assert!(!stdout.contains(secret), "collect output must not echo the secret key"); + } + } + + kill(pid); + let _ = std::fs::remove_dir_all(&home); +} + +/// Wait up to `timeout` for a child to exit; returns its status, or `None` if still running (then +/// kills it so it never leaks). +fn wait_child(child: &mut std::process::Child, timeout: Duration) -> Option { + let deadline = Instant::now() + timeout; + loop { + match child.try_wait().expect("try_wait") { + Some(status) => return Some(status), + None if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + return None; + } + None => std::thread::sleep(Duration::from_millis(50)), + } + } +} From c02d2f9940d36b785300e0a8667dfc0134dfc966 Mon Sep 17 00:00:00 2001 From: orveth Date: Thu, 23 Jul 2026 21:46:16 -0700 Subject: [PATCH 2/3] buyer: apply max_sats on manual award + surface reconcile report MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two #136 wiring obligations: - Manual award (named claim_id) now applies the SAME hard filters as auto-award via lifecycle::named_claim_awardable — max_sats, price, mint, liveness — so max_sats is enforced, not dead input, on the manual path. Refuses OverMax / NotFound / NotLive / Unpayable with a structured reason. - reconcile-on-start no longer discards its ReconcileReport: it is kept on the daemon context and surfaced in `status.reconcile` {released, converted, kept} so kept-uncertain reservations (funds committed to an ambiguous payment) are visible. The reserved-job → disposition mapping is factored into a pure `plan_reconcile`. A reserved job with a local delivery bind stays payable even if the relay expired its events (a delivered job is not dead — cf #140). Tests: named_claim_awardable (5 cases), scan_payment_progress (absent / intent folds to None / unfoldable folds to Uncertain — the reconcile fail-safe), plan_reconcile mapping + conservative default, merge_progress ordering. Co-Authored-By: Claude Opus 4.8 --- crates/mobee-core/src/buyer/lifecycle.rs | 127 +++++++++++ crates/mobee-core/src/buyer/mod.rs | 256 ++++++++++++++++++++--- 2 files changed, 349 insertions(+), 34 deletions(-) diff --git a/crates/mobee-core/src/buyer/lifecycle.rs b/crates/mobee-core/src/buyer/lifecycle.rs index ea24b4c9..f00635b6 100644 --- a/crates/mobee-core/src/buyer/lifecycle.rs +++ b/crates/mobee-core/src/buyer/lifecycle.rs @@ -55,6 +55,69 @@ pub fn select_awardable_claim(view: &JobView, filters: &AwardFilters) -> Option< .map(|claim| claim.claim_id.clone()) } +/// Why a specifically-named (manual) award was refused. The manual path names a `claim_id` instead +/// of auto-selecting, so it must apply the SAME hard filters `select_awardable_claim` applies — +/// otherwise `max_sats` and mint/price compatibility would be dead input on the manual path. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NamedAwardRefused { + /// The offer price exceeds the buyer's `max_sats` ceiling for this award. + OverMax { offer_amount_sats: u64, max_sats: u64 }, + /// No claim with that id is on the relay for this job. + NotFound { claim_id: String }, + /// The named claim is not live (expired / superseded / past deadline) — nothing to award. + NotLive { claim_id: String }, + /// The named claim cannot be paid (missing/malformed creq, price ≠ offer amount, wrong unit, or + /// no mutually-payable mint) — awarding it would commit to something the buyer cannot settle. + Unpayable { claim_id: String }, +} + +impl std::fmt::Display for NamedAwardRefused { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::OverMax { offer_amount_sats, max_sats } => write!( + formatter, + "award refused: offer price {offer_amount_sats} sat exceeds max_sats {max_sats}" + ), + Self::NotFound { claim_id } => write!(formatter, "award refused: claim {claim_id} not found for this job"), + Self::NotLive { claim_id } => write!(formatter, "award refused: claim {claim_id} is not live"), + Self::Unpayable { claim_id } => write!( + formatter, + "award refused: claim {claim_id} is not payable (price/mint/creq incompatible — the buyer could not settle it)" + ), + } + } +} + +impl std::error::Error for NamedAwardRefused {} + +/// Verify a specifically-named claim is awardable under the hard filters — the manual-award +/// counterpart of [`select_awardable_claim`], so `max_sats` and mint/price compatibility are applied +/// on the manual path rather than ignored. Pure: relay truth + filters in, verdict out. +pub fn named_claim_awardable( + view: &JobView, + claim_id: &str, + filters: &AwardFilters, +) -> Result<(), NamedAwardRefused> { + if filters.offer_amount_sats > filters.max_sats { + return Err(NamedAwardRefused::OverMax { + offer_amount_sats: filters.offer_amount_sats, + max_sats: filters.max_sats, + }); + } + let claim = view + .claims + .iter() + .find(|claim| claim.claim_id == claim_id) + .ok_or_else(|| NamedAwardRefused::NotFound { claim_id: claim_id.to_owned() })?; + if !claim.live { + return Err(NamedAwardRefused::NotLive { claim_id: claim_id.to_owned() }); + } + if !claim_is_payable(&view.job_id, claim.creq.as_deref(), filters) { + return Err(NamedAwardRefused::Unpayable { claim_id: claim_id.to_owned() }); + } + Ok(()) +} + /// True when a claim's `creq` is present, well-formed, priced at the offer amount within the /// budget ceiling, denominated in sats for this job, and quotes a mint the buyer can pay from. fn claim_is_payable(job_id: &str, creq: Option<&str>, filters: &AwardFilters) -> bool { @@ -349,6 +412,70 @@ mod tests { assert_eq!(select_awardable_claim(&view, &filters(10, 100)), None); } + // MANUAL-AWARD max_sats tooth: a named claim applies the SAME hard filters as auto-award, so + // max_sats is enforced (not dead input) on the manual path. + #[test] + fn named_claim_awardable_accepts_live_payable_within_max() { + let job = "a".repeat(64); + let claim_id = "c".repeat(64); + let view = view_with(&job, 10, vec![claim(&job, true, 10, &[DEFAULT_MINT_URL.into()])]); + assert_eq!(named_claim_awardable(&view, &claim_id, &filters(10, 100)), Ok(())); + } + + // Over the ceiling: a manual award of a claim whose offer price exceeds max_sats is refused — + // the check that was missing (max_sats was ignored) on the manual path. + #[test] + fn named_claim_over_max_sats_refused() { + let job = "a".repeat(64); + let claim_id = "c".repeat(64); + let view = view_with(&job, 50, vec![claim(&job, true, 50, &[DEFAULT_MINT_URL.into()])]); + assert_eq!( + named_claim_awardable(&view, &claim_id, &filters(50, 40)), + Err(NamedAwardRefused::OverMax { offer_amount_sats: 50, max_sats: 40 }) + ); + } + + // A named claim that is not on the relay is refused as NotFound. + #[test] + fn named_claim_not_found_refused() { + let job = "a".repeat(64); + let view = view_with(&job, 10, vec![claim(&job, true, 10, &[DEFAULT_MINT_URL.into()])]); + let missing = "d".repeat(64); + assert_eq!( + named_claim_awardable(&view, &missing, &filters(10, 100)), + Err(NamedAwardRefused::NotFound { claim_id: missing }) + ); + } + + // A named but non-live claim is refused (nothing live to award). + #[test] + fn named_claim_not_live_refused() { + let job = "a".repeat(64); + let claim_id = "c".repeat(64); + let view = view_with(&job, 10, vec![claim(&job, false, 10, &[DEFAULT_MINT_URL.into()])]); + assert_eq!( + named_claim_awardable(&view, &claim_id, &filters(10, 100)), + Err(NamedAwardRefused::NotLive { claim_id }) + ); + } + + // A named claim quoting only a mint the buyer cannot settle at is refused as Unpayable — the + // manual path never awards a claim it cannot pay. + #[test] + fn named_claim_unpayable_mint_refused() { + let job = "a".repeat(64); + let claim_id = "c".repeat(64); + let view = view_with( + &job, + 10, + vec![claim(&job, true, 10, &["https://foreign.testnut.example".into()])], + ); + assert_eq!( + named_claim_awardable(&view, &claim_id, &filters(10, 100)), + Err(NamedAwardRefused::Unpayable { claim_id }) + ); + } + // AWARD-REFUSED tooth: when the reservation is refused, `publish` is NEVER called and NO // reservation row is written. Red-on-revert: reserving AFTER the publish would fire the // publish closure here (the flag flips), failing the "publish must not run" assertion. diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index 867d07f5..d9c322d7 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -51,7 +51,7 @@ use crate::payment::{PaymentMachine, PaymentRecord, PaymentState}; use lifecycle::{AwardError, AwardFilters, PaymentProgress, SettleError}; use lock::{HomeLock, LockError}; use protocol::{CODE_INTERNAL, CODE_METHOD_NOT_FOUND, CODE_NOT_IMPLEMENTED, Request, Response}; -use reservations::Dispositions; +use reservations::{Dispositions, ReconcileReport}; use signer::SignerHandle; use store::{BuyerStore, StoreError}; use wallet_actor::WalletHandle; @@ -124,6 +124,9 @@ struct BuyerContext { /// reservation's balance/spent snapshot is never read while a concurrent collect is melting. /// The wallet actor's balance reads run independently (reads never race a serialized send). money_lock: Mutex<()>, + /// The last reconcile-on-start report, surfaced in `status` so kept-uncertain reservations + /// (funds committed to an ambiguous payment) are visible rather than silently discarded. + last_reconcile: Mutex>, } fn now_unix() -> i64 { @@ -161,6 +164,7 @@ async fn bootstrap(home: MobeeHome) -> Result<(HomeLock, Arc, Path signer, started_at_unix, money_lock: Mutex::new(()), + last_reconcile: Mutex::new(None), }); Ok((lock, context, socket_path)) } @@ -191,8 +195,11 @@ pub async fn run(home: MobeeHome) -> Result<(), BuyerError> { // daemon starts from a converged ledger. A failure is logged, not fatal — an unreachable relay // must not keep the daemon from coming up (the stale reservation is conservative until the next // reconcile). - if let Err(error) = reconcile_on_start(&context).await { - eprintln!("buyer: reconcile-on-start did not complete ({error}); serving with the ledger as-is"); + match reconcile_on_start(&context).await { + Ok(report) => *context.last_reconcile.lock().await = Some(report), + Err(error) => eprintln!( + "buyer: reconcile-on-start did not complete ({error}); serving with the ledger as-is" + ), } let listener = bind_socket(&socket_path)?; accept_loop(listener, context).await @@ -426,28 +433,33 @@ async fn award(context: &BuyerContext, id: Value, params: Value) -> Response { }; let offer_amount = offer.amount_sats; let max_sats = params.max_sats.unwrap_or(offer_amount); + let filters = AwardFilters { + offer_amount_sats: offer_amount, + max_sats, + buyer_mint: context.home.config.default_mint(), + allow_real_mints: context.home.config.allow_real_mints, + }; - // Manual award names the claim; auto-award selects the first payable one. + // Manual award names the claim but applies the SAME hard filters (max_sats, price, mint) as + // auto-award — max_sats is enforced, not ignored, on the manual path. Auto-award selects the + // first live payable claim. let claim_id = match params.claim_id { - Some(claim_id) => claim_id, - None => { - let filters = AwardFilters { - offer_amount_sats: offer_amount, - max_sats, - buyer_mint: context.home.config.default_mint(), - allow_real_mints: context.home.config.allow_real_mints, - }; - match lifecycle::select_awardable_claim(&view, &filters) { - Some(claim_id) => claim_id, - None => { - return Response::err( - id, - CODE_REFUSED, - format!("no awardable claim for job {} (none live/payable/mint-compatible)", params.job_id), - ); - } + Some(claim_id) => { + if let Err(refused) = lifecycle::named_claim_awardable(&view, &claim_id, &filters) { + return Response::err(id, CODE_REFUSED, refused.to_string()); } + claim_id } + None => match lifecycle::select_awardable_claim(&view, &filters) { + Some(claim_id) => claim_id, + None => { + return Response::err( + id, + CODE_REFUSED, + format!("no awardable claim for job {} (none live/payable/mint-compatible)", params.job_id), + ); + } + }, }; let (balance, total_cap, spent) = match money_snapshot(context).await { @@ -574,27 +586,31 @@ fn buyer_keys(home: &MobeeHome) -> Result { /// a `Closed` attempt is converted to `spent`; an ambiguous (Sent-not-Closed) payment is KEPT (the /// phase-3 saga owns it). Pure classification is [`lifecycle::classify_disposition`]; this gathers /// its inputs and applies the batch through [`BuyerStore::reconcile`]. -async fn reconcile_on_start(context: &BuyerContext) -> Result<(), String> { +async fn reconcile_on_start(context: &BuyerContext) -> Result { let reserved = context .store .reserved_job_ids() .map_err(|error| error.to_string())?; if reserved.is_empty() { - return Ok(()); + return Ok(ReconcileReport::default()); } let keys = buyer_keys(&context.home)?; let progress = scan_payment_progress(&context.home); - let mut dispositions: Dispositions = BTreeMap::new(); - for job_id in reserved { - let payment = progress.get(&job_id).copied().unwrap_or(PaymentProgress::None); - // Relay liveness: is a claim still live/deliverable for this job? An unreachable relay is - // treated as still-payable (conservative — never release a reservation we cannot verify is - // dead). A Closed/Uncertain payment ignores liveness in the classifier anyway. - let claim_payable = match job_lifecycle::fetch_job_view_async( + // Gather each reserved job's "still payable" signal (the only I/O). A job is payable if a claim + // is still live on the relay OR a local delivery bind exists (#140: a delivered job whose relay + // events expired is not dead — its bind + retained git objects still let collect settle it). An + // unreachable relay is treated as still-payable (conservative — never release what we cannot + // verify is dead). + let mut payable: BTreeMap = BTreeMap::new(); + for job_id in &reserved { + let has_bind = job_lifecycle::load_accepted_bind(&context.home, job_id) + .map(|bind| bind.is_some()) + .unwrap_or(false); + let claim_live = match job_lifecycle::fetch_job_view_async( &context.home, &keys, - &job_id, + job_id, RELAY_TIMEOUT, now_unix() as u64, ) @@ -603,14 +619,35 @@ async fn reconcile_on_start(context: &BuyerContext) -> Result<(), String> { Ok(view) => view.live_claim_id.is_some(), Err(_) => true, }; - dispositions.insert(job_id, lifecycle::classify_disposition(payment, claim_payable)); + payable.insert(job_id.clone(), claim_live || has_bind); } + let dispositions = plan_reconcile(&reserved, &progress, &payable); context .store .reconcile(&dispositions, now_unix()) - .map_err(|error| error.to_string())?; - Ok(()) + .map_err(|error| error.to_string()) +} + +/// Pure reconcile planning: map each reserved job to a disposition from its folded payment progress +/// and whether it is still payable. Kept pure (no relay/disk I/O) so the reserved-job → disposition +/// mapping is exhaustively testable; [`reconcile_on_start`] gathers the inputs. A job absent from +/// `payable` defaults to payable (conservative — never release without positive evidence of death). +fn plan_reconcile( + reserved: &[String], + progress: &BTreeMap, + payable: &BTreeMap, +) -> Dispositions { + let mut dispositions: Dispositions = BTreeMap::new(); + for job_id in reserved { + let payment = progress.get(job_id).copied().unwrap_or(PaymentProgress::None); + let claim_payable = payable.get(job_id).copied().unwrap_or(true); + dispositions.insert( + job_id.clone(), + lifecycle::classify_disposition(payment, claim_payable), + ); + } + dispositions } /// Fold every payment-journal attempt under the home into a `job_id → progress` map. Each record @@ -703,6 +740,16 @@ async fn status(context: &BuyerContext, id: Value) -> Response { Err(error) => json!({ "mint": mint, "error": error.to_string() }), }; + // Surface the last reconcile-on-start outcome so kept-uncertain reservations (funds committed to + // an ambiguous payment the crash-safe saga still owns) are visible, not silently discarded. + let reconcile = context.last_reconcile.lock().await.as_ref().map(|report| { + json!({ + "released": report.released, + "converted": report.converted, + "kept": report.kept, + }) + }); + Response::ok( id, json!({ @@ -718,6 +765,7 @@ async fn status(context: &BuyerContext, id: Value) -> Response { "schema_version": schema_version, "jobs": jobs, }, + "reconcile": reconcile, }), ) } @@ -818,4 +866,144 @@ mod tests { let _ = std::fs::remove_dir_all(&root); } + + // ---- reconcile plumbing (scan_payment_progress / plan_reconcile / merge_progress) ---------- + + use std::str::FromStr; + use crate::payment::{ + DeliveryIntegrityHash, JobHash, JobId, PaymentKey, PaymentRecord, PaymentState, PaymentTerms, + ResultId, + }; + + fn payment_key(job_id: &str) -> PaymentKey { + let cashu_pk = cashu::SecretKey::from_slice(&[7u8; 32]).expect("secret").public_key(); + let nostr_pk = nostr_sdk::PublicKey::from_hex(&cashu_pk.to_string()[2..]).expect("nostr pk"); + let terms = PaymentTerms::new( + cashu::MintUrl::from_str("https://testnut.cashu.space").expect("mint"), + cashu::Amount::from(7), + cashu::CurrencyUnit::Sat, + nostr_pk, + cashu_pk, + ); + PaymentKey::new( + JobId::new(job_id).expect("job id"), + ResultId::new("result").expect("result id"), + DeliveryIntegrityHash::from_hex("11".repeat(32)).expect("dih"), + JobHash::from_hex("22".repeat(32)).expect("job hash"), + &terms, + None, + ) + } + + fn write_journal(root: &std::path::Path, filename: &str, records: &[PaymentRecord]) { + let dir = root.join("payment-journal"); + std::fs::create_dir_all(&dir).expect("journal dir"); + let mut body = String::new(); + for record in records { + body.push_str(&serde_json::to_string(record).expect("record json")); + body.push('\n'); + } + std::fs::write(dir.join(filename), body).expect("write journal"); + } + + // No payment-journal directory ⇒ no payments (empty map). scan must never fabricate progress. + #[test] + fn scan_payment_progress_absent_journal_is_empty() { + let root = temp_home("scan-absent"); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).expect("home dir"); + assert!(scan_payment_progress_at(&root).is_empty()); + let _ = std::fs::remove_dir_all(&root); + } + + // An Intent-only journal folds to Intent ⇒ no funds have left ⇒ PaymentProgress::None. Proves + // scan walks the dir, parses real PaymentRecords, folds them, and maps by job id. + #[test] + fn scan_payment_progress_folds_intent_to_none() { + let root = temp_home("scan-intent"); + let _ = std::fs::remove_dir_all(&root); + let job = "a".repeat(64); + let key = payment_key(&job); + write_journal( + &root, + "attempt.jsonl", + &[PaymentRecord { key: key.clone(), value: PaymentState::Intent { attempt_id: key.attempt_id() } }], + ); + let progress = scan_payment_progress_at(&root); + assert_eq!(progress.get(&job), Some(&PaymentProgress::None)); + let _ = std::fs::remove_dir_all(&root); + } + + // A journal that does not fold (a lone Locked with no preceding Intent is an illegal transition) + // is treated as Uncertain — reconcile must KEEP such a reservation, never release it on ambiguous + // evidence. This is the fail-safe the money path depends on. + #[test] + fn scan_payment_progress_unfoldable_journal_is_uncertain() { + let root = temp_home("scan-uncertain"); + let _ = std::fs::remove_dir_all(&root); + let job = "b".repeat(64); + let key = payment_key(&job); + write_journal( + &root, + "attempt.jsonl", + &[PaymentRecord { key: key.clone(), value: PaymentState::Locked { attempt_id: key.attempt_id() } }], + ); + let progress = scan_payment_progress_at(&root); + assert_eq!(progress.get(&job), Some(&PaymentProgress::Uncertain)); + let _ = std::fs::remove_dir_all(&root); + } + + /// scan_payment_progress reads `/payment-journal`; the fn takes a `MobeeHome`, so drive it + /// through a bootstrapped home rooted at `root`. + fn scan_payment_progress_at(root: &std::path::Path) -> BTreeMap { + let home = bootstrap_home(root).expect("home"); + scan_payment_progress(&home) + } + + // plan_reconcile maps each reserved job by (payment progress, still-payable) via the classifier: + // Closed ⇒ Paid; Uncertain ⇒ Payable (kept); None+payable ⇒ Payable; None+not-payable ⇒ Dead. + #[test] + fn plan_reconcile_maps_each_reserved_job() { + let paid = "a".repeat(64); + let uncertain = "b".repeat(64); + let live = "c".repeat(64); + let dead = "d".repeat(64); + let reserved = vec![paid.clone(), uncertain.clone(), live.clone(), dead.clone()]; + + let mut progress = BTreeMap::new(); + progress.insert(paid.clone(), PaymentProgress::Closed); + progress.insert(uncertain.clone(), PaymentProgress::Uncertain); + // `live` and `dead` have no payment progress (None). + + let mut payable = BTreeMap::new(); + payable.insert(paid.clone(), false); // a Closed payment is Paid regardless of liveness + payable.insert(uncertain.clone(), false); // Uncertain is kept regardless of liveness + payable.insert(live.clone(), true); + payable.insert(dead.clone(), false); + + let dispositions = plan_reconcile(&reserved, &progress, &payable); + assert_eq!(dispositions[&paid], reservations::JobDisposition::Paid); + assert_eq!(dispositions[&uncertain], reservations::JobDisposition::Payable); + assert_eq!(dispositions[&live], reservations::JobDisposition::Payable); + assert_eq!(dispositions[&dead], reservations::JobDisposition::Dead); + } + + // A job missing from `payable` defaults to payable — never released without positive evidence of + // death (the #140 conservative posture: a relay re-read that lost the job is not proof it died). + #[test] + fn plan_reconcile_missing_payable_defaults_to_kept() { + let job = "e".repeat(64); + let dispositions = plan_reconcile(&[job.clone()], &BTreeMap::new(), &BTreeMap::new()); + assert_eq!(dispositions[&job], reservations::JobDisposition::Payable); + } + + // merge_progress keeps the MORE-advanced progress so a Closed attempt is never masked by an + // earlier Intent/Uncertain (Closed > Uncertain > None). + #[test] + fn merge_progress_keeps_the_more_advanced() { + assert_eq!(merge_progress(Some(PaymentProgress::None), PaymentProgress::Closed), PaymentProgress::Closed); + assert_eq!(merge_progress(Some(PaymentProgress::Closed), PaymentProgress::None), PaymentProgress::Closed); + assert_eq!(merge_progress(Some(PaymentProgress::Uncertain), PaymentProgress::None), PaymentProgress::Uncertain); + assert_eq!(merge_progress(None, PaymentProgress::Uncertain), PaymentProgress::Uncertain); + } } From 20965f2c8130ac7401fe3c597a5e705598fc8d25 Mon Sep 17 00:00:00 2001 From: orveth Date: Sat, 25 Jul 2026 01:04:26 -0700 Subject: [PATCH 3/3] buyer: background auto-award (2-call loop) + reserve-then-award re-arm; fix #152 debug panic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the 2-call trade loop and folds in #140/#152. Background auto-award (the daemon-drives-the-award half of post_job → collect): - post_job records a pending_award intent (buyer.sqlite schema v3: pending_awards) and spawns a bounded task that waits for a payable claim (bounded by the offer deadline), then reserves-then-awards it. Re-armed on restart (run() → list_pending_awards → spawn), so a job posted before a crash still gets its award with zero manual commands. - Invariant A (never award twice): lifecycle::plan_rearm checks BOTH the relay (has_award_async: kind-3405 by this buyer, #e=offer) AND the local reservation. Skip if an award is on the relay OR the reservation is already Spent; else Attempt. A Reserved-but-not-published row (crash window) re-attempts — award_with_reservation's reserve is idempotent, so it republishes without a double reserve. - Invariant B (reserve-then-award only): every award path (manual RPC, auto, re-arm) goes through award_with_reservation. A refused reservation parks the intent with a surfaced reason (status.parked_awards) — never a silent drop. #152 (collect verify-fetch panics in DEBUG): reqwest::blocking runs a debug-only guard that builds+drops a throwaway runtime on every request; dropping a runtime inside an async context panics (release makes the guard a no-op, masking it). The buyer verify-fetch runs synchronously inside authorize_pay_async, so GitDeliveryVerifier::fetch/fetch_base now run on a dedicated OS thread via git_transport::off_runtime. Verified: collect_refuses_pay_when_delivered_tip_ differs_from_bound_oid was RED in debug before, GREEN after. #140: reconcile keeps a delivered job (local accept-bind) payable even if the relay expired its events (a delivered job is not dead). Tests: plan_rearm decision table, pending_award lifecycle (park-with-reason + re-arm), off_runtime blocking-request regression, schema v3 migration. Co-Authored-By: Claude Opus 4.8 --- crates/mobee-core/src/buyer/lifecycle.rs | 50 +++++- crates/mobee-core/src/buyer/mod.rs | 205 ++++++++++++++++++++++- crates/mobee-core/src/buyer/store.rs | 190 ++++++++++++++++++++- crates/mobee-core/src/delivery_git.rs | 57 ++++--- crates/mobee-core/src/git_transport.rs | 44 +++++ crates/mobee-core/src/job_lifecycle.rs | 35 +++- 6 files changed, 544 insertions(+), 37 deletions(-) diff --git a/crates/mobee-core/src/buyer/lifecycle.rs b/crates/mobee-core/src/buyer/lifecycle.rs index f00635b6..2a1c8f4b 100644 --- a/crates/mobee-core/src/buyer/lifecycle.rs +++ b/crates/mobee-core/src/buyer/lifecycle.rs @@ -20,7 +20,7 @@ use cashu::{Amount, CurrencyUnit}; use crate::authorize_pay::resolve_realized_mint; use crate::job_lifecycle::{AwardClaimOutcome, JobLifecycleError, JobView}; -use super::reservations::{Converted, JobDisposition, ReserveRefused}; +use super::reservations::{Converted, JobDisposition, ReservationState, ReserveRefused}; use super::store::{BuyerStore, StoreError}; /// Hard filters an awardable claim must pass (issue #126). Grounded in the wire the offer/claim @@ -269,6 +269,35 @@ pub enum PaymentProgress { None, } +/// Whether re-arming a pending auto-award should re-attempt the reserve-then-award, or skip because +/// the job is already awarded. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RearmAction { + /// The job is already awarded — do not award again. + Skip, + /// No award exists yet — run the reserve-then-award path. + Attempt, +} + +/// The idempotent re-arm decision (#126/#127 invariant A: never award twice). A job is already +/// awarded iff a buyer AWARD (3405) is on the relay OR its reservation is already `Spent` (collect +/// paid it). A `Reserved` row with NO relay award is the crash window between reserve and publish — +/// the award never went out, so it must be re-attempted (republished); [`award_with_reservation`]'s +/// reserve is idempotent for the same amount, so the re-attempt reserves once and publishes. +/// +/// Checking BOTH the relay AND the local ledger is load-bearing: the relay catches a 3405 that +/// published before a crash (local ledger alone would miss it → double award); the `Spent` check +/// catches an already-paid job. +pub fn plan_rearm(award_on_relay: bool, reservation: Option) -> RearmAction { + if award_on_relay { + return RearmAction::Skip; + } + if matches!(reservation, Some(ReservationState::Spent)) { + return RearmAction::Skip; + } + RearmAction::Attempt +} + /// Classify a reserved job for [`BuyerStore::reconcile`] from its payment progress + relay /// liveness. The payment journal is authoritative over relay liveness: a `Closed` payment is /// `Paid` regardless of whether the claim still looks live, and an ambiguous payment is KEPT @@ -676,4 +705,23 @@ mod tests { assert_eq!(store.available(100, u64::MAX, 0).expect("avail"), 100, "dead job's funds reclaimed"); let _ = std::fs::remove_file(&path); } + + // IDEMPOTENT RE-ARM tooth (invariant A): never award twice. An award already on the relay ⇒ + // Skip regardless of local state; an already-`Spent` reservation ⇒ Skip. No relay award ⇒ + // Attempt — including the `Reserved`-but-not-published crash window (republish) and the + // Released/None cases. + #[test] + fn plan_rearm_skips_only_when_already_awarded() { + // Relay award present ⇒ Skip regardless of local reservation state. + assert_eq!(plan_rearm(true, None), RearmAction::Skip); + assert_eq!(plan_rearm(true, Some(ReservationState::Reserved)), RearmAction::Skip); + assert_eq!(plan_rearm(true, Some(ReservationState::Spent)), RearmAction::Skip); + // Already paid ⇒ Skip. + assert_eq!(plan_rearm(false, Some(ReservationState::Spent)), RearmAction::Skip); + // Crash window: reserved but no relay award ⇒ Attempt (republish, reserve is idempotent). + assert_eq!(plan_rearm(false, Some(ReservationState::Reserved)), RearmAction::Attempt); + // Released or never reserved, no relay award ⇒ Attempt. + assert_eq!(plan_rearm(false, Some(ReservationState::Released)), RearmAction::Attempt); + assert_eq!(plan_rearm(false, None), RearmAction::Attempt); + } } diff --git a/crates/mobee-core/src/buyer/mod.rs b/crates/mobee-core/src/buyer/mod.rs index d9c322d7..d5788ef9 100644 --- a/crates/mobee-core/src/buyer/mod.rs +++ b/crates/mobee-core/src/buyer/mod.rs @@ -48,7 +48,7 @@ use crate::job_lifecycle::{ self, AwardClaimRequest, ContributionSpec, GetJobRequest, JobKind, PostJobRequest, WaitFor, }; use crate::payment::{PaymentMachine, PaymentRecord, PaymentState}; -use lifecycle::{AwardError, AwardFilters, PaymentProgress, SettleError}; +use lifecycle::{AwardError, AwardFilters, PaymentProgress, RearmAction, SettleError}; use lock::{HomeLock, LockError}; use protocol::{CODE_INTERNAL, CODE_METHOD_NOT_FOUND, CODE_NOT_IMPLEMENTED, Request, Response}; use reservations::{Dispositions, ReconcileReport}; @@ -60,6 +60,9 @@ use wallet_actor::WalletHandle; pub const CODE_REFUSED: i64 = -32002; /// Timeout for the daemon's relay fetches (job view / auto-award selection / reconcile liveness). const RELAY_TIMEOUT: Duration = Duration::from_secs(5); +/// How often the background auto-award task re-checks the relay for a payable claim, until one +/// appears or the offer deadline passes. Bounded polling (no tight spin on a live-but-unpayable claim). +const AUTO_AWARD_POLL_INTERVAL: Duration = Duration::from_secs(5); /// Lock file leaf under the home. pub const LOCK_FILE: &str = "buyer.lock"; @@ -201,6 +204,17 @@ pub async fn run(home: MobeeHome) -> Result<(), BuyerError> { "buyer: reconcile-on-start did not complete ({error}); serving with the ledger as-is" ), } + // Re-arm pending auto-awards left by a prior run: a job posted before a crash still gets its + // award with zero manual commands. Each task re-checks the relay for an existing award first + // (invariant A), so re-arming never double-awards. + match context.store.list_pending_awards() { + Ok(pending) => { + for intent in pending { + spawn_auto_award(context.clone(), intent.job_id, intent.max_sats); + } + } + Err(error) => eprintln!("buyer: could not list pending auto-awards to re-arm: {error}"), + } let listener = bind_socket(&socket_path)?; accept_loop(listener, context).await } @@ -248,7 +262,7 @@ async fn handle_connection(stream: UnixStream, context: Arc) -> st /// Map a request to a response. `status`/`health` is live; the buyer trade /// methods are recognized but deferred to later phases (they return a structured /// not-implemented error rather than silently succeeding). -async fn dispatch(context: &BuyerContext, request: Request) -> Response { +async fn dispatch(context: &Arc, request: Request) -> Response { let id = request.id.clone(); match request.method.as_str() { "status" | "health" => status(context, id).await, @@ -296,6 +310,16 @@ struct PostJobParams { base_oid: Option, #[serde(default)] accepts: Option>, + /// Per-job spend ceiling for the background auto-award (defaults to `amount_sats`). The daemon + /// never auto-awards a claim it cannot pay or priced above this. + #[serde(default)] + max_sats: Option, + /// Auto-award preferences recorded with the intent. Not yet hard filters — no offer/claim wire + /// field carries harness/model, so they are stored, not matched (added when the wire does). + #[serde(default)] + harness: Option, + #[serde(default)] + model: Option, } /// Resolve the offer kind from the contribution pins: all four present ⇒ contribution; none ⇒ @@ -326,8 +350,10 @@ fn post_job_kind(params: &PostJobParams) -> Result { } /// Publish an offer (reuses [`job_lifecycle::post_job_async`], the same money-checked post path the -/// CLI/MCP use). No reservation is taken at post — funds are reserved at award. -async fn post_job(context: &BuyerContext, id: Value, params: Value) -> Response { +/// 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}")), @@ -336,6 +362,9 @@ async fn post_job(context: &BuyerContext, id: Value, params: Value) -> Response Ok(job) => job, Err(message) => return Response::err(id, CODE_METHOD_NOT_FOUND, message), }; + let max_sats = params.max_sats.unwrap_or(params.amount_sats); + let harness = params.harness.clone(); + let model = params.model.clone(); let request = PostJobRequest { task: params.task, output: params.output, @@ -348,7 +377,24 @@ async fn post_job(context: &BuyerContext, id: Value, params: Value) -> Response job, }; match job_lifecycle::post_job_async(&context.home, request).await { - Ok(outcome) => Response::ok(id, json!(outcome)), + Ok(outcome) => { + // Record the intent BEFORE spawning so a crash right after post still re-arms on restart. + if let Err(error) = context.store.put_pending_award( + &outcome.job_id, + max_sats, + harness.as_deref(), + model.as_deref(), + now_unix(), + ) { + eprintln!( + "buyer: could not record auto-award intent for {}: {error}", + outcome.job_id + ); + } else { + spawn_auto_award(context.clone(), outcome.job_id.clone(), max_sats); + } + Response::ok(id, json!(outcome)) + } Err(error) => Response::err(id, CODE_INTERNAL, error.to_string()), } } @@ -581,6 +627,144 @@ fn buyer_keys(home: &MobeeHome) -> Result { nostr_sdk::Keys::parse(&secret).map_err(|error| format!("buyer key parse: {error}")) } +/// Spawn the background auto-award task for a posted job — the daemon-drives-the-award half of the +/// 2-call trade loop. A task failure never affects the daemon; the intent stays `pending` and is +/// re-armed on the next start. +fn spawn_auto_award(context: Arc, job_id: String, max_sats: u64) { + tokio::spawn(async move { + if let Err(error) = drive_auto_award(&context, &job_id, max_sats).await { + eprintln!("buyer: auto-award for {job_id} did not complete ({error}); left pending for re-arm"); + } + }); +} + +/// Drive one posted job's award under the hood: wait (bounded by the offer deadline) for a payable +/// claim, then reserve-then-award. Honors both #126/#127 invariants: +/// +/// - **A (idempotent re-arm):** before doing anything, skip if a buyer AWARD is already on the relay +/// OR the reservation is already `Spent` — never award twice (see [`lifecycle::plan_rearm`]). This +/// is why re-arming on restart is safe: the task re-checks the relay first. +/// - **B (reserve-then-award only):** the award goes exclusively through +/// [`lifecycle::award_with_reservation`] (reserve first, publish second). A refused reservation +/// (e.g. funds shrank) PARKS the intent with a surfaced reason — never a silent drop. +/// +/// Returns `Err` only on a transient relay/wallet failure; the intent then stays `pending` and is +/// re-armed on the next start. +async fn drive_auto_award( + context: &Arc, + job_id: &str, + max_sats: u64, +) -> Result<(), String> { + let keys = buyer_keys(&context.home)?; + + // Invariant A: never award twice. A relay error is "unknown" (unwrap_or false) — do not skip; + // the reserve-then-award path below is itself idempotent on the reserve. + let award_on_relay = job_lifecycle::has_award_async(&context.home, &keys, job_id, RELAY_TIMEOUT) + .await + .unwrap_or(false); + let reservation = context + .store + .reservation(job_id) + .ok() + .flatten() + .map(|(state, _)| state); + if lifecycle::plan_rearm(award_on_relay, reservation) == RearmAction::Skip { + let _ = context.store.mark_award_awarded(job_id, now_unix()); + return Ok(()); + } + + loop { + let view = job_lifecycle::fetch_job_view_async( + &context.home, + &keys, + job_id, + RELAY_TIMEOUT, + now_unix() as u64, + ) + .await + .map_err(|error| error.to_string())?; + let Some(offer) = view.offer.as_ref() else { + let _ = context + .store + .mark_award_parked(job_id, "offer no longer on the relay", now_unix()); + return Ok(()); + }; + if now_unix() as u64 > offer.deadline_unix { + let _ = context.store.mark_award_parked( + job_id, + "offer deadline passed before an awardable claim appeared", + now_unix(), + ); + return Ok(()); + } + + let filters = AwardFilters { + offer_amount_sats: offer.amount_sats, + max_sats, + buyer_mint: context.home.config.default_mint(), + allow_real_mints: context.home.config.allow_real_mints, + }; + if let Some(claim_id) = lifecycle::select_awardable_claim(&view, &filters) { + return finalize_auto_award(context, job_id, offer.amount_sats, claim_id).await; + } + + // No awardable claim yet — re-check after a bounded interval (no tight spin on a + // live-but-unpayable claim). The deadline check above bounds the total wait. + tokio::time::sleep(AUTO_AWARD_POLL_INTERVAL).await; + } +} + +/// Reserve-then-award a selected claim (invariant B), serialized on the money lock so the reserve +/// snapshot never races a collect melt. Marks the intent `awarded` on success, or `parked` (with a +/// surfaced reason) on a refused reservation / publish failure — never a silent drop. +async fn finalize_auto_award( + context: &Arc, + job_id: &str, + offer_amount: u64, + claim_id: String, +) -> Result<(), String> { + let _guard = context.money_lock.lock().await; + let (balance, total_cap, spent) = money_snapshot(context).await?; + let home = context.home.clone(); + let job = job_id.to_owned(); + let publish_claim = claim_id.clone(); + let result = lifecycle::award_with_reservation( + &context.store, + job_id, + offer_amount, + balance, + total_cap, + spent, + now_unix(), + move || async move { + job_lifecycle::award_claim_async(&home, AwardClaimRequest { job_id: job, claim_id: publish_claim }) + .await + }, + ) + .await; + + match result { + Ok(_) => { + let _ = context.store.mark_award_awarded(job_id, now_unix()); + Ok(()) + } + Err(AwardError::Reserve(refused)) => { + let _ = context.store.mark_award_parked( + job_id, + &format!("reservation refused: {refused}"), + now_unix(), + ); + Ok(()) + } + Err(error) => { + let _ = context + .store + .mark_award_parked(job_id, &format!("award failed: {error}"), now_unix()); + Ok(()) + } + } +} + /// Reconcile every still-`Reserved` job against relay + payment-journal truth: a job the relay no /// longer shows payable (and that has left no funds) is released; a job whose payment journal shows /// a `Closed` attempt is converted to `spent`; an ambiguous (Sent-not-Closed) payment is KEPT (the @@ -750,6 +934,16 @@ async fn status(context: &BuyerContext, id: Value) -> Response { }) }); + // Surface parked auto-awards (a claim could not be awarded — e.g. funds shrank) so a buyer sees + // jobs whose award was not placed rather than silently losing them. + let parked_awards: Vec = context + .store + .parked_awards() + .unwrap_or_default() + .into_iter() + .map(|(job_id, reason)| json!({ "job_id": job_id, "reason": reason })) + .collect(); + Response::ok( id, json!({ @@ -766,6 +960,7 @@ async fn status(context: &BuyerContext, id: Value) -> Response { "jobs": jobs, }, "reconcile": reconcile, + "parked_awards": parked_awards, }), ) } diff --git a/crates/mobee-core/src/buyer/store.rs b/crates/mobee-core/src/buyer/store.rs index e487bae1..c76e37d4 100644 --- a/crates/mobee-core/src/buyer/store.rs +++ b/crates/mobee-core/src/buyer/store.rs @@ -26,7 +26,10 @@ use super::reservations::{ /// - v2 — the reservation ledger: the `reservations` table (#123). Upgrade is forward-only and /// additive (a new `CREATE TABLE IF NOT EXISTS` + a monotone version bump); a v1 DB opened by a /// v2 binary gains the table and the version moves to 2 with no data migration. -pub const SCHEMA_VERSION: i64 = 2; +/// - v3 — the pending-award intents: the `pending_awards` table (#126/#127 background auto-award). +/// One row per posted job the daemon still owes an award; re-armed on restart. Same additive +/// forward-only upgrade. +pub const SCHEMA_VERSION: i64 = 3; /// A cloneable handle to the daemon-owned SQLite state. #[derive(Clone)] @@ -100,6 +103,20 @@ impl BuyerStore { state TEXT NOT NULL CHECK (state IN ('reserved','spent','released')), created_at_unix INTEGER NOT NULL, updated_at_unix INTEGER NOT NULL + ); + -- v3: pending auto-award intents. One row per posted job the daemon still owes an + -- award. `pending` = awaiting a payable claim; `awarded` = a 3405 was published; + -- `parked` = could not award (reason surfaced). Re-armed on restart so a job posted + -- before a crash still gets its award with zero manual commands. + CREATE TABLE IF NOT EXISTS pending_awards ( + job_id TEXT PRIMARY KEY, + max_sats INTEGER NOT NULL CHECK (max_sats >= 0), + harness TEXT, + model TEXT, + state TEXT NOT NULL CHECK (state IN ('pending','awarded','parked')), + reason TEXT, + created_at_unix INTEGER NOT NULL, + updated_at_unix INTEGER NOT NULL );", )?; // Forward-only, monotone schema-version bump. A fresh DB is stamped at SCHEMA_VERSION; a @@ -411,6 +428,122 @@ impl BuyerStore { let conn = self.lock()?; read_reservation(&conn, job_id) } + + // ---- Pending auto-award intents (#126/#127) ------------------------------------------------ + + /// Record (or reset to `pending`) a job's auto-award intent — its `max_sats` ceiling and optional + /// harness/model preferences. Re-posting the same job resets it to `pending` and clears any prior + /// parked reason so the daemon re-drives it. + pub fn put_pending_award( + &self, + job_id: &str, + max_sats: u64, + harness: Option<&str>, + model: Option<&str>, + now_unix: i64, + ) -> Result<(), StoreError> { + let conn = self.lock()?; + conn.execute( + "INSERT INTO pending_awards + (job_id, max_sats, harness, model, state, reason, created_at_unix, updated_at_unix) + VALUES (?1, ?2, ?3, ?4, 'pending', NULL, ?5, ?5) + ON CONFLICT(job_id) DO UPDATE SET + max_sats = excluded.max_sats, + harness = excluded.harness, + model = excluded.model, + state = 'pending', + reason = NULL, + updated_at_unix = excluded.updated_at_unix", + params![job_id, max_sats as i64, harness, model, now_unix], + )?; + Ok(()) + } + + /// Every still-`pending` auto-award intent — the set the daemon re-arms on restart. + pub fn list_pending_awards(&self) -> Result, StoreError> { + let conn = self.lock()?; + let mut stmt = conn.prepare( + "SELECT job_id, max_sats, harness, model FROM pending_awards + WHERE state = 'pending' ORDER BY created_at_unix", + )?; + let rows = stmt.query_map([], |row| { + Ok(PendingAward { + job_id: row.get::<_, String>(0)?, + max_sats: row.get::<_, i64>(1)?.max(0) as u64, + harness: row.get::<_, Option>(2)?, + model: row.get::<_, Option>(3)?, + }) + })?; + let mut intents = Vec::new(); + for row in rows { + intents.push(row?); + } + Ok(intents) + } + + /// Mark a job's auto-award intent `awarded` (a 3405 published, or already present on the relay). + /// No-op if the job has no intent row. + pub fn mark_award_awarded(&self, job_id: &str, now_unix: i64) -> Result<(), StoreError> { + let conn = self.lock()?; + conn.execute( + "UPDATE pending_awards SET state = 'awarded', reason = NULL, updated_at_unix = ?2 + WHERE job_id = ?1", + params![job_id, now_unix], + )?; + Ok(()) + } + + /// Mark a job's auto-award intent `parked` with a surfaced `reason` (e.g. the reservation was + /// refused because funds shrank) — never a silent drop. No-op if the job has no intent row. + pub fn mark_award_parked(&self, job_id: &str, reason: &str, now_unix: i64) -> Result<(), StoreError> { + let conn = self.lock()?; + conn.execute( + "UPDATE pending_awards SET state = 'parked', reason = ?2, updated_at_unix = ?3 + WHERE job_id = ?1", + params![job_id, reason, now_unix], + )?; + Ok(()) + } + + /// The `(state, reason)` of a job's auto-award intent, if any. Inspection / tests. + pub fn pending_award_state(&self, job_id: &str) -> Result)>, StoreError> { + let conn = self.lock()?; + Ok(conn + .query_row( + "SELECT state, reason FROM pending_awards WHERE job_id = ?1", + [job_id], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, Option>(1)?)), + ) + .optional()?) + } + + /// Every `parked` auto-award intent as `(job_id, reason)` — surfaced in `status` so a buyer sees + /// jobs whose award could not be placed rather than silently losing them. + pub fn parked_awards(&self) -> Result, StoreError> { + let conn = self.lock()?; + let mut stmt = conn.prepare( + "SELECT job_id, COALESCE(reason, '') FROM pending_awards + WHERE state = 'parked' ORDER BY updated_at_unix", + )?; + let rows = stmt.query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })?; + let mut parked = Vec::new(); + for row in rows { + parked.push(row?); + } + Ok(parked) + } +} + +/// A still-pending auto-award intent: the job the daemon owes an award, its spend ceiling, and the +/// (not-yet-a-filter) harness/model preferences captured at post time. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingAward { + pub job_id: String, + pub max_sats: u64, + pub harness: Option, + pub model: Option, } /// Read a job's `(state, amount)` from a live transaction. `None` when no row exists. @@ -522,9 +655,10 @@ mod tests { } // A v1 database (buyer_meta + jobs only, schema_version = 1) is upgraded forward on open: the - // reservations table appears and schema_version moves to 2. No data migration, idempotent. + // reservations AND pending_awards tables appear and schema_version moves to the current version. + // No data migration, idempotent. #[test] - fn open_migrates_v1_db_forward_to_v2() { + fn open_migrates_v1_db_forward() { let path = temp_db("migrate"); let _ = std::fs::remove_file(&path); { @@ -537,15 +671,59 @@ mod tests { .expect("seed v1"); } let store = BuyerStore::open(&path).expect("open upgrades"); - assert_eq!(store.health().expect("health").schema_version, 2); + assert_eq!(store.health().expect("health").schema_version, SCHEMA_VERSION); // The reservations table is now usable. assert_eq!(store.reserved_in_flight().expect("reserved"), 0); store .reserve(&"a".repeat(64), 10, 100, NO_BUDGET, 0, 1) .expect("reserve on upgraded db"); - // Re-open is idempotent (still v2). + // The pending_awards table is now usable too. + store + .put_pending_award(&"a".repeat(64), 10, None, None, 1) + .expect("pending award on upgraded db"); + assert_eq!(store.list_pending_awards().expect("list").len(), 1); + // Re-open is idempotent (still current version). let store2 = BuyerStore::open(&path).expect("reopen"); - assert_eq!(store2.health().expect("health").schema_version, 2); + assert_eq!(store2.health().expect("health").schema_version, SCHEMA_VERSION); + let _ = std::fs::remove_file(&path); + } + + // Pending auto-award intent lifecycle: put ⇒ listed as pending; mark_parked ⇒ off the pending + // list, state=parked with the surfaced reason (invariant B: park, never silent drop); + // mark_awarded ⇒ awarded; re-put ⇒ back to pending with the reason cleared. + #[test] + fn pending_award_lifecycle_parks_with_reason_and_rearms() { + let (store, path) = fresh_store("pending-award"); + let job = "a".repeat(64); + + store.put_pending_award(&job, 21, Some("claude"), Some("opus"), 1).expect("put"); + let pending = store.list_pending_awards().expect("list"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].job_id, job); + assert_eq!(pending[0].max_sats, 21); + assert_eq!(pending[0].harness.as_deref(), Some("claude")); + + // Park with a reason — off the pending list, visible as parked. + store.mark_award_parked(&job, "reservation refused: funds shrank", 2).expect("park"); + assert!(store.list_pending_awards().expect("list").is_empty(), "parked is not pending"); + assert_eq!( + store.pending_award_state(&job).expect("state"), + Some(("parked".to_owned(), Some("reservation refused: funds shrank".to_owned()))) + ); + assert_eq!( + store.parked_awards().expect("parked"), + vec![(job.clone(), "reservation refused: funds shrank".to_owned())] + ); + + // Awarding clears the parked reason. + store.mark_award_awarded(&job, 3).expect("awarded"); + assert_eq!(store.pending_award_state(&job).expect("state"), Some(("awarded".to_owned(), None))); + assert!(store.parked_awards().expect("parked").is_empty()); + + // Re-posting the same job re-arms it (back to pending, reason cleared). + store.put_pending_award(&job, 30, None, None, 4).expect("re-put"); + assert_eq!(store.pending_award_state(&job).expect("state"), Some(("pending".to_owned(), None))); + assert_eq!(store.list_pending_awards().expect("list")[0].max_sats, 30); let _ = std::fs::remove_file(&path); } diff --git a/crates/mobee-core/src/delivery_git.rs b/crates/mobee-core/src/delivery_git.rs index 8cf955d6..12714c53 100644 --- a/crates/mobee-core/src/delivery_git.rs +++ b/crates/mobee-core/src/delivery_git.rs @@ -93,20 +93,26 @@ impl GitDeliveryVerifier { } fn fetch(&self, delivery: &GitDelivery) -> Result { - let repo = self.open_store()?; - let fetched_ref = format!("refs/mobee/deliveries/{}", delivery.commit_oid().as_str()); - let refspec = format!("+refs/heads/{}:{fetched_ref}", delivery.branch()); - // short_timeout=true: a hung fetch must not own the MCP stdio loop past the client timeout, - // and must fail CLOSED before authorize_pay burns budget (verify-before-pay). - git_transport::fetch_refspecs(&repo, delivery.repo(), &[&refspec], self.read_auth(), true) - .map_err(|_| DeliveryError::GitCommandFailed("fetch"))?; - - let fetched_object = format!("{fetched_ref}^{{commit}}"); - let commit = repo - .revparse_single(&fetched_object) - .and_then(|object| object.peel_to_commit()) - .map_err(|_| DeliveryError::MissingFetchedTip)?; - CommitOid::parse(commit.id().to_string()).map_err(|_| DeliveryError::MissingFetchedTip) + // The git2 fetch drives the reqwest::blocking smart-HTTP subtransport, which must not run on + // a Tokio worker thread (reqwest's debug-only runtime-guard would panic — #152). The buyer + // verify-fetch is invoked synchronously inside `authorize_pay_async`, so run it off any + // ambient runtime, on a dedicated OS thread. + git_transport::off_runtime(|| { + let repo = self.open_store()?; + let fetched_ref = format!("refs/mobee/deliveries/{}", delivery.commit_oid().as_str()); + let refspec = format!("+refs/heads/{}:{fetched_ref}", delivery.branch()); + // short_timeout=true: a hung fetch must not own the MCP stdio loop past the client + // timeout, and must fail CLOSED before authorize_pay burns budget (verify-before-pay). + git_transport::fetch_refspecs(&repo, delivery.repo(), &[&refspec], self.read_auth(), true) + .map_err(|_| DeliveryError::GitCommandFailed("fetch"))?; + + let fetched_object = format!("{fetched_ref}^{{commit}}"); + let commit = repo + .revparse_single(&fetched_object) + .and_then(|object| object.peel_to_commit()) + .map_err(|_| DeliveryError::MissingFetchedTip)?; + CommitOid::parse(commit.id().to_string()).map_err(|_| DeliveryError::MissingFetchedTip) + }) } fn require_local_object(&self, oid: &CommitOid) -> Result<(), DeliveryError> { @@ -130,16 +136,19 @@ impl GitDeliveryVerifier { base_oid: &CommitOid, ) -> Result<(), DeliveryError> { self.check_branch(base_branch)?; - let repo = self.open_store()?; - let fetched_ref = format!("refs/mobee/bases/{}", base_oid.as_str()); - let refspec = format!("+refs/heads/{base_branch}:{fetched_ref}"); - git_transport::fetch_refspecs(&repo, base_clone_url, &[&refspec], self.read_auth(), true) - .map_err(|_| DeliveryError::GitCommandFailed("fetch-base"))?; - // The pinned target MUST actually contain base_oid — resolve it as a commit in the store. - let parsed = Self::parse_oid(base_oid)?; - repo.find_commit(parsed) - .map(|_| ()) - .map_err(|_| DeliveryError::MissingBaseObject) + // Same #152 constraint as `fetch`: run the git2 base fetch off any ambient async runtime. + git_transport::off_runtime(|| { + let repo = self.open_store()?; + let fetched_ref = format!("refs/mobee/bases/{}", base_oid.as_str()); + let refspec = format!("+refs/heads/{base_branch}:{fetched_ref}"); + git_transport::fetch_refspecs(&repo, base_clone_url, &[&refspec], self.read_auth(), true) + .map_err(|_| DeliveryError::GitCommandFailed("fetch-base"))?; + // The pinned target MUST actually contain base_oid — resolve it as a commit in the store. + let parsed = Self::parse_oid(base_oid)?; + repo.find_commit(parsed) + .map(|_| ()) + .map_err(|_| DeliveryError::MissingBaseObject) + }) } /// Contribution: refuse unless `commit_oid` descends from `base_oid`, in-process via diff --git a/crates/mobee-core/src/git_transport.rs b/crates/mobee-core/src/git_transport.rs index c9724d04..666d29c5 100644 --- a/crates/mobee-core/src/git_transport.rs +++ b/crates/mobee-core/src/git_transport.rs @@ -88,6 +88,28 @@ fn accept_invalid_certs() -> bool { std::env::var_os("GIT_SSL_NO_VERIFY").is_some() } +/// Run a blocking git2 fetch/push off any ambient async runtime, on a dedicated OS thread. +/// +/// The smart-HTTP subtransport ([`HttpStream`]) uses `reqwest::blocking`. In a DEBUG build reqwest +/// guards every request by building and immediately dropping a throwaway Tokio runtime — and +/// dropping a runtime inside another runtime's context panics ("Cannot drop a runtime in a context +/// where blocking is not allowed"); a release build makes that guard a no-op, which is why the bug +/// was masked (#152). The buyer verify-fetch runs synchronously inside `authorize_pay_async` (a +/// Tokio worker), so the git2 fetch that drives those requests must run on a plain thread where no +/// ambient runtime is present. Works under any caller runtime flavor (unlike `block_in_place`). +pub(crate) fn off_runtime(work: F) -> T +where + T: Send, + F: FnOnce() -> T + Send, +{ + std::thread::scope(|scope| { + scope + .spawn(work) + .join() + .unwrap_or_else(|_| panic!("git transport worker thread panicked")) + }) +} + /// Long-running client for pushes and seller base fetches (large packs are legitimate). fn client_default() -> &'static reqwest::blocking::Client { static CLIENT: OnceLock = OnceLock::new(); @@ -578,4 +600,26 @@ mod tests { Err(TransportError::Transport(_)) )); } + + // #152 regression: a `reqwest::blocking` REQUEST executed on a Tokio worker hits reqwest's + // debug-only guard, which builds and drops a throwaway runtime — a debug panic ("Cannot drop a + // runtime in a context where blocking is not allowed"). The buyer verify-fetch runs inside + // `authorize_pay_async`, so it must go through `off_runtime` (a plain thread). This drives a real + // blocking request from within a Tokio runtime via `off_runtime` and asserts it returns (the + // request fails — nothing listens on port 9 — but must NOT panic). + // + // Red-on-revert (strong form): call `.send()` DIRECTLY here (drop the `off_runtime` wrapper) and + // this test panics in a debug build. + #[tokio::test] + async fn blocking_request_runs_off_the_async_runtime() { + let result = off_runtime(|| { + reqwest::blocking::Client::builder() + .timeout(std::time::Duration::from_millis(100)) + .build() + .expect("client") + .get("http://127.0.0.1:9/") + .send() + }); + assert!(result.is_err(), "the request should fail to connect, but must not panic"); + } } diff --git a/crates/mobee-core/src/job_lifecycle.rs b/crates/mobee-core/src/job_lifecycle.rs index fe57cd36..8f11d824 100644 --- a/crates/mobee-core/src/job_lifecycle.rs +++ b/crates/mobee-core/src/job_lifecycle.rs @@ -22,7 +22,7 @@ use sha2::{Digest, Sha256}; use crate::gateway::{ self, award_draft, parse_git_result_delivery, parse_offer, EventDraft, OfferDraft, TagSpec, - JOB_CLAIM_KIND, JOB_FEEDBACK_KIND, JOB_OFFER_KIND, JOB_RESULT_KIND, + JOB_AWARD_KIND, JOB_CLAIM_KIND, JOB_FEEDBACK_KIND, JOB_OFFER_KIND, JOB_RESULT_KIND, }; use crate::home::{self, HomeError, MobeeHome}; #[cfg(feature = "wallet")] @@ -1530,6 +1530,39 @@ fn derive_claim_liveness( /// against `now` (a `processing` claim past the offer deadline is EXPIRED, not live). Exposed /// `pub(crate)` so the seller daemon can run the backfill money-safety pre-claim check /// (already-delivered / live-claimed-by-another) without duplicating the relay read. +/// True when a buyer AWARD (kind-3405) authored by this buyer already exists on the relay for +/// `job_id` — the relay half of the idempotent re-arm check (a 3405 may have published before a +/// crash, so the local ledger alone is insufficient). A relay error propagates so the caller treats +/// it as "unknown" and does not falsely mark the intent awarded. +pub(crate) async fn has_award_async( + home: &MobeeHome, + keys: &nostr_sdk::Keys, + job_id: &str, + timeout: Duration, +) -> Result { + use nostr_sdk::prelude::{Client, EventId, Filter, Kind}; + + let offer_id = EventId::from_hex(job_id) + .map_err(|error| JobLifecycleError::Input(format!("job_id: {error}")))?; + let client = Client::new(keys.clone()); + client + .add_relay(&home.config.relay_url) + .await + .map_err(|error| JobLifecycleError::Relay(format!("add relay: {error}")))?; + client.connect().await; + + let filter = Filter::new() + .kind(Kind::Custom(JOB_AWARD_KIND)) + .author(keys.public_key()) + .event(offer_id) + .hashtag(gateway::MOBEE_TAG); + let events = client + .fetch_events(filter, timeout) + .await + .map_err(|error| JobLifecycleError::Relay(format!("fetch award: {error}")))?; + Ok(!events.is_empty()) +} + pub(crate) async fn fetch_job_view_async( home: &MobeeHome, keys: &nostr_sdk::Keys,