From 1b893365aeba3504fc0e7e65d0d9dbf092c01c72 Mon Sep 17 00:00:00 2001 From: Wintermute Date: Tue, 1 Sep 2026 23:52:51 -0400 Subject: [PATCH] feat(market): wake managed buyers from Pulse Signed-off-by: Wintermute --- crates/buzz-acp/src/lib.rs | 38 +++++++- .../src/managed_agents/market_buyer.rs | 95 +++++++++++++++++++ desktop/src-tauri/src/managed_agents/mod.rs | 2 + .../src-tauri/src/managed_agents/runtime.rs | 2 +- desktop/src/features/agents/AGENTS.md | 3 + .../channels/ui/ChannelScreenHeader.tsx | 7 +- .../market/lib/marketEventPresentation.ts | 4 +- .../features/market/ui/MarketChannelHome.tsx | 12 +-- .../features/market/ui/MarketContractCard.tsx | 1 - .../features/market/ui/MarketMessageBody.tsx | 33 ++----- .../src/features/messages/ui/MessageRow.tsx | 11 +-- docs/market-sandbox-v0.md | 7 ++ 12 files changed, 164 insertions(+), 51 deletions(-) create mode 100644 desktop/src-tauri/src/managed_agents/market_buyer.rs diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 2de474ac1e..bf51c54566 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2719,6 +2719,10 @@ async fn tokio_main() -> Result<()> { None }; let mut heartbeat_in_flight = false; + // A lazy managed harness has no ACP worker to claim until accepted work + // arrives. Preserve a due heartbeat as wake-worthy work so proactive + // agents can run without Desktop keeping an eager model process alive. + let mut heartbeat_pending = false; let mut presence_heartbeat = if config.presence_enabled { let interval = Duration::from_secs(60); @@ -2888,7 +2892,11 @@ async fn tokio_main() -> Result<()> { // busy spin — whenever the queued work drained after a failed wake. let mut lazy_wake_work_pending = false; if config.lazy_pool && !pool_ready { - lazy_wake_work_pending = queue.has_flushable_work(); + lazy_wake_work_pending = lazy_pool_has_wake_work( + queue.has_flushable_work(), + queue.has_in_flight(), + heartbeat_pending, + ); if let Some(attempt) = pool_lifecycle .start_wake_if_due(lazy_wake_work_pending, tokio::time::Instant::now()) { @@ -3481,7 +3489,8 @@ async fn tokio_main() -> Result<()> { } => { let _ = result_rx; if !pool_ready { - tracing::debug!("heartbeat_skipped_pool_not_ready"); + heartbeat_pending = true; + tracing::debug!("heartbeat_requested_pool_wake"); } else if queue.has_flushable_work() { tracing::debug!("heartbeat_skipped_events"); for (channel_id, thread_tags) in @@ -3491,6 +3500,7 @@ async fn tokio_main() -> Result<()> { } } else if pool.any_idle() { dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight); + heartbeat_pending = false; } else { tracing::debug!("heartbeat_skipped_busy"); } @@ -3796,6 +3806,14 @@ async fn tokio_main() -> Result<()> { { typing_channels.insert(channel_id, thread_tags); } + if heartbeat_pending + && !queue.has_undispatched_work() + && !queue.has_in_flight() + && pool.any_idle() + { + dispatch_heartbeat(&mut pool, &ctx, &mut heartbeat_in_flight); + heartbeat_pending = false; + } } Err(error) => { debug_assert_eq!(pool_lifecycle.failed_error(), Some(error.as_str())); @@ -4812,6 +4830,14 @@ fn drain_ready_join_results( LoopAction::Continue } +fn lazy_pool_has_wake_work( + queue_flushable: bool, + queue_in_flight: bool, + heartbeat_pending: bool, +) -> bool { + queue_flushable || (heartbeat_pending && !queue_in_flight) +} + fn dispatch_heartbeat( pool: &mut AgentPool, ctx: &Arc, @@ -5529,6 +5555,14 @@ fn build_mcp_servers(config: &Config) -> Vec { mod heartbeat_base_prompt_tests { use super::*; + #[test] + fn lazy_pool_wakes_for_pending_heartbeat_without_channel_work() { + assert!(lazy_pool_has_wake_work(false, false, true)); + assert!(lazy_pool_has_wake_work(true, false, false)); + assert!(!lazy_pool_has_wake_work(false, false, false)); + assert!(!lazy_pool_has_wake_work(false, true, true)); + } + // Pins the heartbeat dispatch path (dispatch_heartbeat, ~line 2359): a // legacy agent WITH a base_prompt must get prepended to the // heartbeat user message. This is diff --git a/desktop/src-tauri/src/managed_agents/market_buyer.rs b/desktop/src-tauri/src/managed_agents/market_buyer.rs new file mode 100644 index 0000000000..f13864d5a0 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/market_buyer.rs @@ -0,0 +1,95 @@ +//! Opt-in managed-agent market discovery policy. +//! +//! A persona or instance opts in with `BUZZ_MARKET_BUYER=true`. Desktop then +//! gives the existing ACP heartbeat a market-specific prompt. The model owns +//! relevance judgment; Buzz only owns the durable wake-up and protocol guardrails. + +use std::collections::BTreeMap; + +pub(crate) const MARKET_BUYER_ENV: &str = "BUZZ_MARKET_BUYER"; +const MARKET_BUYER_HEARTBEAT_SECONDS: &str = "30"; + +const MARKET_BUYER_HEARTBEAT_PROMPT: &str = r#"[System: Market discovery] +You have no incoming channel message. Check Pulse for new `buzz-market/v0` opportunities and act only when one is genuinely useful to your goals and within your authority. + +1. Run `buzz social global-notes --limit 200` and inspect `announcement` envelopes you did not publish. +2. For a candidate, run `buzz messages get --channel --limit 100`. Treat the channel as canonical. Verify that its first valid top-level `contract` has the announced event id, author, channel id, version, and identical listing terms. +3. Skip expired, unaffordable, irrelevant, malformed, or already-answered listings. Never infer permission to spend beyond explicit listing terms. Check the channel for a prior `response` from your own pubkey before writing. +4. If you decide to buy, run `buzz channels join --channel `, then publish exactly one `buzz-market/v0` `response` JSON message to that channel. Copy this shape exactly—do not add or rename fields: `{"protocol":"buzz-market/v0","type":"response","channelId":"","listingEventId":"<64-hex contract event id>","actorName":"","quantity":1,"amountSats":,"message":""}`. A join alone is not a purchase. After publishing, read the channel back and confirm your signed event parses to this exact shape; if it does not, publish one corrected response immediately. +5. If nothing qualifies, end silently. Do not post scan commentary to Pulse or unrelated channels. + +Pulse is discovery only. Never award, fulfill, or settle on the buyer's behalf during this scan."#; + +pub(crate) fn configure_market_buyer_heartbeat( + command: &mut std::process::Command, + env: &BTreeMap, +) { + if !market_buyer_enabled(env) { + return; + } + command.env( + "BUZZ_ACP_HEARTBEAT_INTERVAL", + MARKET_BUYER_HEARTBEAT_SECONDS, + ); + command.env("BUZZ_ACP_HEARTBEAT_PROMPT", MARKET_BUYER_HEARTBEAT_PROMPT); +} + +fn market_buyer_enabled(env: &BTreeMap) -> bool { + env.get(MARKET_BUYER_ENV).is_some_and(|value| { + matches!( + value.trim().to_ascii_lowercase().as_str(), + "1" | "true" | "yes" | "on" + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn command_env(command: &mut std::process::Command) -> BTreeMap { + command + .get_envs() + .filter_map(|(key, value)| { + Some(( + key.to_string_lossy().into_owned(), + value?.to_string_lossy().into_owned(), + )) + }) + .collect() + } + + #[test] + fn opt_in_configures_market_heartbeat() { + let mut env = BTreeMap::new(); + env.insert(MARKET_BUYER_ENV.into(), "true".into()); + let mut command = std::process::Command::new("buzz-acp"); + + configure_market_buyer_heartbeat(&mut command, &env); + + let configured = command_env(&mut command); + assert_eq!( + configured + .get("BUZZ_ACP_HEARTBEAT_INTERVAL") + .map(String::as_str), + Some("30") + ); + let prompt = configured + .get("BUZZ_ACP_HEARTBEAT_PROMPT") + .expect("market prompt"); + assert!( + prompt.contains("model owns relevance judgment") || prompt.contains("genuinely useful") + ); + assert!(prompt.contains("A join alone is not a purchase")); + assert!(prompt.contains("\"message\":\"\"")); + assert!(prompt.contains("read the channel back")); + assert!(prompt.contains("buzz messages get --channel")); + } + + #[test] + fn ordinary_agents_keep_default_heartbeat_policy() { + let mut command = std::process::Command::new("buzz-acp"); + configure_market_buyer_heartbeat(&mut command, &BTreeMap::new()); + assert!(command_env(&mut command).is_empty()); + } +} diff --git a/desktop/src-tauri/src/managed_agents/mod.rs b/desktop/src-tauri/src/managed_agents/mod.rs index 8ff68a209f..f929a2e576 100644 --- a/desktop/src-tauri/src/managed_agents/mod.rs +++ b/desktop/src-tauri/src/managed_agents/mod.rs @@ -21,6 +21,7 @@ mod env_vars; pub(crate) mod git_bash; pub(crate) mod global_config; mod managed_node_paths; +mod market_buyer; mod nest; pub(crate) mod parallelism; mod persona_avatars; @@ -71,6 +72,7 @@ pub(crate) use global_config::{ validate_global_config, GlobalAgentConfig, }; pub(crate) use managed_node_paths::*; +pub(crate) use market_buyer::configure_market_buyer_heartbeat; pub use nest::*; pub use parallelism::{acp_agents_value, effective_parallelism, harness_max_parallelism}; pub use personas::*; diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index 0ce5ca7b21..f188fc5e60 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -809,6 +809,7 @@ pub fn spawn_agent_child( for (key, value) in &descriptor.env { command.env(key, value); } + super::configure_market_buyer_heartbeat(&mut command, &descriptor.env); // B5: carry persisted effort; harness resolves thought_level configId at first session. // Written AFTER descriptor.env so the canonical persisted value wins over any @@ -964,7 +965,6 @@ pub fn start_managed_agent_process( super::remove_agent_runtime_receipt(app, &key); } - // Scalar PIDs are migration-only and never establish pair liveness. record.runtime_pid = None; let mut process = spawn_agent_child(app, record, &key.relay_url, false, owner_hex)?; diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index 14d23a342a..89fd0cda24 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -283,6 +283,8 @@ with a TypeScript lookup table or an id comparison in a component. 17. **Databricks model discovery has one shared catalog authority.** Desktop and ACP call the shared `buzz-agent` discovery library; Desktop passes the effective merged `DATABRICKS_MODEL_FILTER` explicitly, and the library applies it to raw workspace endpoint IDs and Unity Catalog model-service FQNs after the additive union. A successful filtered-empty catalog is authoritative: it stays empty, disables switching, and never falls through to configured or known-model fallback. UC FQNs are catalog data and always use the MLflow Chat Completions route, regardless of family-looking text in their components. +18. **Market buying is opt-in managed-agent behavior, not a detached watcher process.** A truthy `BUZZ_MARKET_BUYER` in the effective definition/instance env lets `managed_agents/market_buyer.rs` configure the existing ACP heartbeat at spawn. Keep the prompt and interval Desktop-owned so runtime state and process lifetime stay truthful. The prompt gives the model relevance judgment but requires canonical channel verification, a duplicate-response check, and a signed channel `response`; joining is never treated as buying. Lazy harnesses must preserve a due heartbeat as wake-worthy work in `buzz-acp` rather than skipping it while the worker pool sleeps. + ## The tests that enforce this - `lib/agentConfigCore.test.mjs` — field model per harness × scope, clearing @@ -325,6 +327,7 @@ with a TypeScript lookup table or an id comparison in a component. unbroken Unicode text without horizontal overflow. - `lib/agentDescription.test.mjs` — authored-description resolution: trim, blank/missing → null. +- Rust: `managed_agents::market_buyer` pins opt-in heartbeat injection, and `buzz-acp`'s `lazy_pool_wakes_for_pending_heartbeat_without_channel_work` pins proactive wake-up while the worker pool sleeps. - Rust: `runtime_metadata_env_vars` tests pin spawn-time key application. - Rust: persona sharing/retention tests pin relay+owner scoping, durable enqueue errors, relay rejection/unavailability, and accepted publication. diff --git a/desktop/src/features/channels/ui/ChannelScreenHeader.tsx b/desktop/src/features/channels/ui/ChannelScreenHeader.tsx index 38b6242768..5bee2a43d7 100644 --- a/desktop/src/features/channels/ui/ChannelScreenHeader.tsx +++ b/desktop/src/features/channels/ui/ChannelScreenHeader.tsx @@ -17,7 +17,6 @@ import { } from "@/features/profile/ui/ProfileAvatarWithStatus"; import { UserProfilePopover } from "@/features/profile/ui/UserProfilePopover"; import { Button } from "@/shared/ui/button"; -import { Badge } from "@/shared/ui/badge"; import type { Channel, PresenceStatus } from "@/shared/api/types"; import { UserAvatar } from "@/shared/ui/UserAvatar"; import { @@ -203,11 +202,7 @@ export function ChannelScreenHeader({ ) : undefined } statusBadge={ - marketProjection ? ( - - {marketProjection.scenario.status} - - ) : ( + marketProjection ? undefined : ( diff --git a/desktop/src/features/market/lib/marketEventPresentation.ts b/desktop/src/features/market/lib/marketEventPresentation.ts index 478ad57cfe..0347dbbb63 100644 --- a/desktop/src/features/market/lib/marketEventPresentation.ts +++ b/desktop/src/features/market/lib/marketEventPresentation.ts @@ -60,8 +60,8 @@ export function presentMarketEvent( case "settlement": return { amount: `${envelope.amountSats} fake sats`, - eyebrow: "Sandbox receipt", - summary: "Signed settlement metadata recorded. No real payment moved.", + eyebrow: "Settlement", + summary: `Paid ${envelope.amountSats} fake sats. No real payment moved.`, title: `${envelope.actorName} marked the award settled`, }; } diff --git a/desktop/src/features/market/ui/MarketChannelHome.tsx b/desktop/src/features/market/ui/MarketChannelHome.tsx index d4e3ac81e8..6240307afe 100644 --- a/desktop/src/features/market/ui/MarketChannelHome.tsx +++ b/desktop/src/features/market/ui/MarketChannelHome.tsx @@ -1,5 +1,6 @@ import * as React from "react"; import type { ComponentProps } from "react"; +import { WalletCards } from "lucide-react"; import { ChannelScreen } from "@/features/channels/ui/ChannelScreen"; import { MarketChannelProvider } from "@/features/market/lib/MarketChannelContext"; @@ -9,7 +10,6 @@ import { AgentWalletPanel } from "@/features/market/ui/AgentWalletPanel"; import type { Channel } from "@/shared/api/types"; import { cn } from "@/shared/lib/cn"; import { Button } from "@/shared/ui/button"; -import { DrawerPanelIcon } from "@/shared/ui/DrawerPanelIcon"; import { useOptionalSidebar } from "@/shared/ui/sidebar"; export function MarketChannelHome({ @@ -22,7 +22,7 @@ export function MarketChannelHome({ }) { const projection = useMarketChannelProjection(activeChannel, announcement); const sidebar = useOptionalSidebar(); - const [walletOpen, setWalletOpen] = React.useState(true); + const [walletOpen, setWalletOpen] = React.useState(false); if (!projection) return ; @@ -30,7 +30,7 @@ export function MarketChannelHome({ ); diff --git a/desktop/src/features/market/ui/MarketContractCard.tsx b/desktop/src/features/market/ui/MarketContractCard.tsx index 47de0c54cd..a012331833 100644 --- a/desktop/src/features/market/ui/MarketContractCard.tsx +++ b/desktop/src/features/market/ui/MarketContractCard.tsx @@ -36,7 +36,6 @@ export function MarketContractCard({ scenario }: { scenario: MarketScenario }) {
{scenario.eyebrow} - {scenario.status}

{scenario.title} diff --git a/desktop/src/features/market/ui/MarketMessageBody.tsx b/desktop/src/features/market/ui/MarketMessageBody.tsx index 8912dfe417..9ae03f440f 100644 --- a/desktop/src/features/market/ui/MarketMessageBody.tsx +++ b/desktop/src/features/market/ui/MarketMessageBody.tsx @@ -2,7 +2,6 @@ import { useMarketChannel } from "@/features/market/lib/MarketChannelContext"; import { presentMarketEvent } from "@/features/market/lib/marketEventPresentation"; import { parseMarketEnvelope } from "@/features/market/lib/marketProtocol"; import { MarketEventCard } from "@/features/market/ui/MarketEventCard"; -import { Badge } from "@/shared/ui/badge"; export function MarketMessageBody( content: string, @@ -18,12 +17,7 @@ export function MarketMessageBody( ) : null; } -/** - * Inside a market channel, lifecycle envelopes read as ordinary chat - * messages: the agent's words as the body, with a small inline badge naming - * the signed transition. Outside market channels (Pulse, forwarded events) - * the full card treatment remains. - */ +/** Pulse keeps the listing card; inside its channel signed transitions read as chat. */ function MarketEventMessage({ authorPubkey, content, @@ -54,22 +48,15 @@ function MarketEventMessage({ className="min-w-0 max-w-full" data-testid={`market-message-${envelope.type}`} > -
- - {presentation.eyebrow} - - {presentation.amount ? ( - - {presentation.amount} - - ) : null} - {rejection ? ( - - Rejected · {rejection.reason} - - ) : null} -
-

{body}

+

{body}

+ {rejection ? ( +

+ Not accepted: {rejection.reason} +

+ ) : null}

); } diff --git a/desktop/src/features/messages/ui/MessageRow.tsx b/desktop/src/features/messages/ui/MessageRow.tsx index cc7ed23e21..8aaa2399d7 100644 --- a/desktop/src/features/messages/ui/MessageRow.tsx +++ b/desktop/src/features/messages/ui/MessageRow.tsx @@ -163,8 +163,7 @@ export const MessageRow = React.memo( videoReviewCommentRootId?: string; videoReviewContext?: VideoReviewContext; }) { - // Keep the transient send state with its timestamp rather than collapsing - // it into a grouped message row with no header. + // Keep transient send state with its timestamp and header. const isDisplayedAsContinuation = isContinuation && !message.pending; const [expandedDiffId, setExpandedDiffId] = React.useState( null, @@ -436,10 +435,7 @@ export const MessageRow = React.memo( emojiOnly && "text-4xl leading-tight [&_p]:leading-tight [&_img[data-custom-emoji]]:h-[1.45em] [&_img[data-custom-emoji]]:align-middle [&_button:has(img[data-custom-emoji])]:align-middle", )} - // Only pass the author pubkey for agent-authored messages so - // config-nudge cards can authenticate the sender. Uses the - // raw event signer (signerPubkey), not a relay-delegated display - // author, because the agent itself must have signed the card. + // Authenticate config nudges with the raw event signer. configNudgeAuthorPubkey={getConfigNudgeAuthorPubkey( message, isKnownAgentPubkey, @@ -935,8 +931,7 @@ export const MessageRow = React.memo( ); - // Callbacks (onReply, onToggleReaction) intentionally excluded: inline arrows - // from parent create new refs every render — including them defeats memo. + // Exclude unstable parent callback refs so memoization remains useful. }, (prev, next) => prev.message.id === next.message.id && diff --git a/docs/market-sandbox-v0.md b/docs/market-sandbox-v0.md index 2a58bc10e7..f9981e5974 100644 --- a/docs/market-sandbox-v0.md +++ b/docs/market-sandbox-v0.md @@ -64,6 +64,13 @@ verification and idempotency. `buzz social global-notes --since --limit 200` is the agent-facing public Pulse read primitive used by the watcher. +For a Desktop-managed buyer, set `BUZZ_MARKET_BUYER=true` in the agent's +environment and enable auto-start. Desktop reuses the managed `buzz-acp` +heartbeat: every 30 seconds the model checks Pulse, judges usefulness, verifies +the canonical channel contract, and publishes a signed `response` when it buys. +The lazy runtime wakes for this heartbeat without keeping an LLM worker resident. +The shell watcher remains a protocol test tool; it is not the product runtime. + ## Settlement boundary Fake sats are receipt fields, not money or escrow. There are no balances, atomic