Fix coin-ops blocked by shared Direct maker p2 watches

docs/decisions/0019-coinset-ws-local-watches.md

@@ -31,13 +31,15 @@ never the operator transport.
    HTTP `get_transaction` polling supplements WS by confirming prepared
    `cancel_submitted` transaction ids during recovery and every cycle preamble.
 3. **Watches:** durable SQLite `offer_coin_watches` registered atomically at post
-   (maker coins + on-chain maker puzzle hashes — CAT outer or XCH p2 — not the
-   fixed delegated CONDITIONS hash), sourced from required `OfferCancelFields` on
-   every successful create (Direct: single exact-size maker coin; presplit: split
-   coin + fixed CONDITIONS hash). Schema migration backfills missing watches
-   from cancel metadata for pre-upgrade rows. Shared market inventory
-   receive/CAT outer p2s are **not** stored on per-offer watches; `InventoryP2Index`
-   still drives WS filters and inventory freshness. Optional coin-id fields on
+   (maker **coin ids** always; per-offer maker **p2** only when cancel metadata is
+   presplit-like via `StoredOfferCancelMetadata::is_presplit_like` — explicit
+   Presplit, or legacy NULL `execution_mode` + non-empty `fixed_delegated_puzzle_hash`).
+   Direct `maker_puzzle_hash` (CAT outer / XCH receive inner) is cancel metadata only —
+   shared inventory hashes stay on `InventoryP2Index`, not per-offer watches.
+   Sourced from required `OfferCancelFields` on every successful create (Direct:
+   single exact-size maker coin; presplit: split coin + fixed CONDITIONS hash).
+   Schema migration backfills missing watches from cancel metadata for pre-upgrade
+   rows. Optional coin-id fields on
    transaction frames are matched when present. WS offer events and watch hits
    drive lifecycle transitions directly. Cancel submit prepares `cancel_submitted`
    before broadcast (watches kept), then observes the cancel tx after successful
@@ -62,8 +64,8 @@ never the operator transport.
    `watch_venue_backfill_v2`). Missing watches are healed each reconcile via a
    single `prepare_market_reconcile_local` scan: cancel-submitted collection,
    cancel-metadata heal, and Dexie role classify (`DexieWatchRoles`), then Dexie
-   payloads for heal-only NULL-venue gaps (`fetch_and_ensure_watches` seeds both
-   maker coin ids and on-chain maker p2s from cancellable offer inputs; when a
+   payloads for heal-only NULL-venue gaps (`fetch_and_ensure_watches` seeds maker
+   coin ids always and maker p2s only when local `execution_mode` is presplit; when a
    list row lacks a decodable `offer1…`, heal calls `get_offer` so watches are
    not stuck coin-only; no Dexie lifecycle). Dexie lifecycle remains
    `publish_venue=dexie` only and applies through the same
@@ -97,9 +99,11 @@ never the operator transport.
    registered atomically at post, backfilled once on schema open via
    `schema_meta` (`watch_venue_backfill_v2`), and healed each reconcile via
    `prepare_market_reconcile_local` + heal-only Dexie fetch. Coin-ops excludes durable
-   `kind='coin'` watch ids and durable `kind='p2'` maker puzzle hashes inside
-   `list_spendable_coins` (local on-chain `puzzle_hash`; no network expand).
-   Explicit CLI coin ids are refused when they match durable maker coin watches.
+   `kind='coin'` watch ids only inside `list_spendable_coins`. Per-offer `kind='p2'`
+   watches are WS lifecycle / inventory-stale signals: seeded only when metadata is
+   `is_presplit_like` (unique CONDITIONS puzzle hash). Direct never seeds them because
+   `maker_puzzle_hash` is the shared vault inventory hash. Explicit CLI coin ids are
+   refused when they match durable maker coin watches.
 
 ## Consequences
 

greenfloor-engine/src/coin_ops/execution/context.rs

@@ -24,23 +24,24 @@ pub struct CoinOpExecContext {
     pub base_unit_mojo_multiplier: i64,
     pub combine_input_cap: i64,
     pub watched_coin_ids: HashSet<String>,
-    /// Durable maker puzzle hashes (`kind='p2'` watches) for local spend exclusion.
-    pub watched_p2s: HashSet<String>,
     #[cfg(test)]
     pub test_overrides: CoinOpTestOverrides,
 }
 
 impl CoinOpExecContext {
     /// Build execution context from an owned gated operator market.
     ///
+    /// Coin-ops spendable exclusion uses durable **coin-id** watches only.
+    /// Per-offer `kind='p2'` watches are for WS lifecycle; Direct inventory
+    /// hashes must never lock coin-ops selection (ADR 0019).
+    ///
     /// # Errors
     ///
     /// Returns an error if asset resolution fails.
     pub async fn from_gated_market(
         gated: GatedOperatorMarket,
         canonical_base_asset: Option<&str>,
         watched_coin_ids: HashSet<String>,
-        watched_p2s: HashSet<String>,
         #[cfg(test)] test_overrides: CoinOpTestOverrides,
     ) -> SignerResult<Self> {
         let resolver = gated.asset_resolver();
@@ -54,7 +55,6 @@ impl CoinOpExecContext {
             gated,
             resolved_base_asset_id,
             watched_coin_ids,
-            watched_p2s,
             #[cfg(test)]
             test_overrides,
         })
@@ -87,7 +87,7 @@ impl CoinOpExecContext {
             .await
     }
 
-    /// List spendable coins, excluding durable maker coin-id and p2 watches.
+    /// List spendable coins, excluding durable maker **coin-id** watches.
     ///
     /// # Errors
     ///
@@ -98,7 +98,6 @@ impl CoinOpExecContext {
             return Ok(exclude_watched_spendable(
                 coins.iter().cloned(),
                 &self.watched_coin_ids,
-                &self.watched_p2s,
             ));
         }
         let coins = list_wallet_unspent_coins_for_signer(
@@ -111,7 +110,6 @@ impl CoinOpExecContext {
         Ok(exclude_watched_spendable(
             wallet_coins_to_spendable(&coins, self.gated.market_row.base_asset.trim()),
             &self.watched_coin_ids,
-            &self.watched_p2s,
         ))
     }
 

greenfloor-engine/src/coin_ops/execution/helpers.rs

@@ -2,7 +2,6 @@ use std::collections::HashSet;
 
 use crate::coin_ops::{coin_op_min_amount_mojos, SpendableCoin};
 use crate::coinset::WalletUnspentCoin;
-use crate::hex::normalize_hex_id;
 
 pub(crate) fn wallet_coins_to_spendable(
     coins: &[WalletUnspentCoin],
@@ -22,28 +21,18 @@ pub(crate) fn wallet_coins_to_spendable(
         .collect()
 }
 
-/// Drop durable maker watches (coin id or on-chain p2) from a spendable set.
+/// Drop durable maker **coin-id** watches from a spendable set.
 ///
-/// When `watched_p2s` is non-empty, coins with an empty `puzzle_hash` are excluded
-/// (fail closed) until the wallet path populates on-chain puzzle hashes.
+/// Coin-ops does not exclude by `kind='p2'` watches: Direct maker puzzle hashes
+/// are shared vault inventory hashes and would lock the entire asset balance.
+/// Offer locks are coin-id watches registered at post (ADR 0019).
 pub(crate) fn exclude_watched_spendable(
     coins: impl IntoIterator<Item = SpendableCoin>,
     watched_coin_ids: &HashSet<String>,
-    watched_p2s: &HashSet<String>,
 ) -> Vec<SpendableCoin> {
     coins
         .into_iter()
-        .filter(|coin| {
-            let id = coin.id.to_ascii_lowercase();
-            if watched_coin_ids.contains(&id) {
-                return false;
-            }
-            if watched_p2s.is_empty() {
-                return true;
-            }
-            let p2 = normalize_hex_id(&coin.puzzle_hash);
-            !p2.is_empty() && !watched_p2s.contains(&p2)
-        })
+        .filter(|coin| !watched_coin_ids.contains(&coin.id.to_ascii_lowercase()))
         .collect()
 }
 
@@ -89,26 +78,17 @@ mod tests {
     }
 
     #[test]
-    fn exclude_watched_spendable_drops_coin_id_and_p2_matches() {
+    fn exclude_watched_spendable_drops_coin_id_matches_only() {
         let watched_coins = HashSet::from(["aa".repeat(32)]);
-        let watched_p2s = HashSet::from(["bb".repeat(32)]);
         let coins = vec![
             SpendableCoin::new("aa".repeat(32), 1000),
             SpendableCoin::with_puzzle_hash("cc".repeat(32), 2000, "bb".repeat(32)),
             SpendableCoin::with_puzzle_hash("dd".repeat(32), 3000, "ee".repeat(32)),
             SpendableCoin::new("ff".repeat(32), 4000),
         ];
-        let kept = exclude_watched_spendable(coins, &watched_coins, &watched_p2s);
-        assert_eq!(kept.len(), 1);
-        assert_eq!(kept[0].id, "dd".repeat(32));
-    }
-
-    #[test]
-    fn exclude_watched_spendable_empty_p2_fails_closed_when_p2_watches_exist() {
-        let watched_p2s = HashSet::from(["bb".repeat(32)]);
-        let coins = vec![SpendableCoin::new("ff".repeat(32), 4000)];
-        let kept = exclude_watched_spendable(coins, &HashSet::default(), &watched_p2s);
-        assert!(kept.is_empty());
+        let kept = exclude_watched_spendable(coins, &watched_coins);
+        assert_eq!(kept.len(), 3);
+        assert!(kept.iter().all(|coin| coin.id != "aa".repeat(32)));
     }
 
     #[test]

greenfloor-engine/src/coin_ops/execution/managed/runner.rs

@@ -54,13 +54,11 @@ pub fn execute_managed_coin_op_plans<'a>(
     gated: GatedOperatorMarket,
     plans: &'a [CoinOpPlan],
     watched_coin_ids: &'a HashSet<String>,
-    watched_p2s: &'a HashSet<String>,
 ) -> ManagedCoinOpPlansFuture<'a> {
     Box::pin(execute_managed_coin_op_plans_async(
         gated,
         plans,
         watched_coin_ids,
-        watched_p2s,
         #[cfg(test)]
         CoinOpTestOverrides::default(),
     ))
@@ -73,14 +71,12 @@ pub fn execute_managed_coin_op_plans_with_test_overrides<'a>(
     gated: GatedOperatorMarket,
     plans: &'a [CoinOpPlan],
     watched_coin_ids: &'a HashSet<String>,
-    watched_p2s: &'a HashSet<String>,
     test_overrides: CoinOpTestOverrides,
 ) -> ManagedCoinOpPlansFuture<'a> {
     Box::pin(execute_managed_coin_op_plans_async(
         gated,
         plans,
         watched_coin_ids,
-        watched_p2s,
         test_overrides,
     ))
 }
@@ -89,7 +85,6 @@ async fn execute_managed_coin_op_plans_async(
     gated: GatedOperatorMarket,
     plans: &[CoinOpPlan],
     watched_coin_ids: &HashSet<String>,
-    watched_p2s: &HashSet<String>,
     #[cfg(test)] test_overrides: CoinOpTestOverrides,
 ) -> CoinOpExecutionResult {
     if gated.market_row.receive_address.trim().is_empty() {
@@ -110,7 +105,6 @@ async fn execute_managed_coin_op_plans_async(
         gated,
         None,
         watched_coin_ids.iter().cloned().collect(),
-        watched_p2s.iter().cloned().collect(),
         #[cfg(test)]
         test_overrides,
     )

greenfloor-engine/src/coin_ops/execution/managed/tests.rs

@@ -68,7 +68,6 @@ fn test_exec_context(
         base_unit_mojo_multiplier: 1_000,
         combine_input_cap: resolve_combine_input_cap(),
         watched_coin_ids: HashSet::new(),
-        watched_p2s: HashSet::new(),
         test_overrides: CoinOpTestOverrides::new(
             Some(spendable),
             mixed_split_operation_id.map(str::to_string),
@@ -112,7 +111,6 @@ async fn execute_managed_coin_op_plans_skips_when_receive_address_missing() {
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
         &HashSet::<String>::default(),
-        &HashSet::<String>::default(),
     )
     .await;
 
@@ -142,7 +140,6 @@ async fn execute_managed_coin_op_plans_dry_run_plans_without_execution() {
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
         &HashSet::<String>::default(),
-        &HashSet::<String>::default(),
     )
     .await;
 
@@ -169,7 +166,6 @@ async fn execute_managed_coin_op_plans_skips_invalid_plans() {
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
         &HashSet::<String>::default(),
-        &HashSet::<String>::default(),
     )
     .await;
 
@@ -194,7 +190,6 @@ async fn execute_managed_coin_op_plans_executes_split_and_combine_via_runner_ove
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
         &HashSet::<String>::default(),
-        &HashSet::<String>::default(),
         CoinOpTestOverrides::new(
             Some(vec![
                 SpendableCoin::new(test_coin_id('a'), 100_000),
@@ -223,7 +218,7 @@ async fn execute_managed_coin_op_plans_executes_split_and_combine_via_runner_ove
 }
 
 #[tokio::test]
-async fn execute_managed_coin_op_plans_skips_when_spendable_matches_watched_p2() {
+async fn execute_managed_coin_op_plans_ignores_p2_watches_for_spendable_exclusion() {
     let dir = tempfile::tempdir().expect("tempdir");
     let bundle = minimal_program_bundle(&dir);
     let mut market = sample_market("xch1test");
@@ -232,71 +227,70 @@ async fn execute_managed_coin_op_plans_skips_when_spendable_matches_watched_p2()
         sample_plan(CoinOpKind::Split),
         sample_plan(CoinOpKind::Combine),
     ];
-    let maker_p2 = "ef".repeat(32);
-    let watched_p2s = HashSet::from([maker_p2.clone()]);
+    // Shared inventory-style p2 must not lock every coin; coin-id watches do.
+    let shared_p2 = "ef".repeat(32);
 
     let empty_index = empty_cat_ticker_index();
     let result = execute_managed_coin_op_plans_with_test_overrides(
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
         &HashSet::<String>::default(),
-        &watched_p2s,
         CoinOpTestOverrides::new(
             Some(vec![
-                SpendableCoin::with_puzzle_hash(test_coin_id('a'), 100_000, maker_p2.clone()),
-                SpendableCoin::with_puzzle_hash(test_coin_id('b'), 10_000, maker_p2.clone()),
-                SpendableCoin::with_puzzle_hash(test_coin_id('c'), 10_000, maker_p2),
+                SpendableCoin::with_puzzle_hash(test_coin_id('a'), 100_000, shared_p2.clone()),
+                SpendableCoin::with_puzzle_hash(test_coin_id('b'), 10_000, shared_p2.clone()),
+                SpendableCoin::with_puzzle_hash(test_coin_id('c'), 10_000, shared_p2),
             ]),
             Some("managed-op-test".to_string()),
         ),
     )
     .await;
 
-    assert_eq!(result.executed_count, 0);
-    assert!(result.items.iter().any(|item| {
-        item.op_type == "split" && item.reason == "no_spendable_split_coin_available"
-    }));
-    assert!(result.items.iter().any(|item| {
-        item.op_type == "combine" && item.reason == "no_spendable_combine_coin_available"
-    }));
+    assert_eq!(result.executed_count, 2);
+    assert!(result
+        .items
+        .iter()
+        .any(|item| item.reason == "signer_split_submitted"));
+    assert!(result
+        .items
+        .iter()
+        .any(|item| item.reason == "signer_combine_submitted"));
 }
 
 #[tokio::test]
-async fn execute_managed_coin_op_plans_excludes_empty_puzzle_hash_when_p2_watches_exist() {
+async fn execute_managed_combine_locks_only_watched_coin_ids() {
     let dir = tempfile::tempdir().expect("tempdir");
     let bundle = minimal_program_bundle(&dir);
     let mut market = sample_market("xch1test");
     market.base_asset = test_coin_id('f');
-    let plans = vec![
-        sample_plan(CoinOpKind::Split),
-        sample_plan(CoinOpKind::Combine),
-    ];
-    let watched_p2s = HashSet::from(["ef".repeat(32)]);
+    let locked = test_coin_id('a');
+    let free_b = test_coin_id('b');
+    let free_c = test_coin_id('c');
+    let watched_coins = HashSet::from([locked.clone()]);
+    let shared_p2 = "ef".repeat(32);
+    let plans = vec![sample_plan(CoinOpKind::Combine)];
 
     let empty_index = empty_cat_ticker_index();
     let result = execute_managed_coin_op_plans_with_test_overrides(
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
-        &HashSet::<String>::default(),
-        &watched_p2s,
+        &watched_coins,
         CoinOpTestOverrides::new(
             Some(vec![
-                // Empty puzzle_hash must fail closed when maker p2 watches exist.
-                SpendableCoin::new(test_coin_id('a'), 100_000),
-                SpendableCoin::new(test_coin_id('b'), 10_000),
-                SpendableCoin::new(test_coin_id('c'), 10_000),
+                SpendableCoin::with_puzzle_hash(locked, 10_000, shared_p2.clone()),
+                SpendableCoin::with_puzzle_hash(free_b, 10_000, shared_p2.clone()),
+                SpendableCoin::with_puzzle_hash(free_c, 10_000, shared_p2),
             ]),
             Some("managed-op-test".to_string()),
         ),
     )
     .await;
 
-    assert_eq!(result.executed_count, 0);
-    assert!(result.items.iter().any(|item| {
-        item.op_type == "split" && item.reason == "no_spendable_split_coin_available"
-    }));
+    assert_eq!(result.executed_count, 1);
     assert!(result.items.iter().any(|item| {
-        item.op_type == "combine" && item.reason == "no_spendable_combine_coin_available"
+        item.op_type == "combine"
+            && item.status == "executed"
+            && item.reason == "signer_combine_submitted"
     }));
 }
 
@@ -317,7 +311,6 @@ async fn execute_managed_coin_op_plans_skips_when_all_spendable_coins_are_watche
         sample_gated_market(bundle.program, bundle.signer, &market, empty_index),
         &plans,
         &watched,
-        &HashSet::<String>::default(),
         CoinOpTestOverrides::new(
             Some(vec![
                 SpendableCoin::new(test_coin_id('a'), 100_000),