Improve watcher idempotency for duplicate closure polls

Author: warmidris

packages/stackflow-agent/README.md

@@ -132,4 +132,5 @@ const result = await agent.acceptIncomingTransfer({
 3. `HourlyClosureWatcher` supports two sources:
    - `getPipeState` (recommended): per-pipe read-only polling (`get-pipe`)
    - `listClosureEvents`: event scan mode
-4. For production hardening, add alerting, signer balance checks, and idempotency audit logs.
+4. Watcher retries are idempotent for already-disputed closures (same closure txid is skipped on later polls).
+5. For production hardening, add alerting, signer balance checks, and idempotency audit logs.

packages/stackflow-agent/src/db.js

@@ -197,6 +197,9 @@ export class AgentStateStore {
           dispute_txid = ?
       WHERE txid = ?
     `);
+    this.getClosureStmt = this.db.prepare(`
+      SELECT * FROM closures WHERE txid = ?
+    `);
 
     this.getCursorStmt = this.db.prepare(`
       SELECT last_block_height FROM watcher_cursor WHERE id = 1
@@ -386,6 +389,29 @@ export class AgentStateStore {
     );
   }
 
+  getClosure(txid) {
+    this.assertOpen();
+    const row = this.getClosureStmt.get(assertNonEmptyString(txid, "txid"));
+    if (!row) {
+      return null;
+    }
+    return {
+      txid: row.txid,
+      contractId: row.contract_id,
+      pipeId: row.pipe_id,
+      pipeKey: JSON.parse(row.pipe_key_json),
+      eventName: row.event_name,
+      nonce: row.nonce,
+      closer: row.closer,
+      blockHeight: row.block_height,
+      expiresAt: row.expires_at,
+      closureMyBalance: row.closure_my_balance,
+      disputed: row.disputed === 1,
+      disputeTxid: row.dispute_txid,
+      createdAt: row.created_at,
+    };
+  }
+
   getWatcherCursor() {
     this.assertOpen();
     const row = this.getCursorStmt.get();

packages/stackflow-agent/src/watcher.js

@@ -205,6 +205,7 @@ export class HourlyClosureWatcher {
 
       let closuresFound = 0;
       let disputesSubmitted = 0;
+      let skippedAlreadyDisputed = 0;
       let pipesScanned = 0;
       for (const trackedPipe of trackedPipes) {
         pipesScanned += 1;
@@ -231,7 +232,13 @@ export class HourlyClosureWatcher {
         }
 
         closuresFound += 1;
+        const existingClosure = this.agentService.stateStore.getClosure(closure.txid);
         this.agentService.stateStore.recordClosure(closure);
+        if (existingClosure?.disputed) {
+          skippedAlreadyDisputed += 1;
+          continue;
+        }
+
         const disputeResult = await this.agentService.disputeClosure({
           closureEvent: closure,
           walletPassword: this.walletPassword,
@@ -247,6 +254,7 @@ export class HourlyClosureWatcher {
         pipesScanned,
         closuresFound,
         disputesSubmitted,
+        skippedAlreadyDisputed,
       };
     } finally {
       this.running = false;
@@ -280,6 +288,7 @@ export class HourlyClosureWatcher {
 
       let highestBlock = parseUnsignedBigInt(fromBlockHeight, "fromBlockHeight");
       let disputesSubmitted = 0;
+      let skippedAlreadyDisputed = 0;
       let scanned = 0;
 
       for (const rawEvent of events) {
@@ -290,14 +299,18 @@ export class HourlyClosureWatcher {
           continue;
         }
         scanned += 1;
+        const existingClosure = this.agentService.stateStore.getClosure(closure.txid);
         this.agentService.stateStore.recordClosure(closure);
-
-        const disputeResult = await this.agentService.disputeClosure({
-          closureEvent: closure,
-          walletPassword: this.walletPassword,
-        });
-        if (disputeResult.submitted) {
-          disputesSubmitted += 1;
+        if (existingClosure?.disputed) {
+          skippedAlreadyDisputed += 1;
+        } else {
+          const disputeResult = await this.agentService.disputeClosure({
+            closureEvent: closure,
+            walletPassword: this.walletPassword,
+          });
+          if (disputeResult.submitted) {
+            disputesSubmitted += 1;
+          }
         }
 
         const block = parseUnsignedBigInt(closure.blockHeight, "blockHeight");
@@ -311,6 +324,7 @@ export class HourlyClosureWatcher {
         ok: true,
         scanned,
         disputesSubmitted,
+        skippedAlreadyDisputed,
         fromBlockHeight,
         toBlockHeight: highestBlock.toString(10),
       };

tests/stackflow-agent.test.ts

@@ -208,6 +208,87 @@ describe("stackflow agent", () => {
     store.close();
   });
 
+  it("skips duplicate disputes for closures already marked disputed", async () => {
+    const dbFile = tempDbFile("agent-duplicate-dispute");
+    const store = new AgentStateStore({ dbFile });
+
+    const contractId = "ST1TESTABC.contract";
+    const pipeKey = {
+      "principal-1": "ST1LOCAL",
+      "principal-2": "ST1OTHER",
+      token: null,
+    };
+    const pipeId = buildPipeId({ contractId, pipeKey });
+
+    store.upsertTrackedPipe({
+      pipeId,
+      contractId,
+      pipeKey,
+      localPrincipal: "ST1LOCAL",
+      counterpartyPrincipal: "ST1OTHER",
+      token: null,
+    });
+    store.upsertSignatureState({
+      contractId,
+      pipeKey,
+      forPrincipal: "ST1LOCAL",
+      withPrincipal: "ST1OTHER",
+      token: null,
+      myBalance: "90",
+      theirBalance: "10",
+      nonce: "8",
+      action: "1",
+      actor: "ST1LOCAL",
+      mySignature: "0x" + "11".repeat(65),
+      theirSignature: "0x" + "22".repeat(65),
+      secret: null,
+      validAfter: null,
+      beneficialOnly: false,
+    });
+
+    let disputeCalls = 0;
+    const agent = new StackflowAgentService({
+      stateStore: store,
+      signer: {
+        async submitDispute() {
+          disputeCalls += 1;
+          return { txid: "0xdispute-dup" };
+        },
+        async sip018Sign() {
+          return "0x" + "44".repeat(65);
+        },
+        async callContract() {
+          return { ok: true };
+        },
+      },
+      network: "devnet",
+      disputeOnlyBeneficial: true,
+    });
+
+    const watcher = new HourlyClosureWatcher({
+      agentService: agent,
+      getPipeState: async () => ({
+        "balance-1": "20",
+        "balance-2": "80",
+        "expires-at": "200",
+        nonce: "5",
+        closer: "ST1OTHER",
+      }),
+    });
+
+    const first = await watcher.runOnce();
+    expect(first.disputesSubmitted).toBe(1);
+    expect(first.skippedAlreadyDisputed).toBe(0);
+
+    const second = await watcher.runOnce();
+    expect(second.disputesSubmitted).toBe(0);
+    expect(second.skippedAlreadyDisputed).toBe(1);
+    expect(disputeCalls).toBe(1);
+
+    watcher.stop();
+    store.close();
+  });
+
   it("validates and signs incoming transfer requests", async () => {
     const dbFile = tempDbFile("agent-sign");
     const store = new AgentStateStore({ dbFile });