package com.github.btakita.agentdoc

import com.intellij.openapi.Disposable
import com.intellij.openapi.application.ApplicationManager
import com.intellij.openapi.application.ex.ApplicationEx
import com.intellij.openapi.command.CommandProcessor
import com.intellij.openapi.command.UndoConfirmationPolicy
import com.intellij.openapi.editor.Document
import com.intellij.openapi.editor.EditorFactory
import com.intellij.openapi.editor.event.DocumentEvent
import com.intellij.openapi.editor.event.DocumentListener
import com.intellij.openapi.fileEditor.FileDocumentManager
import com.intellij.openapi.fileEditor.FileDocumentManagerListener
import com.intellij.openapi.fileEditor.FileEditorManager
import com.intellij.openapi.project.Project
import com.intellij.openapi.vfs.LocalFileSystem
import com.intellij.openapi.vfs.VirtualFile
import io.github.lazily.IngressOutcome
import io.github.lazily.MergePolicy
import io.github.lazily.ThreadSafeContext
import java.io.File
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Executors
import java.util.concurrent.RejectedExecutionException
import java.util.concurrent.ScheduledFuture
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference
import javax.swing.SwingUtilities

private const val CRDT_LISTENER_WARN_MS = 10L
private const val CRDT_WORKER_WARN_MS = 100L
private const val NATIVE_RELOAD_WORKER_TIMEOUT_MS = 5_000L
private const val CRDT_EDT_WARN_MS = 50L
private const val CRDT_AWAIT_ATTACH_TIMEOUT_MS = 750L
private const val CRDT_AWAIT_CLOSE_PUBLISH_TIMEOUT_MS = 2_000L
private const val CRDT_REGISTER_FAILURE_BASE_BACKOFF_MS = 1_000L
private const val CRDT_REGISTER_FAILURE_MAX_BACKOFF_MS = 30_000L
private const val LOCAL_EDITOR_FLUSH_QUIET_MS = 16L
private const val LOCAL_EDITOR_READ_RETRY_MS = 25L
private const val CRDT_DRAIN_NOOP_RESCHEDULE_BASE_BACKOFF_MS = 100L

/**
 * `#ctrlkillreregister` Tier 3: minimum gap between whole-editor missing-replica
 * pulls. A controller kill makes every open document report transport loss at once,
 * and one pull already answers for all of them.
 */
private const val PEER_REPLICA_PULL_MIN_INTERVAL_MS = 5_000L
// `#crdt-drain-idle-quiet`: back off work that arrived while a no-op drain was
// already running. The resume callback must consume that retained work without
// manufacturing another drain-all request; doing so turns one overlap into a
// permanent workspace-wide socket polling loop.
private const val CRDT_DRAIN_NOOP_RESCHEDULE_MAX_BACKOFF_MS = 30_000L
private const val CRDT_IDLE_WORKER_WARN_MS = 1_000L
private const val CRDT_DRAIN_BACKOFF_REASON = "backoff-resume"
private const val PROJECTION_RECOVERY_REREGISTER_MIN_INTERVAL_MS = 5_000L
// Delivery routability is a component-level fact, not merely an IDE-process
// fact. Refresh from the same serialized executor that pulls/applies/ACKs CRDT
// deliveries: if that worker stalls, this heartbeat stalls too and Rust stops
// targeting it while continuing to protect its possibly-unsaved buffer.

private data class RemoteEditorEffectToken(
    val generation: Long,
    val endpoint: CrdtReplicaForwarder,
)

internal fun remoteEditorEffectTokenCurrentUtil(
    tokenGeneration: Long,
    liveGeneration: Long?,
    endpointMatches: Boolean,
    endpointBacked: Boolean,
): Boolean =
    liveGeneration == tokenGeneration &&
        endpointMatches &&
        endpointBacked

/** A retained delivery frontier owns the retry cadence. File-watcher and editor
 * events may add work while backoff is active, but must not bypass that gate
 * and hammer the controller with the same unresolved projection. */
internal fun shouldStartRemoteDrainUtil(backoffScheduled: Boolean): Boolean = !backoffScheduled

/**
 * `#crdtpushdrain`: a controller-published CRDT remote event is positive evidence
 * that the CP already holds a frontier for this document, so it must bypass the
 * speculative no-op drain backoff instead of being suppressed by it.
 *
 * The no-op backoff (see [scheduleRemoteDrainAfterBackoff]) exists to stop a
 * *self-driven* drain spin when there is nothing to pull; on an idle document it
 * climbs to [CRDT_DRAIN_NOOP_RESCHEDULE_MAX_BACKOFF_MS]. That is exactly the state a
 * document sits in when the operator triggers Compact Exchange, so every
 * controller projection event drains eagerly. The controller's keyed retained
 * Lazily projection provides the coalescing boundary.
 */
internal fun shouldUrgentDrainForRemoteEventUtil(@Suppress("UNUSED_PARAMETER") reasonToken: String?): Boolean = true

internal fun projectionRecoveryReregisterDueUtil(
    lastStartedMs: Long?,
    nowMs: Long,
    minIntervalMs: Long = PROJECTION_RECOVERY_REREGISTER_MIN_INTERVAL_MS,
): Boolean =
    lastStartedMs == null ||
        nowMs < lastStartedMs ||
        nowMs - lastStartedMs >= minIntervalMs

@Suppress("UNUSED_PARAMETER")
internal fun shouldProjectVisibleRemoteDeliveryUtil(
    editorText: String?,
    targetText: String,
    diskPersisted: Boolean,
): Boolean = editorText == targetText

internal enum class RemoteCrdtProjectionMode {
    Reject,
    MemoryOnly,
    Persist,
}

internal enum class RemotePersistReconciliation {
    Persisted,
    PersistedEditorNormalization,
    RollbackToBefore,
    PreserveAdvancedEditor,
}

/**
 * A failed save is transactional only while both observable planes still prove
 * the attempted mutation. Roll back when the editor remains at the remote
 * target and disk remains at the exact pre-apply text; otherwise preserve the
 * advanced plane and fail closed.
 */
internal fun remotePersistReconciliationUtil(
    beforeText: String,
    targetText: String,
    editorAfterSave: String,
    diskAfterSave: String?,
): RemotePersistReconciliation = when {
    editorAfterSave == targetText && diskAfterSave == targetText ->
        RemotePersistReconciliation.Persisted
    editorAfterSave == targetText && diskAfterSave == beforeText ->
        RemotePersistReconciliation.RollbackToBefore
    diskAfterSave != null && editorAfterSave == diskAfterSave ->
        RemotePersistReconciliation.PersistedEditorNormalization
    else -> RemotePersistReconciliation.PreserveAdvancedEditor
}

internal data class ReplicaRegistrationRetryProjection(
    val failureCount: Int,
    val retryAfterMs: Long,
    val backoffMs: Long,
)

internal fun nextReplicaRegistrationRetryProjection(
    previous: ReplicaRegistrationRetryProjection?,
    nowMs: Long,
    baseBackoffMs: Long = CRDT_REGISTER_FAILURE_BASE_BACKOFF_MS,
    maxBackoffMs: Long = CRDT_REGISTER_FAILURE_MAX_BACKOFF_MS,
): ReplicaRegistrationRetryProjection {
    val failureCount = ((previous?.failureCount ?: 0) + 1).coerceAtMost(16)
    val step = (failureCount - 1).coerceAtLeast(0).coerceAtMost(3)
    val backoffMs = (baseBackoffMs * (1L shl step)).coerceAtMost(maxBackoffMs)
    return ReplicaRegistrationRetryProjection(
        failureCount = failureCount,
        retryAfterMs = nowMs + backoffMs,
        backoffMs = backoffMs,
    )
}

internal fun replicaRegistrationAttemptDueUtil(
    projection: ReplicaRegistrationRetryProjection?,
    nowMs: Long,
): Boolean = projection == null || nowMs >= projection.retryAfterMs

private data class RemotePersistOutcome(
    val diskPersisted: Boolean,
    val editorTextForProjection: String?,
    val editorNormalizedText: String? = null,
)

/**
 * The CRDT canonical is the durable effect sink. An unsaved editor can safely
 * accept a causally fenced remote delta in memory, but the plugin must not
 * refresh or save behind the operator. Clean editors still require exact disk
 * proof before projection and persistence.
 */
internal fun remoteCrdtProjectionModeUtil(
    documentUnsaved: Boolean,
    diskCanPersist: Boolean,
): RemoteCrdtProjectionMode = when {
    documentUnsaved -> RemoteCrdtProjectionMode.MemoryOnly
    diskCanPersist -> RemoteCrdtProjectionMode.Persist
    else -> RemoteCrdtProjectionMode.Reject
}

internal enum class TemplateStructureProjectionState {
    Exact,
    RepairRequired,
    Invalid,
}

internal fun templateStructureProjectionStateUtil(
    text: String,
    normalized: String?,
): TemplateStructureProjectionState = when {
    normalized == null -> TemplateStructureProjectionState.Invalid
    normalized == text -> TemplateStructureProjectionState.Exact
    else -> TemplateStructureProjectionState.RepairRequired
}

internal fun remoteReplaceStructureAcceptedUtil(
    remoteState: TemplateStructureProjectionState,
): Boolean = remoteState == TemplateStructureProjectionState.Exact

internal enum class RemoteTemplateProjectionDecision {
    QueueRemote,
    RetryFailClosed,
}

internal fun remoteTemplateProjectionDecisionUtil(
    remoteState: TemplateStructureProjectionState,
    editorState: TemplateStructureProjectionState?,
    editorMatchesExpected: Boolean,
    recoveryInFlight: Boolean,
): RemoteTemplateProjectionDecision = when {
    remoteState == TemplateStructureProjectionState.Exact ->
        RemoteTemplateProjectionDecision.QueueRemote
    else -> RemoteTemplateProjectionDecision.RetryFailClosed
}

internal enum class ReplicaBaselineDecision {
    ApplyRemote,
    ApplyRemoteRepair,
    ProjectRemoteTarget,
    RebootstrapVisibleRemoteTarget,
    ReplayRemoteTarget,
    RealignShadow,
    RetryFailClosed,
}

internal fun matchingRemoteTargetGenerationUtil(
    updates: List<ReplicaRemoteUpdate>,
    contentHash: String?,
): Long? =
    contentHash?.let { targetHash ->
        updates
            .asSequence()
            .filter { it.expectedContentHash == targetHash }
            .maxOfOrNull { it.generation }
    }

/**
 * The visible editor is the operator-authoritative plane. A save or a stale
 * native replica must never turn that text into a second logical mutation or
 * project an older replica snapshot over it.
 */
internal fun replicaBaselineDecisionUtil(
    editorState: TemplateStructureProjectionState?,
    editorMatchesExpected: Boolean,
    replicaMatchesExpected: Boolean,
    replicaMatchesEditor: Boolean,
    editorMatchesRemoteTarget: Boolean,
    replicaMatchesRemoteTarget: Boolean,
    recoveryInFlight: Boolean,
    canonicalProjectionRetained: Boolean = false,
): ReplicaBaselineDecision = when {
    recoveryInFlight ->
        ReplicaBaselineDecision.RetryFailClosed
    editorMatchesRemoteTarget && replicaMatchesEditor ->
        ReplicaBaselineDecision.ProjectRemoteTarget
    editorMatchesRemoteTarget ->
        ReplicaBaselineDecision.RebootstrapVisibleRemoteTarget
    canonicalProjectionRetained && replicaMatchesRemoteTarget ->
        ReplicaBaselineDecision.ReplayRemoteTarget
    canonicalProjectionRetained && replicaMatchesExpected ->
        ReplicaBaselineDecision.ApplyRemote
    canonicalProjectionRetained ->
        ReplicaBaselineDecision.RetryFailClosed
    editorMatchesExpected && replicaMatchesRemoteTarget ->
        ReplicaBaselineDecision.ReplayRemoteTarget
    editorState != TemplateStructureProjectionState.Exact && editorMatchesExpected ->
        ReplicaBaselineDecision.ApplyRemoteRepair
    editorState != TemplateStructureProjectionState.Exact ->
        ReplicaBaselineDecision.RetryFailClosed
    editorMatchesExpected && replicaMatchesExpected -> ReplicaBaselineDecision.ApplyRemote
    replicaMatchesEditor -> ReplicaBaselineDecision.RealignShadow
    else -> ReplicaBaselineDecision.RetryFailClosed
}

internal fun shouldForwardLocalDeltaUtil(replicaText: String?, shadowText: String): Boolean =
    replicaText == shadowText

internal enum class LocalReplicaBaselineDecision {
    ForwardLocal,
    RebootstrapCanonicalThenForward,
}

internal fun localReplicaBaselineDecisionUtil(
    replicaText: String?,
    capturedBaseText: String,
): LocalReplicaBaselineDecision =
    if (shouldForwardLocalDeltaUtil(replicaText, capturedBaseText)) {
        LocalReplicaBaselineDecision.ForwardLocal
    } else {
        LocalReplicaBaselineDecision.RebootstrapCanonicalThenForward
    }

internal data class CapturedLocalEditorEdit(
    val offsetUtf16: Int,
    val oldFragment: String,
    val newFragment: String,
    val projectionEpoch: Long,
)

internal data class PreparedLocalEditorEdit(
    val offsetCodePoints: Int,
    val deleteCodePoints: Int,
    val insert: String,
    val resultingText: String,
)

private enum class LocalEditorForwardResult {
    Applied,
    Fenced,
    Retry,
}

/**
 * Validate and translate the editor's causal splice stream against the retained
 * CRDT shadow. Ordinary editor events already carry the changed UTF-16 range;
 * never widen them into a whole-buffer diff.
 */
internal fun prepareLocalEditorEditsUtil(
    before: String,
    edits: List<CapturedLocalEditorEdit>,
): List<PreparedLocalEditorEdit>? {
    var current = before
    val prepared = ArrayList<PreparedLocalEditorEdit>(edits.size)
    for (edit in edits) {
        val start = edit.offsetUtf16
        val end = start + edit.oldFragment.length
        if (start < 0 || start > current.length || end > current.length) return null
        if (current.substring(start, end) != edit.oldFragment) return null
        val resultingText =
            current.substring(0, start) + edit.newFragment + current.substring(end)
        prepared.add(
            PreparedLocalEditorEdit(
                offsetCodePoints = current.codePointCount(0, start),
                deleteCodePoints = edit.oldFragment.codePointCount(0, edit.oldFragment.length),
                insert = edit.newFragment,
                resultingText = resultingText,
            ),
        )
        current = resultingText
    }
    return prepared
}

internal fun pullDeliveryRequestsReplicaRefreshUtil(delivery: ReplicaPullDelivery): Boolean =
    delivery is ReplicaPullDelivery.Unavailable

private data class PendingRemoteEditorApply(
    val filePath: String,
    val expectedText: String,
    val targetText: String,
    val effectToken: RemoteEditorEffectToken,
)

private data class RemoteEditorApplyOutcome(
    val diskPersisted: Boolean,
    val editorText: String?,
    val editorNormalizedText: String? = null,
)

private enum class RemoteTextApplyDisposition {
    Queued,
    Recovered,
    RetryFailClosed,
}

/**
 * Oldest baseline + newest converged text. Full visible-state projection makes
 * the final text itself the cumulative delivery proof.
 */
private val REMOTE_EDITOR_APPLY_MERGE = MergePolicy(
    name = "RemoteEditorApply",
    merge = { old: PendingRemoteEditorApply, latest: PendingRemoteEditorApply ->
        latest.copy(
            expectedText = old.expectedText,
        )
    },
    commutative = false,
    idempotent = true,
)

/**
 * Production editor-as-CRDT-replica wiring (`#crdtauth5`, realtime phase 3).
 *
 * The manager is intentionally thin: local edits are forwarded to [CrdtReplicaForwarder],
 * remote updates are pulled from the CP document model, and document mutation uses the same
 * minimal-edit helper as IPC patches. A remote mutation is projected as disk-persisted only
 * when raw disk still equals the guarded editor baseline or the converged target; novel external
 * disk text rejects the apply instead of being overwritten.
 */
class CrdtReplicaManager(private val project: Project) : Disposable, DocumentListener {
    private val log = com.intellij.openapi.diagnostic.Logger.getInstance(CrdtReplicaManager::class.java)
    // Project-wide scheduling owns fan-out/backoff only. Every operation that
    // reads or mutates one replica runs on that document's serialized lane.
    private val executor = Executors.newSingleThreadScheduledExecutor { r ->
        Thread(r, "agent-doc-crdt-replica-control").apply { isDaemon = true }
    }
    private val documentWorkers = DocumentReplicaWorkers()
    private val forwarders = ConcurrentHashMap<String, CrdtReplicaForwarder>()
    // Every document ownership chart joins the manager's project-lifetime graph.
    // This keeps cross-thread projections composable instead of creating a
    // private reactive island per open editor.
    private val ownershipContext = ThreadSafeContext()
    private val templateValidation =
        TemplateValidationPlane(
            ownershipContext,
            NativePatching::normalizeTemplateStructure,
            ::contentHash,
        )
    private val shadows = ConcurrentHashMap<String, String>()
    private val applyingRemote = ConcurrentHashMap.newKeySet<String>()
    private val fileContentReloadingPaths = ConcurrentHashMap.newKeySet<String>()
    private val pendingLocalEdits = ConcurrentHashMap<String, AtomicInteger>()
    private val localEditorFlushVersions = ConcurrentHashMap<String, AtomicLong>()
    private val localEditorFlushTasks = ConcurrentHashMap<String, ScheduledFuture<*>>()
    private val pendingLocalEditorEdits =
        ConcurrentHashMap<String, MutableList<CapturedLocalEditorEdit>>()
    private val remoteEditorEffectGenerations = ConcurrentHashMap<String, AtomicLong>()
    private val localEditorFlushPendingPaths = ConcurrentHashMap.newKeySet<String>()
    private val drainQueued = AtomicBoolean(false)
    private val drainAllRequested = AtomicBoolean(false)
    private val drainRequestedPaths = ConcurrentHashMap.newKeySet<String>()
    private val registerFailureCounts = ConcurrentHashMap<String, Int>()
    private val registerRetryAfterMs = ConcurrentHashMap<String, Long>()
    private val registerRetryProjections =
        ConcurrentHashMap<String, ReplicaRegistrationRetryProjection>()
    private val registerRetryTasks = ConcurrentHashMap<String, ScheduledFuture<*>>()
    private val projectionRecoveryReregisterStartedAtMs = ConcurrentHashMap<String, Long>()
    private val consecutiveNoOpReschedules = AtomicInteger(0)
    private val remoteDrainBackoffScheduled = AtomicBoolean(false)
    private val remoteEditorApplies =
        KeyedCoalescingRelay<String, PendingRemoteEditorApply>(REMOTE_EDITOR_APPLY_MERGE)
    private val remoteEditorApplyScheduled = AtomicBoolean(false)
    private val remoteEditorApplyPaths = ConcurrentHashMap.newKeySet<String>()
    private val retainedCanonicalProjectionPaths = ConcurrentHashMap.newKeySet<String>()
    private val templateGuardRecoveryPaths = ConcurrentHashMap.newKeySet<String>()
    private val templateGuardRecoveryRetryPaths = ConcurrentHashMap.newKeySet<String>()
    private val templateGuardRecoveryFailureCounts = ConcurrentHashMap<String, Int>()
    private val refreshConnectionEpoch = AtomicLong(0)
    // `#ctrlkillreregister` Tier 3: transport loss is reported per document, but a
    // dead controller strands every document at once. One pull answers for all of
    // them, so the second and third file to notice must not each start their own.
    private val lastPeerReplicaPullAtMs = AtomicLong(0)
    private val disposed = AtomicBoolean(false)

    fun start() {
        EditorFactory.getInstance().eventMulticaster.addDocumentListener(this, this)
        ApplicationManager.getApplication().messageBus.connect(this).subscribe(
            FileDocumentManagerListener.TOPIC,
            object : FileDocumentManagerListener {
                override fun beforeFileContentReload(file: VirtualFile, document: Document) {
                    val filePath = file.path
                    if (!file.name.endsWith(".md") || managerForFilePath(filePath) !== this@CrdtReplicaManager) {
                        return
                    }
                    fileContentReloadingPaths.add(filePath)
                    advanceNonOperatorMutationEpoch(filePath)
                }

                override fun fileContentReloaded(file: VirtualFile, document: Document) {
                    fileContentReloadingPaths.remove(file.path)
                }
            },
        )
    }

    override fun dispose() {
        disposed.set(true)
        executor.shutdownNow()
        documentWorkers.shutdownNow()
        localEditorFlushTasks.clear()
        localEditorFlushVersions.clear()
        pendingLocalEditorEdits.clear()
        localEditorFlushPendingPaths.clear()
        pendingLocalEdits.clear()
        fileContentReloadingPaths.clear()
        remoteEditorApplies.clear()
        remoteEditorApplyPaths.clear()
        retainedCanonicalProjectionPaths.clear()
        remoteEditorEffectGenerations.clear()
        templateGuardRecoveryPaths.clear()
        templateGuardRecoveryRetryPaths.clear()
        templateGuardRecoveryFailureCounts.clear()
        drainRequestedPaths.clear()
        registerFailureCounts.clear()
        registerRetryAfterMs.clear()
        registerRetryProjections.clear()
        registerRetryTasks.values.forEach { it.cancel(false) }
        registerRetryTasks.clear()
        projectionRecoveryReregisterStartedAtMs.clear()
        forwarders.values.forEach { it.deregister() }
        forwarders.clear()
        shadows.clear()
        templateValidation.clear()
    }

    private fun awaitWorkerTermination(timeoutMs: Long): Boolean {
        val deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs)
        val controlStopped = try {
            executor.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)
        } catch (_: InterruptedException) {
            Thread.currentThread().interrupt()
            false
        }
        if (!controlStopped) return false
        val remaining = deadline - System.nanoTime()
        return remaining > 0L &&
            documentWorkers.awaitTermination(TimeUnit.NANOSECONDS.toMillis(remaining).coerceAtLeast(1L))
    }

    /**
     * Claim the next Tier 3 pull window, or report that a recent pull already covers
     * this caller. Reuses [projectionRecoveryReregisterDueUtil]'s interval rule; only the
     * window differs, because one pull is authoritative for every document.
     */
    private fun beginPeerReplicaPull(): Boolean {
        val nowMs = System.currentTimeMillis()
        val lastMs = lastPeerReplicaPullAtMs.get()
        if (!projectionRecoveryReregisterDueUtil(
                lastStartedMs = lastMs.takeIf { it > 0L },
                nowMs = nowMs,
                minIntervalMs = PEER_REPLICA_PULL_MIN_INTERVAL_MS,
            )
        ) {
            return false
        }
        return lastPeerReplicaPullAtMs.compareAndSet(lastMs, nowMs)
    }

    private fun beginProjectionRecoveryReregister(filePath: String): Boolean {
        val nowMs = System.currentTimeMillis()
        var due = false
        projectionRecoveryReregisterStartedAtMs.compute(filePath) { _, lastStartedMs ->
            if (projectionRecoveryReregisterDueUtil(lastStartedMs, nowMs)) {
                due = true
                nowMs
            } else {
                lastStartedMs
            }
        }
        return due
    }

    override fun documentChanged(event: DocumentEvent) {
        val started = System.nanoTime()
        var loggedFilePath: String? = null
        try {
            val file = FileDocumentManager.getInstance().getFile(event.document) ?: return
            if (!file.name.endsWith(".md")) return
            val filePath = file.path
            loggedFilePath = filePath
            if (managerForFilePath(filePath) !== this) return
            if (!CrdtReplicaManager.isOperatorDocumentEvent(filePath, event)) {
                advanceNonOperatorMutationEpoch(filePath)
                // A remote CRDT apply mutates the IntelliJ Document before its
                // visible-content projection reaches this worker. Replacing
                // the replica from this listener would run ahead of that projection,
                // retire its retained frontier, and make the controller deliver
                // the same canonical edit again. The remote apply path updates
                // the shadow and schedules its own post-projection drain.
                if (CrdtReplicaManager.isApplyingRemote(filePath)) return
                // A clean File Cache Conflict reload may be the operator
                // accepting a Lazily-retained external disk candidate. Resolve
                // on the worker; exact CAS/rebootstrap rules keep ordinary
                // remote CRDT applies as no-ops here.
                ensureOpenDocumentReplica(filePath, event.document, forceRefresh = true)
                requestRemoteDrain(filePath, "non-operator-editor-event")
                return
            }
            val projectionEpoch = nonOperatorMutationEpoch(filePath)
            val newFragment = event.newFragment.toString()
            val oldFragment = event.oldFragment.toString()
            if (newFragment.isEmpty() && oldFragment.isEmpty()) return
            if (!shadows.containsKey(filePath)) {
                seedAndAttachFromDocument(filePath, event.document)
                return
            }
            recordLocalEditorEdit(
                filePath,
                CapturedLocalEditorEdit(
                    offsetUtf16 = event.offset,
                    oldFragment = oldFragment,
                    newFragment = newFragment,
                    projectionEpoch = projectionEpoch,
                ),
            )
            scheduleLocalEditorFlush(filePath)
        } finally {
            loggedFilePath?.let { logSlow("documentChanged-listener", it, started, warnMs = CRDT_LISTENER_WARN_MS) }
        }
    }

    /**
     * Keep at most one running and one latest queued splice flush per document.
     *
     * A native/controller round trip can be slower than typing. Enqueuing every
     * DocumentEvent then makes a one-second operation repeat once per character.
     * Exact event splices accumulate for one transport batch. The retained
     * version owns one pending-local fence for the whole burst; a superseded
     * task never clears that fence out from under its successor.
     */
    private fun scheduleLocalEditorFlush(
        filePath: String,
        retryVersion: Long? = null,
    ) {
        val versionCounter =
            localEditorFlushVersions.computeIfAbsent(filePath) { AtomicLong(0L) }
        val version = retryVersion ?: versionCounter.incrementAndGet()
        if (retryVersion == null && localEditorFlushPendingPaths.add(filePath)) {
            markLocalPending(filePath)
        }
        val delayMs =
            if (retryVersion == null) LOCAL_EDITOR_FLUSH_QUIET_MS else LOCAL_EDITOR_READ_RETRY_MS
        lateinit var scheduled: ScheduledFuture<*>
        try {
            scheduled =
                documentWorkers.forDocument(filePath).schedule(
                    Runnable {
                        if (disposed.get() || versionCounter.get() != version) {
                            return@Runnable
                        }
                        val workerStarted = System.nanoTime()
                        val capturedEdits = drainLocalEditorEdits(filePath)
                        var retrySplices = false
                        try {
                            retrySplices =
                                forwardLocalEditsFromShadow(
                                    filePath,
                                    capturedEdits,
                                ) == LocalEditorForwardResult.Retry
                        } catch (error: Exception) {
                            retrySplices = true
                            log.warn(
                                "[crdt-replica] local editor splice flush failed for $filePath: ${error.message}",
                                error,
                            )
                        } finally {
                            if (retrySplices) {
                                prependLocalEditorEdits(filePath, capturedEdits)
                            }
                            if (versionCounter.get() == version) {
                                if (retrySplices && !disposed.get()) {
                                    scheduleLocalEditorFlush(
                                        filePath,
                                        retryVersion = version,
                                    )
                                } else {
                                    localEditorFlushTasks.remove(filePath, scheduled)
                                    localEditorFlushVersions.remove(filePath, versionCounter)
                                    if (localEditorFlushPendingPaths.remove(filePath)) {
                                        clearLocalPending(filePath)
                                    }
                                    logSlow(
                                    "local-delta-worker",
                                    filePath,
                                    workerStarted,
                                    details = "splices=${capturedEdits.size}",
                                    )
                                    requestRemoteDrain(filePath, "local-delta")
                                }
                            }
                        }
                    },
                    delayMs,
                    TimeUnit.MILLISECONDS,
                )
            localEditorFlushTasks.put(filePath, scheduled)?.cancel(false)
        } catch (error: RejectedExecutionException) {
            if (versionCounter.get() == version) {
                localEditorFlushVersions.remove(filePath, versionCounter)
                if (localEditorFlushPendingPaths.remove(filePath)) {
                    clearLocalPending(filePath)
                }
            }
            if (!disposed.get()) {
                log.warn("[crdt-replica] local editor flush scheduling rejected for $filePath", error)
            }
        }
    }

    private fun recordLocalEditorEdit(filePath: String, edit: CapturedLocalEditorEdit) {
        pendingLocalEditorEdits.compute(filePath) { _, existing ->
            (existing ?: mutableListOf()).also { it.add(edit) }
        }
    }

    private fun drainLocalEditorEdits(filePath: String): List<CapturedLocalEditorEdit> {
        var drained: List<CapturedLocalEditorEdit> = emptyList()
        pendingLocalEditorEdits.compute(filePath) { _, existing ->
            if (existing != null) drained = existing.toList()
            null
        }
        return drained
    }

    private fun prependLocalEditorEdits(
        filePath: String,
        edits: List<CapturedLocalEditorEdit>,
    ) {
        if (edits.isEmpty()) return
        pendingLocalEditorEdits.compute(filePath) { _, existing ->
            mutableListOf<CapturedLocalEditorEdit>().also { retained ->
                retained.addAll(edits)
                if (existing != null) retained.addAll(existing)
            }
        }
    }

    private fun seedAndAttachFromDocument(filePath: String, document: Document) {
        markLocalPending(filePath)
        documentWorkers.forDocument(filePath).execute {
            val started = System.nanoTime()
            var chars = -1
            try {
                val text = tryReadDocumentText(document)
                    ?: return@execute
                chars = text.length
                forwarderFor(filePath, text)
                requestRemoteDrain(filePath, "seed")
            } catch (e: Exception) {
                log.debug("[crdt-replica] seed skipped for $filePath: ${e.message}")
            } finally {
                clearLocalPending(filePath)
                logSlow("seed-and-attach", filePath, started, details = "chars=$chars")
            }
        }
    }

    fun ensureOpenDocumentReplica(
        filePath: String,
        document: Document,
        editorText: String? = null,
        await: Boolean = false,
        forceRefresh: Boolean = false,
    ): Boolean {
        // TypingTracker reports the Lazily current-document projection after
        // each coalesced edit burst. Once this document already owns a live
        // replica, that observation must not enqueue another attach, normalize
        // the whole markdown document, or overwrite the incremental shadow with
        // a future editor cut. The latter used to advance the shadow past queued
        // DocumentEvents and manufacture the stale-baseline recovery loop.
        if (!forceRefresh && forwarders[filePath]?.attached == true) {
            return true
        }
        val attach = attach@{
            val started = System.nanoTime()
            var chars = -1
            try {
                val text = editorText ?: tryReadDocumentText(document)
                    ?: return@attach false
                chars = text.length
                // `#replicarefusalstorm` (the editor half). The controller refuses a
                // non-agent-doc markdown file terminally and cheaply, and its comment
                // names this as the durable stop: "the editor only registering
                // agent-doc documents". Until now the editor registered every `.md`
                // it opened and retried the refusal on an 8s backoff forever —
                // measured 2026-08-12 at ~150 retries per file per 20 minutes across
                // four plain markdown files.
                //
                // No sticky state: the test runs on the CURRENT text every time, so a
                // plain file that later gains `<!-- agent:` markers registers on the
                // next observation without anything to invalidate.
                if (!isAgentDocDocumentTextUtil(text)) {
                    // Recorded so an operator-facing command can say WHY, but
                    // deliberately NOT through `recordRegisterFailure`: that arms a
                    // retry, and there is nothing here to converge on.
                    attachFailureReasons[filePath] = "not_agent_doc_document"
                    return@attach false
                }
                // Registration always opens the controller bootstrap. An existing
                // editor buffer is a downstream consumer until subsequent DocumentEvent
                // deltas prove new operator intent.
                val registrationText = text
                if (!NativePatching.isAvailable()) {
                    log.warn(
                        "[crdt-replica] open-document replica registration deferred for ${File(filePath).name}; " +
                            "native FFI unavailable",
                    )
                    recordRegisterFailure(filePath, "native-ffi-unavailable")
                    return@attach false
            }
                chars = registrationText.length
                val forwarder = forwarderFor(
                    filePath,
                    registrationText,
                    bypassRegisterBackoff = forceRefresh,
                    replaceCached = forceRefresh,
                    expectedEditorTextAtSwap = if (forceRefresh) registrationText else null,
                )
                (forwarder != null).also { attached ->
                    if (attached) {
                        // Queue retained semantic replay behind this registration task.
                        // In await mode the caller can therefore observe registration
                        // before any native reconstruction or EDT mutation begins.
                        scheduleDeferredWriteReplayAfterRegistration(
                            filePath,
                            document,
                            registrationText,
                            forwarder!!,
                        )
                        requestRemoteDrain(filePath, "open-document")
                    }
                }
            } catch (e: Exception) {
                log.debug("[crdt-replica] open-document attach skipped for $filePath: ${e.message}")
                recordRegisterFailure(filePath, "attach-exception")
                false
            } finally {
                logSlow("open-document-attach", filePath, started, details = "chars=$chars force_refresh=$forceRefresh")
            }
        }
        if (await) {
            return try {
                documentWorkers.forDocument(filePath).submit<Boolean> { attach() }
                    .get(CRDT_AWAIT_ATTACH_TIMEOUT_MS, TimeUnit.MILLISECONDS)
            } catch (e: TimeoutException) {
                log.warn("[crdt-replica] open-document attach timed out for $filePath after ${CRDT_AWAIT_ATTACH_TIMEOUT_MS}ms")
                false
            } catch (e: Exception) {
                log.debug("[crdt-replica] open-document attach failed for $filePath: ${e.message}")
                false
            }
        }
        documentWorkers.forDocument(filePath).execute { attach() }
        return true
    }

    private fun rebindOpenDocumentPath(
        oldPath: String,
        newPath: String,
        document: Document,
        editorText: String,
    ): Boolean {
        if (oldPath == newPath) {
            return ensureOpenDocumentReplica(
                newPath,
                document,
                editorText = editorText,
                await = true,
                forceRefresh = false,
            )
        }
        return try {
            documentWorkers.forDocument(newPath).submit<Boolean> {
                val existingNew = forwarders[newPath]
                val oldForwarder = forwarders[oldPath]
                val activeNew =
                    if (existingNew?.attached == true) {
                        existingNew
                    } else {
                        val root = resolveProjectRoot(newPath) ?: return@submit false
                        val resume = oldForwarder?.captureResumeState()
                        val replacement =
                            CrdtReplicaForwarder(
                                filePath = newPath,
                                identity = "${EditorIdentity.id}:$newPath:path-transition-${refreshConnectionEpoch.incrementAndGet()}",
                                node = NativeReplicaNode(),
                                transport = CpSocketReplicaTransport(root),
                                ownershipContext = ownershipContext,
                                resumeState = resume,
                            )
                        if (!replacement.register()) {
                            recordRegisterFailure(newPath, "path-transition-register")
                            return@submit false
                        }
                    val raced = forwarders.putIfAbsent(newPath, replacement)
                        if (raced != null) {
                            replacement.deregister()
                            raced
                    } else {
                        retainCanonicalProjectionAfterRegistration(newPath, editorText, replacement)
                        replacement
                        }
            }

            if (!activeNew.attached) return@submit false
            retainCanonicalProjectionAfterRegistration(newPath, editorText, activeNew)
            clearRegisterFailure(newPath)
                scheduleDeferredWriteReplayAfterRegistration(
                    newPath,
                    document,
                    editorText,
                    activeNew,
                )

                if (oldForwarder != null && forwarders.remove(oldPath, oldForwarder)) {
                    oldForwarder.deregister()
                }
                shadows.remove(oldPath)
                localEditorFlushTasks.remove(oldPath)?.cancel(false)
                localEditorFlushVersions.remove(oldPath)
                pendingLocalEditorEdits.remove(oldPath)
                if (localEditorFlushPendingPaths.remove(oldPath)) {
                    clearLocalPending(oldPath)
                }
                pendingLocalEdits.remove(oldPath)
                remoteEditorEffectGenerations.remove(oldPath)
                drainRequestedPaths.remove(oldPath)
                registerFailureCounts.remove(oldPath)
                registerRetryAfterMs.remove(oldPath)
                registerRetryProjections.remove(oldPath)
                registerRetryTasks.remove(oldPath)?.cancel(false)
                projectionRecoveryReregisterStartedAtMs.remove(oldPath)
                remoteEditorApplyPaths.remove(oldPath)
                retainedCanonicalProjectionPaths.remove(oldPath)
                templateGuardRecoveryPaths.remove(oldPath)
                templateGuardRecoveryRetryPaths.remove(oldPath)
                templateGuardRecoveryFailureCounts.remove(oldPath)
                nonOperatorMutationEpochs.remove(oldPath)?.let { oldEpoch ->
                    nonOperatorMutationEpochs
                        .computeIfAbsent(newPath) { AtomicLong(0L) }
                        .updateAndGet { current -> maxOf(current, oldEpoch.get()) }
                }
                TypingTracker.rekeyDocumentPath(oldPath, newPath, document)
                requestRemoteDrain(newPath, "document-path-transition")
                true
            }.get(CRDT_AWAIT_CLOSE_PUBLISH_TIMEOUT_MS, TimeUnit.MILLISECONDS)
        } catch (error: Exception) {
            log.warn(
                "[crdt-replica] path transition rebind deferred $oldPath → $newPath: ${error.message}",
            )
            false
        }
    }

    private fun scheduleDeferredWriteReplayAfterRegistration(
        filePath: String,
        document: Document,
        registrationText: String,
        forwarder: CrdtReplicaForwarder,
    ) {
        try {
            documentWorkers.forDocument(filePath).execute {
                if (disposed.get() || forwarders[filePath] !== forwarder) {
                    return@execute
                }
            replayDeferredWriteAfterRegistration(
                filePath,
                document,
                registrationText,
                forwarder,
            )
        }
        } catch (error: RejectedExecutionException) {
            if (!disposed.get()) {
                log.warn(
                    "[crdt-replica] deferred write replay scheduling rejected for $filePath",
                    error,
                )
            }
        }
    }

    private fun replayDeferredWriteAfterRegistration(
        filePath: String,
        document: Document,
        registrationText: String,
        forwarder: CrdtReplicaForwarder,
    ): Boolean {
        if (
            forwarders[filePath] !== forwarder ||
            hasPendingLocal(filePath) ||
            tryReadDocumentText(document) != registrationText
        ) {
            scheduleTemplateGuardRecoveryRetry(filePath, "post-register-replay-replica-raced")
            return false
        }
        if (!NativePatching.projectDeferredWritePostRegister(filePath, registrationText)) {
            scheduleTemplateGuardRecoveryRetry(filePath, "post-register-projection-unavailable")
            return false
        }
        // FFI only wakes/projects the retained semantic intent into the
        // controller-owned CRDT authority. The ordinary remote-delivery path
        // below is the sole owner of editor mutation and persistence.
        requestUrgentRemoteDrain(filePath, "post-register-projected-intent")
        log.info(
            "[crdt-replica] projected deferred write after exact editor registration for $filePath; " +
                "baseline_hash=${contentHash(registrationText)} editor_mutation=remote_delivery_only",
        )
        return true
    }

    /**
     * Publish the exact closing editor cut through the same serialized Lazily
     * replica worker as every preceding local delta, then retire that replica.
     * The reliable-sync close fact is emitted only after this returns true, so
     * the controller can hand authority to disk without losing a last unsaved
     * deletion that was still waiting behind the debounce worker.
     */
    private fun publishClosingDocumentCut(filePath: String, document: Document): Boolean {
        val closingText = tryReadDocumentText(document) ?: return false
        return try {
            documentWorkers.forDocument(filePath).submit<Boolean> {
                val forwarder = forwarderFor(filePath, closingText) ?: return@submit false
                forwarder.ensureEditorText(closingText)
                shadows[filePath] = closingText
                clearLocalPending(filePath)
                if (forwarders.remove(filePath, forwarder)) {
                    forwarder.deregister()
                }
                templateValidation.remove(filePath)
                true
            }.get(CRDT_AWAIT_CLOSE_PUBLISH_TIMEOUT_MS, TimeUnit.MILLISECONDS)
        } catch (e: TimeoutException) {
            log.warn(
                "[crdt-replica] closing editor cut publish timed out for $filePath after ${CRDT_AWAIT_CLOSE_PUBLISH_TIMEOUT_MS}ms",
            )
            false
        } catch (e: Exception) {
            log.warn("[crdt-replica] closing editor cut publish failed for $filePath: ${e.message}")
            false
        }
    }

    private fun forwardLocalEditsFromShadow(
        filePath: String,
        capturedEdits: List<CapturedLocalEditorEdit>,
    ): LocalEditorForwardResult {
        val started = System.nanoTime()
        if (capturedEdits.isEmpty()) return LocalEditorForwardResult.Applied
        val currentEpoch = nonOperatorMutationEpoch(filePath)
        val currentEdits = capturedEdits.filter { it.projectionEpoch == currentEpoch }
        if (currentEdits.isEmpty()) {
            log.debug(
                "[crdt-replica] dropped stale operator splice batch for $filePath after a newer non-operator projection",
            )
            requestRemoteDrain(filePath, "stale-operator-event-fenced")
            return LocalEditorForwardResult.Fenced
        }
        val beforeText = shadows[filePath] ?: return LocalEditorForwardResult.Retry
        val edits =
            prepareLocalEditorEditsUtil(beforeText, currentEdits)
                ?: run {
                    log.debug(
                        "[crdt-replica] retained local splice batch for $filePath because its exact shadow range no longer matches",
                    )
                    requestRemoteDrain(filePath, "captured-local-splice-baseline-mismatch")
                    return LocalEditorForwardResult.Retry
                }
        if (currentEpoch != nonOperatorMutationEpoch(filePath)) {
            requestRemoteDrain(filePath, "stale-operator-event-fenced")
            return LocalEditorForwardResult.Fenced
        }
        val editorText = edits.lastOrNull()?.resultingText ?: beforeText
        val forwarder = forwarderFor(filePath, beforeText)
        if (forwarder == null) {
            shadows[filePath] = beforeText
            return LocalEditorForwardResult.Retry
        }
        val replicaText = forwarder.replicaText()
        val forwarded =
            when (localReplicaBaselineDecisionUtil(replicaText, beforeText)) {
                LocalReplicaBaselineDecision.ForwardLocal -> {
                    currentEpoch == nonOperatorMutationEpoch(filePath) &&
                        forwarder.forwardLocalEdits(edits) &&
                        forwarder.replicaText() == editorText
                }
                LocalReplicaBaselineDecision.RebootstrapCanonicalThenForward ->
                    rebootstrapCanonicalAndForwardCapturedLocalEdit(
                        filePath = filePath,
                        capturedBaseText = beforeText,
                        visibleEditorText = editorText,
                        staleForwarder = forwarder,
                        edits = edits,
                    )
            }
        if (!forwarded) {
            // Retain the exact captured base. Advancing the shadow before the
            // delta reaches a canonical replica loses the only safe rebase
            // proof and turns the next retry into whole-editor adoption.
            shadows[filePath] = beforeText
            if (replicaText != beforeText) {
                retainedCanonicalProjectionPaths.add(filePath)
                log.warn(
                    "[crdt-replica] local delta found a stale native baseline for ${File(filePath).name}; " +
                        "shadow_hash=${contentHash(beforeText)} " +
                        "replica_hash=${replicaText?.let(::contentHash) ?: "missing"} " +
                        "recovery=canonical-rebootstrap-captured-local-delta",
                )
            }
            requestRemoteDrain(filePath, "captured-local-delta-retry")
            return LocalEditorForwardResult.Retry
        }
        shadows[filePath] = editorText
        if (!forwarders[filePath]!!.projectVisibleState(editorText)) {
            requestRemoteDrain(filePath, "local-visible-projection-retry")
        }
        logSlow(
            "forward-local-delta",
            filePath,
            started,
            details =
                "splices=${edits.size} before_chars=${beforeText.length} after_chars=${editorText.length}",
        )
        return LocalEditorForwardResult.Applied
    }

    private fun rebootstrapCanonicalAndForwardCapturedLocalEdit(
        filePath: String,
        capturedBaseText: String,
        visibleEditorText: String,
        staleForwarder: CrdtReplicaForwarder,
        edits: List<PreparedLocalEditorEdit>,
    ): Boolean {
        if (forwarders[filePath] !== staleForwarder) return false
        val replacement =
            forwarderFor(
                filePath = filePath,
                bypassRegisterBackoff = true,
                replaceCached = true,
                expectedEditorTextAtSwap = visibleEditorText,
                allowPendingLocalAtSwap = true,
                bootstrapFromControllerCanonical = true,
                expectedCanonicalTextAtSwap = capturedBaseText,
            )
        if (replacement == null || replacement === staleForwarder) return false
        if (
            editorBufferText(filePath) != visibleEditorText ||
            replacement.replicaText() != capturedBaseText
        ) {
            return false
        }
        return replacement.forwardLocalEdits(edits) &&
            replacement.replicaText() == visibleEditorText
    }

    fun requestRemoteDrain(filePath: String? = null, reason: String = "event") {
        if (filePath == null) {
            drainAllRequested.set(true)
        } else {
            drainRequestedPaths.add(filePath)
        }
        queueRemoteDrain(reason)
    }

    /** Queue retained drain flags without creating new work. */
    private fun queueRemoteDrain(reason: String) {
        if (!shouldStartRemoteDrainUtil(remoteDrainBackoffScheduled.get())) return
        if (!drainQueued.compareAndSet(false, true)) return
        if (!shouldStartRemoteDrainUtil(remoteDrainBackoffScheduled.get())) {
            drainQueued.set(false)
            return
        }
        executor.execute {
            var appliedTotal = 0
            try {
                appliedTotal = drainRemoteUpdates(reason)
            } catch (e: Exception) {
                log.debug("[crdt-replica] remote drain skipped: ${e.message}")
            } finally {
                val moreWorkRequested = drainAllRequested.get() || drainRequestedPaths.isNotEmpty()
                if (moreWorkRequested && appliedTotal == 0) {
                    // #crdt-drain-backoff: when a drain cycle applied zero useful
                    // updates (notably when the CP socket is unavailable and every
                    // pullDelivery returns empty deltas), delay the reschedule with
                    // exponential backoff instead of re-executing immediately. A
                    // tight no-op spin generated ~70MB/min of logs and froze the IDE.
                    val delayMs = nextNoOpRescheduleBackoffMs()
                    log.debug("[crdt-replica] no-op drain cycle; backing off reschedule by ${delayMs}ms (consecutive=${consecutiveNoOpReschedules.get()})")
                    // Publish the backoff gate before releasing drainQueued so an
                    // external CRDT event cannot win the gap and start immediately.
                    scheduleRemoteDrainAfterBackoff(delayMs)
                    drainQueued.set(false)
                } else if (moreWorkRequested) {
                    consecutiveNoOpReschedules.set(0)
                    drainQueued.set(false)
                    queueRemoteDrain("rescheduled")
                } else {
                    consecutiveNoOpReschedules.set(0)
                    drainQueued.set(false)
                    // Close the request-vs-release race: a request can arrive after the
                    // moreWorkRequested snapshot but before drainQueued is released.
                    if (drainAllRequested.get() || drainRequestedPaths.isNotEmpty()) {
                        queueRemoteDrain("handoff")
                    }
                }
            }
        }
    }

    /**
     * Foreground delivery recovery for a controller write that is already
     * retained in the existing replica's delivery frontier. This deliberately
     * bypasses only the background no-op drain timer: it neither clears that
     * timer nor replaces the replica. Re-registering from the visible editor
     * here would publish the pre-delivery buffer back into canonical and undo
     * the controller write before the editor had a chance to apply it.
     */
    fun requestUrgentRemoteDrain(filePath: String, reason: String) {
        documentWorkers.forDocument(filePath).execute {
            val forwarder = forwarders[filePath] ?: return@execute
            var applied = 0
            try {
                applied = drainRemoteUpdatesFor(filePath, forwarder)
            } catch (e: Exception) {
                log.debug("[crdt-replica] urgent remote drain skipped for $filePath: ${e.message}")
            } finally {
                log.debug(
                    "[crdt-replica] urgent remote drain completed for ${File(filePath).name}; " +
                        "reason=$reason applied=$applied",
                )
                if (applied == 0 && !disposed.get()) {
                    requestRemoteDrain(filePath, "$reason-follow-up")
                } else if (applied > 0) {
                    // #crdtpushdrain: useful work proves the document is live again,
                    // so the escalated no-op backoff is stale. Without this reset the
                    // gate stays parked at its previous (up to 30s) delay and the
                    // *next* controller push is suppressed all over again.
                    consecutiveNoOpReschedules.set(0)
                }
            }
        }
    }

    private fun nextNoOpRescheduleBackoffMs(): Long {
        val n = consecutiveNoOpReschedules.incrementAndGet()
        val shifted = 1L shl minOf(n - 1, 12)
        return minOf(
            CRDT_DRAIN_NOOP_RESCHEDULE_BASE_BACKOFF_MS * shifted,
            CRDT_DRAIN_NOOP_RESCHEDULE_MAX_BACKOFF_MS,
        )
    }

    private fun scheduleRemoteDrainAfterBackoff(delayMs: Long, retryFilePath: String? = null) {
        if (retryFilePath != null) drainRequestedPaths.add(retryFilePath)
        if (!remoteDrainBackoffScheduled.compareAndSet(false, true)) return
        executor.schedule(
            {
                remoteDrainBackoffScheduled.set(false)
                if (!disposed.get()) queueRemoteDrain(CRDT_DRAIN_BACKOFF_REASON)
            },
            delayMs,
            TimeUnit.MILLISECONDS,
        )
    }

    private fun drainRemoteUpdates(reason: String): Int {
        val started = System.nanoTime()
        val drainAll = drainAllRequested.getAndSet(false)
        val paths = if (drainAll) {
            // Every currently requested path is already covered by this all-replica
            // snapshot. Leaving those flags behind rearms the no-op backoff forever.
            drainRequestedPaths.clear()
            forwarders.keys().toList()
        } else {
            drainRequestedPaths.toList().also { drained ->
                drained.forEach { drainRequestedPaths.remove(it) }
            }
        }
        if (paths.isEmpty()) return 0
        log.debug("[crdt-replica] draining ${paths.size} replica(s) via $reason")
        val drains = paths.mapNotNull { filePath ->
            val forwarder = forwarders[filePath] ?: return@mapNotNull null
            filePath to documentWorkers.forDocument(filePath).submit<Int> {
                if (forwarders[filePath] !== forwarder) {
                    0
                } else {
                    drainRemoteUpdatesFor(filePath, forwarder)
                }
            }
        }
        var appliedTotal = 0
        for ((filePath, drain) in drains) {
            try {
                appliedTotal += drain.get()
            } catch (e: Exception) {
                log.debug("[crdt-replica] document drain skipped for $filePath: ${e.message}")
            }
        }
        logSlow(
            "remote-drain",
            paths.firstOrNull() ?: "(none)",
            started,
            warnMs = if (appliedTotal == 0) CRDT_IDLE_WORKER_WARN_MS else CRDT_WORKER_WARN_MS,
            details = "paths=${paths.size} reason=$reason drain_all=$drainAll applied_total=$appliedTotal",
        )
        return appliedTotal
    }

    private fun drainRemoteUpdatesFor(filePath: String, forwarder: CrdtReplicaForwarder): Int {
        val started = System.nanoTime()
        var updateCount = 0
        var selfEchoCount = 0
        var peerUpdateCount = 0
        var queuedForEditor = false
        var deliveryKind = "deltas"
        var usefulWork = 0
        if (hasPendingLocal(filePath) || remoteEditorApplyPaths.contains(filePath)) return 0
        try {
            val expectedText = shadows[filePath] ?: return 0
            // D2: a replace delivery (out-of-band deletion re-bootstrap) installs
            // the corrected canonical only when the editor buffer still matches
            // the local replica baseline; normal deltas are merged into the native
            // replica first, then applied to the editor in one EDT command.
            val delivery = forwarder.pullRemoteDelivery()
            if (pullDeliveryRequestsReplicaRefreshUtil(delivery)) {
                val reason = (delivery as ReplicaPullDelivery.Unavailable).reason
                refreshReplicaAfterTransportLoss(filePath, forwarder, expectedText, reason)
                return usefulWork
            }
            if (delivery is ReplicaPullDelivery.Replace) {
                deliveryKind = "replace"
            queuedForEditor = applyReplaceDelivery(filePath, forwarder, expectedText, delivery.text)
                usefulWork += if (queuedForEditor) 1 else 0
                return usefulWork
            }
            val updates = (delivery as ReplicaPullDelivery.Deltas).updates
            updateCount = updates.size
            usefulWork = updateCount
            if (updates.isEmpty()) return usefulWork

        if (!editorReplicaBaselineMatches(filePath, forwarder, expectedText, updates)) {
            return usefulWork
        }
            val appliedRemoteUpdates = mutableListOf<ReplicaRemoteUpdate>()
            var converged: String? = null
            for (update in updates) {
                if (hasPendingLocal(filePath)) break
                if (!shouldApplyRemoteCrdtUpdateUtil(update, forwarder.clientId)) {
                    selfEchoCount++
                    val visibleText = editorBufferText(filePath) ?: expectedText
                    if (!forwarder.projectVisibleState(visibleText)) {
                        requestRemoteDrain(filePath, "self-echo-projection-retry")
                    }
                    continue
                }
                peerUpdateCount++
                converged = forwarder.applyRemoteUpdate(update.update) ?: break
                appliedRemoteUpdates.add(update)
            }

            val targetText = converged
            if (targetText != null && appliedRemoteUpdates.isNotEmpty() && !hasPendingLocal(filePath)) {
                val projectionExpectedText =
                    if (
                        retainedCanonicalProjectionPaths.contains(filePath) &&
                        !TypingTracker.hasUnsyncedOperatorEdits(filePath)
                    ) {
                        editorBufferText(filePath) ?: expectedText
                    } else {
                        expectedText
                    }
                when (
                    queueRemoteTextApply(
                        filePath,
                        projectionExpectedText,
                        targetText,
                        forwarder,
                        appliedRemoteUpdates,
                    )
                ) {
                    RemoteTextApplyDisposition.Queued -> queuedForEditor = true
                    RemoteTextApplyDisposition.Recovered -> usefulWork++
                    RemoteTextApplyDisposition.RetryFailClosed -> {
                        editorBufferText(filePath)?.let { current -> shadows[filePath] = current }
                    }
                }
            }
            usefulWork = peerUpdateCount
        } finally {
            logSlow(
                "remote-drain-file",
                filePath,
                started,
                warnMs = if (usefulWork == 0) CRDT_IDLE_WORKER_WARN_MS else CRDT_WORKER_WARN_MS,
                details = "delivery=$deliveryKind updates=$updateCount peer=$peerUpdateCount self=$selfEchoCount queued=$queuedForEditor",
            )
        }
        return usefulWork
    }

    /**
     * D2 — apply a REPLACE delivery: install the corrected canonical text into the
     * buffer wholesale (an out-of-band deletion the additive CRDT delta cannot
     * express), then re-bootstrap the local replica node so later deltas are
     * relative to the corrected state. Never clobbers editor-buffer text that
     * has advanced past the local replica baseline; in that case the buffer is
     * published back through the relay and the replacement is dropped.
     */
    private fun applyReplaceDelivery(
        filePath: String,
        forwarder: CrdtReplicaForwarder,
        expectedText: String,
        canonical: String,
    ): Boolean {
        val remoteState =
            templateStructureState(
                filePath,
                canonical,
                TemplateValidationPlane.Lane.RemoteCandidate,
                "replace-remote",
            )
        if (!remoteReplaceStructureAcceptedUtil(remoteState)) {
            recoverRejectedRemoteCanonical(
                filePath = filePath,
                expectedText = expectedText,
                remoteText = canonical,
                staleForwarder = forwarder,
                remoteState = remoteState,
            )
            return false
        }
        if (hasPendingLocal(filePath)) return false
        val started = System.nanoTime()
        var installed = false
        var deferredEditorText: String? = null
        val replicaText = forwarder.replicaText()
        if (!prepareNonOperatorEditorMutationOnWorker(filePath)) {
            log.warn("[crdt-replica] replace delivery retained because native op-capture fencing failed for $filePath")
            scheduleTemplateGuardRecoveryRetry(filePath, "replace-delivery-op-epoch")
            return false
        }
        try {
            ApplicationManager.getApplication().invokeAndWait {
                val edtStarted = System.nanoTime()
                try {
                    val targetFile = LocalFileSystem.getInstance().findFileByPath(filePath) ?: return@invokeAndWait
                    val document = FileDocumentManager.getInstance().getDocument(targetFile) ?: return@invokeAndWait
                    if (!refreshCleanDocumentBeforeRemoteApply(filePath, targetFile, document)) {
                        deferredEditorText = document.text
                        return@invokeAndWait
                    }
                    val before = document.text
                    if (before == canonical) {
                        shadows[filePath] = canonical
                        installed = persistRemoteCrdtTextIfSafe(
                            filePath,
                            document,
                            expectedText,
                            canonical,
                            before,
                        ).diskPersisted
                        return@invokeAndWait
                    }
                    if (hasPendingLocal(filePath)) return@invokeAndWait
                    if (!remoteCrdtReplaceStillCurrentUtil(expectedText, before, replicaText)) {
                        val editorHash = contentHash(before)
                        val expectedHash = contentHash(expectedText)
                        val replicaHash = replicaText?.let(::contentHash) ?: "missing"
                        log.warn(
                            "[crdt-replica] replace delivery observed non-operator editor divergence for $filePath: " +
                                "editor_hash=$editorHash expected_hash=$expectedHash replica_hash=$replicaHash canonical_hash=${contentHash(canonical)}"
                        )
                        deferredEditorText = before
                        return@invokeAndWait
                    }
                    if (!remoteCrdtDiskCanPersistUtil(expectedText, canonical, readRawDiskText(filePath))) {
                        log.warn(
                            "[crdt-replica] replace delivery rejected because disk contains novel external text for $filePath; " +
                                "expected_hash=${contentHash(expectedText)} canonical_hash=${contentHash(canonical)}"
                        )
                        return@invokeAndWait
                    }
                    advanceNonOperatorMutationEpoch(filePath)
                    applyingRemote.add(filePath)
                    try {
                        runUndoableRemoteUpdateCommand(document) {
                            applyMinimalDocumentEditUtil(document, before, canonical)
                        }
                        shadows[filePath] = canonical
                        installed = persistRemoteCrdtTextIfSafe(
                            filePath,
                            document,
                            expectedText,
                            canonical,
                            before,
                        ).diskPersisted
                        if (installed) {
                            log.info("[crdt-replica] applied and saved REPLACE re-bootstrap for $filePath (${canonical.length} chars)")
                        } else {
                            scheduleTemplateGuardRecoveryRetry(
                                filePath,
                                "replace-delivery-persist-rollback",
                            )
                        }
                    } finally {
                        applyingRemote.remove(filePath)
                    }
                } finally {
                    logSlow("replace-apply-edt", filePath, edtStarted, warnMs = CRDT_EDT_WARN_MS, details = "target_chars=${canonical.length}")
                }
            }
        } finally {
            logSlow("replace-apply-total", filePath, started, details = "target_chars=${canonical.length} installed=$installed deferred=${deferredEditorText != null}")
        }
        deferredEditorText?.let { editorText ->
            log.warn(
                "[crdt-replica] CP replace retained while the live editor diverges for $filePath; " +
                    "editor_hash=${contentHash(editorText)} canonical_hash=${contentHash(canonical)}",
            )
            retainedCanonicalProjectionPaths.add(filePath)
            requestRemoteDrain(filePath, "replace-delivery-lazy-canonical-projection")
            return false
        }
        if (installed) {
            // Re-open from the canonical bootstrap. Editing the divergent local
            // CRDT until its *text* matches would mint replacement ops and merge
            // them back into a canonical that already contains the response,
            // potentially duplicating content. A true rebootstrap discards the
            // divergent lineage and also retires its stale pending delivery.
            if (forwarders[filePath] === forwarder) {
                val reattached = forwarderFor(
                    filePath,
                    canonical,
                    bypassRegisterBackoff = true,
                    expectedEditorTextAtSwap = canonical,
                )
                if (reattached == null) {
                    log.warn("[crdt-replica] canonical re-bootstrap could not reattach ${File(filePath).name}; the normal attach path will retry")
                }
            }
        }
        return installed
    }

    private fun queueRemoteTextApply(
        filePath: String,
        expectedText: String,
        converged: String,
        forwarder: CrdtReplicaForwarder,
        updates: List<ReplicaRemoteUpdate>,
    ): RemoteTextApplyDisposition {
        val remoteState =
            templateStructureState(
                filePath,
                converged,
                TemplateValidationPlane.Lane.RemoteCandidate,
                "remote",
            )
        if (remoteState != TemplateStructureProjectionState.Exact) {
            return recoverRejectedRemoteCanonical(
                filePath = filePath,
                expectedText = expectedText,
                remoteText = converged,
                staleForwarder = forwarder,
                remoteState = remoteState,
            )
        }
        // Retain authority before the native op-capture fence. That fence may
        // time out behind a compact/write RPC, but the already-pulled canonical
        // target must still block request-full-state and stale-baseline adoption.
        retainedCanonicalProjectionPaths.add(filePath)
        if (!prepareNonOperatorEditorMutationOnWorker(filePath)) {
            log.warn("[crdt-replica] remote editor apply retained because native op-capture fencing failed for $filePath")
            scheduleTemplateGuardRecoveryRetry(filePath, "remote-editor-apply-op-epoch")
            return RemoteTextApplyDisposition.RetryFailClosed
        }
        remoteEditorApplyPaths.add(filePath)
        val effectCounter =
            remoteEditorEffectGenerations
                .computeIfAbsent(filePath) { AtomicLong(0L) }
        val effectGeneration = effectCounter.get() + 1L
        val outcome = remoteEditorApplies.ingress(
            filePath,
            PendingRemoteEditorApply(
                filePath = filePath,
                expectedText = expectedText,
                targetText = converged,
                effectToken =
                    RemoteEditorEffectToken(
                        generation = effectGeneration,
                        endpoint = forwarder,
                    ),
            ),
        )
        if (outcome != IngressOutcome.Blocked && outcome != IngressOutcome.Dropped) {
            effectCounter.set(effectGeneration)
        }
        log.debug(
            "[crdt-replica] remote editor apply ${outcome.name.lowercase()} for ${File(filePath).name}; " +
                "pending_keys=${remoteEditorApplies.pendingKeyCount()} updates=${updates.size}",
        )
        scheduleRemoteEditorApply()
        return if (outcome != IngressOutcome.Blocked && outcome != IngressOutcome.Dropped) {
            RemoteTextApplyDisposition.Queued
        } else {
            scheduleTemplateGuardRecoveryRetry(filePath, "remote-editor-apply-${outcome.name.lowercase()}")
            RemoteTextApplyDisposition.RetryFailClosed
        }
    }

    private fun recoverRejectedRemoteCanonical(
        filePath: String,
        expectedText: String,
        remoteText: String,
        staleForwarder: CrdtReplicaForwarder,
        remoteState: TemplateStructureProjectionState,
    ): RemoteTextApplyDisposition {
        retainedCanonicalProjectionPaths.add(filePath)
        log.warn(
            "[crdt-replica] rejected malformed remote projection for ${File(filePath).name}; " +
                "remote_state=$remoteState expected_hash=${contentHash(expectedText)} " +
                "remote_hash=${contentHash(remoteText)} recovery=lazy-controller-canonical-projection",
        )
        requestRemoteDrain(filePath, "template-guard-lazy-canonical-projection")
        return RemoteTextApplyDisposition.RetryFailClosed
    }

    private fun scheduleTemplateGuardRecoveryRetry(filePath: String, reason: String) {
        if (!templateGuardRecoveryRetryPaths.add(filePath)) return
        val failureCount = templateGuardRecoveryFailureCounts.merge(filePath, 1) { current, one ->
            current + one
        } ?: 1
        val shifted = 1L shl minOf(failureCount - 1, 12)
        val delayMs = minOf(
            CRDT_DRAIN_NOOP_RESCHEDULE_BASE_BACKOFF_MS * shifted,
            CRDT_DRAIN_NOOP_RESCHEDULE_MAX_BACKOFF_MS,
        )
        log.debug(
            "[crdt-replica] template-guard recovery retry scheduled for ${File(filePath).name}; " +
                "reason=$reason delay_ms=$delayMs failures=$failureCount",
        )
        documentWorkers.forDocument(filePath).schedule(
            {
                templateGuardRecoveryRetryPaths.remove(filePath)
                if (!disposed.get()) requestRemoteDrain(filePath, "template-guard-retry")
            },
            delayMs,
            TimeUnit.MILLISECONDS,
        )
    }

    private fun clearTemplateGuardRecoveryBackoff(filePath: String) {
        templateGuardRecoveryFailureCounts.remove(filePath)
        templateGuardRecoveryRetryPaths.remove(filePath)
    }

    private fun scheduleRemoteEditorApply() {
        if (disposed.get() || project.isDisposed) return
        if (!remoteEditorApplyScheduled.compareAndSet(false, true)) return
        try {
            ApplicationManager.getApplication().invokeLater {
                try {
                    if (disposed.get() || project.isDisposed) {
                        remoteEditorApplies.clear()
                    } else {
                        remoteEditorApplies.drainOne()?.second?.let(::applyRemoteTextOnEdt)
                    }
                } finally {
                    remoteEditorApplyScheduled.set(false)
                    if (remoteEditorApplies.hasPending()) scheduleRemoteEditorApply()
                }
            }
        } catch (e: RuntimeException) {
            remoteEditorApplyScheduled.set(false)
            throw e
        }
    }

    private fun applyRemoteTextOnEdt(pending: PendingRemoteEditorApply) {
        val started = System.nanoTime()
        val liveEffectGeneration = remoteEditorEffectGenerations[pending.filePath]?.get()
        if (
            !remoteEditorEffectTokenCurrentUtil(
                tokenGeneration = pending.effectToken.generation,
                liveGeneration = liveEffectGeneration,
                endpointMatches = forwarders[pending.filePath] === pending.effectToken.endpoint,
                endpointBacked = pending.effectToken.endpoint.attached,
            )
        ) {
            log.warn(
                "[crdt-replica] remote editor effect refused for ${pending.filePath}; " +
                    "reason=retired_or_superseded_or_model_less " +
                    "token_generation=${pending.effectToken.generation} live_generation=$liveEffectGeneration",
            )
            return completeRemoteEditorApply(
                pending,
                RemoteEditorApplyOutcome(false, null),
                started,
            )
        }
        val outcome = try {
            val targetFile = LocalFileSystem.getInstance().findFileByPath(pending.filePath)
                ?: return completeRemoteEditorApply(pending, RemoteEditorApplyOutcome(false, null), started)
            val document = FileDocumentManager.getInstance().getDocument(targetFile)
                ?: return completeRemoteEditorApply(pending, RemoteEditorApplyOutcome(false, null), started)
            val fileDocumentManager = FileDocumentManager.getInstance()
            val documentWasUnsaved = fileDocumentManager.isDocumentUnsaved(document)
            if (
                !documentWasUnsaved &&
                !refreshCleanDocumentBeforeRemoteApply(pending.filePath, targetFile, document)
            ) {
                return completeRemoteEditorApply(
                    pending,
                    RemoteEditorApplyOutcome(false, document.text),
                    started,
                )
            }
            val before = document.text
            val projectionMode =
                if (documentWasUnsaved) {
                    RemoteCrdtProjectionMode.MemoryOnly
                } else {
                    remoteCrdtProjectionModeUtil(
                        documentUnsaved = false,
                        diskCanPersist =
                            remoteCrdtDiskCanPersistUtil(
                                pending.expectedText,
                                pending.targetText,
                                readRawDiskText(pending.filePath),
                            ),
                    )
                }
            if (before == pending.targetText) {
                shadows[pending.filePath] = pending.targetText
                if (projectionMode == RemoteCrdtProjectionMode.Persist) {
                    val persisted =
                        persistRemoteCrdtTextIfSafe(
                            pending.filePath,
                            document,
                            pending.expectedText,
                            pending.targetText,
                            before,
                        )
                    RemoteEditorApplyOutcome(
                        persisted.diskPersisted,
                        persisted.editorTextForProjection,
                        persisted.editorNormalizedText,
                    )
                } else {
                    RemoteEditorApplyOutcome(false, before)
                }
            } else if (hasPendingLocal(pending.filePath)) {
                RemoteEditorApplyOutcome(false, before)
            } else if (!remoteCrdtApplyStillCurrentUtil(pending.expectedText, before, pending.targetText)) {
                log.warn("[crdt-replica] stale coalesced remote update rejected for ${pending.filePath}; editor text advanced before apply")
                RemoteEditorApplyOutcome(false, before)
            } else if (projectionMode == RemoteCrdtProjectionMode.Reject) {
                log.warn(
                    "[crdt-replica] coalesced remote update rejected because disk contains novel external text for ${pending.filePath}; " +
                        "expected_hash=${contentHash(pending.expectedText)} target_hash=${contentHash(pending.targetText)}"
                )
                RemoteEditorApplyOutcome(false, before)
            } else {
                if (projectionMode == RemoteCrdtProjectionMode.MemoryOnly) {
                    log.debug(
                        "[crdt-replica] projecting remote delta into unsaved editor memory for ${pending.filePath}; " +
                            "durable_sink=crdt_canonical disk_write=deferred",
                    )
                }
                advanceNonOperatorMutationEpoch(pending.filePath)
                applyingRemote.add(pending.filePath)
                try {
                    runUndoableRemoteUpdateCommand(document) {
                        applyMinimalDocumentEditUtil(document, before, pending.targetText)
                        shadows[pending.filePath] = pending.targetText
                    }
                    if (projectionMode == RemoteCrdtProjectionMode.Persist) {
                        val persisted =
                            persistRemoteCrdtTextIfSafe(
                                pending.filePath,
                                document,
                                pending.expectedText,
                                pending.targetText,
                                before,
                            )
                        RemoteEditorApplyOutcome(
                            persisted.diskPersisted,
                            persisted.editorTextForProjection,
                            persisted.editorNormalizedText,
                        )
                    } else {
                        RemoteEditorApplyOutcome(false, pending.targetText)
                    }
                } finally {
                    applyingRemote.remove(pending.filePath)
                }
            }
        } catch (e: RuntimeException) {
            log.warn("[crdt-replica] coalesced remote editor apply failed for ${pending.filePath}", e)
            RemoteEditorApplyOutcome(false, null)
        }
        completeRemoteEditorApply(pending, outcome, started)
    }

    private fun readRawDiskText(filePath: String): String? =
        try {
            File(filePath).readText()
        } catch (e: Exception) {
            log.warn("[crdt-replica] raw disk read failed for $filePath: ${e.message}")
            null
        }

    /**
     * Synchronize the target file's VFS stamp before a remote CRDT edit is
     * installed and saved. IntelliJ resolves save conflicts by modification
     * stamp, not by comparing the bytes that [readRawDiskText] validated. An
     * external agent-doc write can therefore leave a clean Document with a
     * stale VirtualFile stamp; editing first and saving second would arm the
     * File Cache Conflict dialog even when disk still equals [expectedText].
     *
     * Refreshing an unsaved Document is the inverse hazard: it immediately asks
     * IntelliJ to choose between operator memory and disk. Delta deliveries skip
     * this helper for unsaved buffers, project only into the Document, and leave
     * persistence to the durable CRDT canonical. Replace deliveries still fail
     * closed here because they do not carry merge semantics.
     */
    private fun refreshCleanDocumentBeforeRemoteApply(
        filePath: String,
        targetFile: VirtualFile,
        document: Document,
    ): Boolean {
        val fileDocumentManager = FileDocumentManager.getInstance()
        if (!shouldRefreshVfsBeforeApplyUtil(fileDocumentManager.isDocumentUnsaved(document))) {
            log.debug("[crdt-replica] remote apply deferred before VFS refresh because the editor is unsaved for $filePath")
            return false
        }
        targetFile.refresh(false, false)
        if (fileDocumentManager.isDocumentUnsaved(document)) {
            log.warn("[crdt-replica] remote apply deferred because the editor became unsaved during the clean VFS refresh for $filePath")
            return false
        }
        return true
    }

    private fun persistRemoteCrdtTextIfSafe(
        filePath: String,
        document: Document,
        expectedText: String,
        targetText: String,
        beforeText: String,
    ): RemotePersistOutcome {
        val diskText = readRawDiskText(filePath)
        if (!remoteCrdtDiskCanPersistUtil(expectedText, targetText, diskText)) {
            return reconcileRemotePersistence(
                filePath,
                document,
                beforeText,
                targetText,
                diskText,
            )
        }
        return try {
            val fileDocumentManager = FileDocumentManager.getInstance()
            fileDocumentManager.saveDocument(document)
            val diskAfterSave = readRawDiskText(filePath)
            val outcome = reconcileRemotePersistence(
                filePath,
                document,
                beforeText,
                targetText,
                diskAfterSave,
            )
            if (
                outcome.diskPersisted &&
                fileDocumentManager.isDocumentUnsaved(document)
            ) {
                log.debug(
                    "[crdt-replica] remote editor apply reached exact disk bytes while IntelliJ save state still lagged for $filePath",
                )
            }
            outcome
        } catch (e: RuntimeException) {
            log.warn("[crdt-replica] remote editor apply save failed for $filePath", e)
            reconcileRemotePersistence(
                filePath,
                document,
                beforeText,
                targetText,
                readRawDiskText(filePath),
            )
        }
    }

    private fun reconcileRemotePersistence(
        filePath: String,
        document: Document,
        beforeText: String,
        targetText: String,
        diskAfterSave: String?,
    ): RemotePersistOutcome {
        return when (
            remotePersistReconciliationUtil(
                beforeText,
                targetText,
                document.text,
                diskAfterSave,
            )
        ) {
            RemotePersistReconciliation.Persisted -> {
                shadows[filePath] = targetText
                RemotePersistOutcome(true, targetText)
            }
            RemotePersistReconciliation.PersistedEditorNormalization -> {
                val normalizedText = document.text
                shadows[filePath] = normalizedText
                log.info(
                    "[crdt-replica] editor save normalized a remote projection for $filePath; " +
                        "target_hash=${contentHash(targetText)} normalized_hash=${contentHash(normalizedText)} " +
                        "disk_hash=${diskAfterSave?.let(::contentHash) ?: "missing"} " +
                        "recovery=project_normalized_editor_text",
                )
                RemotePersistOutcome(
                    diskPersisted = true,
                    editorTextForProjection = normalizedText,
                    editorNormalizedText = normalizedText,
                )
            }
            RemotePersistReconciliation.RollbackToBefore -> {
                log.warn(
                    "[crdt-replica] remote editor apply did not persist; rolling the exact attempted projection back for $filePath: " +
                        "before_hash=${contentHash(beforeText)} target_hash=${contentHash(targetText)} " +
                        "editor_hash=${contentHash(document.text)} disk_hash=${diskAfterSave?.let(::contentHash) ?: "missing"}",
                )
                if (document.text == targetText) {
                    runUndoableRemoteUpdateCommand(document) {
                        applyMinimalDocumentEditUtil(document, targetText, beforeText)
                    }
                }
                shadows[filePath] = beforeText
                if (document.text == beforeText && diskAfterSave == beforeText) {
                    FileDocumentManager.getInstance().reloadFromDisk(document)
                }
                RemotePersistOutcome(false, beforeText)
            }
            RemotePersistReconciliation.PreserveAdvancedEditor -> {
                val editorText = document.text
                shadows[filePath] = editorText
                log.warn(
                    "[crdt-replica] remote editor persistence diverged from both exact planes; preserving the advanced editor and withholding disk-persisted projection for $filePath: " +
                        "before_hash=${contentHash(beforeText)} target_hash=${contentHash(targetText)} " +
                        "editor_hash=${contentHash(editorText)} disk_hash=${diskAfterSave?.let(::contentHash) ?: "missing"} " +
                        "document_unsaved=${FileDocumentManager.getInstance().isDocumentUnsaved(document)} " +
                        "pending_local=${hasPendingLocal(filePath)}",
                )
                RemotePersistOutcome(false, null)
            }
        }
    }

    private fun completeRemoteEditorApply(
        pending: PendingRemoteEditorApply,
        outcome: RemoteEditorApplyOutcome,
        started: Long,
    ) {
        val projectionVisible = shouldProjectVisibleRemoteDeliveryUtil(
            outcome.editorText,
            pending.targetText,
            outcome.diskPersisted,
        )
        logSlow(
            "remote-apply-edt",
            pending.filePath,
            started,
            warnMs = CRDT_EDT_WARN_MS,
            details = "target_chars=${pending.targetText.length} visible=$projectionVisible disk_persisted=${outcome.diskPersisted}",
        )
        if (disposed.get()) return
        try {
            documentWorkers.forDocument(pending.filePath).execute {
                var projectionPublished = false
                try {
                    val normalizedText = outcome.editorNormalizedText
                    if (normalizedText != null) {
                        retainedCanonicalProjectionPaths.add(pending.filePath)
                        log.info(
                            "[crdt-replica] editor normalization retained for controller reprojection for " +
                                "${File(pending.filePath).name}; normalized_hash=${contentHash(normalizedText)}",
                        )
                    }
                    projectionPublished =
                        projectionVisible &&
                            outcome.editorText?.let { visibleText ->
                                pending.effectToken.endpoint.projectVisibleState(
                                    visibleText,
                                    outcome.diskPersisted,
                                )
                            } == true
                    log.debug(
                        "[crdt-replica] remote editor apply completed for ${File(pending.filePath).name}; " +
                            "visible=$projectionVisible disk_persisted=${outcome.diskPersisted} projection_published=$projectionPublished",
                    )
                    if (!projectionPublished && projectionVisible) {
                        requestRemoteDrain(
                            pending.filePath,
                            "remote-editor-state-projection-retry",
                        )
                    }
                } finally {
                    remoteEditorApplyPaths.remove(pending.filePath)
                    if (projectionPublished && outcome.editorNormalizedText == null) {
                        retainedCanonicalProjectionPaths.remove(pending.filePath)
                        consecutiveNoOpReschedules.set(0)
                    } else if (outcome.editorNormalizedText != null) {
                        requestRemoteDrain(
                            pending.filePath,
                            "remote-editor-normalization-canonical-reproject",
                        )
                    } else {
                        val delayMs = nextNoOpRescheduleBackoffMs()
                        log.debug(
                            "[crdt-replica] remote editor projection not yet visible for ${File(pending.filePath).name}; " +
                                "backing off retry by ${delayMs}ms",
                        )
                        scheduleRemoteDrainAfterBackoff(delayMs, pending.filePath)
                    }
                }
            }
        } catch (_: RejectedExecutionException) {
            remoteEditorApplyPaths.remove(pending.filePath)
            if (projectionVisible) {
                retainedCanonicalProjectionPaths.remove(pending.filePath)
            }
        }
    }

    private fun editorReplicaBaselineMatches(
        filePath: String,
        forwarder: CrdtReplicaForwarder,
        expectedText: String,
        updates: List<ReplicaRemoteUpdate>,
    ): Boolean {
        val editorText = editorBufferText(filePath) ?: return false
        val replicaText = forwarder.replicaText()
        val editorState =
            templateStructureState(
                filePath,
                editorText,
                TemplateValidationPlane.Lane.Editor,
                "editor-baseline",
            )
        val editorHash = contentHash(editorText)
        val replicaHash = replicaText?.let(::contentHash)
        val editorRemoteGeneration = matchingRemoteTargetGenerationUtil(updates, editorHash)
        val replicaRemoteGeneration = matchingRemoteTargetGenerationUtil(updates, replicaHash)
        val canonicalProjectionRetained = retainedCanonicalProjectionPaths.contains(filePath)
        val decision = replicaBaselineDecisionUtil(
            editorState = editorState,
            editorMatchesExpected = editorText == expectedText,
            replicaMatchesExpected = replicaText == expectedText,
            replicaMatchesEditor = replicaText == editorText,
            editorMatchesRemoteTarget = editorRemoteGeneration != null,
            replicaMatchesRemoteTarget = replicaRemoteGeneration != null,
            recoveryInFlight = templateGuardRecoveryPaths.contains(filePath),
            canonicalProjectionRetained = canonicalProjectionRetained,
        )
        if (
            decision == ReplicaBaselineDecision.ApplyRemote ||
            decision == ReplicaBaselineDecision.ApplyRemoteRepair
        ) {
            if (decision == ReplicaBaselineDecision.ApplyRemoteRepair) {
                log.warn(
                    "[crdt-replica] applying a structurally exact remote correction over the exact expected editor baseline for $filePath: " +
                        "editor_state=$editorState editor_hash=${contentHash(editorText)} expected_hash=${contentHash(expectedText)}",
                )
            }
            return true
        }
        val expectedHash = contentHash(expectedText)
        val visibleReplicaHash = replicaHash ?: "missing"
        if (
            decision == ReplicaBaselineDecision.ProjectRemoteTarget &&
            editorRemoteGeneration != null
        ) {
            shadows[filePath] = editorText
            if (!forwarder.projectVisibleState(editorText)) {
                requestRemoteDrain(filePath, "already-visible-state-projection-retry")
            }
            log.info(
                "[crdt-replica] projected an already-visible remote target for $filePath: " +
                    "editor_hash=$editorHash generation=$editorRemoteGeneration",
            )
            return false
        }
        if (
            decision == ReplicaBaselineDecision.RebootstrapVisibleRemoteTarget &&
            editorRemoteGeneration != null
        ) {
            val replacement = forwarderFor(
                filePath = filePath,
                initialEditorText = editorText,
                bypassRegisterBackoff = true,
                replaceCached = true,
                expectedEditorTextAtSwap = editorText,
                bootstrapFromControllerCanonical = true,
            )
            if (replacement == null || replacement === forwarder) {
                requestRemoteDrain(filePath, "visible-target-canonical-rebootstrap-retry")
                return false
            }
            val replacementText = replacement.replicaText()
            if (replacementText != editorText) {
                retainedCanonicalProjectionPaths.add(filePath)
                log.info(
                    "[crdt-replica] visible remote target advanced during canonical rebootstrap for $filePath: " +
                        "editor_hash=$editorHash replacement_hash=${replacementText?.let(::contentHash) ?: "missing"} " +
                        "generation=$editorRemoteGeneration",
                )
                requestRemoteDrain(filePath, "visible-target-canonical-advanced")
                return false
            }
            shadows[filePath] = editorText
            if (!replacement.projectVisibleState(editorText)) {
                retainedCanonicalProjectionPaths.add(filePath)
                requestRemoteDrain(filePath, "visible-target-projection-retry")
                return false
            }
            retainedCanonicalProjectionPaths.remove(filePath)
            log.info(
                "[crdt-replica] acknowledged an exact visible remote target after canonical replica rebootstrap for $filePath: " +
                    "editor_hash=$editorHash generation=$editorRemoteGeneration",
            )
            return false
        }
        if (
            decision == ReplicaBaselineDecision.ReplayRemoteTarget &&
            replicaText != null &&
            replicaRemoteGeneration != null
        ) {
            val replayedUpdates =
                updates.filter { it.generation <= replicaRemoteGeneration }
            val replayExpectedText =
                if (
                    canonicalProjectionRetained &&
                    !TypingTracker.hasUnsyncedOperatorEdits(filePath)
                ) {
                    editorText
                } else {
                    expectedText
                }
            log.info(
                "[crdt-replica] replaying a native remote target over its exact editor baseline for $filePath: " +
                    "editor_hash=$editorHash replica_hash=$visibleReplicaHash generation=$replicaRemoteGeneration " +
                    "retained_canonical=$canonicalProjectionRetained",
            )
            queueRemoteTextApply(
                filePath,
                replayExpectedText,
                replicaText,
                forwarder,
                replayedUpdates,
            )
            return false
        }
        if (decision == ReplicaBaselineDecision.RealignShadow) {
            log.warn(
                "[crdt-replica] incoming update deferred after shadow realignment for $filePath: " +
                    "editor_hash=$editorHash expected_hash=$expectedHash replica_hash=$visibleReplicaHash"
            )
            shadows[filePath] = editorText
            requestRemoteDrain(filePath, "shadow-realigned")
            return false
        }
        log.warn(
            "[crdt-replica] incoming update retained for lazy canonical projection because the baselines diverged for $filePath: " +
                "editor_state=$editorState editor_hash=$editorHash expected_hash=$expectedHash replica_hash=$visibleReplicaHash",
        )
        retainedCanonicalProjectionPaths.add(filePath)
        requestRemoteDrain(filePath, "baseline-diverged-lazy-canonical-projection")
        return false
    }

    private fun editorBufferText(filePath: String): String? {
        val targetFile = LocalFileSystem.getInstance().findFileByPath(filePath) ?: return null
        val application = ApplicationManager.getApplication()
        if (SwingUtilities.isEventDispatchThread() || application.isReadAccessAllowed) {
            return FileDocumentManager.getInstance().getDocument(targetFile)?.text
        }
        val applicationEx = application as? ApplicationEx ?: return null
        val text = AtomicReference<String?>()
        return if (
            applicationEx.tryRunReadAction {
                text.set(FileDocumentManager.getInstance().getDocument(targetFile)?.text)
            }
        ) {
            text.get()
        } else {
            null
        }
    }

    /**
     * Native/replica workers must never queue indefinitely behind IDEA's
     * write-intent permit. EDT callers already own safe editor access; workers
     * use ApplicationEx's immediate read attempt and let their retained event
     * retry when a writer has priority.
     */
    private fun tryReadDocumentText(document: Document): String? {
        val application = ApplicationManager.getApplication()
        if (SwingUtilities.isEventDispatchThread() || application.isReadAccessAllowed) {
            return document.text
        }
        val applicationEx = application as? ApplicationEx ?: return null
        val text = AtomicReference<String?>()
        return if (applicationEx.tryRunReadAction { text.set(document.text) }) {
            text.get()
        } else {
            null
        }
    }

    private fun contentHash(text: String): String =
        java.security.MessageDigest.getInstance("SHA-256")
            .digest(text.toByteArray(Charsets.UTF_8))
            .joinToString("") { "%02x".format(it) }

    private fun templateStructureState(
        filePath: String,
        text: String,
        lane: TemplateValidationPlane.Lane,
        source: String,
    ): TemplateStructureProjectionState {
        val started = System.nanoTime()
        return try {
            templateValidation.publish(filePath, lane, text).state.also { state ->
                if (state != TemplateStructureProjectionState.Exact) {
                    log.warn(
                        "[crdt-replica] $source text rejected by template-structure guard for $filePath; " +
                            "state=$state lane=$lane revision_hash=${contentHash(text)}",
                    )
                }
            }
        } finally {
            logSlow(
                "template-validation-computed",
                filePath,
                started,
                details = "source=$source lane=$lane target_chars=${text.length}",
            )
        }
    }

    private fun runUndoableRemoteUpdateCommand(document: Document, body: () -> Unit) {
        CommandProcessor.getInstance().executeCommand(
            project,
            {
                ApplicationManager.getApplication().runWriteAction {
                    body()
                }
            },
            "Agent Doc CRDT Remote Update",
            null,
            UndoConfirmationPolicy.DEFAULT,
            document,
        )
    }

    private fun forwarderFor(
        filePath: String,
        initialEditorText: String? = null,
        bypassRegisterBackoff: Boolean = false,
        replaceCached: Boolean = bypassRegisterBackoff,
        expectedEditorTextAtSwap: String? = null,
        allowPendingLocalAtSwap: Boolean = false,
        bootstrapFromControllerCanonical: Boolean = false,
        expectedCanonicalTextAtSwap: String? = null,
    ): CrdtReplicaForwarder? {
        val cached = forwarders[filePath]
        if (bypassRegisterBackoff) {
            // A refresh is register -> swap -> retire. Never create an authority
            // gap by deregistering the working member before its replacement has
            // accepted the canonical bootstrap.
            clearRegisterFailure(filePath)
        } else if (!replaceCached) {
            cached?.let { return it }
        }
        if (!bypassRegisterBackoff && !shouldAttemptRegister(filePath)) return cached
        val root = resolveProjectRoot(filePath) ?: return null
        val baseIdentity = "${EditorIdentity.id}:$filePath"
        val identity = if (replaceCached && cached != null) {
            "$baseIdentity:refresh-${refreshConnectionEpoch.incrementAndGet()}"
        } else {
            baseIdentity
        }
        val retainedResumeState =
            if (bootstrapFromControllerCanonical) {
                null
            } else {
                cached?.captureResumeState() ?: nativeReloadResumeStates[filePath]
            }
        val forwarder = CrdtReplicaForwarder(
            filePath = filePath,
            identity = identity,
            node = NativeReplicaNode(),
            transport = CpSocketReplicaTransport(root),
            ownershipContext = ownershipContext,
            resumeState = retainedResumeState,
        )
        if (!forwarder.register()) {
            recordRegisterFailure(filePath, forwarder.lastRegisterFailureReason ?: "controller-register")
            forwarder.lastRegisterFailureReason?.let { attachFailureReasons[filePath] = it }
            if (cached != null) {
                log.warn("[crdt-replica] replacement register failed for ${File(filePath).name}; retained cached forwarder")
            }
            return null
        }
        if (
            expectedCanonicalTextAtSwap != null &&
            forwarder.replicaText() != expectedCanonicalTextAtSwap
        ) {
            // The captured operator delta is only meaningful relative to its
            // exact base. Keep the cached endpoint and retry; never derive a
            // replacement edit from a different controller generation.
            forwarder.deregister()
            return null
        }
        clearRegisterFailure(filePath)
        if (forwarder.canonicalProjectionRetained) {
            // Controller state survives an IDEA/plugin restart; this local set
            // does not. Restore the fail-closed baseline before any whole-editor
            // synchronization can publish a stale restarted buffer.
            //
            // `#retainedprojectionclobbersoperatortext`: that reasoning holds for
            // a RESTARTED buffer and inverts for a live one. When registration is
            // refused for hours (see `#registeridentitypid`) the operator keeps
            // typing into a buffer that reaches nothing, and a retained projection
            // computed before those keystrokes then adopts over them on attach.
            // Measured 2026-08-12: a compaction computed at 04:14:47 against disk
            // was retained undelivered, and at 04:23 it replaced a buffer holding a
            // prompt typed in between. The prompt existed only in that buffer, so
            // it was not in git, not in the CRDT, and not in the compaction
            // archive — it was simply gone.
            //
            // The shadow separates the two cases exactly, and needs no new state:
            // it is the last text this plugin PROVED reached canonical, and it
            // lives in memory. A restarted IDE has none, so that path is unchanged
            // and still fail-closed. A live IDE whose buffer has moved past its
            // shadow is holding operator text the controller has never seen, and
            // adopting canonical over it is data loss, not recovery.
            // The live buffer, not the caller's captured cut: this decides whether
            // operator text is about to be destroyed, so it must read what the
            // operator can currently see.
            val bufferText = editorBufferText(filePath) ?: initialEditorText
            val publishedShadow = shadows[filePath]
            if (retainedCanonicalWouldClobberOperatorTextUtil(publishedShadow, bufferText)) {
                log.warn(
                    "[crdt-replica] registration retained a controller canonical projection for " +
                        "${File(filePath).name} while the buffer holds unpublished operator text; " +
                        "keeping the buffer and forwarding it as a local delta. " +
                        "shadow_hash=${contentHash(publishedShadow!!)} " +
                        "buffer_hash=${contentHash(bufferText!!)} " +
                        "canonical_hash=${forwarder.canonicalContentHash ?: "unknown"}",
                )
                // Do NOT arm the retained-canonical suppression. The ordinary
                // local-delta path owns this text and must publish it.
                retainedCanonicalProjectionPaths.remove(filePath)
            } else {
                retainedCanonicalProjectionPaths.add(filePath)
                log.info(
                    "[crdt-replica] registration retained the controller canonical projection for " +
                        "${File(filePath).name}; canonical_hash=${forwarder.canonicalContentHash ?: "unknown"}",
                )
            }
        }
        if (bootstrapFromControllerCanonical) {
            nativeReloadResumeStates.remove(filePath)
        } else if (retainedResumeState != null) {
            nativeReloadResumeStates.remove(filePath, retainedResumeState)
        }
        if (
            expectedEditorTextAtSwap != null &&
            (editorBufferText(filePath) != expectedEditorTextAtSwap ||
                (!allowPendingLocalAtSwap && hasPendingLocal(filePath)))
        ) {
            // Registration can block while the operator keeps typing. Reject
            // the captured editor cut before swapping the endpoint; the
            // still-authoritative cached forwarder will drain those local
            // deltas and a later refresh can retry from a fresh cut.
            forwarder.deregister()
            return null
        }
        if (
            expectedEditorTextAtSwap != null &&
            (editorBufferText(filePath) != expectedEditorTextAtSwap ||
                (!allowPendingLocalAtSwap && hasPendingLocal(filePath)))
        ) {
            // Recheck at the actual swap so a raced local delta keeps the cached
            // forwarder authoritative until the serialized local-delta worker
            // has drained it.
            forwarder.deregister()
            return null
        }
        if (replaceCached && cached != null) {
            if (forwarders.replace(filePath, cached, forwarder)) {
                cached.deregister()
                log.info(
                    "[crdt-replica] atomically replaced cached forwarder for ${File(filePath).name}",
                )
                retainCanonicalProjectionAfterRegistration(filePath, initialEditorText, forwarder)
                return forwarder
            }
            // The manager worker is serialized, but preserve a concurrently
            // installed winner without sending a false document-close event.
            forwarder.deregister()
            return forwarders[filePath]
        }
        val existing = forwarders.putIfAbsent(filePath, forwarder)
        if (existing != null) {
            forwarder.deregister()
            return existing
        }
        log.info("[crdt-replica] attached ${File(filePath).name} as $identity")
        retainCanonicalProjectionAfterRegistration(filePath, initialEditorText, forwarder)
        return forwarder
    }

    /**
     * Registration is controller -> editor projection only. A pre-existing
     * editor buffer cannot become a whole-document recovery baseline because it
     * may predate retained queue additions or another editor's projected ops.
     */
    private fun retainCanonicalProjectionAfterRegistration(
        filePath: String,
        editorText: String?,
        forwarder: CrdtReplicaForwarder,
    ) {
        val canonical = forwarder.replicaText() ?: return
        shadows[filePath] = canonical
        if (editorText == null) return
        if (editorText == canonical) {
            // Registration proves which canonical generation the replacement
            // replica opened, but not what the IDE buffer currently displays.
            // Publish the exact post-swap view so the controller can discharge
            // this generation's delivery receipt.
            if (!forwarder.projectVisibleState(canonical)) {
                requestRemoteDrain(filePath, "registration-visible-projection-retry")
            }
            return
        }
        retainedCanonicalProjectionPaths.add(filePath)
        log.info(
            "[crdt-replica] projecting controller bootstrap into ${File(filePath).name}; " +
                "editor_hash=${contentHash(editorText)} canonical_hash=${contentHash(canonical)} " +
                "driver=lazy-controller-canonical-projection",
        )
        queueRemoteTextApply(
            filePath = filePath,
            expectedText = editorText,
            converged = canonical,
            forwarder = forwarder,
            updates = emptyList(),
        )
    }

    private fun refreshReplicaAfterTransportLoss(
        filePath: String,
        staleForwarder: CrdtReplicaForwarder,
        editorText: String,
        reason: String,
    ) {
        if (forwarders[filePath] !== staleForwarder) return
        val replacement = forwarderFor(
            filePath = filePath,
            initialEditorText = editorText,
            bypassRegisterBackoff = false,
            replaceCached = true,
            expectedEditorTextAtSwap = editorText,
        )
        if (replacement != null && replacement !== staleForwarder) {
            log.info(
                "[crdt-replica] controller transport recovered for ${File(filePath).name}; " +
                    "reason=$reason",
            )
            requestRemoteDrain(filePath, "controller-transport-reregistered")
            // `#ctrlkillreregister` Tier 3: this file recovered by noticing its own
            // loss, but a controller that lost its hub stranded every registration
            // this editor holds — including documents nothing is currently draining,
            // which would otherwise wait for an operator to touch them. Ask once,
            // about ourselves, and repair the rest now.
            pullMissingReplicas(project, "controller-transport-recovered")
        } else {
            log.debug(
                "[crdt-replica] controller transport unavailable for ${File(filePath).name}; " +
                    "reason=$reason",
            )
        }
    }

    private fun shouldAttemptRegister(filePath: String): Boolean {
        val projection = registerRetryProjections[filePath] ?: return true
        val now = System.currentTimeMillis()
        if (replicaRegistrationAttemptDueUtil(projection, now)) return true
        log.debug(
            "[crdt-replica] register skipped for $filePath; " +
                "retry_after_ms=${projection.retryAfterMs - now}",
        )
        return false
    }

    private fun recordRegisterFailure(filePath: String, reason: String = "controller-register") {
        val now = System.currentTimeMillis()
        val projection =
            registerRetryProjections.compute(filePath) { _, previous ->
                nextReplicaRegistrationRetryProjection(previous, now)
            }!!
        registerFailureCounts[filePath] = projection.failureCount
        registerRetryAfterMs[filePath] = projection.retryAfterMs
        log.warn(
            "[crdt-replica] register failed for ${File(filePath).name}; " +
                "reason=$reason failure_count=${projection.failureCount} " +
                "retry_backoff_ms=${projection.backoffMs}",
        )
        scheduleRegisterRetry(filePath, projection.backoffMs)
    }

    private fun clearRegisterFailure(filePath: String) {
        attachFailureReasons.remove(filePath)
        registerFailureCounts.remove(filePath)
        registerRetryAfterMs.remove(filePath)
        registerRetryProjections.remove(filePath)
        registerRetryTasks.remove(filePath)?.cancel(false)
    }

    /**
     * Retain desired open-document registration across controller/native startup
     * races. The VFS event is only the observation edge; this keyed retry owns
     * convergence until the open document has an attached replica.
     */
    private fun scheduleRegisterRetry(filePath: String, delayMs: Long) {
        if (disposed.get()) return
        registerRetryTasks.compute(filePath) { _, existing ->
            if (existing != null && !existing.isDone) {
                existing
            } else {
                executor.schedule(
                    {
                        registerRetryTasks.remove(filePath)
                        if (
                            disposed.get() ||
                            project.isDisposed ||
                            forwarders[filePath]?.attached == true
                        ) {
                            return@schedule
                        }
                        runOnEdtNonBlocking {
                            if (disposed.get() || project.isDisposed) return@runOnEdtNonBlocking
                            val file =
                                FileEditorManager.getInstance(project).openFiles
                                    .firstOrNull { it.path == filePath }
                                    ?: return@runOnEdtNonBlocking
                            val document =
                                FileDocumentManager.getInstance().getDocument(file)
                                    ?: return@runOnEdtNonBlocking
                            ensureOpenDocumentReplica(file.path, document, await = false)
                        }
                    },
                    delayMs,
                    TimeUnit.MILLISECONDS,
                )
            }
        }
    }

    private fun markLocalPending(filePath: String) {
        pendingLocalEdits.computeIfAbsent(filePath) { AtomicInteger(0) }.incrementAndGet()
    }

    private fun clearLocalPending(filePath: String) {
        val counter = pendingLocalEdits[filePath] ?: return
        if (counter.decrementAndGet() <= 0) {
            pendingLocalEdits.remove(filePath, counter)
        }
    }

    private fun hasPendingLocal(filePath: String): Boolean =
        (pendingLocalEdits[filePath]?.get() ?: 0) > 0

    private fun resolveProjectRoot(filePath: String): String? {
        var dir: File? = File(filePath).absoluteFile.parentFile
        while (dir != null) {
            if (File(dir, ".agent-doc").isDirectory) return dir.absolutePath
            dir = dir.parentFile
        }
        return project.basePath?.takeIf { File(it, ".agent-doc").isDirectory }
    }

    private fun logSlow(
        operation: String,
        filePath: String,
        startedNanos: Long,
        warnMs: Long = CRDT_WORKER_WARN_MS,
        details: String = "",
    ) {
        val elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos)
        val suffix = if (details.isBlank()) "" else " $details"
        val name = if (filePath == "(none)") filePath else File(filePath).name
        val message = "[crdt-perf] $operation file=$name elapsed_ms=$elapsedMs thread=${Thread.currentThread().name}$suffix"
        if (elapsedMs >= warnMs) {
            log.warn(message)
        } else {
            log.debug(message)
        }
    }

    companion object {
        private val instances = ConcurrentHashMap<Project, CrdtReplicaManager>()
        private val applyingAgentMutations = ConcurrentHashMap.newKeySet<String>()
        private val nonOperatorMutationEpochs = ConcurrentHashMap<String, AtomicLong>()
        private val nativeReloadResumeStates =
            ConcurrentHashMap<String, ReplicaResumeState>()

        fun getInstance(project: Project): CrdtReplicaManager =
            instances.getOrPut(project) {
                CrdtReplicaManager(project).also { it.start() }
            }

        fun disposeProject(project: Project) {
            instances.remove(project)?.dispose()
        }

        internal fun quiesceAllForNativeReload(): Pair<List<Project>, Boolean> {
            val managers = instances.entries.mapNotNull { (project, manager) ->
                if (instances.remove(project, manager)) project to manager else null
            }
            // Stop the control worker and every document lane first, but keep
            // native replicas open long enough to copy their encoded state into
            // JVM-owned memory.
            managers.forEach { (_, manager) ->
                manager.disposed.set(true)
                manager.executor.shutdownNow()
                manager.documentWorkers.shutdownNow()
            }
            val quiesced = managers.map { (_, manager) ->
                manager.awaitWorkerTermination(NATIVE_RELOAD_WORKER_TIMEOUT_MS)
            }.all { it }
            if (quiesced) {
                managers.forEach { (_, manager) ->
                    manager.forwarders.forEach { (filePath, forwarder) ->
                        forwarder.captureResumeState()?.let { state ->
                            nativeReloadResumeStates[filePath] = state
                        }
                    }
                }
            }
            managers.forEach { (_, manager) -> manager.dispose() }
            return managers.map { it.first } to quiesced
        }

        internal fun restartAfterNativeReload(projects: List<Project>) {
            projects
                .filterNot { it.isDisposed }
                .forEach { project ->
                    runOnEdtNonBlocking {
                        if (project.isDisposed) return@runOnEdtNonBlocking
                        getInstance(project)
                        forceRefreshOpenDocumentReplicas(project, "native-generation-handoff")
                    }
                }
        }

        fun requestRemoteDrain(project: Project, filePath: String? = null, reason: String = "event") {
            instances[project]?.requestRemoteDrain(filePath, reason)
        }

        fun requestUrgentRemoteDrain(project: Project, filePath: String, reason: String) {
            instances[project]?.requestUrgentRemoteDrain(filePath, reason)
        }

        /**
         * `#ctrlkillreregister` Tier 3 — ask the controller which of this editor's
         * registrations it holds no replica for, and rebuild exactly those.
         *
         * Replaces the blind "force-refresh every open markdown document" sweep at
         * startup and after a transport recovery. The sweep was both too much and too
         * little: it drops and rebuilds healthy CRDT baselines (the lossiest thing
         * this manager can do), while still missing a registration whose document is
         * not currently open in a tab.
         *
         * `held` is deliberately EMPTY. This editor's own forwarder map is the wrong
         * evidence: after a controller kill the forwarders still look live here, and
         * passing them as held would suppress precisely the documents that need
         * repair. The controller subtracts what its process-local hub can actually
         * serve, which is the only fact that separates "registered" from "registered
         * and backed". Re-register storms are already bounded by
         * [beginProjectionRecoveryReregister]'s coalescing window.
         *
         * A null answer means the question could not be asked (old cdylib, controller
         * unreachable). Only then does this fall back to the compatibility sweep, so
         * an editor is never left stranded by the pull's own unavailability.
         */
        fun pullMissingReplicas(project: Project, reason: String) {
            val manager = instances[project] ?: return
            val root = project.basePath ?: return
            if (!manager.beginPeerReplicaPull()) {
                manager.log.debug("[crdt-replica] coalesced peer replica pull; reason=$reason")
                return
            }
            ApplicationManager.getApplication().executeOnPooledThread {
                if (project.isDisposed) return@executeOnPooledThread
                val pid = ProcessHandle.current().pid()
                val json = PeerReplicaPull.missingRegistrationsJson(root, pid, emptyList())
                val paths = PeerReplicaPull.rebuildPaths(json, pid)
                if (paths == null) {
                    manager.log.info(
                        "[crdt-replica] peer replica pull unavailable; reason=$reason " +
                            "recovery=controller_tier1_fan_out",
                    )
                    forceRefreshOpenDocumentReplicas(project, "$reason-pull-unavailable")
                    return@executeOnPooledThread
                }
                if (paths.isEmpty()) {
                    manager.log.info("[crdt-replica] peer replica pull found nothing to rebuild; reason=$reason")
                    return@executeOnPooledThread
                }
                manager.log.info(
                    "[crdt-replica] peer replica pull names ${paths.size} stranded registration(s); reason=$reason",
                )
                paths.forEach { filePath ->
                    forceRefreshOpenDocumentReplica(project, filePath, "$reason-peer-replica-pull")
                }
            }
        }

        /**
         * Observe every currently open markdown document without replacing a
         * healthy replica. Missing registrations retain their own bounded retry.
         */
        fun ensureOpenDocumentReplicas(project: Project, reason: String) {
            runOnEdtNonBlocking {
                if (project.isDisposed) return@runOnEdtNonBlocking
                val manager = instances[project] ?: return@runOnEdtNonBlocking
                val fileDocumentManager = FileDocumentManager.getInstance()
                FileEditorManager.getInstance(project).openFiles
                    .asSequence()
                    .filter { it.name.endsWith(".md") }
                    .forEach { file ->
                        val document = fileDocumentManager.getDocument(file) ?: return@forEach
                        manager.log.debug(
                            "[crdt-replica] ensuring open-document registration for ${file.name}; reason=$reason",
                        )
                        manager.ensureOpenDocumentReplica(
                            file.path,
                            document,
                            await = false,
                            forceRefresh = false,
                        )
                    }
            }
        }

        fun ensureOpenDocumentReplica(project: Project, filePath: String, reason: String) {
            runOnEdtNonBlocking {
                if (project.isDisposed) return@runOnEdtNonBlocking
                val manager = instances[project] ?: return@runOnEdtNonBlocking
                val file =
                    FileEditorManager.getInstance(project).openFiles
                        .firstOrNull { it.path == filePath }
                        ?: return@runOnEdtNonBlocking
                val document =
                    FileDocumentManager.getInstance().getDocument(file)
                        ?: return@runOnEdtNonBlocking
                manager.log.debug(
                    "[crdt-replica] ensuring open-document registration for ${file.name}; reason=$reason",
                )
                manager.ensureOpenDocumentReplica(
                    file.path,
                    document,
                    await = false,
                    forceRefresh = false,
                )
            }
        }

        fun rebindOpenDocumentPath(
            project: Project,
            oldPath: String,
            newPath: String,
        ): Boolean {
            if (project.isDisposed) return false
            val manager = instances[project] ?: return false
            val documentRef = AtomicReference<Document?>()
            val textRef = AtomicReference<String?>()
            val capture = {
                val file =
                    FileEditorManager.getInstance(project).openFiles
                        .firstOrNull { it.path == newPath }
                        ?: LocalFileSystem.getInstance().findFileByPath(newPath)
                val document = file?.let { FileDocumentManager.getInstance().getDocument(it) }
                documentRef.set(document)
                textRef.set(document?.text)
            }
            if (SwingUtilities.isEventDispatchThread()) {
                capture()
            } else {
                ApplicationManager.getApplication().invokeAndWait { capture() }
            }
            val document = documentRef.get() ?: return false
            val editorText = textRef.get() ?: return false
            return manager.rebindOpenDocumentPath(oldPath, newPath, document, editorText)
        }

        fun forceRefreshOpenDocumentReplicas(project: Project, reason: String) {
            runOnEdtNonBlocking {
                if (project.isDisposed) return@runOnEdtNonBlocking
                val manager = instances[project] ?: return@runOnEdtNonBlocking
                val fileDocumentManager = FileDocumentManager.getInstance()
                val openDocuments =
                    FileEditorManager.getInstance(project).openFiles
                        .asSequence()
                        .filter { it.name.endsWith(".md") }
                        .mapNotNull { file ->
                            fileDocumentManager.getDocument(file)?.let { document ->
                                Triple(file.path, file.name, document)
                            }
                        }
                        .toList()
                ApplicationManager.getApplication().executeOnPooledThread {
                    if (project.isDisposed) return@executeOnPooledThread
                    openDocuments.forEach { (filePath, fileName, document) ->
                        manager.log.info("[crdt-replica] forcing open-document re-register for $fileName reason=$reason")
                        manager.ensureOpenDocumentReplica(
                            filePath,
                            document,
                            await = false,
                            forceRefresh = true,
                        )
                    }
                }
            }
        }

        fun <T> withAgentAppliedEditorMutation(filePath: String, block: () -> T): T {
            advanceNonOperatorMutationEpoch(filePath)
            applyingAgentMutations.add(filePath)
            return try {
                block()
            } finally {
                applyingAgentMutations.remove(filePath)
            }
        }

        /**
         * Close the Rust-owned operator-op epoch before an editor projection is
         * handed to the EDT. The native generation proxy deliberately rejects
         * calls from the event-dispatch thread, so this durable transition and
         * the local editor mutation are a two-stage handoff.
         *
         * This operation is idempotent. Returning false keeps the projection
         * retained for retry instead of applying text without its causal fence.
         */
        fun prepareNonOperatorEditorMutationOnWorker(filePath: String): Boolean {
            check(!javax.swing.SwingUtilities.isEventDispatchThread()) {
                "native op-capture fencing must complete before dispatching an editor mutation to the EDT"
            }
            val lib = AgentDocLib.get() ?: return true
            return lib.agent_doc_clear_editor_op_epoch(filePath) == 1
        }

        fun forceRefreshOpenDocumentReplica(project: Project, filePath: String, reason: String) {
            runOnEdtNonBlocking {
                if (project.isDisposed) return@runOnEdtNonBlocking
                val manager = instances[project] ?: return@runOnEdtNonBlocking
                val file =
                    LocalFileSystem.getInstance().findFileByPath(filePath)
                        ?: return@runOnEdtNonBlocking
                val document =
                    FileDocumentManager.getInstance().getDocument(file)
                        ?: return@runOnEdtNonBlocking
                val resolvedFilePath = file.path
                val fileName = file.name
                ApplicationManager.getApplication().executeOnPooledThread {
                    if (project.isDisposed) return@executeOnPooledThread
                    if (!manager.beginProjectionRecoveryReregister(resolvedFilePath)) {
                        manager.log.info(
                            "[crdt-replica] coalesced projection-recovery re-register for $fileName reason=$reason",
                        )
                        manager.requestUrgentRemoteDrain(
                            resolvedFilePath,
                            "projection-recovery-reregister-coalesced",
                        )
                        return@executeOnPooledThread
                    }
                    manager.log.info(
                        "[crdt-replica] forcing projection-recovery re-register for $fileName reason=$reason",
                    )
                    manager.ensureOpenDocumentReplica(
                        resolvedFilePath,
                        document,
                        await = false,
                        forceRefresh = true,
                    )
                }
            }
        }

        private fun runOnEdtNonBlocking(block: () -> Unit) {
            if (javax.swing.SwingUtilities.isEventDispatchThread()) {
                block()
            } else {
                ApplicationManager.getApplication().invokeLater(block)
            }
        }

        fun ensureReplicaForOpenDocument(
            filePath: String,
            document: Document,
            editorText: String? = null,
            await: Boolean = false,
            forceRefresh: Boolean = false,
        ): Boolean {
            val manager = managerForFilePath(filePath)
                ?: return false
            return manager.ensureOpenDocumentReplica(filePath, document, editorText, await, forceRefresh)
        }

        fun publishClosingDocumentCut(filePath: String, document: Document): Boolean {
            val manager = managerForFilePath(filePath) ?: return false
            return manager.publishClosingDocumentCut(filePath, document)
        }

        fun isApplyingRemote(filePath: String): Boolean =
            instances.values.any { it.applyingRemote.contains(filePath) }

        private fun isReloadingFileContent(filePath: String): Boolean =
            instances.values.any { it.fileContentReloadingPaths.contains(filePath) }

        /**
         * #ensurereregister: true when some open project already holds a CRDT
         * replica (forwarder) for [filePath].
         *
         * This guard prevents needless endpoint replacement while a live
         * consumer is already projecting the controller bootstrap. When there
         * is no forwarder, registration reconstructs that downstream consumer
         * without promoting its editor buffer to whole-document authority.
         */
        fun hasOpenDocumentReplica(filePath: String): Boolean =
            instances.values.any { it.forwarders.containsKey(filePath) }

        /**
         * Why the most recent replica attach for [filePath] was refused, or `null`
         * when the last attempt succeeded or none has run (`#replicarefusalreason`).
         *
         * Operator-facing commands gate on `ensureReplicaForOpenDocument` returning
         * false. Reporting only that fact sends the operator to the controller,
         * which is typically healthy — the actual cause lives here.
         */
        fun lastAttachFailureReason(filePath: String): String? =
            instances.values.firstNotNullOfOrNull { it.attachFailureReasons[filePath] }

        /**
         * A one-line operator remedy for [reason], or `null` when there is no
         * specific advice beyond the reason itself.
         *
         * `detached_authority` on a *registration* means the controller saw no
         * live editor PID for the request. Registration is what establishes editor
         * authority, so the controller does not refuse it for lacking authority —
         * it refuses when the request carries no live PID, which is what a plugin
         * generation loaded before that field existed sends. The IDE keeps the
         * generation it started with, so the installed jar can read current on disk
         * while the loaded one is not: only an IDE restart replaces it. This is the
         * same conclusion the controller reaches in its refusal-storm advisory.
         */
        fun attachFailureRemedy(reason: String): String? = when {
            reason.contains("detached_authority") ->
                "Restart the IDE. The controller saw no live editor PID on the registration, which is " +
                    "what a plugin generation loaded before the current install sends. Reloading the " +
                    "native library or recycling the controller cannot replace a loaded plugin generation."
            reason.contains("socket_unavailable") ->
                "No project controller is listening for that document's project root; start or recycle it."
            reason.contains("not_agent_doc_document") ->
                "That file is not an agent-doc session document (no agent_doc_* frontmatter, no <!-- agent: markers)."
            else -> null
        }

        fun isApplyingNonOperatorMutation(filePath: String): Boolean =
            applyingAgentMutations.contains(filePath) ||
                isApplyingRemote(filePath) ||
                isReloadingFileContent(filePath)

        fun isOperatorDocumentEvent(filePath: String, event: DocumentEvent): Boolean =
            isOperatorDocumentEventUtil(
                nonOperatorMutation = isApplyingNonOperatorMutation(filePath),
                wholeTextReplaced = event.isWholeTextReplaced,
                documentUnsaved =
                    FileDocumentManager.getInstance().isDocumentUnsaved(event.document),
            )

        private fun managerForFilePath(filePath: String): CrdtReplicaManager? =
            instances.values
                .filter { it.ownsFilePath(filePath) }
                .maxWithOrNull(
                    compareBy<CrdtReplicaManager> { it.project.basePath?.length ?: 0 }
                        .thenBy { it.project.basePath.orEmpty() },
                )

        private fun nonOperatorMutationEpoch(filePath: String): Long =
            nonOperatorMutationEpochs[filePath]?.get() ?: 0L

        private fun advanceNonOperatorMutationEpoch(filePath: String): Long {
            return nonOperatorMutationEpochs
                .computeIfAbsent(filePath) { AtomicLong(0L) }
                .incrementAndGet()
        }
    }

    /** Last refusal reason per document path; cleared on a successful register. */
    private val attachFailureReasons = java.util.concurrent.ConcurrentHashMap<String, String>()

    private fun ownsFilePath(filePath: String): Boolean {
        val base = project.basePath ?: return false
        return try {
            File(filePath).absoluteFile.toPath().startsWith(File(base).absoluteFile.toPath())
        } catch (_: Exception) {
            false
        }
    }
}

/**
 * Whether [text] is an agent-doc session document.
 *
 * Mirrors the controller's `is_agent_doc_document`: a session document carries
 * `agent_doc_*` frontmatter or an `<!-- agent:` component marker. A README,
 * CONTRIBUTING, plan note, or draft has neither, never attaches an editor
 * authority, and is refused terminally — so registering one is work that cannot
 * succeed. Kept as a plain content test, deliberately identical to the
 * controller's, so the two ends cannot disagree about what a session document is.
 */
internal fun isAgentDocDocumentTextUtil(text: String): Boolean =
    text.contains("agent_doc_session") ||
        text.contains("agent_doc_format") ||
        text.contains("agent_doc_write") ||
        text.contains("<!-- agent:")

/**
 * Whether adopting a retained controller canonical projection would destroy
 * operator text (`#retainedprojectionclobbersoperatortext`).
 *
 * [publishedShadow] is the last text this plugin PROVED reached canonical, held
 * in memory; [bufferText] is what the operator can currently see. The two cases
 * that reach the retained-canonical branch are opposite and must not be
 * conflated:
 *
 *  - a RESTARTED IDE has no shadow (the map does not survive a restart), so its
 *    buffer is a stale reconstruction and adopting canonical is the recovery;
 *  - a LIVE IDE whose buffer has moved past its shadow is holding text the
 *    controller has never seen — because registration was being refused — and
 *    adopting canonical over it is data loss.
 *
 * Unknown buffer text is not divergence: without both facts this returns false
 * and the historical fail-closed adoption stands.
 */
internal fun retainedCanonicalWouldClobberOperatorTextUtil(
    publishedShadow: String?,
    bufferText: String?,
): Boolean = publishedShadow != null && bufferText != null && publishedShadow != bufferText

internal fun shouldApplyRemoteCrdtUpdateUtil(update: ReplicaRemoteUpdate, clientId: Long): Boolean =
    update.origin != clientId

internal fun isOperatorDocumentEventUtil(
    nonOperatorMutation: Boolean,
    wholeTextReplaced: Boolean,
    documentUnsaved: Boolean,
): Boolean = !nonOperatorMutation && !wholeTextReplaced && documentUnsaved

internal fun remoteCrdtApplyStillCurrentUtil(
    expectedText: String,
    currentText: String,
    targetText: String,
): Boolean =
    currentText == expectedText || currentText == targetText

internal fun remoteCrdtDiskCanPersistUtil(
    expectedText: String,
    targetText: String,
    diskText: String?,
): Boolean = diskText == expectedText || diskText == targetText

internal fun remoteCrdtReplaceStillCurrentUtil(
    expectedText: String,
    currentText: String,
    replicaText: String?,
): Boolean =
    currentText == expectedText && replicaText == expectedText
