contractcourt+itest: tests for coop close insta-dispatch

diff --git a/contractcourt/chain_watcher_early_dispatch_test.go b/contractcourt/chain_watcher_early_dispatch_test.go
new file mode 100644
index 0000000..3a71f68
--- /dev/null
+++ b/contractcourt/chain_watcher_early_dispatch_test.go
@@ -0,0 +1,224 @@
+package contractcourt
+
+import (
+	"testing"
+	"time"
+
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/stretchr/testify/require"
+)
+
+// TestEarlyDispatchCoopClose verifies the headline behavior: when a
+// cooperative close spend is first detected on chain in the async path
+// (numConfs > 1), the chain watcher fires the early-notify callback exactly
+// once with a summary that carries IsPending=true. The full N-conf flow
+// still completes normally and produces the regular CooperativeCloseInfo
+// downstream.
+func TestEarlyDispatchCoopClose(t *testing.T) {
+	t.Parallel()
+
+	harness := newChainWatcherTestHarness(
+		t, withRequiredConfs(3), withEarlyCoopCloseCapture(),
+	)
+
+	tx := harness.createCoopCloseTx(5000)
+
+	harness.sendSpend(tx)
+	harness.waitForConfRegistration()
+
+	// The early-dispatch callback must have fired exactly once with a
+	// preliminary close summary that identifies the right tx, channel
+	// point, and close type.
+	harness.waitForEarlyCoopClose(1, time.Second)
+	require.Equal(t, 1, harness.earlyCoopCloseCount(),
+		"exactly one early dispatch expected on first spend detection")
+
+	earlySummary := harness.earlyCoopCloseAt(0)
+	require.True(t, earlySummary.IsPending,
+		"early dispatched summary must have IsPending=true")
+	require.Equal(t, channeldb.CooperativeClose, earlySummary.CloseType)
+	require.Equal(t, tx.TxHash(), earlySummary.ClosingTXID)
+	require.Equal(t, harness.aliceChannel.State().FundingOutpoint,
+		earlySummary.ChanPoint)
+
+	// Drive the close to N confs so the regular post-N-conf dispatch
+	// path also completes; the resulting CooperativeCloseInfo must
+	// reference the same tx.
+	harness.mineBlocks(1)
+	harness.confirmTx(tx, harness.currentHeight)
+
+	closeInfo := harness.waitForCoopClose(5 * time.Second)
+	harness.assertCoopCloseTx(closeInfo, tx)
+}
+
+// TestEarlyDispatchForceCloseNotInvoked verifies that force-close spends do
+// NOT trigger the early-dispatch callback. Force-close paths intentionally
+// stay on the N-confirmation dispatch contract; their CLOSED_CHANNEL event
+// fires from the channel arbitrator's MarkChannelClosed callback at N
+// confs.
+func TestEarlyDispatchForceCloseNotInvoked(t *testing.T) {
+	t.Parallel()
+
+	harness := newChainWatcherTestHarness(
+		t, withRequiredConfs(3), withEarlyCoopCloseCapture(),
+	)
+
+	tx := harness.createRemoteForceCloseTx()
+	harness.sendSpend(tx)
+
+	// processDetectedSpend evaluates the early-dispatch path before
+	// registering the conf ntfn, so once that registration lands the
+	// decision is final: no need for a separate sleep window.
+	harness.waitForConfRegistration()
+	require.Equal(t, 0, harness.earlyCoopCloseCount(),
+		"remote force close must not trigger early dispatch")
+}
+
+// TestEarlyDispatchSkippedOnFastPath verifies that when the chain watcher is
+// in the fast (single-confirmation) path, the early-dispatch callback is NOT
+// invoked. The fast-path's existing dispatchCooperativeClose ->
+// MarkChannelClosed flow already fires CLOSED_CHANNEL at first conf with a
+// fully populated summary (including close initiator from the historical
+// bucket), so an early dispatch here would deliver a duplicate event with an
+// unknown initiator and break the SubscribeChannelEvents contract.
+// EarlyCoopCloseDispatched() must stay false so the channel arbitrator's
+// suppression gate does not fire.
+func TestEarlyDispatchSkippedOnFastPath(t *testing.T) {
+	t.Parallel()
+
+	harness := newChainWatcherTestHarness(
+		t, withRequiredConfs(1), withEarlyCoopCloseCapture(),
+	)
+
+	tx := harness.createCoopCloseTx(5000)
+
+	harness.sendSpend(tx)
+
+	// The fast path dispatches the regular CooperativeCloseInfo
+	// synchronously from processDetectedSpend, so once the coop close
+	// event lands we know the early-dispatch branch (which runs in the
+	// same goroutine, just below the fast-path return) was either taken
+	// or skipped: no sleep window required.
+	closeInfo := harness.waitForCoopClose(5 * time.Second)
+	harness.assertCoopCloseTx(closeInfo, tx)
+
+	require.Equal(t, 0, harness.earlyCoopCloseCount(),
+		"single-conf fast path must not invoke early dispatch")
+	require.False(t, harness.chainWatcher.EarlyCoopCloseDispatched(),
+		"EarlyCoopCloseDispatched must stay false on fast path so the "+
+			"arbitrator still fires NotifyClosedChannel at "+
+			"MarkChannelClosed time")
+}
+
+// TestEarlyDispatchFlagSetAfterAsyncDispatch verifies that
+// EarlyCoopCloseDispatched flips to true once the chain watcher fires the
+// preliminary CLOSED_CHANNEL on the async multi-conf path. This is the gate
+// the channel arbitrator reads in MarkChannelClosed to suppress the duplicate
+// notify; if it didn't flip, subscribers would see two CLOSED_CHANNEL events
+// for the same close.
+func TestEarlyDispatchFlagSetAfterAsyncDispatch(t *testing.T) {
+	t.Parallel()
+
+	harness := newChainWatcherTestHarness(
+		t, withRequiredConfs(3), withEarlyCoopCloseCapture(),
+	)
+
+	require.False(t, harness.chainWatcher.EarlyCoopCloseDispatched(),
+		"flag must be false before any spend is observed")
+
+	tx := harness.createCoopCloseTx(5000)
+	harness.sendSpend(tx)
+	harness.waitForConfRegistration()
+	harness.waitForEarlyCoopClose(1, time.Second)
+
+	require.True(t, harness.chainWatcher.EarlyCoopCloseDispatched(),
+		"flag must flip to true once the early dispatch fires so the "+
+			"arbitrator suppresses the duplicate notify at "+
+			"MarkChannelClosed time")
+}
+
+// TestEarlyDispatchReorgRefiresOnReReplacement verifies the reorg recovery
+// path: once a deep reorg removes the close, the early-dispatch flag is
+// cleared, and the next coop close re-fires the early event with its own
+// summary. This is the contract that lets a subscriber observe each
+// distinct close attempt rather than only the first one.
+func TestEarlyDispatchReorgRefiresOnReReplacement(t *testing.T) {
+	t.Parallel()
+
+	harness := newChainWatcherTestHarness(
+		t, withRequiredConfs(3), withEarlyCoopCloseCapture(),
+	)
+
+	tx1 := harness.createCoopCloseTx(5000)
+	tx2 := harness.createCoopCloseTx(4900)
+
+	// First close detected → early dispatch #1.
+	harness.sendSpend(tx1)
+	harness.waitForConfRegistration()
+	harness.waitForEarlyCoopClose(1, time.Second)
+
+	// Reorg flushes the conf ntfn and resets the flag.
+	harness.triggerReorg(tx1, 2)
+	harness.waitForSpendRegistration()
+
+	// Replacement close detected → early dispatch #2 with the new tx.
+	harness.sendSpend(tx2)
+	harness.waitForConfRegistration()
+	harness.waitForEarlyCoopClose(2, 2*time.Second)
+
+	require.Equal(t, 2, harness.earlyCoopCloseCount(),
+		"reorg + replacement close must re-fire the early dispatch")
+
+	first := harness.earlyCoopCloseAt(0)
+	second := harness.earlyCoopCloseAt(1)
+	require.Equal(t, tx1.TxHash(), first.ClosingTXID,
+		"first early dispatch must reference tx1")
+	require.Equal(t, tx2.TxHash(), second.ClosingTXID,
+		"second early dispatch must reference the replacement tx2")
+}
+
+// TestEarlyDispatchRefiresOnReplacementBeforeNegConf verifies the narrow
+// reorg race where a replacement coop close is processed *before* the old
+// confirmation subscription's NegativeConf has been drained. Without the
+// reconciliation guard in processDetectedSpend, the stale
+// coopCloseEarlyDispatched flag from the first spend would suppress the
+// second early dispatch, and MarkChannelClosed's CloseType-gated
+// suppression would then drop the final notify too — leaving subscribers
+// with only the stale event for the no-longer-tracked txid. The test
+// simulates that ordering by feeding the replacement spend directly
+// without first draining the reorg path.
+func TestEarlyDispatchRefiresOnReplacementBeforeNegConf(t *testing.T) {
+	t.Parallel()
+
+	harness := newChainWatcherTestHarness(
+		t, withRequiredConfs(3), withEarlyCoopCloseCapture(),
+	)
+
+	tx1 := harness.createCoopCloseTx(5000)
+	tx2 := harness.createCoopCloseTx(4900)
+
+	// First close detected → early dispatch #1 with tx1.
+	harness.sendSpend(tx1)
+	harness.waitForConfRegistration()
+	harness.waitForEarlyCoopClose(1, time.Second)
+
+	// A different coop close lands while the watcher is still tracking
+	// tx1 — this is the ordering ziggie flagged: the replacement arrives
+	// before the prior spend's NegativeConf is observed. The watcher
+	// must clear the stale flag and re-fire the early dispatch so the
+	// new tx surfaces over the channel notifier.
+	harness.sendSpend(tx2)
+	harness.waitForConfRegistration()
+	harness.waitForEarlyCoopClose(2, 2*time.Second)
+
+	require.Equal(t, 2, harness.earlyCoopCloseCount(),
+		"replacement spend must re-fire the early dispatch even when "+
+			"NegativeConf has not yet drained")
+
+	require.Equal(t, tx2.TxHash(), harness.earlyCoopCloseAt(1).ClosingTXID,
+		"second early dispatch must reference the replacement tx2")
+	require.True(t, harness.chainWatcher.EarlyCoopCloseDispatched(),
+		"flag must remain set after the replacement's early dispatch "+
+			"so the arbitrator's MarkChannelClosed suppression "+
+			"still fires for the new tx")
+}
diff --git a/contractcourt/chain_watcher_test_harness.go b/contractcourt/chain_watcher_test_harness.go
index 09ab035..e67b84a 100644
--- a/contractcourt/chain_watcher_test_harness.go
+++ b/contractcourt/chain_watcher_test_harness.go
@@ -1,6 +1,7 @@
 package contractcourt
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -46,6 +47,14 @@ type chainWatcherTestHarness struct {
 	// blockbeatProcessed is a channel that signals when a blockbeat has
 	// been processed.
 	blockbeatProcessed chan struct{}
+
+	// earlyCoopCloseMu guards earlyCoopCloseSummaries.
+	earlyCoopCloseMu sync.Mutex
+
+	// earlyCoopCloseSummaries records every invocation of the
+	// notifyEarlyCoopClose callback when captureEarlyCoopClose was
+	// enabled. Tests assert against length and contents.
+	earlyCoopCloseSummaries []*channeldb.ChannelCloseSummary
 }
 
 // mockChainNotifier extends the standard mock with additional channels for
@@ -136,6 +145,11 @@ type harnessOpt func(*harnessConfig)
 // harnessConfig holds configuration for the test harness.
 type harnessConfig struct {
 	requiredConfs fn.Option[uint32]
+
+	// captureEarlyCoopClose, when true, wires a notifyEarlyCoopClose
+	// callback into the chain watcher that records each invocation onto
+	// the harness's earlyCoopCloseSummaries slice for assertions.
+	captureEarlyCoopClose bool
 }
 
 // withRequiredConfs sets the number of confirmations required for channel
@@ -146,6 +160,15 @@ func withRequiredConfs(confs uint32) harnessOpt {
 	}
 }
 
+// withEarlyCoopCloseCapture enables recording of every invocation of the
+// chain watcher's notifyEarlyCoopClose callback on the harness so tests can
+// assert when, how often, and with which summary the early-dispatch fires.
+func withEarlyCoopCloseCapture() harnessOpt {
+	return func(cfg *harnessConfig) {
+		cfg.captureEarlyCoopClose = true
+	}
+}
+
 // newChainWatcherTestHarness creates a new test harness for chain watcher
 // tests.
 func newChainWatcherTestHarness(t *testing.T,
@@ -194,13 +217,38 @@ func newChainWatcherTestHarnessFromReporter(t *testing.T,
 		spendRegistered: make(chan struct{}, 10),
 	}
 
+	harness := &chainWatcherTestHarness{
+		t:                  reporter,
+		aliceChannel:       aliceChannel,
+		bobChannel:         bobChannel,
+		notifier:           notifier,
+		currentHeight:      100,
+		blockbeatProcessed: make(chan struct{}),
+	}
+
+	// If the test wants to observe early-dispatch invocations, install
+	// a callback that records each summary onto the harness.
+	var notifyEarlyCoopClose func(*channeldb.ChannelCloseSummary)
+	if cfg.captureEarlyCoopClose {
+		notifyEarlyCoopClose = func(
+			s *channeldb.ChannelCloseSummary) {
+
+			harness.earlyCoopCloseMu.Lock()
+			harness.earlyCoopCloseSummaries = append(
+				harness.earlyCoopCloseSummaries, s,
+			)
+			harness.earlyCoopCloseMu.Unlock()
+		}
+	}
+
 	// Create chain watcher.
 	chainWatcher, err := newChainWatcher(chainWatcherConfig{
-		chanState:           aliceChannel.State(),
-		notifier:            notifier,
-		signer:              aliceChannel.Signer,
-		extractStateNumHint: lnwallet.GetStateNumHint,
-		chanCloseConfs:      cfg.requiredConfs,
+		chanState:            aliceChannel.State(),
+		notifier:             notifier,
+		signer:               aliceChannel.Signer,
+		extractStateNumHint:  lnwallet.GetStateNumHint,
+		chanCloseConfs:       cfg.requiredConfs,
+		notifyEarlyCoopClose: notifyEarlyCoopClose,
 		contractBreach: func(
 			retInfo *lnwallet.BreachRetribution,
 		) error {
@@ -222,16 +270,8 @@ func newChainWatcherTestHarnessFromReporter(t *testing.T,
 	// Subscribe to channel events.
 	chanEvents := chainWatcher.SubscribeChannelEvents()
 
-	harness := &chainWatcherTestHarness{
-		t:                  reporter,
-		aliceChannel:       aliceChannel,
-		bobChannel:         bobChannel,
-		chainWatcher:       chainWatcher,
-		notifier:           notifier,
-		chanEvents:         chanEvents,
-		currentHeight:      100,
-		blockbeatProcessed: make(chan struct{}),
-	}
+	harness.chainWatcher = chainWatcher
+	harness.chanEvents = chanEvents
 
 	// Wait for the initial spend registration that happens in Start().
 	harness.waitForSpendRegistration()
@@ -407,6 +447,56 @@ func (h *chainWatcherTestHarness) mineBlocks(n int32) {
 	h.currentHeight += n
 }
 
+// earlyCoopCloseCount returns how many times the early-dispatch callback has
+// fired since the harness started.
+func (h *chainWatcherTestHarness) earlyCoopCloseCount() int {
+	h.earlyCoopCloseMu.Lock()
+	defer h.earlyCoopCloseMu.Unlock()
+
+	return len(h.earlyCoopCloseSummaries)
+}
+
+// earlyCoopCloseAt returns the early-dispatch summary recorded at the given
+// index. The harness fails the test if the index is out of range.
+func (h *chainWatcherTestHarness) earlyCoopCloseAt(
+	idx int) *channeldb.ChannelCloseSummary {
+
+	h.earlyCoopCloseMu.Lock()
+	defer h.earlyCoopCloseMu.Unlock()
+
+	if idx >= len(h.earlyCoopCloseSummaries) {
+		h.t.Fatalf("expected early-dispatch index %d, only %d "+
+			"summaries recorded", idx,
+			len(h.earlyCoopCloseSummaries))
+	}
+
+	return h.earlyCoopCloseSummaries[idx]
+}
+
+// waitForEarlyCoopClose blocks until at least the supplied count of
+// early-dispatch invocations have been recorded, or the timeout elapses.
+func (h *chainWatcherTestHarness) waitForEarlyCoopClose(want int,
+	timeout time.Duration) {
+
+	h.t.Helper()
+
+	deadline := time.Now().Add(timeout)
+	for {
+		if h.earlyCoopCloseCount() >= want {
+			return
+		}
+		if time.Now().After(deadline) {
+			h.t.Fatalf("expected %d early-dispatch invocations, "+
+				"got %d after %v", want,
+				h.earlyCoopCloseCount(), timeout)
+
+			return
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
 // waitForCoopClose waits for a cooperative close event and returns it.
 func (h *chainWatcherTestHarness) waitForCoopClose(
 	timeout time.Duration) *CooperativeCloseInfo {
diff --git a/docs/release-notes/release-notes-0.21.0.md b/docs/release-notes/release-notes-0.21.0.md
index f8ca9fc..8540d5f 100644
--- a/docs/release-notes/release-notes-0.21.0.md
+++ b/docs/release-notes/release-notes-0.21.0.md
@@ -103,6 +103,16 @@
   to use independent probe payment hashes when probing multiple LSPs, preventing
   later probes from reusing the first probe's CLTV delta.
 
+* [Restored insta-dispatch of `CLOSED_CHANNEL` on the first confirmation of a
+  cooperative close](https://github.com/lightningnetwork/lnd/pull/10794).
+  After the multi-conf reorg-aware close dispatch landed,
+  `SubscribeChannelEvents` no longer emitted `CLOSED_CHANNEL` until the full
+  required confirmation depth was reached. The chain watcher now fires an
+  early `CLOSED_CHANNEL` event over the channel notifier as soon as the coop
+  close spend lands on chain, restoring the v0.20.1 behavior, while the
+  channel arbitrator suppresses the duplicate event that would otherwise be
+  emitted from `MarkChannelClosed` at the final confirmation depth.
+
 # New Features
 
 - [Basic Support](https://github.com/lightningnetwork/lnd/pull/9868) for onion
diff --git a/itest/list_on_test.go b/itest/list_on_test.go
index c8e4244..5b9a752 100644
--- a/itest/list_on_test.go
+++ b/itest/list_on_test.go
@@ -463,6 +463,10 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "zero conf channel open",
 		TestFunc: testZeroConfChannelOpen,
 	},
+	{
+		Name:     "zero conf coop close subscribe events",
+		TestFunc: testZeroConfCoopCloseSubscribeEvents,
+	},
 	{
 		Name:     "option scid alias",
 		TestFunc: testOptionScidAlias,
diff --git a/itest/lnd_zero_conf_close_event_test.go b/itest/lnd_zero_conf_close_event_test.go
new file mode 100644
index 0000000..40c06d1
--- /dev/null
+++ b/itest/lnd_zero_conf_close_event_test.go
@@ -0,0 +1,310 @@
+package itest
+
+import (
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lntest"
+	"github.com/lightningnetwork/lnd/lntest/rpc"
+	"github.com/lightningnetwork/lnd/lntest/wait"
+	"github.com/stretchr/testify/require"
+)
+
+// testZeroConfCoopCloseSubscribeEvents exercises the regression that was
+// reported when production builds switched to a multi-confirmation
+// reorg-aware close dispatch: SubscribeChannelEvents stopped emitting
+// CLOSED_CHANNEL on cooperative closes for zero-conf channels until the
+// close had reached the full confirmation depth, instead of firing at first
+// detection like v0.20.1.
+//
+// The fix wires an early-dispatch callback into the chain watcher that fires
+// a preliminary CLOSED_CHANNEL event over the channel notifier as soon as a
+// coop close spend lands on chain. This test asserts:
+//
+//  1. CLOSED_CHANNEL fires on the SubscribeChannelEvents stream after only
+//     the first confirmation of the close tx (not after the full N=3
+//     depth required by --dev.force-channel-close-confs=3).
+//  2. FULLY_RESOLVED_CHANNEL fires once the close has reached N confs and
+//     the channel arbitrator has finished its resolution flow.
+//  3. Exactly one CLOSED_CHANNEL is delivered — the suppression logic in
+//     the channel arbitrator drops the duplicate that would otherwise fire
+//     from MarkChannelClosed at N confs.
+func testZeroConfCoopCloseSubscribeEvents(ht *lntest.HarnessTest) {
+	// Force coop close to require 3 confs so we exercise the async path
+	// in the chain watcher (the same path production hits via
+	// CloseConfsForCapacity).
+	const requiredConfs = 3
+
+	// Zero-conf channels need option-scid-alias and anchors; force-confs
+	// is what flips us out of the numConfs==1 fast-path so we can verify
+	// the early dispatch is what surfaces the CLOSED_CHANNEL event.
+	nodeArgs := []string{
+		"--protocol.option-scid-alias",
+		"--protocol.zero-conf",
+		"--protocol.anchors",
+		"--dev.force-channel-close-confs=3",
+	}
+
+	alice := ht.NewNode("Alice", nodeArgs)
+	bob := ht.NewNode("Bob", nodeArgs)
+
+	ht.FundCoins(btcutil.SatoshiPerBitcoin, alice)
+	ht.EnsureConnected(alice, bob)
+
+	// A channel acceptor on Bob is needed to allow the zero-conf
+	// negotiation to succeed.
+	acceptStream, cancelAcceptor := bob.RPC.ChannelAcceptor()
+	go acceptChannel(ht.T, true, acceptStream)
+
+	const chanAmt = btcutil.Amount(1_000_000)
+	openParams := lntest.OpenChannelParams{
+		Amt:            chanAmt,
+		Private:        true,
+		CommitmentType: lnrpc.CommitmentType_ANCHORS,
+		ZeroConf:       true,
+	}
+	stream := ht.OpenChannelAssertStream(alice, bob, openParams)
+	cancelAcceptor()
+
+	// Wait for the channel-open update — for zero-conf this arrives
+	// without any blocks needing to be mined.
+	chanPoint := ht.WaitForChannelOpenEvent(stream)
+	ht.AssertChannelInGraph(alice, chanPoint)
+	ht.AssertChannelInGraph(bob, chanPoint)
+
+	// Subscribe Alice to channel events BEFORE we initiate the close so
+	// we capture the full close lifecycle on the wire.
+	chanSub := alice.RPC.SubscribeChannelEvents()
+
+	// Alice initiates the cooperative close; NoWait so the closing tx
+	// just lands in the mempool.
+	closeStream, _ := ht.CloseChannelAssertPending(alice, chanPoint, false)
+
+	// Mine a single block so the close tx confirms once. For a zero-conf
+	// channel the funding tx is still unconfirmed when we initiate the
+	// close, so the mempool holds both the funding tx and the close tx
+	// when we mine the first block. With the fix in place, the chain
+	// watcher's processDetectedSpend should insta-dispatch
+	// CLOSED_CHANNEL over the notifier as soon as the close spend lands,
+	// even though the async path is still waiting on two more confs
+	// before driving MarkChannelClosed.
+	ht.MineBlocksAndAssertNumTxes(1, 2)
+
+	closedSeen := waitForChannelEventOfType(
+		ht, chanSub,
+		lnrpc.ChannelEventUpdate_CLOSED_CHANNEL,
+	)
+	require.NotNil(
+		ht, closedSeen,
+		"CLOSED_CHANNEL must fire after the first conf of the "+
+			"close tx (regression: production was waiting for "+
+			"the full 3-conf depth)",
+	)
+
+	// The CLOSED_CHANNEL summary must reflect a cooperative close
+	// initiated by the local node.
+	closedSummary := closedSeen.GetClosedChannel()
+	require.NotNil(ht, closedSummary,
+		"CLOSED_CHANNEL update must carry a close summary")
+	require.Equal(ht,
+		lnrpc.ChannelCloseSummary_COOPERATIVE_CLOSE,
+		closedSummary.CloseType,
+	)
+	require.Equal(ht,
+		lnrpc.Initiator_INITIATOR_LOCAL,
+		closedSummary.CloseInitiator,
+	)
+
+	// Mine the remaining confs needed to take the close to its final
+	// resolution state.
+	ht.MineBlocksAndAssertNumTxes(requiredConfs-1, 0)
+
+	// Drain the close-channel client stream so the test cleanup path
+	// doesn't hang on it.
+	go func() {
+		for {
+			if _, err := closeStream.Recv(); err != nil {
+				return
+			}
+		}
+	}()
+
+	// FULLY_RESOLVED_CHANNEL must arrive after the close advances
+	// through the channel arbitrator at full N confs. Crucially, no
+	// second CLOSED_CHANNEL event must arrive between the early one and
+	// the FULLY_RESOLVED_CHANNEL — the channel-arbitrator-side
+	// suppression drops the duplicate that MarkChannelClosed would
+	// otherwise emit. We use a stream walker here (rather than
+	// waitForChannelEventOfType, which silently discards every event
+	// that isn't the one being waited for) so that a regression which
+	// re-fires CLOSED_CHANNEL while we wait for FULLY_RESOLVED_CHANNEL
+	// is surfaced as a test failure instead of being swallowed.
+	resolvedSeen := waitForChannelEventForbidClosed(ht, chanSub)
+	require.NotNil(ht, resolvedSeen,
+		"FULLY_RESOLVED_CHANNEL must fire after the close reaches "+
+			"the required confirmation depth")
+
+	// Belt-and-suspenders: also drain a small quiet window after
+	// FULLY_RESOLVED to catch any late-arriving duplicate.
+	assertNoMoreClosedEvents(ht, chanSub, 500*time.Millisecond)
+}
+
+// waitForChannelEventOfType drains the channel events subscription until
+// one of the supplied type lands or the harness's default timeout elapses.
+// Other event types (PENDING_OPEN, OPEN, ACTIVE, INACTIVE, CHANNEL_UPDATE)
+// are expected during a normal close flow and must be tolerated rather than
+// fail the test.
+func waitForChannelEventOfType(ht *lntest.HarnessTest,
+	sub rpc.ChannelEventsClient,
+	want lnrpc.ChannelEventUpdate_UpdateType) *lnrpc.ChannelEventUpdate {
+
+	type result struct {
+		event *lnrpc.ChannelEventUpdate
+		err   error
+	}
+
+	results := make(chan result, 1)
+	deadline := time.After(wait.DefaultTimeout)
+
+	go func() {
+		for {
+			ev, err := sub.Recv()
+			if err != nil {
+				results <- result{err: err}
+				return
+			}
+			if ev.Type == want {
+				results <- result{event: ev}
+				return
+			}
+		}
+	}()
+
+	select {
+	case r := <-results:
+		require.NoErrorf(ht, r.err,
+			"error from channel event stream while waiting "+
+				"for %v", want)
+
+		return r.event
+
+	case <-deadline:
+		ht.Fatalf("timed out waiting for channel event %v", want)
+		return nil
+	}
+}
+
+// waitForChannelEventForbidClosed drains the channel events subscription
+// until FULLY_RESOLVED_CHANNEL lands, failing the test immediately if a
+// second CLOSED_CHANNEL is observed along the way. This is the strict
+// variant of waitForChannelEventOfType for the gap between the early
+// CLOSED_CHANNEL (fired by the chain watcher at first conf) and the
+// FULLY_RESOLVED_CHANNEL (fired by the channel arbitrator at N confs):
+// any CLOSED_CHANNEL in that window would be the duplicate that the
+// MarkChannelClosed suppression in the arbitrator is meant to drop.
+func waitForChannelEventForbidClosed(ht *lntest.HarnessTest,
+	sub rpc.ChannelEventsClient) *lnrpc.ChannelEventUpdate {
+
+	type result struct {
+		event *lnrpc.ChannelEventUpdate
+		err   error
+	}
+
+	results := make(chan result, 1)
+	deadline := time.After(wait.DefaultTimeout)
+
+	go func() {
+		for {
+			ev, err := sub.Recv()
+			if err != nil {
+				results <- result{err: err}
+				return
+			}
+
+			switch ev.Type {
+			case lnrpc.ChannelEventUpdate_FULLY_RESOLVED_CHANNEL:
+				results <- result{event: ev}
+				return
+
+			case lnrpc.ChannelEventUpdate_CLOSED_CHANNEL:
+				results <- result{event: ev}
+				return
+			}
+		}
+	}()
+
+	select {
+	case r := <-results:
+		require.NoErrorf(ht, r.err,
+			"error from channel event stream while waiting "+
+				"for FULLY_RESOLVED_CHANNEL")
+
+		require.Equal(ht,
+			lnrpc.ChannelEventUpdate_FULLY_RESOLVED_CHANNEL,
+			r.event.Type,
+			"unexpected duplicate CLOSED_CHANNEL event observed "+
+				"between the early dispatch and "+
+				"FULLY_RESOLVED_CHANNEL: %v", r.event,
+		)
+
+		return r.event
+
+	case <-deadline:
+		ht.Fatalf("timed out waiting for FULLY_RESOLVED_CHANNEL")
+		return nil
+	}
+}
+
+// assertNoMoreClosedEvents reads from the subscription for the supplied
+// quiet window and fails the test if a CLOSED_CHANNEL event is observed.
+// FULLY_RESOLVED_CHANNEL has already been observed by this point so a
+// second CLOSED_CHANNEL would represent the duplicate-notify regression
+// that the suppression logic is meant to prevent.
+func assertNoMoreClosedEvents(ht *lntest.HarnessTest,
+	sub rpc.ChannelEventsClient, window time.Duration) {
+
+	done := make(chan struct{})
+	defer close(done)
+
+	errs := make(chan error, 1)
+	dups := make(chan *lnrpc.ChannelEventUpdate, 1)
+
+	go func() {
+		for {
+			ev, err := sub.Recv()
+			if err != nil {
+				select {
+				case errs <- err:
+				case <-done:
+				}
+
+				return
+			}
+			if ev.Type ==
+				lnrpc.ChannelEventUpdate_CLOSED_CHANNEL {
+
+				select {
+				case dups <- ev:
+				case <-done:
+				}
+
+				return
+			}
+		}
+	}()
+
+	select {
+	case dup := <-dups:
+		ht.Fatalf("unexpected duplicate CLOSED_CHANNEL event: %v",
+			dup)
+
+	case err := <-errs:
+		// Stream EOF or context cancel is fine; we just want the
+		// quiet window to elapse without a duplicate.
+		_ = err
+
+	case <-time.After(window):
+		// Quiet window elapsed without a duplicate. Pass.
+	}
+}