supplier: in-process wallet auto-consolidation

Periodic setInterval ticker + on-failure backstop from the buildClaimTx
catch in server.ts. Both reuse the existing single-slot lock and call
runConsolidateWallet under the hood — planConsolidate short-circuits to
already-healthy in the steady state so no tx, no fee on a clean wallet.

Driven by deepseek-lane outage 2026-05-28 (supplier wallet drifted to 14
UTxOs with 9 dust 2-AP3X outputs; Claim tx coin-selector picked dust as
working input → chain_submit_failed; lane stalled until manual
tx:consolidate-wallet).

buyer/scripts/monitor-wallets: surface deepseek + kimi as operator wallets.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

buyer/scripts/monitor-wallets.ts

@@ -69,6 +69,20 @@ const OPERATOR_SOURCES: OperatorSource[] = [
     keyVar: "SUPPLIER_PRIV_KEY_HEX",
     role: "supplier",
   },
+  {
+    label: "supplier-deepseek@vector-marketplace",
+    host: "root@91.98.147.172",
+    envPath: "/root/agents-marketplace/supplier/.env.deepseek",
+    keyVar: "SUPPLIER_PRIV_KEY_HEX",
+    role: "supplier",
+  },
+  {
+    label: "supplier-kimi@vector-marketplace",
+    host: "root@91.98.147.172",
+    envPath: "/root/agents-marketplace/supplier/.env.kimi",
+    keyVar: "SUPPLIER_PRIV_KEY_HEX",
+    role: "supplier",
+  },
   {
     label: "buyer@open-webui",
     host: "open-webui",

supplier/src/config.ts

@@ -37,6 +37,10 @@
  *   LIVE_CHAIN             — "1" opts in to live-chain submit/await (default off).
  *                            Any other value (including "true", "yes", "TRUE")
  *                            keeps the supplier in read-only mode for safety.
+ *   WALLET_HEALTH_INTERVAL_MS
+ *                          — periodic wallet self-consolidate tick interval
+ *                            (default 600_000 = 10 min). "0" disables the
+ *                            ticker. Only fires while LIVE_CHAIN=1.
  *
  * On error: throws Error whose message names the offending field (matches
  * supplier-config.test.ts assertion that error message mentions field name).
@@ -87,6 +91,11 @@ export interface SupplierConfig {
    * Only the literal env value "1" sets this to true.
    */
   liveChain: boolean;
+  /**
+   * Periodic wallet self-consolidate tick interval, ms. 0 disables the ticker.
+   * Default 600_000 (10 min). Only fires while liveChain=true.
+   */
+  walletHealthIntervalMs: number;
 }
 
 function requireField(env: Record<string, string | undefined>, name: string): string {
@@ -215,6 +224,16 @@ export function loadConfig(env: Record<string, string | undefined>): SupplierCon
   // in safe read-only mode. Real submissions require explicit opt-in.
   const liveChain = env.LIVE_CHAIN === "1";
 
+  // WALLET_HEALTH_INTERVAL_MS accepts any non-negative integer; "0" disables.
+  const whIntervalStr = env.WALLET_HEALTH_INTERVAL_MS;
+  let walletHealthIntervalMs = 600_000;
+  if (whIntervalStr !== undefined && whIntervalStr !== "") {
+    if (!NON_NEG_INT_RE.test(whIntervalStr)) {
+      throw new Error("loadConfig: WALLET_HEALTH_INTERVAL_MS must be a non-negative integer");
+    }
+    walletHealthIntervalMs = Number(whIntervalStr);
+  }
+
   return {
     supplierPrivKeyHex,
     ogmiosUrl,
@@ -231,5 +250,6 @@ export function loadConfig(env: Record<string, string | undefined>): SupplierCon
     capabilityKind,
     llmBackend,
     liveChain,
+    walletHealthIntervalMs,
   };
 }

supplier/src/index.ts

@@ -21,6 +21,7 @@ import { SupplierState } from "./state.js";
 import { JobStore } from "./jobs.js";
 import { createApp, type SupplierDeps } from "./server.js";
 import { buildChainProvider } from "./chain.js";
+import { startWalletHealthTicker } from "./walletHealth.js";
 
 async function main(): Promise<void> {
   const cfg = loadConfig(process.env);
@@ -61,10 +62,22 @@ async function main(): Promise<void> {
   );
   evictInterval.unref();
 
+  // Wallet self-consolidation: only meaningful when chain writes are enabled.
+  // 0 disables. planConsolidate short-circuits to already-healthy in the
+  // common case so this is a no-op on a clean wallet.
+  const walletHealthTicker =
+    cfg.liveChain && cfg.walletHealthIntervalMs > 0
+      ? startWalletHealthTicker(
+          { chain, state, supplierKey },
+          { intervalMs: cfg.walletHealthIntervalMs },
+        )
+      : null;
+
   const shutdown = (signal: string) => {
     // eslint-disable-next-line no-console
     console.log(`received ${signal}, shutting down`);
     clearInterval(evictInterval);
+    walletHealthTicker?.stop();
     state.markOffline();
     server.close(() => process.exit(0));
     setTimeout(() => process.exit(1), 5_000).unref();

supplier/src/server.ts

@@ -45,6 +45,7 @@ import type { SupplierConfig } from "./config.js";
 import { JobStore } from "./jobs.js";
 import { runChatJob, runTtsJob } from "./jobRunner.js";
 import { healthzRouter } from "./routes/healthz.js";
+import { triggerOnFailureConsolidate } from "./walletHealth.js";
 
 export interface SupplierDeps {
   chain: ChainProvider;
@@ -307,6 +308,15 @@ function makeChatHandler(deps: ResolvedDeps) {
         });
       } catch (err) {
         deps.state.release();
+        // Backstop the periodic wallet-health ticker: most Claim build failures
+        // we see in practice are wallet-shape problems (fragmentation, dust).
+        // Fire-and-forget a debounced consolidate so the NEXT buyer retry
+        // finds a clean 2-UTxO wallet. Safe no-op if already healthy.
+        triggerOnFailureConsolidate({
+          chain: deps.chain,
+          state: deps.state,
+          supplierKey: deps.supplierKey,
+        });
         return jsonError(res, 503, "chain_submit_failed",
           `Claim tx submit failed: ${(err as Error).message}`);
       }

supplier/src/walletHealth.ts

@@ -0,0 +1,151 @@
+/**
+ * walletHealth.ts — Periodic + on-failure wallet auto-consolidation.
+ *
+ * The supplier wallet drifts into a fragmented shape over time as Claim/
+ * Submit/Accept change outputs accumulate. Once a dust UTxO becomes the
+ * coin-selected working input, buildClaimTx fails with `chain_submit_failed`
+ * and the lane stalls until an operator runs `tx:consolidate-wallet`.
+ *
+ * Two layers of self-healing:
+ *   startWalletHealthTicker — periodic background tick (every N ms). When
+ *     the supplier is idle, runs runConsolidateWallet. planConsolidate
+ *     returns `already-healthy` cheaply for clean 2-UTxO wallets, so this
+ *     is a no-op in steady state — no tx, no fee.
+ *   triggerOnFailureConsolidate — fire-and-forget consolidate from the
+ *     Claim-failure path. Debounced so back-to-back failures don't queue
+ *     multiple consolidates. Returns 503 to the buyer for THIS request; the
+ *     next buyer retry (~30-60s later) finds the wallet healthy.
+ *
+ * State-lock discipline: both layers use state.tryAcquire with a synthetic
+ * escrowRef so they share the existing single-slot lock with real Claims.
+ * If a Claim is in flight, the consolidation skips this tick.
+ */
+
+import type { ChainProvider } from "@marketplace/shared/chain";
+import type { WalletKey } from "@marketplace/shared/tx";
+import {
+  runConsolidateWallet,
+  type ConsolidateWalletFlowParams,
+  type ConsolidateWalletFlowResult,
+} from "@marketplace/shared/tx/server";
+
+import type { SupplierState } from "./state.js";
+
+const WALLET_HEALTH_LOCK_REF = "__wallet_consolidate__";
+const DEFAULT_COLLATERAL_LOVELACE = 5_000_000n;
+const DEFAULT_AWAIT_TIMEOUT_MS = 90_000;
+const DEFAULT_ON_FAILURE_DEBOUNCE_MS = 5 * 60_000;
+
+export type ConsolidateFn = (
+  params: ConsolidateWalletFlowParams,
+) => Promise<ConsolidateWalletFlowResult>;
+
+export interface WalletHealthDeps {
+  chain: ChainProvider;
+  state: SupplierState;
+  supplierKey: WalletKey;
+  /** Test-injectable. Defaults to runConsolidateWallet from @marketplace/shared. */
+  consolidate?: ConsolidateFn;
+}
+
+export interface WalletHealthOptions {
+  intervalMs: number;
+  collateralLovelace?: bigint;
+  awaitTimeoutMs?: number;
+  /** Override for tests; defaults to console.log. */
+  log?: (line: string) => void;
+}
+
+export interface WalletHealthTicker {
+  stop(): void;
+}
+
+const inFlight = new WeakSet<SupplierState>();
+const lastOnFailureAt = new WeakMap<SupplierState, number>();
+
+function defaultLog(line: string): void {
+  // eslint-disable-next-line no-console
+  console.log(`[wallet-health] ${line}`);
+}
+
+async function consolidateOnce(
+  deps: WalletHealthDeps,
+  opts: { collateralLovelace: bigint; awaitTimeoutMs: number; log: (line: string) => void },
+): Promise<void> {
+  const { state } = deps;
+  if (inFlight.has(state)) {
+    opts.log("skip: another consolidate already in flight");
+    return;
+  }
+  if (!state.tryAcquire(WALLET_HEALTH_LOCK_REF)) {
+    opts.log("skip: supplier busy");
+    return;
+  }
+  inFlight.add(state);
+  const consolidate: ConsolidateFn = deps.consolidate ?? runConsolidateWallet;
+  try {
+    const result = await consolidate({
+      chain: deps.chain,
+      walletKey: deps.supplierKey,
+      collateralLovelace: opts.collateralLovelace,
+      awaitTimeoutMs: opts.awaitTimeoutMs,
+      log: opts.log,
+    });
+    if (result.reason === "already-healthy") {
+      opts.log("already-healthy");
+    } else {
+      opts.log(`consolidated: ${result.reason} txHash=${result.txHash}`);
+    }
+  } catch (err) {
+    opts.log(`consolidate failed: ${(err as Error).message}`);
+  } finally {
+    inFlight.delete(state);
+    state.release();
+  }
+}
+
+export function startWalletHealthTicker(
+  deps: WalletHealthDeps,
+  opts: WalletHealthOptions,
+): WalletHealthTicker {
+  const collateralLovelace = opts.collateralLovelace ?? DEFAULT_COLLATERAL_LOVELACE;
+  const awaitTimeoutMs = opts.awaitTimeoutMs ?? DEFAULT_AWAIT_TIMEOUT_MS;
+  const log = opts.log ?? defaultLog;
+
+  const handle = setInterval(() => {
+    void consolidateOnce(deps, { collateralLovelace, awaitTimeoutMs, log });
+  }, opts.intervalMs);
+  handle.unref();
+
+  return {
+    stop(): void {
+      clearInterval(handle);
+    },
+  };
+}
+
+export function triggerOnFailureConsolidate(
+  deps: WalletHealthDeps,
+  opts: Partial<WalletHealthOptions> & { debounceMs?: number } = {},
+): void {
+  const { state } = deps;
+  const debounceMs = opts.debounceMs ?? DEFAULT_ON_FAILURE_DEBOUNCE_MS;
+  const now = Date.now();
+  const last = lastOnFailureAt.get(state) ?? 0;
+  if (now - last < debounceMs) return;
+  lastOnFailureAt.set(state, now);
+
+  const log = opts.log ?? defaultLog;
+  log("on-failure trigger: scheduling consolidate");
+  void consolidateOnce(deps, {
+    collateralLovelace: opts.collateralLovelace ?? DEFAULT_COLLATERAL_LOVELACE,
+    awaitTimeoutMs: opts.awaitTimeoutMs ?? DEFAULT_AWAIT_TIMEOUT_MS,
+    log,
+  });
+}
+
+/** Test-only: reset the in-flight + debounce maps between tests. */
+export function _resetWalletHealthForTests(state: SupplierState): void {
+  inFlight.delete(state);
+  lastOnFailureAt.delete(state);
+}

tests/unit/supplier-walletHealth.test.ts

@@ -0,0 +1,221 @@
+/**
+ * supplier-walletHealth.test.ts — Tests for the periodic + on-failure
+ * wallet auto-consolidation hooks.
+ *
+ * Coverage:
+ *   - Ticker calls consolidate when idle, logs already-healthy
+ *   - Ticker skips when the supplier state is busy
+ *   - Ticker releases the lock around runConsolidateWallet (success + failure)
+ *   - On-failure trigger fires consolidate the first time
+ *   - On-failure trigger debounces back-to-back invocations
+ *   - On-failure trigger respects the state lock
+ *
+ * The consolidate function is injected via deps so the test never touches
+ * a real Ogmios.
+ */
+
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+
+import { SupplierState } from "../../supplier/src/state.js";
+import {
+  startWalletHealthTicker,
+  triggerOnFailureConsolidate,
+  _resetWalletHealthForTests,
+  type ConsolidateFn,
+} from "../../supplier/src/walletHealth.js";
+
+const FAKE_CHAIN = {} as never;
+const FAKE_KEY = {
+  privateKeyHex: "0".repeat(64),
+  pubKeyHex: "1".repeat(64),
+  pubKeyHash: "2".repeat(56),
+  address: "addr1_fake",
+} as never;
+const ESCROW_REF = `${"f".repeat(64)}#0`;
+
+const healthyResult = {
+  txHash: null,
+  collateralRef: null,
+  workingRef: null,
+  reason: "already-healthy" as const,
+  inputCount: 2,
+  totalLovelaceIn: 200_000_000n,
+  collateralOutputLovelace: 5_000_000n,
+  workingOutputLovelace: 195_000_000n,
+};
+
+const consolidatedResult = {
+  txHash: "a".repeat(64),
+  collateralRef: `${"a".repeat(64)}#0`,
+  workingRef: `${"a".repeat(64)}#1`,
+  reason: "consolidate" as const,
+  inputCount: 14,
+  totalLovelaceIn: 328_530_235n,
+  collateralOutputLovelace: 5_000_000n,
+  workingOutputLovelace: 322_530_235n,
+};
+
+describe("walletHealth — ticker", () => {
+  let state: SupplierState;
+  let logLines: string[];
+  let log: (line: string) => void;
+  let consolidate: ReturnType<typeof vi.fn> & ConsolidateFn;
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+    state = new SupplierState();
+    _resetWalletHealthForTests(state);
+    logLines = [];
+    log = (line) => logLines.push(line);
+    consolidate = vi.fn() as never;
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("calls consolidate when idle and logs already-healthy", async () => {
+    consolidate.mockResolvedValue(healthyResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    await vi.advanceTimersByTimeAsync(1_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
+
+    expect(logLines).toContain("already-healthy");
+    expect(state.snapshot().status).toBe("free");
+
+    ticker.stop();
+  });
+
+  it("skips when supplier state is busy", async () => {
+    state.tryAcquire(ESCROW_REF);
+    consolidate.mockResolvedValue(healthyResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    await vi.advanceTimersByTimeAsync(1_000);
+    await vi.advanceTimersByTimeAsync(0);
+
+    expect(consolidate).not.toHaveBeenCalled();
+    expect(logLines).toContain("skip: supplier busy");
+    expect(state.snapshot().status).toBe("working");
+
+    ticker.stop();
+  });
+
+  it("releases the lock after a successful consolidate", async () => {
+    consolidate.mockResolvedValue(consolidatedResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    await vi.advanceTimersByTimeAsync(1_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
+
+    expect(state.snapshot().status).toBe("free");
+    expect(logLines.some((l) => l.startsWith("consolidated:"))).toBe(true);
+
+    ticker.stop();
+  });
+
+  it("releases the lock when consolidate throws", async () => {
+    consolidate.mockRejectedValue(new Error("ogmios unreachable"));
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    await vi.advanceTimersByTimeAsync(1_000);
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
+
+    expect(state.snapshot().status).toBe("free");
+    expect(logLines.some((l) => l.includes("consolidate failed"))).toBe(true);
+
+    ticker.stop();
+  });
+
+  it("stop() prevents further ticks", async () => {
+    consolidate.mockResolvedValue(healthyResult);
+
+    const ticker = startWalletHealthTicker(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { intervalMs: 1_000, log },
+    );
+
+    ticker.stop();
+    await vi.advanceTimersByTimeAsync(5_000);
+
+    expect(consolidate).not.toHaveBeenCalled();
+  });
+});
+
+describe("walletHealth — triggerOnFailureConsolidate", () => {
+  let state: SupplierState;
+  let logLines: string[];
+  let log: (line: string) => void;
+  let consolidate: ReturnType<typeof vi.fn> & ConsolidateFn;
+
+  beforeEach(() => {
+    state = new SupplierState();
+    _resetWalletHealthForTests(state);
+    logLines = [];
+    log = (line) => logLines.push(line);
+    consolidate = vi.fn() as never;
+  });
+
+  it("fires consolidate on first call", async () => {
+    consolidate.mockResolvedValue(consolidatedResult);
+
+    triggerOnFailureConsolidate(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { log, debounceMs: 60_000 },
+    );
+
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
+    expect(state.snapshot().status).toBe("free");
+  });
+
+  it("debounces back-to-back invocations", async () => {
+    consolidate.mockResolvedValue(healthyResult);
+
+    triggerOnFailureConsolidate(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { log, debounceMs: 60_000 },
+    );
+    await vi.waitFor(() => expect(consolidate).toHaveBeenCalledTimes(1));
+
+    triggerOnFailureConsolidate(
+      { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+      { log, debounceMs: 60_000 },
+    );
+    // No second call — second invocation was inside the debounce window.
+    expect(consolidate).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not throw when supplier is busy — just skips", async () => {
+    state.tryAcquire(ESCROW_REF);
+    consolidate.mockResolvedValue(healthyResult);
+
+    expect(() =>
+      triggerOnFailureConsolidate(
+        { chain: FAKE_CHAIN, state, supplierKey: FAKE_KEY, consolidate },
+        { log, debounceMs: 60_000 },
+      ),
+    ).not.toThrow();
+
+    await Promise.resolve();
+    expect(consolidate).not.toHaveBeenCalled();
+    expect(logLines).toContain("skip: supplier busy");
+    expect(state.snapshot().status).toBe("working");
+  });
+});