contractcourt+server: insta-dispatch CLOSED_CHANNEL on first conf

diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go
index eac63cb..1b8b00e 100644
--- a/contractcourt/chain_arbitrator.go
+++ b/contractcourt/chain_arbitrator.go
@@ -165,6 +165,15 @@ type ChainArbitratorConfig struct {
 	// will use to notify the ChannelNotifier about a newly closed channel.
 	NotifyClosedChannel func(wire.OutPoint)
 
+	// NotifyEarlyClosedChannel is invoked by the chain watcher when a
+	// cooperative close spend is first detected on chain, before the close
+	// summary has been persisted to the closed-channel bucket. It allows
+	// the channel notifier to dispatch a CLOSED_CHANNEL event over RPC at
+	// the same depth it did before the multi-confirmation reorg-aware
+	// dispatch was introduced. The follow-up persist + state advance still
+	// waits for the full required confirmation count.
+	NotifyEarlyClosedChannel func(*channeldb.ChannelCloseSummary)
+
 	// NotifyFullyResolvedChannel is a function closure that the
 	// ChainArbitrator will use to notify the ChannelNotifier about a newly
 	// resolved channel. The main difference to NotifyClosedChannel is that
@@ -424,6 +433,18 @@ func (a *arbChannel) ForceCloseChan() (*wire.MsgTx, error) {
 	return closeSummary.CloseTx, nil
 }
 
+// shouldSuppressClosedChannelNotify reports whether MarkChannelClosed should
+// skip firing NotifyClosedChannel because the chain watcher already emitted a
+// preliminary CLOSED_CHANNEL via the early-dispatch path. Only the
+// cooperative-close path is gated; force, breach, and abandon closes never
+// take the early-dispatch path, so their NotifyClosedChannel must always
+// fire from MarkChannelClosed.
+func shouldSuppressClosedChannelNotify(closeType channeldb.ClosureType,
+	earlyDispatched bool) bool {
+
+	return closeType == channeldb.CooperativeClose && earlyDispatched
+}
+
 // newActiveChannelArbitrator creates a new instance of an active channel
 // arbitrator given the state of the target channel.
 func newActiveChannelArbitrator(channel *channeldb.OpenChannel,
@@ -451,7 +472,35 @@ func newActiveChannelArbitrator(channel *channeldb.OpenChannel,
 			if err != nil {
 				return err
 			}
+
+			// In the async multi-conf path the chain watcher
+			// already fires a preliminary CLOSED_CHANNEL event
+			// over the channel notifier as soon as the coop
+			// close spend lands on chain. Suppressing the
+			// duplicate notify here keeps the
+			// SubscribeChannelEvents stream emitting a single
+			// CLOSED_CHANNEL per close, matching the v0.20.1
+			// surface. In the fast path (numConfs == 1) no early
+			// dispatch fires, so we still need to fire the
+			// CLOSED_CHANNEL event from here. Force/breach
+			// closes never take the early-dispatch path and so
+			// always notify here.
+			c.Lock()
+			w := c.activeWatchers[summary.ChanPoint]
+			c.Unlock()
+
+			earlyDispatched := w != nil &&
+				w.EarlyCoopCloseDispatched()
+
+			if shouldSuppressClosedChannelNotify(
+				summary.CloseType, earlyDispatched,
+			) {
+
+				return nil
+			}
+
 			c.cfg.NotifyClosedChannel(summary.ChanPoint)
+
 			return nil
 		},
 		IsPendingClose:        false,
@@ -1145,11 +1194,12 @@ func (c *ChainArbitrator) WatchNewChannel(newChan *channeldb.OpenChannel) error
 					chanPoint, retInfo,
 				)
 			},
-			extractStateNumHint: lnwallet.GetStateNumHint,
-			auxLeafStore:        c.cfg.AuxLeafStore,
-			auxResolver:         c.cfg.AuxResolver,
-			auxCloser:           c.cfg.AuxCloser,
-			chanCloseConfs:      c.cfg.ChannelCloseConfs,
+			extractStateNumHint:  lnwallet.GetStateNumHint,
+			auxLeafStore:         c.cfg.AuxLeafStore,
+			auxResolver:          c.cfg.AuxResolver,
+			auxCloser:            c.cfg.AuxCloser,
+			chanCloseConfs:       c.cfg.ChannelCloseConfs,
+			notifyEarlyCoopClose: c.cfg.NotifyEarlyClosedChannel,
 		},
 	)
 	if err != nil {
@@ -1317,18 +1367,20 @@ func (c *ChainArbitrator) loadOpenChannels() error {
 			return c.cfg.ContractBreach(chanPoint, ret)
 		}
 
+		notifyEarlyClose := c.cfg.NotifyEarlyClosedChannel
 		chainWatcher, err := newChainWatcher(
 			chainWatcherConfig{
-				chanState:           channel,
-				notifier:            c.cfg.Notifier,
-				signer:              c.cfg.Signer,
-				isOurAddr:           c.cfg.IsOurAddress,
-				contractBreach:      breachClosure,
-				extractStateNumHint: lnwallet.GetStateNumHint,
-				auxLeafStore:        c.cfg.AuxLeafStore,
-				auxResolver:         c.cfg.AuxResolver,
-				auxCloser:           c.cfg.AuxCloser,
-				chanCloseConfs:      c.cfg.ChannelCloseConfs,
+				chanState:            channel,
+				notifier:             c.cfg.Notifier,
+				signer:               c.cfg.Signer,
+				isOurAddr:            c.cfg.IsOurAddress,
+				contractBreach:       breachClosure,
+				extractStateNumHint:  lnwallet.GetStateNumHint,
+				auxLeafStore:         c.cfg.AuxLeafStore,
+				auxResolver:          c.cfg.AuxResolver,
+				auxCloser:            c.cfg.AuxCloser,
+				chanCloseConfs:       c.cfg.ChannelCloseConfs,
+				notifyEarlyCoopClose: notifyEarlyClose,
 			},
 		)
 		if err != nil {
diff --git a/contractcourt/chain_arbitrator_test.go b/contractcourt/chain_arbitrator_test.go
index 622686f..d859d2d 100644
--- a/contractcourt/chain_arbitrator_test.go
+++ b/contractcourt/chain_arbitrator_test.go
@@ -219,3 +219,69 @@ func TestResolveContract(t *testing.T) {
 	err = chainArb.ResolveContract(channel.FundingOutpoint)
 	require.NoError(t, err, "second resolve call shouldn't fail")
 }
+
+// TestShouldSuppressClosedChannelNotify pins down the gate that prevents
+// MarkChannelClosed from firing a duplicate NotifyClosedChannel after the
+// chain watcher has already emitted a preliminary CLOSED_CHANNEL via the
+// early-dispatch path. Only the cooperative-close path can be suppressed;
+// every other CloseType (force, breach, abandon) must always notify here
+// regardless of the early-dispatched flag. The fast path (numConfs==1)
+// never sets the early-dispatched flag, so cooperative closes on that path
+// also fall through to NotifyClosedChannel.
+func TestShouldSuppressClosedChannelNotify(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name            string
+		closeType       channeldb.ClosureType
+		earlyDispatched bool
+		wantSuppress    bool
+	}{
+		{
+			name:            "coop close with early dispatch",
+			closeType:       channeldb.CooperativeClose,
+			earlyDispatched: true,
+			wantSuppress:    true,
+		},
+		{
+			name: "coop close without early dispatch " +
+				"(fast path or no watcher)",
+			closeType:       channeldb.CooperativeClose,
+			earlyDispatched: false,
+			wantSuppress:    false,
+		},
+		{
+			name:            "local force close",
+			closeType:       channeldb.LocalForceClose,
+			earlyDispatched: true,
+			wantSuppress:    false,
+		},
+		{
+			name:            "remote force close",
+			closeType:       channeldb.RemoteForceClose,
+			earlyDispatched: true,
+			wantSuppress:    false,
+		},
+		{
+			name:            "breach close",
+			closeType:       channeldb.BreachClose,
+			earlyDispatched: true,
+			wantSuppress:    false,
+		},
+		{
+			name:            "abandoned close",
+			closeType:       channeldb.Abandoned,
+			earlyDispatched: true,
+			wantSuppress:    false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := shouldSuppressClosedChannelNotify(
+				tc.closeType, tc.earlyDispatched,
+			)
+			require.Equal(t, tc.wantSuppress, got)
+		})
+	}
+}
diff --git a/contractcourt/chain_watcher.go b/contractcourt/chain_watcher.go
index e45bb3d..577dcc3 100644
--- a/contractcourt/chain_watcher.go
+++ b/contractcourt/chain_watcher.go
@@ -279,6 +279,16 @@ type chainWatcherConfig struct {
 	// the normal capacity-based scaling. This is only available in
 	// dev/integration builds for testing purposes.
 	chanCloseConfs fn.Option[uint32]
+
+	// notifyEarlyCoopClose, if set, is invoked with a synthesized
+	// ChannelCloseSummary the first time a cooperative close spend is
+	// detected on chain. It dispatches a CLOSED_CHANNEL event over the
+	// channel notifier so RPC subscribers see the close at the same
+	// block depth they did before the multi-confirmation reorg-aware
+	// dispatch was introduced. The follow-up state transition (DB persist
+	// + state machine advance + FULLY_RESOLVED_CHANNEL) still waits for
+	// the full required confirmation depth via the existing async path.
+	notifyEarlyCoopClose func(*channeldb.ChannelCloseSummary)
 }
 
 // chainWatcher is a system that's assigned to every active channel. The duty
@@ -329,6 +339,26 @@ type chainWatcher struct {
 	// ensure that the outpoint+pkscript pair is confirmed before calling
 	// `RegisterSpendNtfn`.
 	fundingConfirmedNtfn *chainntnfs.ConfirmationEvent
+
+	// coopCloseEarlyDispatched is set when we have already insta-dispatched
+	// a preliminary CLOSED_CHANNEL event for a coop close upon first spend
+	// detection. It is cleared on a deep reorg of the close so a re-mined
+	// close still re-fires the early event. The closeObserver goroutine is
+	// the only writer, but the channel arbitrator's MarkChannelClosed
+	// callback reads it from its own goroutine to decide whether to
+	// suppress the duplicate notify at full conf depth, so the field is an
+	// atomic.Bool to make that cross-goroutine read race-free.
+	coopCloseEarlyDispatched atomic.Bool
+}
+
+// EarlyCoopCloseDispatched reports whether the chain watcher already fired the
+// preliminary CLOSED_CHANNEL event for the in-flight cooperative close. The
+// channel arbitrator uses this to gate the duplicate CLOSED_CHANNEL that
+// MarkChannelClosed would otherwise fire at full conf depth. The flag is only
+// ever set on the async multi-conf path; the fast-path (numConfs == 1) leaves
+// it false and so the regular MarkChannelClosed-driven notify still fires.
+func (c *chainWatcher) EarlyCoopCloseDispatched() bool {
+	return c.coopCloseEarlyDispatched.Load()
 }
 
 // newChainWatcher returns a new instance of a chainWatcher for a channel given
@@ -711,13 +741,21 @@ type spendProcessResult struct {
 //
 // For single-confirmation mode (numConfs == 1), it immediately dispatches the
 // close event and returns empty result. For multi-confirmation mode, it
-// registers for confirmations and returns the new pending state.
+// registers for confirmations and returns the new pending state. In the
+// async path, a coop close also triggers an early CLOSED_CHANNEL event over
+// the channel notifier so RPC subscribers see the close at the same depth
+// they did before the multi-confirmation reorg-aware dispatch was introduced.
 func (c *chainWatcher) processDetectedSpend(
 	spend *chainntnfs.SpendDetail, source string,
 	currentPendingSpend *chainntnfs.SpendDetail,
 	currentConfNtfn *chainntnfs.ConfirmationEvent) spendProcessResult {
 
-	// FAST PATH: Single confirmation mode dispatches immediately.
+	// FAST PATH: Single confirmation mode dispatches immediately. In this
+	// mode the existing flow already drives MarkChannelClosed at the
+	// single conf, which fires CLOSED_CHANNEL with a fully populated
+	// summary (including close initiator from the historical bucket), so
+	// the early dispatch is not needed and would actually deliver a
+	// summary with an unknown close initiator to subscribers.
 	if c.handleSpendDispatch(spend, source) {
 		if currentConfNtfn != nil {
 			currentConfNtfn.Cancel()
@@ -727,12 +765,22 @@ func (c *chainWatcher) processDetectedSpend(
 	}
 
 	// ASYNC PATH: Multiple confirmations (production).
+	//
 	// STATE TRANSITION: None -> Pending.
 	log.Infof("ChannelPoint(%v): detected spend from %s, "+
 		"transitioning to %v", c.cfg.chanState.FundingOutpoint,
 		source, spendStatePending)
 
-	// Check for duplicate spend detection.
+	// Reconcile against any spend we're already tracking *before* firing
+	// the preliminary CLOSED_CHANNEL. If a replacement coop close arrives
+	// while the previous spend's NegativeConf has not yet been drained
+	// (e.g. a deep reorg removed the old spend and a different coop close
+	// then confirmed), the early-dispatch flag may still be set from the
+	// stale spend. Clearing it on the replacement path lets the next
+	// maybeDispatchEarlyCoopClose call fire a fresh event for the new tx,
+	// so subscribers observe the replacement instead of being left with
+	// the stale event (which the arbitrator's CloseType-gated suppression
+	// would otherwise let stand).
 	if currentPendingSpend != nil {
 		if *currentPendingSpend.SpenderTxHash == *spend.SpenderTxHash {
 			log.Debugf("ChannelPoint(%v): ignoring duplicate "+
@@ -746,7 +794,10 @@ func (c *chainWatcher) processDetectedSpend(
 			}
 		}
 
-		// Different spend detected. Cancel existing confNtfn.
+		// Different spend detected (e.g. an RBF replacement). Cancel
+		// the existing confNtfn so we can re-register for the new tx,
+		// and clear the early-dispatch flag so the replacement's own
+		// preliminary CLOSED_CHANNEL event below can fire.
 		log.Warnf("ChannelPoint(%v): detected different spend tx %v, "+
 			"replacing pending tx %v",
 			c.cfg.chanState.FundingOutpoint,
@@ -755,8 +806,18 @@ func (c *chainWatcher) processDetectedSpend(
 		if currentConfNtfn != nil {
 			currentConfNtfn.Cancel()
 		}
+
+		c.coopCloseEarlyDispatched.Store(false)
 	}
 
+	// Fire a preliminary CLOSED_CHANNEL event over the channel notifier
+	// as soon as the spend is first detected so SubscribeChannelEvents
+	// subscribers see the close at the same depth they did before the
+	// multi-confirmation reorg-aware dispatch was introduced. The
+	// suppression of the duplicate notify at MarkChannelClosed time is
+	// handled in chain_arbitrator.go via a CloseType check.
+	c.maybeDispatchEarlyCoopClose(spend)
+
 	numConfs := c.requiredConfsForSpend()
 	txid := spend.SpenderTxHash
 
@@ -943,6 +1004,11 @@ func (c *chainWatcher) closeObserver() {
 			confNtfn = nil
 			pendingSpend = nil
 
+			// Clear the early-dispatch flag so a re-mined coop
+			// close re-fires the preliminary CLOSED_CHANNEL event
+			// with its own close summary.
+			c.coopCloseEarlyDispatched.Store(false)
+
 			// Reset the close confirmation height since the spend
 			// was reorged out.
 			err := c.cfg.chanState.ResetCloseConfirmationHeight()
@@ -1347,26 +1413,78 @@ func (c *chainWatcher) requiredConfsForSpend() uint32 {
 	})
 }
 
-// dispatchCooperativeClose processed a detect cooperative channel closure.
-// We'll use the spending transaction to locate our output within the
-// transaction, then clean up the database state. We'll also dispatch a
-// notification to all subscribers that the channel has been closed in this
-// manner.
-func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDetail) error {
-	broadcastTx := commitSpend.SpendingTx
+// isCoopCloseSpend reports whether the supplied spending tx looks like a
+// cooperative close. A coop close has a finalized input sequence number
+// (either MaxTxInSequenceNum or MaxRBFSequence); regular commitment txns
+// carry an obfuscated state hint in the sequence + locktime fields and
+// won't match either constant.
+func isCoopCloseSpend(spendingTx *wire.MsgTx) bool {
+	if len(spendingTx.TxIn) == 0 {
+		return false
+	}
 
-	log.Infof("Cooperative closure for ChannelPoint(%v): %v",
-		c.cfg.chanState.FundingOutpoint,
-		lnutils.SpewLogClosure(broadcastTx))
+	switch spendingTx.TxIn[0].Sequence {
+	case wire.MaxTxInSequenceNum:
+		return true
+	case mempool.MaxRBFSequence:
+		return true
+	}
 
-	// If the input *is* final, then we'll check to see which output is
-	// ours.
+	return false
+}
+
+// maybeDispatchEarlyCoopClose fires a preliminary CLOSED_CHANNEL event over
+// the channel notifier the first time a coop close spend is detected on
+// chain. It is a no-op if no early-dispatch callback was wired in, the spend
+// is not a coop close, or an early dispatch has already happened for this
+// close. The flag is cleared on a deep reorg of the close (in the closeObserver
+// negativeConfChan handler) so a re-mined close re-fires.
+func (c *chainWatcher) maybeDispatchEarlyCoopClose(
+	spend *chainntnfs.SpendDetail) {
+
+	if c.coopCloseEarlyDispatched.Load() {
+		return
+	}
+	if c.cfg.notifyEarlyCoopClose == nil {
+		return
+	}
+
+	// We only insta-dispatch for coop closes. Force-close, breach, and
+	// abandon spends intentionally remain on the existing N-confirmation
+	// dispatch contract: their CLOSED_CHANNEL event is driven from the
+	// channel arbitrator's MarkChannelClosed callback at the required
+	// confirmation depth, so an early dispatch here would either deliver
+	// a duplicate event or, worse, surface a "cooperative close" summary
+	// for a unilateral spend.
+	if !isCoopCloseSpend(spend.SpendingTx) {
+		return
+	}
+
+	summary := c.buildCoopCloseSummary(spend)
+
+	log.Infof("ChannelPoint(%v): dispatching early CLOSED_CHANNEL "+
+		"event for coop close tx %v at height %d",
+		c.cfg.chanState.FundingOutpoint, spend.SpenderTxHash,
+		spend.SpendingHeight)
+
+	c.cfg.notifyEarlyCoopClose(summary)
+	c.coopCloseEarlyDispatched.Store(true)
+}
+
+// buildCoopCloseSummary constructs a ChannelCloseSummary for a cooperative
+// close from the supplied spend detail. The summary is returned with
+// IsPending=true; the channel arbitrator's MarkChannelClosed callback flips
+// this to false after the close reaches the required confirmation depth. This
+// helper is shared between the early insta-dispatch path (first conf, no DB
+// persist) and the post-N-conf dispatch path so both surfaces produce
+// equivalent summaries.
+func (c *chainWatcher) buildCoopCloseSummary(
+	commitSpend *chainntnfs.SpendDetail) *channeldb.ChannelCloseSummary {
+
+	broadcastTx := commitSpend.SpendingTx
 	localAmt := c.toSelfAmount(broadcastTx)
 
-	// Once this is known, we'll mark the state as fully closed in the
-	// database. For cooperative closes, we wait for a confirmation depth
-	// determined by channel capacity before dispatching this event.
-	closeSummary := &channeldb.ChannelCloseSummary{
+	summary := &channeldb.ChannelCloseSummary{
 		ChanPoint:               c.cfg.chanState.FundingOutpoint,
 		ChainHash:               c.cfg.chanState.ChainHash,
 		ClosingTXID:             *commitSpend.SpenderTxHash,
@@ -1388,9 +1506,28 @@ func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDet
 		log.Errorf("ChannelPoint(%v): unable to create channel sync "+
 			"message: %v", c.cfg.chanState.FundingOutpoint, err)
 	} else {
-		closeSummary.LastChanSyncMsg = chanSync
+		summary.LastChanSyncMsg = chanSync
 	}
 
+	return summary
+}
+
+// dispatchCooperativeClose processed a detect cooperative channel closure.
+// We'll use the spending transaction to locate our output within the
+// transaction, then clean up the database state. We'll also dispatch a
+// notification to all subscribers that the channel has been closed in this
+// manner.
+func (c *chainWatcher) dispatchCooperativeClose(
+	commitSpend *chainntnfs.SpendDetail) error {
+
+	broadcastTx := commitSpend.SpendingTx
+
+	log.Infof("Cooperative closure for ChannelPoint(%v): %v",
+		c.cfg.chanState.FundingOutpoint,
+		lnutils.SpewLogClosure(broadcastTx))
+
+	closeSummary := c.buildCoopCloseSummary(commitSpend)
+
 	// Create a summary of all the information needed to handle the
 	// cooperative closure.
 	closeInfo := &CooperativeCloseInfo{
@@ -1399,7 +1536,7 @@ func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDet
 
 	// If we have an aux closer, finalize the cooperative close now that
 	// it's confirmed.
-	err = fn.MapOptionZ(
+	err := fn.MapOptionZ(
 		c.cfg.auxCloser, func(aux AuxChanCloser) error {
 			return c.finalizeCoopClose(aux, broadcastTx)
 		},
diff --git a/rpcserver.go b/rpcserver.go
index 6319655..8b40192 100644
--- a/rpcserver.go
+++ b/rpcserver.go
@@ -5446,10 +5446,12 @@ func rpcChannelResolution(report *channeldb.ResolverReport) (*lnrpc.Resolution,
 }
 
 // getInitiators returns an initiator enum that provides information about the
-// party that initiated channel's open and close. This information is obtained
-// from the historical channel bucket, so unknown values are returned when the
-// channel is not present (which indicates that it was closed before we started
-// writing channels to the historical close bucket).
+// party that initiated channel's open and close. The information is normally
+// read from the historical channel bucket; for early-dispatched coop closes
+// the channel is still live in the open bucket at notify time (the historical
+// bucket is only populated at MarkChannelClosed time), so we fall back to the
+// open channel state in that case. Unknown values are returned when neither
+// bucket can provide the channel.
 func (r *rpcServer) getInitiators(chanPoint *wire.OutPoint) (
 	lnrpc.Initiator,
 	lnrpc.Initiator, error) {
@@ -5469,10 +5471,20 @@ func (r *rpcServer) getInitiators(chanPoint *wire.OutPoint) (
 	case err == channeldb.ErrNoHistoricalBucket:
 		return openInitiator, closeInitiator, nil
 
-	// The channel was closed before we started storing historical
-	// channels. Do  not return an error, initiator values are unknown.
+	// The channel was either closed before we started storing
+	// historical channels OR the historical bucket has not been
+	// populated yet because this is an early-dispatched
+	// CLOSED_CHANNEL event for a coop close that hasn't reached its
+	// full confirmation depth. Try the open channel bucket so the
+	// early dispatch still carries close-initiator info.
 	case err == channeldb.ErrChannelNotFound:
-		return openInitiator, closeInitiator, nil
+		openChan, openErr := r.server.chanStateDB.FetchChannel(
+			*chanPoint,
+		)
+		if openErr != nil {
+			return openInitiator, closeInitiator, nil
+		}
+		histChan = openChan
 
 	case err != nil:
 		return 0, 0, err
diff --git a/server.go b/server.go
index 45992c4..0a666bc 100644
--- a/server.go
+++ b/server.go
@@ -1406,6 +1406,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 		Sweeper:                       s.sweeper,
 		Registry:                      s.invoices,
 		NotifyClosedChannel:           s.channelNotifier.NotifyClosedChannelEvent,
+		NotifyEarlyClosedChannel:      s.channelNotifier.NotifyEarlyClosedChannelEvent,
 		NotifyFullyResolvedChannel:    s.channelNotifier.NotifyFullyResolvedChannelEvent,
 		OnionProcessor:                s.sphinxPayment,
 		PaymentsExpirationGracePeriod: cfg.PaymentsExpirationGracePeriod,