IJAI-69 replace AIR session RpcFlow with RemoteTopic

Publish surface state as revisioned project-topic deltas repaired by unary snapshots and route outline refresh through an application topic, removing per-editor subscriptions and startup activity.

Keep terminal runtimes project-owned when editor tabs detach. Fence retained-runtime transfer, reconciliation, and lifecycle reports by authority and generation, flush final working-directory state before close, and isolate cold reconnect bootstrap in an optional backend-split module.

Update the owning specs and architecture model for detached status reporting and the new transport.

(cherry picked from commit 0bff1559ff53f7415f46518842460440cadd3c00)

GitOrigin-RevId: 770df72b0d2b8539bf4797cc88444b07261ab32b

plugins/terminal/frontend/src/com/intellij/terminal/frontend/session/StandardTerminalSessionStart.kt

@@ -0,0 +1,92 @@
+// Copyright 2000-2026 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.terminal.frontend.session
+
+import com.intellij.openapi.diagnostic.logger
+import com.intellij.openapi.project.Project
+import com.intellij.platform.util.coroutines.childScope
+import com.jediterm.core.util.TermSize
+import kotlinx.coroutines.CoroutineScope
+import org.jetbrains.annotations.ApiStatus
+import org.jetbrains.plugins.terminal.JBTerminalSystemSettingsProvider
+import org.jetbrains.plugins.terminal.ShellStartupOptions
+import org.jetbrains.plugins.terminal.ShellTerminalWidget
+import org.jetbrains.plugins.terminal.session.impl.TerminalSession
+import org.jetbrains.plugins.terminal.session.impl.TerminalStartupOptionsImpl
+import org.jetbrains.plugins.terminal.session.TerminalStartupOptions
+
+/**
+ * Starts the standard reconnectable terminal session without assigning its ownership to a project service.
+ *
+ * [project] is used synchronously to resolve the startup connector and EEL environment. [statisticsProject]
+ * controls optional project-scoped statistics listeners and may be null for application-owned runtimes.
+ */
+@ApiStatus.Internal
+fun startStandardTerminalSession(
+  project: Project,
+  options: ShellStartupOptions,
+  scope: CoroutineScope,
+  statisticsProject: Project? = project,
+): StandardTerminalSessionStartResult {
+  val termSize = options.initialTermSize ?: run {
+    LOG.warn("No initial terminal size provided, using default 80x24. $options")
+    TermSize(80, 24)
+  }
+  val optionsWithSize = options.builder().initialTermSize(termSize).build()
+  val (ttyConnector, configuredOptions) = startTerminalProcess(project, optionsWithSize)
+  val observableTtyConnector = ObservableTtyConnector(ttyConnector)
+
+  val jediTermScope = scope.childScope("JediTerm session")
+  val delegate = createTerminalSession(
+    project = statisticsProject,
+    ttyConnector = observableTtyConnector,
+    options = configuredOptions,
+    settings = JBTerminalSystemSettingsProvider(),
+    coroutineScope = jediTermScope,
+  )
+  val startupOptions = TerminalStartupOptionsImpl(
+    shellCommand = configuredOptions.shellCommand!!,
+    workingDirectory = configuredOptions.workingDirectory!!,
+    envVariables = configuredOptions.envVariables,
+    processType = configuredOptions.processType,
+    pid = getLocalPid(ttyConnector),
+  )
+  return StandardTerminalSessionStartResult(
+    configuredOptions = configuredOptions,
+    session = createStandardStateAwareTerminalSession(delegate, startupOptions, scope),
+    ttyConnector = observableTtyConnector,
+  )
+}
+
+@ApiStatus.Internal
+data class StandardTerminalSessionStartResult(
+  val configuredOptions: ShellStartupOptions,
+  val session: TerminalSession,
+  val ttyConnector: ObservableTtyConnector,
+)
+
+@ApiStatus.Internal
+fun createStandardStateAwareTerminalSession(
+  delegate: TerminalSession,
+  startupOptions: TerminalStartupOptions,
+  scope: CoroutineScope,
+): TerminalSession = StateAwareTerminalSession(delegate, startupOptions, scope)
+
+private fun getLocalPid(ttyConnector: com.jediterm.terminal.TtyConnector): Long? {
+  val processTtyConnector = ShellTerminalWidget.getProcessTtyConnector(ttyConnector) ?: run {
+    LOG.warn("Failed to get ProcessTtyConnector from $ttyConnector.")
+    return null
+  }
+  return try {
+    processTtyConnector.process.pid()
+  }
+  catch (_: UnsupportedOperationException) {
+    // IjentChildPtyProcessAdapter does not expose a real PID for a remote process.
+    null
+  }
+  catch (t: Throwable) {
+    LOG.warn("Failed to get pid of the started process: $ttyConnector", t)
+    null
+  }
+}
+
+private val LOG = logger<StandardTerminalSessionStartResult>()

plugins/terminal/frontend/src/com/intellij/terminal/frontend/session/TerminalSessionStart.kt

@@ -50,7 +50,7 @@ fun startTerminalProcess(
 @ApiStatus.Internal
 @OptIn(AwaitCancellationAndInvoke::class)
 fun createTerminalSession(
-  project: Project,
+  project: Project?,
   ttyConnector: TtyConnector,
   options: ShellStartupOptions,
   settings: JBTerminalSystemSettingsProviderBase,
@@ -63,7 +63,9 @@ fun createTerminalSession(
 
   val outputScope = coroutineScope.childScope("Terminal output forwarding")
   val shellIntegrationController = TerminalShellIntegrationController(services.controller)
-  shellIntegrationController.addListener(TerminalShellIntegrationStatisticsListener(project))
+  if (project != null) {
+    shellIntegrationController.addListener(TerminalShellIntegrationStatisticsListener(project))
+  }
   val outputFlow = createTerminalOutputFlow(
     services,
     shellIntegrationController,
@@ -154,4 +156,4 @@ internal class JediTermServices(
   val startupOptions: ShellStartupOptions,
 )
 
-private val LOG = fileLogger()
+private val LOG = fileLogger()

plugins/terminal/frontend/src/com/intellij/terminal/frontend/session/TerminalSessionsManager.kt

@@ -2,15 +2,12 @@ package com.intellij.terminal.frontend.session
 
 import com.intellij.openapi.components.Service
 import com.intellij.openapi.components.service
-import com.intellij.openapi.diagnostic.logger
 import com.intellij.openapi.project.Project
 import com.intellij.openapi.project.ProjectManager
 import com.intellij.openapi.project.ProjectManagerListener
 import com.intellij.platform.ide.progress.runWithModalProgressBlocking
-import com.intellij.platform.util.coroutines.childScope
 import com.intellij.util.AwaitCancellationAndInvoke
 import com.intellij.util.awaitCancellationAndInvoke
-import com.jediterm.core.util.TermSize
 import com.jediterm.terminal.TtyConnector
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.async
@@ -18,13 +15,10 @@ import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.job
 import org.jetbrains.annotations.ApiStatus
-import org.jetbrains.plugins.terminal.JBTerminalSystemSettingsProvider
 import org.jetbrains.plugins.terminal.ShellStartupOptions
-import org.jetbrains.plugins.terminal.ShellTerminalWidget
 import org.jetbrains.plugins.terminal.TerminalBundle
 import org.jetbrains.plugins.terminal.block.reworked.session.rpc.TerminalSessionId
 import org.jetbrains.plugins.terminal.session.impl.TerminalSession
-import org.jetbrains.plugins.terminal.session.impl.TerminalStartupOptionsImpl
 import org.jetbrains.plugins.terminal.util.CONNECTOR_CLOSING_TIMEOUT
 import org.jetbrains.plugins.terminal.util.closeAndWaitFor
 import java.util.Collections
@@ -60,38 +54,14 @@ class TerminalSessionsManager(private val project: Project) {
    * then the [scope] will be canceled as well.
    */
   fun startSession(options: ShellStartupOptions, scope: CoroutineScope): TerminalSessionStartResult {
-    val termSize = options.initialTermSize ?: run {
-      LOG.warn("No initial terminal size provided, using default 80x24. $options")
-      TermSize(80, 24)
-    }
-    val optionsWithSize = options.builder().initialTermSize(termSize).build()
-
-    val (ttyConnector, configuredOptions) = startTerminalProcess(project, optionsWithSize)
-    storeConnector(ttyConnector, scope)
-    val observableTtyConnector = ObservableTtyConnector(ttyConnector)
-
-    // Create the JediTerm session scope as the child of the main scope.
-    // If the original session terminates on its own, then StateAwareTerminalSession will handle the TerminalSessionTerminatedEvent
-    // and cancel the main scope.
-    val jediTermScope = scope.childScope("JediTerm session")
-    val jediTermSession = createTerminalSession(project, observableTtyConnector, configuredOptions, JBTerminalSystemSettingsProvider(), jediTermScope)
-
-    // It should be guaranteed that the shell command and working directory are not null.
-    val options = TerminalStartupOptionsImpl(
-      shellCommand = configuredOptions.shellCommand!!,
-      workingDirectory = configuredOptions.workingDirectory!!,
-      envVariables = configuredOptions.envVariables,
-      processType = configuredOptions.processType,
-      pid = getLocalPid(ttyConnector),
-    )
-    val stateAwareSession = StateAwareTerminalSession(jediTermSession, options, scope)
-
-    val sessionId = storeSession(stateAwareSession, scope)
+    val started = startStandardTerminalSession(project = project, options = options, scope = scope)
+    storeConnector(started.ttyConnector, scope)
+    val sessionId = storeSession(started.session, scope)
 
     return TerminalSessionStartResult(
-      configuredOptions,
+      started.configuredOptions,
       sessionId,
-      observableTtyConnector
+      started.ttyConnector,
     )
   }
 
@@ -101,7 +71,7 @@ class TerminalSessionsManager(private val project: Project) {
 
   private fun storeSession(session: TerminalSession, scope: CoroutineScope): TerminalSessionId {
     val sessionId = TerminalSessionId(sessionIdCounter.getAndIncrement())
-    sessionsMap.put(sessionId, session)
+    sessionsMap[sessionId] = session
     scope.awaitCancellationAndInvoke {
       sessionsMap.remove(sessionId)
     }
@@ -115,28 +85,6 @@ class TerminalSessionsManager(private val project: Project) {
     }
   }
 
-  /**
-   * Returns a process ID if this process is running in the same machine as the IDE.
-   */
-  private fun getLocalPid(ttyConnector: TtyConnector): Long? {
-    val processTtyConnector = ShellTerminalWidget.getProcessTtyConnector(ttyConnector) ?: run {
-      LOG.warn("Failed to get ProcessTtyConnector from $ttyConnector.")
-      return null
-    }
-
-    return try {
-      processTtyConnector.process.pid()
-    }
-    catch (_: UnsupportedOperationException) {
-      // IjentChildPtyProcessAdapter doesn't return a real PID of a remote process
-      null
-    }
-    catch (ex: Exception) {
-      LOG.warn("Failed to get pid of the started process: $ttyConnector", ex)
-      null
-    }
-  }
-
   private suspend fun closeTerminalProcesses() = coroutineScope {
     // Perform closing activities asynchronously for every process
     // because closing sequentially may take much more time
@@ -149,8 +97,6 @@ class TerminalSessionsManager(private val project: Project) {
   companion object {
     private val sessionIdCounter = AtomicInteger(0)
 
-    private val LOG = logger<TerminalSessionsManager>()
-
     @JvmStatic
     fun getInstance(project: Project): TerminalSessionsManager = project.service()
   }
@@ -161,4 +107,4 @@ data class TerminalSessionStartResult(
   val configuredOptions: ShellStartupOptions,
   val sessionId: TerminalSessionId,
   val ttyConnector: ObservableTtyConnector,
-)
+)

plugins/terminal/frontend/src/com/intellij/terminal/frontend/session/TransferableTerminalSession.kt

@@ -0,0 +1,417 @@
+// Copyright 2000-2026 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.terminal.frontend.session
+
+import com.intellij.openapi.application.EDT
+import com.intellij.openapi.application.ModalityState
+import com.intellij.openapi.application.asContextElement
+import com.intellij.openapi.components.Service
+import com.intellij.openapi.components.service
+import com.intellij.openapi.project.Project
+import com.intellij.platform.util.coroutines.childScope
+import com.intellij.terminal.frontend.view.TerminalView
+import com.intellij.terminal.frontend.view.TerminalViewSessionState
+import com.intellij.terminal.frontend.view.impl.TerminalViewImpl
+import com.intellij.terminal.frontend.view.portForwarding.installPortForwarding
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.job
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import org.jetbrains.annotations.ApiStatus
+import org.jetbrains.annotations.TestOnly
+import org.jetbrains.plugins.terminal.JBTerminalSystemSettingsProvider
+import org.jetbrains.plugins.terminal.ShellStartupOptions
+import org.jetbrains.plugins.terminal.fus.TerminalStartupFusInfo
+import org.jetbrains.plugins.terminal.session.impl.TerminalSession
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * A terminal process/session whose lifetime is independent of any one project presentation.
+ *
+ * The application scope is only the fallback lifetime parent. Callers own each returned handle and must close it.
+ * Project-bound [TerminalView] instances can be replaced while the exact [TerminalSession] and process keep running.
+ */
+@ApiStatus.Internal
+interface TransferableTerminalSession : AutoCloseable {
+  val runtimeId: Long
+  val coroutineScope: CoroutineScope
+  val sessionState: StateFlow<TerminalViewSessionState>
+  val view: TerminalView
+
+  /** Replaces the project-specific presentation without restarting the process/session. Must be called on EDT. */
+  fun bind(project: Project, sourceNavigationProjectPath: String? = null): TerminalView
+
+  /** Creates and connects a destination presentation without replacing the current binding. */
+  fun prepareBind(project: Project, sourceNavigationProjectPath: String? = null): PreparedTransferableTerminalBinding
+
+  suspend fun processId(): Long
+
+  override fun close()
+}
+
+@ApiStatus.Internal
+interface PreparedTransferableTerminalBinding : AutoCloseable {
+  val view: TerminalView
+
+  /** Returns null when the destination owner was canceled before the atomic swap. */
+  fun commit(): CommittedTransferableTerminalBinding?
+
+  override fun close()
+}
+
+@ApiStatus.Internal
+interface CommittedTransferableTerminalBinding {
+  /** Completes the swap and releases the previous project binding. */
+  fun finish()
+
+  /** Restores the previous project binding and releases the prepared destination. */
+  fun rollback()
+}
+
+/**
+ * Owns an explicitly closed runtime scope and at most one replaceable project binding.
+ *
+ * Binding creation is intentionally performed outside the lifecycle monitor. The only lock order is the lifecycle
+ * monitor by itself; callbacks and scope cancellation never run while it is held.
+ */
+@ApiStatus.Internal
+class TransferableTerminalLifetime(parentScope: CoroutineScope) : AutoCloseable {
+  class Binding<T : Any> internal constructor(
+    val value: T,
+    internal val scope: CoroutineScope,
+    internal val ownerCompletion: kotlinx.coroutines.DisposableHandle,
+  )
+
+  class PreparedBinding<T : Any> internal constructor(
+    private val lifetime: TransferableTerminalLifetime,
+    internal val binding: Binding<T>,
+  ) : AutoCloseable {
+    private val claimed = AtomicBoolean(false)
+
+    fun commit(): CommittedBinding<T>? {
+      check(claimed.compareAndSet(false, true)) { "Prepared terminal binding was already completed" }
+      return lifetime.commitPrepared(binding) ?: run {
+        lifetime.disposeBinding(binding)
+        null
+      }
+    }
+
+    override fun close() {
+      if (claimed.compareAndSet(false, true)) {
+        lifetime.disposeBinding(binding)
+      }
+    }
+  }
+
+  class CommittedBinding<T : Any> internal constructor(
+    private val lifetime: TransferableTerminalLifetime,
+    internal val binding: Binding<T>,
+    internal val previous: Binding<*>?,
+  ) {
+    private val completed = AtomicBoolean(false)
+
+    fun finish() {
+      if (completed.compareAndSet(false, true)) {
+        lifetime.finishCommit(this)
+      }
+    }
+
+    fun rollback() {
+      if (completed.compareAndSet(false, true)) {
+        lifetime.rollbackCommit(this)
+      }
+    }
+  }
+
+  val runtimeScope: CoroutineScope = parentScope.childScope("Transferable terminal runtime")
+
+  private var currentBinding: Binding<*>? = null
+  private var pendingCommit: CommittedBinding<*>? = null
+  private var closed: Boolean = false
+
+  fun <T : Any> prepareBinding(ownerScope: CoroutineScope, create: (CoroutineScope) -> T): PreparedBinding<T> {
+    val bindingScope = runtimeScope.childScope("Transferable terminal presentation")
+    val ownerCompletion = ownerScope.coroutineContext.job.invokeOnCompletion {
+      bindingScope.cancel()
+    }
+    val value = try {
+      create(bindingScope)
+    }
+    catch (t: Throwable) {
+      ownerCompletion.dispose()
+      bindingScope.cancel()
+      throw t
+    }
+    val binding = Binding(value, bindingScope, ownerCompletion)
+    if (isClosed() || !bindingScope.isActive) {
+      disposeBinding(binding)
+      error("Transferable terminal lifetime is already closed")
+    }
+    return PreparedBinding(this, binding)
+  }
+
+  fun <T : Any> replaceBinding(ownerScope: CoroutineScope, create: (CoroutineScope) -> T): Binding<T> {
+    val prepared = prepareBinding(ownerScope, create)
+    val committed = prepared.commit() ?: error("Transferable terminal binding owner is already canceled")
+    committed.finish()
+    return committed.binding
+  }
+
+  fun isClosed(): Boolean = synchronized(this) { closed }
+
+  override fun close() {
+    val bindings = synchronized(this) {
+      if (closed) return
+      closed = true
+      val pendingPrevious = pendingCommit?.previous
+      pendingCommit = null
+      listOfNotNull(currentBinding, pendingPrevious).distinct().also { currentBinding = null }
+    }
+    bindings.forEach(::disposeBinding)
+    runtimeScope.cancel()
+  }
+
+  private fun <T : Any> commitPrepared(binding: Binding<T>): CommittedBinding<T>? = synchronized(this) {
+    if (closed || !binding.scope.isActive || pendingCommit != null) {
+      return@synchronized null
+    }
+    val committed = CommittedBinding(this, binding, currentBinding)
+    currentBinding = binding
+    pendingCommit = committed
+    committed
+  }
+
+  private fun finishCommit(commit: CommittedBinding<*>) {
+    val previous = synchronized(this) {
+      if (pendingCommit !== commit) return
+      pendingCommit = null
+      commit.previous
+    }
+    previous?.let(::disposeBinding)
+  }
+
+  private fun rollbackCommit(commit: CommittedBinding<*>) {
+    val restorePrevious = synchronized(this) {
+      if (pendingCommit !== commit) return
+      pendingCommit = null
+      if (!closed && currentBinding === commit.binding) {
+        currentBinding = commit.previous
+        true
+      }
+      else {
+        false
+      }
+    }
+    disposeBinding(commit.binding)
+    if (!restorePrevious) {
+      commit.previous?.let(::disposeBinding)
+    }
+  }
+
+  private fun disposeBinding(binding: Binding<*>) {
+    binding.ownerCompletion.dispose()
+    binding.scope.cancel()
+  }
+}
+
+@Service(Service.Level.PROJECT)
+private class TransferableTerminalProjectBindingScope(val coroutineScope: CoroutineScope)
+
+@ApiStatus.Internal
+@Service(Service.Level.APP)
+class TransferableTerminalSessionFactory(private val coroutineScope: CoroutineScope) {
+  fun create(
+    project: Project,
+    options: ShellStartupOptions,
+    sourceNavigationProjectPath: String? = null,
+    startupFusInfo: TerminalStartupFusInfo? = null,
+  ): TransferableTerminalSession {
+    return TransferableTerminalSessionImpl(
+      parentScope = coroutineScope,
+      initialProject = project,
+      requestedOptions = options,
+      sourceNavigationProjectPath = sourceNavigationProjectPath,
+      startupFusInfo = startupFusInfo,
+    )
+  }
+
+  companion object {
+    @JvmStatic
+    fun getInstance(): TransferableTerminalSessionFactory = service()
+  }
+}
+
+@ApiStatus.Internal
+@TestOnly
+fun createTransferableTerminalSessionForTest(
+  parentScope: CoroutineScope,
+  initialProject: Project,
+  sessionStarter: (CoroutineScope) -> TerminalSession,
+  stateTransitionObserver: (TerminalViewSessionState) -> Unit,
+): TransferableTerminalSession {
+  return TransferableTerminalSessionImpl(
+    parentScope = parentScope,
+    initialProject = initialProject,
+    requestedOptions = ShellStartupOptions.Builder().build(),
+    sourceNavigationProjectPath = null,
+    startupFusInfo = null,
+    sessionStarter = { _, _, scope -> sessionStarter(scope) },
+    stateTransitionObserver = stateTransitionObserver,
+  )
+}
+
+private class TransferableTerminalSessionImpl(
+  parentScope: CoroutineScope,
+  initialProject: Project,
+  requestedOptions: ShellStartupOptions,
+  sourceNavigationProjectPath: String?,
+  private val startupFusInfo: TerminalStartupFusInfo?,
+  sessionStarter: (Project, ShellStartupOptions, CoroutineScope) -> TerminalSession = { project, options, scope ->
+    startStandardTerminalSession(
+      project = project,
+      options = options,
+      scope = scope,
+      statisticsProject = null,
+    ).session
+  },
+  private val stateTransitionObserver: (TerminalViewSessionState) -> Unit = {},
+) : TransferableTerminalSession {
+  private val lifetime = TransferableTerminalLifetime(parentScope)
+  override val coroutineScope: CoroutineScope
+    get() = lifetime.runtimeScope
+
+  override val runtimeId: Long = nextRuntimeId.getAndIncrement()
+
+  private val mutableSessionState = MutableStateFlow<TerminalViewSessionState>(TerminalViewSessionState.NotStarted)
+  override val sessionState: StateFlow<TerminalViewSessionState> = mutableSessionState.asStateFlow()
+  private val closed = AtomicBoolean(false)
+  private val stateLock = Any()
+  private val viewLock = Any()
+  private val session: TerminalSession
+  @Volatile
+  private lateinit var currentView: TerminalView
+
+  override val view: TerminalView
+    get() = currentView
+
+  init {
+    stateTransitionObserver(TerminalViewSessionState.NotStarted)
+    try {
+      session = sessionStarter(initialProject, requestedOptions, coroutineScope)
+      coroutineScope.coroutineContext.job.invokeOnCompletion {
+        transitionToTerminated()
+      }
+      check(transitionToRunning()) { "Transferable terminal runtime terminated during startup" }
+      bind(initialProject, sourceNavigationProjectPath)
+      check(sessionState.value == TerminalViewSessionState.Running) { "Transferable terminal runtime terminated during startup" }
+    }
+    catch (t: Throwable) {
+      lifetime.close()
+      transitionToTerminated()
+      throw t
+    }
+  }
+
+  private fun transitionToRunning(): Boolean = synchronized(stateLock) {
+    if (mutableSessionState.value != TerminalViewSessionState.NotStarted) return@synchronized false
+    mutableSessionState.value = TerminalViewSessionState.Running
+    stateTransitionObserver(TerminalViewSessionState.Running)
+    true
+  }
+
+  private fun transitionToTerminated() = synchronized(stateLock) {
+    if (mutableSessionState.value == TerminalViewSessionState.Terminated) return@synchronized
+    mutableSessionState.value = TerminalViewSessionState.Terminated
+    stateTransitionObserver(TerminalViewSessionState.Terminated)
+  }
+
+  override fun bind(project: Project, sourceNavigationProjectPath: String?): TerminalView {
+    val prepared = prepareBind(project, sourceNavigationProjectPath)
+    val committed = prepared.commit() ?: run {
+      prepared.close()
+      error("Transferable terminal destination project is already disposed")
+    }
+    committed.finish()
+    return prepared.view
+  }
+
+  override fun prepareBind(project: Project, sourceNavigationProjectPath: String?): PreparedTransferableTerminalBinding {
+    check(!closed.get()) { "Transferable terminal session is already closed" }
+    val projectScope = project.service<TransferableTerminalProjectBindingScope>().coroutineScope
+    val prepared = lifetime.prepareBinding(projectScope) { bindingScope ->
+      TerminalViewImpl(
+        project = project,
+        settings = JBTerminalSystemSettingsProvider(),
+        startupFusInfo = startupFusInfo,
+        coroutineScope = bindingScope,
+        sourceNavigationProjectPath = sourceNavigationProjectPath,
+      )
+    }
+    val binding = prepared.binding
+    binding.scope.launch {
+      withContext(Dispatchers.EDT + ModalityState.any().asContextElement()) {
+        if (binding.scope.isActive) {
+          binding.value.connectToSession(session)
+          installPortForwarding(binding.value, binding.scope.childScope("PortForwarding"))
+        }
+      }
+    }
+    return object : PreparedTransferableTerminalBinding {
+      override val view: TerminalView = binding.value
+      private val completed = AtomicBoolean(false)
+
+      override fun commit(): CommittedTransferableTerminalBinding? {
+        check(completed.compareAndSet(false, true)) { "Prepared terminal binding was already completed" }
+        val lifetimeCommit = prepared.commit() ?: return null
+        val previousView = synchronized(viewLock) {
+          val previous = if (this@TransferableTerminalSessionImpl::currentView.isInitialized) currentView else null
+          currentView = view
+          previous
+        }
+        return object : CommittedTransferableTerminalBinding {
+          private val finished = AtomicBoolean(false)
+
+          override fun finish() {
+            if (finished.compareAndSet(false, true)) {
+              lifetimeCommit.finish()
+            }
+          }
+
+          override fun rollback() {
+            if (!finished.compareAndSet(false, true)) return
+            synchronized(viewLock) {
+              if (currentView === view && previousView != null) {
+                currentView = previousView
+              }
+            }
+            lifetimeCommit.rollback()
+          }
+        }
+      }
+
+      override fun close() {
+        if (completed.compareAndSet(false, true)) {
+          prepared.close()
+        }
+      }
+    }
+  }
+
+  override suspend fun processId(): Long = session.processId
+
+  override fun close() {
+    if (!closed.compareAndSet(false, true)) return
+    transitionToTerminated()
+    lifetime.close()
+  }
+
+  companion object {
+    private val nextRuntimeId = AtomicLong(1)
+  }
+}

plugins/terminal/tests/src/com/intellij/terminal/tests/reworked/frontend/TransferableTerminalLifetimeTest.kt

@@ -0,0 +1,219 @@
+// Copyright 2000-2026 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
+package com.intellij.terminal.tests.reworked.frontend
+
+import com.intellij.platform.eel.EelDescriptor
+import com.intellij.platform.eel.provider.LocalEelDescriptor
+import com.intellij.platform.util.coroutines.childScope
+import com.intellij.terminal.frontend.session.TransferableTerminalLifetime
+import com.intellij.terminal.frontend.session.createStandardStateAwareTerminalSession
+import com.intellij.terminal.frontend.session.createTransferableTerminalSessionForTest
+import com.intellij.terminal.frontend.view.TerminalViewSessionState
+import com.intellij.openapi.project.Project
+import com.intellij.testFramework.junit5.TestApplication
+import com.intellij.testFramework.junit5.fixture.projectFixture
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.yield
+import org.assertj.core.api.Assertions.assertThat
+import org.jetbrains.plugins.terminal.session.impl.TerminalContentUpdatedEvent
+import org.jetbrains.plugins.terminal.session.impl.TerminalInitialStateEvent
+import org.jetbrains.plugins.terminal.session.impl.TerminalInputEvent
+import org.jetbrains.plugins.terminal.session.impl.TerminalOutputEvent
+import org.jetbrains.plugins.terminal.session.impl.TerminalSession
+import org.jetbrains.plugins.terminal.session.impl.TerminalStartupOptionsImpl
+import org.jetbrains.plugins.terminal.startup.TerminalProcessType
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Timeout
+
+@TestApplication
+class TransferableTerminalLifetimeTest {
+  private val project: Project by projectFixture()
+
+  @Test
+  fun sourceOwnerCancellationAfterRebindLeavesRuntimeAndDestinationBindingActive() = runBlocking(Dispatchers.Unconfined) {
+    val applicationScope = childScope("test application")
+    val sourceOwnerScope = childScope("test source project")
+    val destinationOwnerScope = childScope("test destination project")
+    val lifetime = TransferableTerminalLifetime(applicationScope)
+
+    val sourceBinding = lifetime.replaceBinding(sourceOwnerScope) { bindingScope -> bindingScope }
+    val destinationBinding = lifetime.replaceBinding(destinationOwnerScope) { bindingScope -> bindingScope }
+    sourceOwnerScope.cancel()
+
+    assertThat(sourceBinding.value.isActive).isFalse()
+    assertThat(lifetime.runtimeScope.isActive).isTrue()
+    assertThat(destinationBinding.value.isActive).isTrue()
+
+    lifetime.close()
+
+    assertThat(lifetime.runtimeScope.isActive).isFalse()
+    assertThat(destinationBinding.value.isActive).isFalse()
+    destinationOwnerScope.cancel()
+    applicationScope.cancel()
+  }
+
+  @Test
+  fun failedBindingCreationCancelsTemporaryScopeAndPreservesCurrentRuntime() = runBlocking(Dispatchers.Unconfined) {
+    val applicationScope = childScope("test application")
+    val sourceOwnerScope = childScope("test source project")
+    val destinationOwnerScope = childScope("test destination project")
+    val lifetime = TransferableTerminalLifetime(applicationScope)
+    lateinit var failedBindingScope: CoroutineScope
+    try {
+      val sourceBinding = lifetime.replaceBinding(sourceOwnerScope) { bindingScope -> bindingScope }
+
+      val failure = runCatching {
+        lifetime.replaceBinding(destinationOwnerScope) { bindingScope ->
+          failedBindingScope = bindingScope
+          error("binding creation failed")
+        }
+      }
+
+      assertThat(failure.exceptionOrNull()).hasMessage("binding creation failed")
+      assertThat(failedBindingScope.isActive).isFalse()
+      assertThat(sourceBinding.value.isActive).isTrue()
+      assertThat(lifetime.runtimeScope.isActive).isTrue()
+    }
+    finally {
+      lifetime.close()
+      sourceOwnerScope.cancel()
+      destinationOwnerScope.cancel()
+      applicationScope.cancel()
+    }
+    Unit
+  }
+
+  @Test
+  fun immediatelyCompletedRuntimeCannotRegressTerminationOrEscapeConstructor() = runBlocking(Dispatchers.Unconfined) {
+    val applicationScope = childScope("test application")
+    val transitions = mutableListOf<TerminalViewSessionState>()
+    lateinit var runtimeScope: CoroutineScope
+    try {
+      val result = runCatching {
+        createTransferableTerminalSessionForTest(
+          parentScope = applicationScope,
+          initialProject = project,
+          sessionStarter = { startedScope ->
+            runtimeScope = startedScope
+            startedScope.cancel()
+            TestRawTerminalSession(startedScope)
+          },
+          stateTransitionObserver = transitions::add,
+        )
+      }
+
+      assertThat(result.exceptionOrNull()).hasMessage("Transferable terminal runtime terminated during startup")
+      assertThat(result.getOrNull()).isNull()
+      assertThat(runtimeScope.isActive).isFalse()
+      assertThat(transitions).containsExactly(
+        TerminalViewSessionState.NotStarted,
+        TerminalViewSessionState.Terminated,
+      )
+    }
+    finally {
+      applicationScope.cancel()
+    }
+    Unit
+  }
+
+  @Test
+  @Timeout(20)
+  fun destinationCollectorReceivesStandardInitialStateForSameRuntime() = runBlocking(Dispatchers.Unconfined) {
+    val applicationScope = childScope("test application")
+    val sourceOwnerScope = childScope("test source project")
+    val destinationOwnerScope = childScope("test destination project")
+    val lifetime = TransferableTerminalLifetime(applicationScope)
+    try {
+      val rawSession = TestRawTerminalSession(lifetime.runtimeScope)
+      val session = createStandardStateAwareTerminalSession(
+        delegate = rawSession,
+        startupOptions = TerminalStartupOptionsImpl(
+          shellCommand = listOf("test-shell"),
+          workingDirectory = "/test/project",
+          envVariables = emptyMap(),
+          processType = TerminalProcessType.SHELL,
+          pid = TEST_PROCESS_ID,
+        ),
+        scope = lifetime.runtimeScope,
+      )
+      rawSession.emit(
+        TerminalContentUpdatedEvent(
+          text = "",
+          styles = emptyList(),
+          startLineLogicalIndex = 0,
+          cursorLogicalLineIndex = 0,
+          cursorColumnIndex = 0,
+        )
+      )
+      yield()
+      val sourceBinding = lifetime.replaceBinding(sourceOwnerScope) { bindingScope -> bindingScope }
+      val sourceInitial = session.getOutputFlow().first().single() as TerminalInitialStateEvent
+      assertThat(sourceInitial.outputModelState.text).isEmpty()
+
+      rawSession.emit(
+        TerminalContentUpdatedEvent(
+          text = "retained transcript",
+          styles = emptyList(),
+          startLineLogicalIndex = 0,
+          cursorLogicalLineIndex = 0,
+          cursorColumnIndex = 19,
+        )
+      )
+      yield()
+      val destinationBinding = lifetime.replaceBinding(destinationOwnerScope) { bindingScope -> bindingScope }
+      sourceOwnerScope.cancel()
+      val destinationInitial = session.getOutputFlow().first().single() as TerminalInitialStateEvent
+
+      assertThat(destinationInitial.outputModelState.text).isEqualTo("retained transcript")
+      assertThat(destinationInitial.startupOptions.shellCommand).containsExactly("test-shell")
+      assertThat(session.processId).isEqualTo(TEST_PROCESS_ID)
+      assertThat(sourceBinding.value.isActive).isFalse()
+      assertThat(destinationBinding.value.isActive).isTrue()
+      assertThat(lifetime.runtimeScope.isActive).isTrue()
+
+      lifetime.close()
+
+      assertThat(lifetime.runtimeScope.isActive).isFalse()
+      assertThat(destinationBinding.value.isActive).isFalse()
+    }
+    finally {
+      lifetime.close()
+      sourceOwnerScope.cancel()
+      destinationOwnerScope.cancel()
+      applicationScope.cancel()
+    }
+    Unit
+  }
+}
+
+private class TestRawTerminalSession(
+  override val coroutineScope: CoroutineScope,
+) : TerminalSession {
+  private val input = Channel<TerminalInputEvent>(Channel.UNLIMITED)
+  private val output = MutableSharedFlow<List<TerminalOutputEvent>>(replay = 1, extraBufferCapacity = 8)
+
+  suspend fun emit(event: TerminalOutputEvent) {
+    output.emit(listOf(event))
+  }
+
+  override suspend fun getInputChannel(): SendChannel<TerminalInputEvent> = input
+
+  override suspend fun getOutputFlow(): Flow<List<TerminalOutputEvent>> = output
+
+  override val eelDescriptor: EelDescriptor = LocalEelDescriptor
+  override val processId: Long = TEST_PROCESS_ID
+  override val isClosed: Boolean
+    get() = !coroutineScope.isActive
+
+  override suspend fun hasRunningCommands(): Boolean = false
+}
+
+private const val TEST_PROCESS_ID = 42L