// SPDX-FileCopyrightText: 2026 Kaede Akino
// SPDX-License-Identifier: GPL-3.0-only

import Foundation
import TermindApplication
import TermindDomain
import Testing

@testable import TermindRemoteInfrastructure

@Suite
struct ResumableTerminalEngineTests {
    @Test func restoreStartsWithTheLatestViewportAcceptedDuringPreparationWithoutRedundantResize() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["prepare"])
        let client = client(engine, events: events)
        let connecting = Task { try await connect(client) }
        await events.wait(for: "gated:prepare")
        try await client.resize(columns: 100, rows: 30)
        try await client.resize(columns: 132, rows: 43)
        try await client.resize(columns: 132, rows: 43)
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") || $0.hasPrefix("start:") }))
        await engine.release("prepare")
        try await connecting.value
        #expect(await engine.preparedPayload == checkpoint.payload)
        #expect(events.values.filter { $0.hasPrefix("start:") } == ["start:132x43"])
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") }))
        #expect(await engine.appliedGrid == "132x43")
        #expect(events.values.last == "state:connected")
        await client.close()
    }

    @Test func startupPreservesTerminalProtocolRepliesBeforePublishingConnected() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["start"])
        let client = client(engine, events: events)
        let connecting = Task { try await connect(client) }
        await events.wait(for: "gated:start")
        await events.emit(.connected)
        let reply = "\u{1B}[1;1R"
        try await client.write(Data(reply.utf8))
        #expect(events.values.contains("write:\(reply)"))
        #expect(!events.values.contains("state:connected"))
        await engine.release("start")
        try await connecting.value
        #expect(events.values.last == "state:connected")
        await client.close()
    }

    @Test func startupDrainsResizeChangesAcrossStartAndResizeAwaitsBeforeConnected() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["start", "resize"])
        let client = client(engine, events: events)
        let connecting = Task { try await connect(client) }
        await events.wait(for: "gated:start")
        try await client.resize(columns: 100, rows: 30)
        try await client.resize(columns: 132, rows: 43)
        await events.emit(.connected)
        #expect(!events.values.contains("state:connected"))
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") }))
        await engine.release("start")
        await events.wait(for: "gated:resize")
        #expect(await engine.appliedGrid == "80x24")
        try await client.resize(columns: 150, rows: 50)
        try await client.resize(columns: 160, rows: 60)
        try await client.resize(columns: 160, rows: 60)
        await events.emit(.activity)
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:132x43"])
        #expect(!events.values.contains("state:connected"))
        await engine.release("resize")
        try await connecting.value
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:132x43", "resize:160x60"])
        #expect(await engine.appliedGrid == "160x60")
        #expect(events.values.last == "state:connected")
        try await client.write(Data("after".utf8))
        #expect(events.values.last == "write:after")
        await client.close()
    }

    @Test func startupRechecksViewportAfterWritingRestoredIntegrationBeforePublishingConnected() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["start", "write"], restoresIntegration: true)
        let client = client(engine, events: events)
        let connecting = Task { try await connect(client) }
        await events.wait(for: "gated:start")
        await events.emit(.output(Data("__TRMD_RESTORE_SHELL__/bin/bash__\r\n".utf8)))
        #expect(!events.values.contains(where: { $0.hasPrefix("write:") }))
        await engine.release("start")
        await events.wait(for: "gated:write")
        try await client.resize(columns: 140, rows: 50)
        #expect(!events.values.contains("state:connected"))
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") }))
        await engine.release("write")
        try await connecting.value
        #expect(events.values.filter { $0.hasPrefix("write:") && $0.contains("TermindCommand=RESTORE") }.count == 1)
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:140x50"])
        #expect(await engine.appliedGrid == "140x50")
        #expect(events.values.last == "state:connected")
        await client.close()
    }

    @Test func cancellingStartupDuringResizeDrainClosesWithoutAdmittingConnectedOrInput() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["start", "resize"])
        let client = client(engine, events: events)
        let connecting = Task { try await connect(client) }
        await events.wait(for: "gated:start")
        try await client.resize(columns: 132, rows: 43)
        await engine.release("start")
        await events.wait(for: "gated:resize")
        connecting.cancel()
        try await client.resize(columns: 160, rows: 60)
        await engine.release("resize")
        await #expect(throws: CancellationError.self) { try await connecting.value }
        #expect(!events.values.contains("state:connected"))
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:132x43"])
        #expect(Array(events.values.suffix(3)) == ["rollback", "close", "state:closed"])
        await #expect(throws: CancellationError.self) { try await client.write(Data("blocked".utf8)) }
    }

    @Test func routeFactsPassWhileRecoveryIsFrozenWithoutConfirmingProtocolReadiness() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry"])
        let client = client(engine, events: events)
        try await connect(client)
        await events.emit(.recovering)
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "gated:retry")
        let attemptID = UUID()
        await events.emit(.routeChanged(.init(attemptID: attemptID, revision: 1, stage: .opening)))
        await events.emit(.routeChanged(.init(attemptID: attemptID, revision: 2, stage: .ready)))
        #expect(events.values.suffix(2) == ["route:1:opening", "route:2:ready"])
        #expect(events.values.filter { $0.hasPrefix("state:") }.last == "state:recovering")
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        await engine.release("retry")
        try await retrying.value
        #expect(events.values.filter { $0.hasPrefix("state:") }.last == "state:recovering")
        #expect(events.values.filter { $0 == "prepareRetry" }.count == 1)
        #expect(events.values.filter { $0 == "retry" }.count == 1)
        await events.emit(.connected)
        #expect(events.values.last == "state:connected")
        await client.close()
    }

    @Test func reorderedRouteCallbacksAndClosedClientCannotReviveAnOldRoute() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        let old = UUID()
        let next = UUID()
        await events.emit(.routeChanged(.init(attemptID: old, revision: 3, stage: .unavailable)))
        await events.emit(.routeChanged(.init(attemptID: old, revision: 2, stage: .ready)))
        await events.emit(.routeChanged(.init(attemptID: next, revision: 4, stage: .opening)))
        await events.emit(.routeChanged(.init(attemptID: old, revision: 3, stage: .unavailable)))
        #expect(events.values.filter { $0.hasPrefix("route:") } == ["route:3:unavailable", "route:4:opening"])
        await client.close()
        let closed = events.values
        await events.emit(.routeChanged(.init(attemptID: next, revision: 5, stage: .ready)))
        #expect(events.values == closed)
    }

    @Test func recoveryRetryUsesTheExistingEngineWithoutReplayingStartup() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        await events.emit(.recovering)
        try await client.retryNow()
        #expect(Array(events.values.suffix(3)) == ["prepareRetry", "retry", "resize:80x24"])
        #expect(events.values.filter { $0 == "prepare" }.count == 1)
        #expect(events.values.filter { $0.hasPrefix("start:") }.count == 1)
        #expect(!events.values.contains(where: { $0.hasPrefix("write:") }))
        await client.close()
        let closedEvents = events.values
        await events.emit(.recovering)
        await events.emit(.connected)
        #expect(events.values == closedEvents)
    }

    @Test func missingRegistrationFailsBeforeWakeOrBootstrapAndClearsFacadeSecrets() async throws {
        let events = EngineEvents()
        let channel = RemoteConnectionCoordinator(transportEngines: .init(factories: [:])).makePTYChannel(
            output: { _ in }, exit: { _ in })
        await channel.configureTransport(
            .mosh, options: options,
            route: { _ in
                events.record("route")
                throw CancellationError()
            }, stateChanged: { _ in })
        await channel.configureStartup { _ in
            events.record("startup")
            return nil
        }
        let password = SSHSecret("one-use")
        var request = request
        request.configuration.credentials = [.password(password)]
        request.configuration.beforeConnect = { events.record("wake") }
        do {
            try await channel.connect(using: request, columns: 80, rows: 24)
            Issue.record("An unregistered transport must fail before attempting SSH")
        } catch TerminalTransportError.unsupportedRoute {
        }
        #expect(events.values.isEmpty)
        #expect(password.isConsumed)
        await channel.disconnect()
    }

    @Test func coordinatorVariantUsesRegisteredEngineAndRestoresWithoutStartupOrRoute() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let coordinator = RemoteConnectionCoordinator(transportEngines: registry(engine, events: events))
            .withKeyboardInteractive { _ in [] }
        let channel = coordinator.makePTYChannel(
            output: { events.record("output:\(String(decoding: $0, as: UTF8.self))") }, exit: { _ in })
        await channel.configureTransport(
            .mosh, options: options, checkpoint: checkpoint,
            route: { _ in
                events.record("route")
                throw CancellationError()
            }, stateChanged: { events.record("state:\($0)") })
        await channel.configureStartup { _ in
            events.record("startup")
            return nil
        }
        try await channel.connect(using: request, columns: 81, rows: 25)
        #expect(await engine.preparedPayload == checkpoint.payload)
        #expect(await engine.permitsAgentForwarding == false)
        #expect(events.values.contains("start:81x25"))
        #expect(events.values.contains("output:ready"))
        #expect(!events.values.contains("startup"))
        #expect(!events.values.contains("route"))
        await channel.write(Data("hello".utf8))
        await channel.resize(columns: 100, rows: 30, widthPixels: 0, heightPixels: 0)
        await channel.disconnect()
        await channel.disconnect()
        #expect(events.values.contains("write:hello"))
        #expect(events.values.contains("resize:100x30"))
        #expect(events.values.filter { $0 == "close" }.count == 1)
    }

    @Test func suspensionDrainsWritesAndLateActivityCannotUnfreezeTheCheckpoint() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        let writing = Task { try await client.write(Data("hold".utf8)) }
        #expect(await eventually { events.values.contains("write:hold") })
        let suspending = Task { try await client.suspend() }
        #expect(
            await eventually {
                do {
                    try await client.write(Data("probe".utf8))
                    return false
                } catch TerminalTransportError.suspended {
                    return true
                } catch {
                    return false
                }
            })
        #expect(!events.values.contains("checkpoint"))
        await events.emit(.connected)
        await events.emit(.activity)
        await events.emit(.output(Data("tail".utf8)))
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        try await client.resize(columns: 132, rows: 43)
        #expect(!events.values.contains("resize:132x43"))
        await engine.finishWrite()
        try await writing.value
        let saved = try await suspending.value
        #expect(saved.transport == .mosh)
        #expect(saved.payload == Data("saved-state".utf8))
        #expect(events.values.contains("output:tail"))
        #expect(events.values.last == "state:suspended")
        await events.emit(.connected)
        #expect(events.values.last == "state:suspended")
        try await client.retryNow()
        let retry = events.values.suffix(5)
        #expect(Array(retry) == ["prepareRetry", "beforeResume", "state:recovering", "resume", "resize:132x43"])
        try await client.write(Data("after".utf8))
        #expect(events.values.last == "write:after")
        await client.close()
    }

    @Test func resumeCoalescesAndFlushesOnlyTheLatestViewportBeforeIntegrationAndConnected() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["resume"], restoresIntegration: true)
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        let resuming = Task { try await client.resume() }
        await events.wait(for: "gated:resume")
        let secondResume = Task { try await client.resume() }
        await events.wait(for: "lifecycle:queued", count: 3)
        try await client.resize(columns: 100, rows: 30)
        try await client.resize(columns: 132, rows: 43)
        await events.emit(.connected)
        await events.emit(.output(Data("__TRMD_RESTORE_SHELL__/bin/bash__\r\n".utf8)))
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") || $0.hasPrefix("write:") }))
        #expect(events.values.filter { $0.hasPrefix("state:") }.last == "state:recovering")

        await engine.release("resume")
        try await resuming.value
        try await secondResume.value
        #expect(events.values.filter { $0 == "resume" }.count == 1)
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:132x43"])
        let writes = events.values.filter { $0.hasPrefix("write:") }
        #expect(writes.count == 1)
        #expect(writes.first?.contains("TermindCommand=RESTORE") == true)
        let resizeIndex = try #require(events.values.firstIndex(of: "resize:132x43"))
        let writeIndex = try #require(events.values.firstIndex(where: { $0.hasPrefix("write:") }))
        #expect(resizeIndex < writeIndex)
        #expect(events.values.last == "state:connected")
        #expect(!events.values.contains("state:failed"))
        try await client.write(Data("after".utf8))
        #expect(events.values.last == "write:after")
        await client.close()
    }

    @Test func retryDrainsExistingSendsAndCoalescesWhilePreservingBlockedViewportChanges() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry"])
        let client = client(engine, events: events)
        try await connect(client)
        let writing = Task { try await client.write(Data("hold".utf8)) }
        await events.wait(for: "write:hold")
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued")
        let secondRetry = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued", count: 2)
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        try await client.resize(columns: 111, rows: 31)
        #expect(!events.values.contains("prepareRetry"))
        #expect(!events.values.contains("retry"))
        await engine.finishWrite()
        try await writing.value
        await events.wait(for: "gated:retry")
        try await client.resize(columns: 140, rows: 50)
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") }))
        await events.emit(.connected)
        await engine.release("retry")
        try await retrying.value
        try await secondRetry.value
        #expect(events.values.filter { $0 == "prepareRetry" }.count == 1)
        #expect(events.values.filter { $0 == "retry" }.count == 1)
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:140x50"])
        #expect(events.values.last == "state:connected")
        #expect(!events.values.contains("state:failed"))
        await client.close()
    }

    @Test func checkpointAndResumeQueueBehindRetryWithoutOpeningSendAdmissionBetweenThem() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry", "checkpoint", "resume"])
        let client = client(engine, events: events)
        try await connect(client)
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "gated:retry")
        let suspending = Task { try await client.suspend() }
        await events.wait(for: "lifecycle:queued", count: 2)
        let resuming = Task { try await client.resume() }
        await events.wait(for: "lifecycle:queued", count: 3)
        #expect(!events.values.contains("checkpoint"))
        #expect(!events.values.contains("resume"))
        await events.emit(.connected)
        await engine.release("retry")
        try await retrying.value
        await events.wait(for: "gated:checkpoint")
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        #expect(!events.values.contains("resume"))
        await engine.release("checkpoint")
        let saved = try await suspending.value
        #expect(saved.payload == Data("saved-state".utf8))
        await events.wait(for: "gated:resume")
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        await events.emit(.connected)
        await engine.release("resume")
        try await resuming.value
        #expect(
            events.values.filter { ["retry", "checkpoint", "resume"].contains($0) } == [
                "retry", "checkpoint", "resume",
            ])
        #expect(events.values.filter { $0 == "state:connected" }.count == 2)
        #expect(!events.values.contains("state:failed"))
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func queuedResumePreservesTheOnlyConnectedEventFromRetry() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry"])
        let client = client(engine, events: events)
        try await connect(client)
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "gated:retry")
        let resuming = Task { try await client.resume() }
        await events.wait(for: "lifecycle:queued", count: 2)
        await events.emit(.connected)
        await engine.release("retry")
        try await retrying.value
        try await resuming.value
        #expect(events.values.last == "state:connected")
        #expect(events.values.filter { $0 == "resume" }.count == 1)
        #expect(!events.values.contains("state:failed"))
        await client.close()
    }

    @Test func checkpointQueuedDuringInitialStartWaitsForTheLatestViewportToDrain() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["start", "resize"])
        let client = client(engine, events: events)
        let connecting = Task { try await connect(client) }
        await events.wait(for: "gated:start")
        let suspending = Task { try await client.suspend() }
        await events.wait(for: "lifecycle:queued")
        try await client.resize(columns: 132, rows: 43)
        #expect(!events.values.contains("checkpoint"))
        await engine.release("start")
        await events.wait(for: "gated:resize")
        try await client.resize(columns: 160, rows: 60)
        #expect(!events.values.contains("checkpoint"))
        #expect(!events.values.contains("state:connected"))
        await engine.release("resize")
        try await connecting.value
        let saved = try await suspending.value
        #expect(await engine.appliedGrid == "160x60")
        let resizeIndex = try #require(events.values.firstIndex(of: "resize:160x60"))
        let checkpointIndex = try #require(events.values.firstIndex(of: "checkpoint"))
        #expect(resizeIndex < checkpointIndex)
        #expect(saved.transport == .mosh)
        #expect(events.values.last == "state:suspended")
        #expect(!events.values.contains("state:failed"))
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        try await client.resume()
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func checkpointAfterStartDoesNotRequireTheConnectedObserverToHaveCaughtUp() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, reportsConnectedOnStart: false)
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        #expect(events.values.contains("checkpoint"))
        #expect(events.values.last == "state:suspended")
        try await client.resume()
        try await client.write(Data("after".utf8))
        await events.emit(.connected)
        #expect(events.values.last == "state:connected")
        await client.close()
    }

    @Test func resizeChangesDuringRecoveryFlushAreAppliedBeforeUserWritesReopen() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["resize"])
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        let resuming = Task { try await client.resume() }
        await events.wait(for: "gated:resize")
        try await client.resize(columns: 110, rows: 40)
        try await client.resize(columns: 150, rows: 60)
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:80x24"])
        await engine.release("resize")
        try await resuming.value
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:80x24", "resize:150x60"])
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func coalescedResumeFailureReachesEveryCallerAndKeepsAdmissionFrozenUntilRetry() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["resume", "prepareRetry"], failures: ["resume"])
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        let resuming = Task { try await client.resume() }
        await events.wait(for: "gated:resume")
        let secondResume = Task { try await client.resume() }
        await events.wait(for: "lifecycle:queued", count: 3)
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued", count: 4)
        await engine.release("resume")
        await expectUnreachable(resuming)
        await expectUnreachable(secondResume)
        await events.wait(for: "gated:prepareRetry")
        #expect(events.values.filter { $0 == "resume" }.count == 1)
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") }))
        await engine.release("prepareRetry")
        try await retrying.value
        #expect(events.values.filter { $0 == "resume" }.count == 2)
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func failedResumeAdmissionLeavesTheExistingCheckpointEligible() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, failures: ["beforeResume"])
        let client = client(engine, events: events)
        try await connect(client)
        let saved = try await client.suspend()
        await expectUnreachable(Task { try await client.resume() })
        let stillSaved = try await client.suspend()
        #expect(stillSaved.payload == saved.payload)
        #expect(events.values.filter { $0 == "checkpoint" }.count == 1)
        #expect(!events.values.contains("resume"))
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        try await client.retryNow()
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func failedResumePreflightWithoutASuspensionKeepsBufferedStateFrozenUntilRecovery() async throws {
        let events = EngineEvents()
        let engine = TestEngine(
            events: events, gates: ["beforeResume"], restoresIntegration: true, failures: ["beforeResume"])
        let client = client(engine, events: events)
        try await connect(client)
        let resuming = Task { try await client.resume() }
        await events.wait(for: "gated:beforeResume")
        try await client.resize(columns: 150, rows: 60)
        await events.emit(.output(Data("__TRMD_RESTORE_SHELL__/bin/bash__\r\n".utf8)))
        await engine.release("beforeResume")
        await expectUnreachable(resuming)
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        #expect(!events.values.contains(where: { $0.hasPrefix("resize:") }))
        #expect(!events.values.contains(where: { $0.hasPrefix("write:") }))
        try await client.retryNow()
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:150x60"])
        #expect(events.values.filter { $0.hasPrefix("write:") && $0.contains("TermindCommand=RESTORE") }.count == 1)
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func cancellingOneCoalescedRetryDoesNotCancelTheOtherCaller() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry"])
        let client = client(engine, events: events)
        try await connect(client)
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "gated:retry")
        let cancelled = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued", count: 2)
        cancelled.cancel()
        await #expect(throws: CancellationError.self) { try await cancelled.value }
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        await engine.release("retry")
        try await retrying.value
        #expect(events.values.filter { $0 == "retry" }.count == 1)
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func cancelledQueuedCheckpointDoesNotSuspendTheRecoveredEngine() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry"])
        let client = client(engine, events: events)
        try await connect(client)
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "gated:retry")
        let cancelled = Task { try await client.suspend() }
        await events.wait(for: "lifecycle:queued", count: 2)
        cancelled.cancel()
        await #expect(throws: CancellationError.self) { try await cancelled.value }
        await engine.release("retry")
        try await retrying.value
        #expect(!events.values.contains("checkpoint"))
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func cancellingRetryCallerDuringSendDrainStillFinishesBeforeTheQueuedCheckpoint() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        let writing = Task { try await client.write(Data("hold".utf8)) }
        await events.wait(for: "write:hold")
        let cancelled = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued")
        let suspending = Task { try await client.suspend() }
        await events.wait(for: "lifecycle:queued", count: 2)
        cancelled.cancel()
        await #expect(throws: CancellationError.self) { try await cancelled.value }
        #expect(!events.values.contains("checkpoint"))
        await engine.finishWrite()
        try await writing.value
        _ = try await suspending.value
        #expect(
            events.values.filter { ["prepareRetry", "retry", "checkpoint"].contains($0) } == [
                "prepareRetry", "retry", "checkpoint",
            ])
        #expect(events.values.last == "state:suspended")
        await client.close()
    }

    @Test func cancelledRetryCallerStillFlushesAndANewCallerCanJoinItsAdmittedWork() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["retry"], restoresIntegration: true)
        let client = client(engine, events: events)
        try await connect(client)
        await events.emit(.recovering)
        let writing = Task { try await client.write(Data("hold".utf8)) }
        await events.wait(for: "write:hold")
        let cancelled = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued")
        try await client.resize(columns: 150, rows: 60)
        await events.emit(.output(Data("__TRMD_RESTORE_SHELL__/bin/bash__\r\n".utf8)))
        #expect(events.values.filter { $0.hasPrefix("state:") }.last == "state:recovering")
        cancelled.cancel()
        await #expect(throws: CancellationError.self) { try await cancelled.value }
        #expect(!events.values.contains("prepareRetry"))
        let replacement = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued", count: 2)
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        await engine.finishWrite()
        try await writing.value
        await events.wait(for: "gated:retry")
        await events.emit(.connected)
        await engine.release("retry")
        try await replacement.value
        #expect(events.values.filter { $0 == "retry" }.count == 1)
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:150x60"])
        #expect(events.values.filter { $0.hasPrefix("write:") && $0.contains("TermindCommand=RESTORE") }.count == 1)
        #expect(events.values.last == "state:connected")
        try await client.write(Data("after".utf8))
        await client.close()
    }

    @Test func cancelledResumeCallerLeavesProtocolWorkSerializedBeforeTheNextCheckpoint() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: ["resume"])
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        let cancelled = Task { try await client.resume() }
        await events.wait(for: "gated:resume")
        cancelled.cancel()
        await #expect(throws: CancellationError.self) { try await cancelled.value }
        let suspending = Task { try await client.suspend() }
        await events.wait(for: "lifecycle:queued", count: 3)
        #expect(events.values.filter { $0 == "checkpoint" }.count == 1)
        await #expect(throws: TerminalTransportError.self) { try await client.write(Data("blocked".utf8)) }
        await engine.release("resume")
        _ = try await suspending.value
        #expect(events.values.filter { $0 == "checkpoint" }.count == 2)
        #expect(events.values.last == "state:suspended")
        #expect(events.values.filter { $0.hasPrefix("resize:") } == ["resize:80x24"])
        await client.close()
    }

    @Test(arguments: ["resume", "retry", "checkpoint", "beforeResume", "prepareRetry"])
    func closeImmediatelyReleasesLifecycleCallersAndFencesLateEngineCompletion(_ gate: String) async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, gates: [gate])
        let client = client(engine, events: events)
        try await connect(client)
        if gate == "resume" || gate == "beforeResume" { _ = try await client.suspend() }
        let baseline = events.values.filter { $0 == "lifecycle:queued" }.count
        let operation = Task {
            if gate == "checkpoint" {
                _ = try await client.suspend()
            } else if gate == "retry" || gate == "prepareRetry" {
                try await client.retryNow()
            } else {
                try await client.resume()
            }
        }
        await events.wait(for: "gated:\(gate)")
        let queued = Task {
            if gate == "checkpoint" {
                try await client.resume()
            } else {
                _ = try await client.suspend()
            }
        }
        await events.wait(for: "lifecycle:queued", count: baseline + 2)
        await client.close()
        await #expect(throws: CancellationError.self) { try await operation.value }
        await #expect(throws: CancellationError.self) { try await queued.value }
        let closedEvents = events.values
        await engine.release(gate)
        await events.wait(for: "released:\(gate)")
        await events.emit(.output(Data("late".utf8)))
        await events.emit(.connected)
        await events.emit(.activity)
        await #expect(throws: CancellationError.self) { try await client.resize(columns: 200, rows: 80) }
        await #expect(throws: CancellationError.self) { try await client.write(Data("late".utf8)) }
        #expect(events.values == closedEvents + ["released:\(gate)"])
        #expect(events.values.filter { $0 == "close" }.count == 1)
    }

    @Test func closeReleasesARetryWaitingForAnOutstandingWrite() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        let writing = Task { try await client.write(Data("hold".utf8)) }
        await events.wait(for: "write:hold")
        let retrying = Task { try await client.retryNow() }
        await events.wait(for: "lifecycle:queued")
        await client.close()
        await #expect(throws: CancellationError.self) { try await retrying.value }
        try await writing.value
        #expect(!events.values.contains("prepareRetry"))
        #expect(!events.values.contains("retry"))
        #expect(events.values.last == "state:closed")
    }

    @Test func initialHealthyPathDoesNotReconnectButSubsequentHealthyHandoversDo() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        await client.networkPathChanged(available: true)
        #expect(!events.values.contains("networkAvailable"))
        #expect(!events.values.contains("state:recovering"))
        await client.networkPathChanged(available: true)
        await client.networkPathChanged(available: true)
        #expect(events.values.filter { $0 == "networkAvailable" }.count == 2)
        await client.close()
        await client.networkPathChanged(available: true)
        #expect(events.values.filter { $0 == "networkAvailable" }.count == 2)
    }

    @Test func initialUnavailablePathPresentsRecoveryAndRestorationNotifiesTheEngine() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        await client.networkPathChanged(available: false)
        #expect(events.values.last == "state:recovering")
        #expect(!events.values.contains("networkAvailable"))
        await client.networkPathChanged(available: true)
        #expect(events.values.filter { $0 == "networkAvailable" }.count == 1)
        await client.close()
    }

    @Test func frozenNetworkObservationsAreRememberedWithoutContactingTheEngine() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        await client.networkPathChanged(available: true)
        await client.networkPathChanged(available: false)
        #expect(!events.values.contains("networkAvailable"))
        #expect(events.values.last == "state:suspended")
        try await client.resume()
        await client.networkPathChanged(available: true)
        #expect(events.values.filter { $0 == "networkAvailable" }.count == 1)
        await client.close()
    }

    @Test func failedStartRollsBackAndClosesTheSelectedEngine() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, failsStart: true)
        let client = client(engine, events: events)
        await #expect(throws: TerminalTransportError.self) { try await connect(client) }
        #expect(Array(events.values.suffix(3)) == ["rollback", "close", "state:closed"])
        await client.close()
        #expect(events.values.filter { $0 == "close" }.count == 1)
    }

    @Test func checkpointCapturesShellEventsEmittedDuringCheckpointPreparation() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events, emitsShellEventDuringCheckpoint: true)
        let client = client(engine, events: events)
        try await connect(client)
        _ = try await client.suspend()
        var filter = try #require(await engine.capturedShell?.commandFilter)
        let update = filter.receive(Data("hello\u{7}visible".utf8))
        #expect(update.events == [.commandStarted("echo hello")])
        #expect(update.visibleOutput == Data("visible".utf8))
        await client.close()
    }

    @Test func remoteEndClosesOnceAndRejectsLateEngineEvents() async throws {
        let events = EngineEvents()
        let engine = TestEngine(events: events)
        let client = client(engine, events: events)
        try await connect(client)
        await events.emit(.ended)
        let ended = events.values
        await events.emit(.output(Data("late".utf8)))
        await events.emit(.connected)
        await events.emit(.ended)
        await client.close()
        #expect(events.values == ended)
        #expect(events.values.filter { $0 == "state:ended" }.count == 1)
        #expect(events.values.filter { $0 == "close" }.count == 1)
        await #expect(throws: CancellationError.self) { try await client.write(Data("late".utf8)) }
    }

    @Test func incompatibleEnvelopeNeverConstructsAnEngine() async throws {
        for invalid in [
            TerminalTransportCheckpoint(transport: .eternalTerminal, payload: Data()),
            {
                var value = checkpoint
                value.version = 99
                return value
            }(),
        ] {
            let events = EngineEvents()
            let engine = TestEngine(events: events)
            let client = ResumableTerminalClient(
                transport: .mosh, checkpoint: invalid, output: { _ in }, stateChanged: { _ in },
                shellIntegration: { _ in }, registry: registry(engine, events: events))
            await #expect(throws: TerminalTransportError.self) { try await connect(client) }
            #expect(events.values.isEmpty)
        }
    }

    @Test func legacyMoshPayloadRemainsTheUnwrappedVendorSnapshot() throws {
        let literal = Data(
            """
            {"endpoint":{"host":"old.example","port":60001,"keyBase64_22":"AAAAAAAAAAAAAAAAAAAAAA"},
             "transportState":"bGVnYWN5LXN0YXRl","createdAtMs":1700000000123,"schemaVersion":2}
            """.utf8)
        let snapshot = try MoshTerminalEngine.decodeCheckpoint(literal)
        #expect(snapshot.endpoint.host == "old.example")
        #expect(snapshot.endpoint.port == 60001)
        #expect(snapshot.transportState == Data("legacy-state".utf8))
        #expect(try json(JSONEncoder().encode(snapshot)) == json(literal))
    }

    @Test func legacyEternalPayloadRetainsFlatKeysAndRestoredShellInput() throws {
        let literal = Data(
            """
            {"clientID":"legacy-client","passkey":"AAECAwQ=",
             "checkpoint":{"reader":{"nonce":"AQID","sequenceNumber":17},
               "writer":{"nonce":"BAUG","sequenceNumber":19,"serializedBackupPackets":["YWJj"]}},
             "integratedShell":{"shell":"zsh","homeDirectory":"/home/legacy"},
             "pendingIntegrationCommand":"cGVuZGluZw=="}
            """.utf8)
        let payload = try EternalTerminalEngine.decodeCheckpoint(literal)
        #expect(payload.clientID == "legacy-client")
        #expect(payload.passkey == Data([0, 1, 2, 3, 4]))
        #expect(payload.shellState.integratedShell?.homeDirectory == "/home/legacy")
        #expect(payload.shellState.pendingIntegrationCommand == Data("pending".utf8))
        #expect(try json(JSONEncoder().encode(payload)) == json(literal))
    }

    private var options: HostTerminalTransports {
        var options = HostTerminalTransports()
        options.moshEnabled = true
        return options
    }

    private var request: SSHConnectionRequest {
        .init(
            configuration: .init(
                hostID: UUID(), host: "127.0.0.1", port: 1, username: "unused", credentials: [], expectedHostKeys: []))
    }

    private var checkpoint: TerminalTransportCheckpoint {
        .init(transport: .mosh, payload: Data("old-state".utf8))
    }

    private func registry(_ engine: TestEngine, events: EngineEvents) -> ResumableTerminalEngineRegistry {
        .init(factories: [
            .mosh: {
                events.install($0)
                events.record("factory")
                return engine
            }
        ])
    }

    private func client(_ engine: TestEngine, events: EngineEvents) -> ResumableTerminalClient {
        .init(
            transport: .mosh, checkpoint: checkpoint,
            output: { events.record("output:\(String(decoding: $0, as: UTF8.self))") },
            stateChanged: { events.record("state:\($0)") }, shellIntegration: { _ in },
            registry: registry(engine, events: events), pathMonitor: nil,
            lifecycleRequestEnqueued: { events.record("lifecycle:queued") },
            routeChanged: { events.record("route:\($0.revision):\($0.stage)") })
    }

    private func connect(_ client: ResumableTerminalClient) async throws {
        try await client.connect(
            request: request, options: options, environment: [], shellIntegrationEnabled: true,
            columns: 80, rows: 24, route: { _ in throw CancellationError() }, progress: nil,
            startupPreparation: { _ in
                Issue.record("Restoring a checkpoint must not prepare a new shell")
                return nil
            })
    }

    private func json(_ data: Data) throws -> NSDictionary {
        try #require(JSONSerialization.jsonObject(with: data) as? NSDictionary)
    }

    private func expectUnreachable(_ task: Task<Void, any Error>) async {
        do {
            try await task.value
            Issue.record("Expected the injected engine failure")
        } catch TerminalTransportError.unreachable {
        } catch {
            Issue.record("Unexpected lifecycle error: \(error)")
        }
    }

    private func eventually(_ predicate: () async -> Bool) async -> Bool {
        for _ in 0..<200 {
            if await predicate() { return true }
            try? await Task.sleep(for: .milliseconds(10))
        }
        return await predicate()
    }
}

private nonisolated final class EngineEvents: @unchecked Sendable {
    private struct Waiter {
        let value: String
        let count: Int
        let continuation: CheckedContinuation<Void, Never>
    }
    private let lock = NSLock()
    private var recorded: [String] = []
    private var waiters: [Waiter] = []
    private var receive: ResumableTerminalEngine.EventHandler?
    var values: [String] { lock.withLock { recorded } }
    func record(_ value: String) {
        let ready = lock.withLock {
            recorded.append(value)
            let count = recorded.filter { $0 == value }.count
            let ready = waiters.filter { $0.value == value && $0.count <= count }
            waiters.removeAll { $0.value == value && $0.count <= count }
            return ready
        }
        for waiter in ready { waiter.continuation.resume() }
    }
    func wait(for value: String, count: Int = 1) async {
        await withCheckedContinuation { continuation in
            let ready = lock.withLock {
                if recorded.filter({ $0 == value }).count >= count { return true }
                waiters.append(.init(value: value, count: count, continuation: continuation))
                return false
            }
            if ready { continuation.resume() }
        }
    }
    func install(_ receive: @escaping ResumableTerminalEngine.EventHandler) {
        lock.withLock { self.receive = receive }
    }
    func emit(_ event: ResumableTerminalEngineEvent) async {
        let receive = lock.withLock { self.receive }
        await receive?(event)
    }
}

private actor TestEngine: ResumableTerminalEngine {
    let events: EngineEvents
    let failsStart: Bool
    let emitsShellEventDuringCheckpoint: Bool
    let restoresIntegration: Bool
    let reportsConnectedOnStart: Bool
    private(set) var preparedPayload: Data?
    private(set) var appliedGrid: String?
    private(set) var permitsAgentForwarding = true
    private(set) var capturedShell: ResumableTerminalShellState?
    private var writing: CheckedContinuation<Void, Never>?
    private var gates: Set<String>
    private var failures: Set<String>
    private var blocked: [String: CheckedContinuation<Void, Never>] = [:]

    init(
        events: EngineEvents, failsStart: Bool = false, emitsShellEventDuringCheckpoint: Bool = false,
        gates: Set<String> = [], restoresIntegration: Bool = false, failures: Set<String> = [],
        reportsConnectedOnStart: Bool = true
    ) {
        self.events = events
        self.failsStart = failsStart
        self.emitsShellEventDuringCheckpoint = emitsShellEventDuringCheckpoint
        self.gates = gates
        self.restoresIntegration = restoresIntegration
        self.failures = failures
        self.reportsConnectedOnStart = reportsConnectedOnStart
    }
    func prepare(_ context: ResumableTerminalEngineContext) async throws -> ResumableTerminalEnginePreparation {
        preparedPayload = context.checkpoint
        permitsAgentForwarding = context.request.permitsAgentForwarding
        events.record("prepare")
        await waitAtGate("prepare")
        return .init(
            shell: .init(
                integration: restoresIntegration ? RemoteShellIntegrationBootstrap(token: "RESTORE") : nil,
                commandFilter: emitsShellEventDuringCheckpoint ? RemoteCommandEventFilter(token: "capture") : nil))
    }
    func start(columns: UInt16, rows: UInt16) async throws {
        events.record("start:\(columns)x\(rows)")
        if failsStart { throw TerminalTransportError.unreachable }
        await waitAtGate("start")
        appliedGrid = "\(columns)x\(rows)"
        if reportsConnectedOnStart { await events.emit(.connected) }
        await events.emit(.output(Data("ready".utf8)))
    }
    func write(_ data: Data) async throws {
        events.record("write:\(String(decoding: data, as: UTF8.self))")
        await waitAtGate("write")
        if data == Data("hold".utf8) {
            await withCheckedContinuation { writing = $0 }
        }
    }
    func finishWrite() {
        writing?.resume()
        writing = nil
    }
    func resize(columns: UInt16, rows: UInt16) async throws {
        events.record("resize:\(columns)x\(rows)")
        await waitAtGate("resize")
        try failIfRequested("resize")
        appliedGrid = "\(columns)x\(rows)"
    }
    func checkpoint(shell: @Sendable () async -> ResumableTerminalShellState) async throws -> Data {
        events.record("checkpoint")
        await waitAtGate("checkpoint")
        try failIfRequested("checkpoint")
        if emitsShellEventDuringCheckpoint {
            await events.emit(.output(Data("\u{1b}]777;TermindCommand=capture;echo%20".utf8)))
        }
        capturedShell = await shell()
        return Data("saved-state".utf8)
    }
    func beforeResume() async throws {
        events.record("beforeResume")
        await waitAtGate("beforeResume")
        try failIfRequested("beforeResume")
    }
    func resume() async throws {
        events.record("resume")
        await waitAtGate("resume")
        try failIfRequested("resume")
    }
    func prepareRetry() async {
        events.record("prepareRetry")
        await waitAtGate("prepareRetry")
    }
    func retry() async throws {
        events.record("retry")
        await waitAtGate("retry")
        try failIfRequested("retry")
    }
    func networkAvailable() async { events.record("networkAvailable") }
    private func waitAtGate(_ name: String) async {
        guard gates.contains(name) else { return }
        await withCheckedContinuation {
            blocked[name] = $0
            events.record("gated:\(name)")
        }
        events.record("released:\(name)")
    }
    func release(_ name: String) {
        gates.remove(name)
        blocked.removeValue(forKey: name)?.resume()
    }
    private func failIfRequested(_ name: String) throws {
        if failures.remove(name) != nil { throw TerminalTransportError.unreachable }
    }
    func rollback(using bootstrap: SSHConnection) async { events.record("rollback") }
    func close() async {
        events.record("close")
        finishWrite()
    }
}