feat(ios): W2 connection core — URLSessionTermTransport + SessionEngine actor
T-iOS-9: TermTransport impl (Origin single-source, 16MiB from Tunables, re-arm loop, EMSGSIZE(40)/ENOBUFS(55)→typed replayTooLarge, delegate-driven state, RFC6455 scripted test server), 11 tests T-iOS-10: SessionEngine actor (attach-first ordering, generation guard, terminal states never backoff, exactly-once digest, gate canDecide second line, defensive input validation), 20 tests Verified: 209 tests green across 5 packages; SessionCore own-src coverage 96.68%; 4/4 semantic invariants audited in source; frozen contracts untouched
This commit is contained in:
Yaojia Wang
480
ios/Packages/SessionCore/Sources/SessionCore/SessionEngine.swift
Normal file
480
ios/Packages/SessionCore/Sources/SessionCore/SessionEngine.swift
Normal file
@@ -0,0 +1,480 @@
|
||||
import Foundation
|
||||
import WireProtocol
|
||||
|
||||
/// One "open session" connection engine (frozen contract, plan §3.2 —
|
||||
/// T-iOS-10). The foreground session holds exactly one live engine; switching
|
||||
/// sessions = `close()` + a fresh engine (the server replays the ring buffer).
|
||||
///
|
||||
/// Responsibilities:
|
||||
/// - Owns the single WS connection via the injected `TermTransport` (the ONLY
|
||||
/// WS I/O boundary — `FakeTransport` and `URLSessionTermTransport` are
|
||||
/// indistinguishable here, plan §3.2).
|
||||
/// - Enforces the server's first-frame rule: `attach` is ALWAYS the first
|
||||
/// frame on every (re)connection (src/server.ts:707-711); frames submitted
|
||||
/// earlier queue behind it in order, never reordered.
|
||||
/// - Survives disconnects: `ReconnectMachine` backoff on the injected clock;
|
||||
/// `notifyForegrounded` short-circuits the retry timer (the user is looking
|
||||
/// — connect NOW) and re-sends dims (latest-writer-wins sizing, v0.4).
|
||||
/// - Treats the server as an UNTRUSTED input source (standards §4): every
|
||||
/// frame goes through `MessageCodec.decodeServer`; nil → dropped + counted,
|
||||
/// never a crash.
|
||||
/// - Terminal states never re-enter backoff: `exit` (session over, including
|
||||
/// spawn failure -1, M4) and `.replayTooLarge` (deterministic failure).
|
||||
///
|
||||
/// Sizing (documented decision, plan §7 T-iOS-10): the server spawns every PTY
|
||||
/// at 80×24 (src/server.ts:714-717) and the frozen `open` signature carries no
|
||||
/// dims, so the INITIAL resize goes out on the first `notifyForegrounded` /
|
||||
/// `send(.resize)` (SwiftTerm fires `sizeChanged` on first layout, so this is
|
||||
/// immediate in practice). Every LATER attach replays `lastKnownDims` directly
|
||||
/// after the attach frame so a reconnect reclaims full-screen.
|
||||
///
|
||||
/// Gate decisions (documented decision): the wire `approve` frame carries no
|
||||
/// epoch (the server re-parses only `mode`, src/server.ts:91-102), so the
|
||||
/// engine guards decisions with `GateTracker.canDecide` against the epoch of
|
||||
/// the most recently EMITTED gate — a decision arriving after the gate lifted
|
||||
/// (or before any gate rose) is dropped, and decisions are never queued while
|
||||
/// disconnected. The tap-vs-newest-gate epoch comparison lives in the UI layer
|
||||
/// (T-iOS-14), mirroring the web client (public/terminal-session.ts:98-102).
|
||||
public actor SessionEngine {
|
||||
// MARK: - Dependencies (immutable)
|
||||
|
||||
private let transport: any TermTransport
|
||||
private let clock: any Clock<Duration>
|
||||
private let endpoint: HostEndpoint
|
||||
/// Away-digest injection point (plan §3.2): production wraps
|
||||
/// `APIClient.events`; tests inject a fake. Keeps `TermTransport` the only
|
||||
/// WS boundary — the engine never holds an HTTP client.
|
||||
private let eventsSource: @Sendable (UUID) async throws -> [TimelineEvent]
|
||||
|
||||
// MARK: - Event outlet (single consumer: the ViewModel)
|
||||
|
||||
public nonisolated let events: AsyncStream<SessionEvent>
|
||||
private let eventsContinuation: AsyncStream<SessionEvent>.Continuation
|
||||
|
||||
// MARK: - Connection runtime state (actor-confined)
|
||||
|
||||
private var reconnect: ReconnectMachine
|
||||
private var connection: TransportConnection?
|
||||
private var connectionTask: Task<Void, Never>?
|
||||
private var pingTask: Task<Void, Never>?
|
||||
private var digestTask: Task<Void, Never>?
|
||||
/// Bumped by `close()` and every lifecycle restart: stale lifecycle tasks
|
||||
/// compare against it after each await and bow out silently.
|
||||
private var generation = 0
|
||||
/// True once attach (+dims+queue) went out on the CURRENT connection.
|
||||
private var isAttachReady = false
|
||||
private var reconnectAttempt = 0
|
||||
private var pendingOutbound: [ClientMessage] = []
|
||||
|
||||
// MARK: - Session state
|
||||
|
||||
private var currentSessionId: UUID?
|
||||
private var requestedCwd: String?
|
||||
private var lastKnownDims: (cols: Int, rows: Int)?
|
||||
private var gateTracker = GateTracker.initial
|
||||
/// Epoch of the most recently emitted (non-nil) gate — see type doc.
|
||||
private var latestEmittedGateEpoch = 0
|
||||
/// Moment the live connection was lost; consumed by the next adoption to
|
||||
/// reduce the away digest. `nil` = no away window pending.
|
||||
private var disconnectedAt: Date?
|
||||
|
||||
// MARK: - Lifecycle flags
|
||||
|
||||
private var hasOpened = false
|
||||
private var hasExited = false
|
||||
private var hasFailedTerminally = false
|
||||
private var isClosedByClient = false
|
||||
|
||||
private var isTerminal: Bool { hasExited || hasFailedTerminally }
|
||||
|
||||
// MARK: - Diagnostics (internal, test-visible; server input is untrusted)
|
||||
|
||||
private(set) var droppedServerFrameCount = 0
|
||||
private(set) var droppedDecisionCount = 0
|
||||
private(set) var droppedOutboundFrameCount = 0
|
||||
private(set) var digestFetchFailureCount = 0
|
||||
|
||||
/// `AwayDigest.reduce` truncation is a UI concern (T-iOS-14 renders the
|
||||
/// list); no Tunables constant exists for an engine-side cap, so the
|
||||
/// engine passes the no-truncation identity and keeps every post-`since`
|
||||
/// event.
|
||||
private static let digestRecentLimit = Int.max
|
||||
|
||||
// MARK: - Init (frozen signature, plan §3.2)
|
||||
|
||||
public init(
|
||||
transport: any TermTransport,
|
||||
clock: any Clock<Duration>,
|
||||
endpoint: HostEndpoint,
|
||||
reconnect: ReconnectMachine = .initial,
|
||||
eventsSource: @escaping @Sendable (UUID) async throws -> [TimelineEvent]
|
||||
) {
|
||||
self.transport = transport
|
||||
self.clock = clock
|
||||
self.endpoint = endpoint
|
||||
self.reconnect = reconnect
|
||||
self.eventsSource = eventsSource
|
||||
(events, eventsContinuation) = AsyncStream<SessionEvent>.makeStream()
|
||||
}
|
||||
|
||||
// MARK: - Public API (frozen signatures, plan §3.2)
|
||||
|
||||
/// Connect → first-frame attach → replay → live. One-shot: later calls are
|
||||
/// ignored (reconnection is the engine's own job). A `sessionId` that the
|
||||
/// server's SESSION_ID_RE would reject (non-v4) is treated as nil — the
|
||||
/// server would silently drop the attach frame and hang the connection.
|
||||
/// An invalid (relative) `cwd` is stripped for the same reason
|
||||
/// (src/protocol.ts:142-149).
|
||||
public func open(sessionId: UUID?, cwd: String?) async {
|
||||
guard !hasOpened, !isClosedByClient, !isTerminal else { return }
|
||||
hasOpened = true
|
||||
currentSessionId = sessionId.flatMap { id in
|
||||
Validation.isValidSessionId(id.uuidString) ? id : nil
|
||||
}
|
||||
requestedCwd = cwd.flatMap { Validation.isAbsoluteCwd($0) ? $0 : nil }
|
||||
emit(.connection(.connecting))
|
||||
startLifecycle()
|
||||
}
|
||||
|
||||
/// Submit one client frame. Before the current connection's attach went
|
||||
/// out (or while disconnected) frames QUEUE and flush after the next
|
||||
/// attach, order preserved — except gate decisions, which are only valid
|
||||
/// against the gate held right now and are dropped instead (type doc).
|
||||
public func send(_ message: ClientMessage) async {
|
||||
guard !isClosedByClient, !isTerminal else {
|
||||
droppedOutboundFrameCount += 1
|
||||
return
|
||||
}
|
||||
if case let .resize(cols, rows) = message {
|
||||
guard Validation.isValidResize(cols: cols, rows: rows) else {
|
||||
droppedOutboundFrameCount += 1 // server would silently discard it
|
||||
return
|
||||
}
|
||||
lastKnownDims = (cols, rows)
|
||||
}
|
||||
if isGateDecision(message) {
|
||||
guard isAttachReady, gateTracker.canDecide(epoch: latestEmittedGateEpoch) else {
|
||||
droppedDecisionCount += 1
|
||||
return
|
||||
}
|
||||
}
|
||||
guard isAttachReady, let connection else {
|
||||
pendingOutbound = pendingOutbound + [message]
|
||||
return
|
||||
}
|
||||
await transmit(message, over: connection)
|
||||
}
|
||||
|
||||
/// The scene became active: reclaim full-screen (latest-writer-wins).
|
||||
/// Connected → just resize; disconnected → connect NOW (skip the backoff
|
||||
/// timer; the ladder itself only resets on a successful connect) and the
|
||||
/// re-attach replays these dims right after the attach frame.
|
||||
public func notifyForegrounded(dims: (cols: Int, rows: Int)) async {
|
||||
guard Validation.isValidResize(cols: dims.cols, rows: dims.rows) else {
|
||||
droppedOutboundFrameCount += 1
|
||||
return
|
||||
}
|
||||
lastKnownDims = dims
|
||||
guard hasOpened, !isClosedByClient, !isTerminal else { return }
|
||||
if isAttachReady, connection != nil {
|
||||
await send(.resize(cols: dims.cols, rows: dims.rows))
|
||||
return
|
||||
}
|
||||
let (machine, effect) = reconnect.reduce(.foregrounded)
|
||||
reconnect = machine
|
||||
guard case .connectNow = effect else { return }
|
||||
startLifecycle()
|
||||
}
|
||||
|
||||
/// Explicit detach: the server-side PTY keeps running (invariant #2).
|
||||
/// Cancels every engine task, closes the transport, emits `.closed`, and
|
||||
/// finishes the events stream. Idempotent.
|
||||
public func close() async {
|
||||
guard !isClosedByClient else { return }
|
||||
isClosedByClient = true
|
||||
generation += 1 // invalidate in-flight lifecycle resumptions
|
||||
connectionTask?.cancel()
|
||||
digestTask?.cancel()
|
||||
stopPing()
|
||||
if let connection {
|
||||
self.connection = nil
|
||||
await connection.close()
|
||||
}
|
||||
isAttachReady = false
|
||||
emit(.connection(.closed))
|
||||
eventsContinuation.finish()
|
||||
}
|
||||
|
||||
// MARK: - Lifecycle loop
|
||||
|
||||
/// Starts (or restarts) the connect→attach→pump→backoff loop under a fresh
|
||||
/// generation; the previous task is cancelled and neutered by the bump.
|
||||
private func startLifecycle() {
|
||||
generation += 1
|
||||
let gen = generation
|
||||
connectionTask?.cancel()
|
||||
connectionTask = Task { await self.runLifecycle(generation: gen) }
|
||||
}
|
||||
|
||||
private func runLifecycle(generation gen: Int) async {
|
||||
while true {
|
||||
let established: (connection: TransportConnection, pinger: (any ConnectionPinger)?)
|
||||
do {
|
||||
established = try await establishConnection()
|
||||
} catch {
|
||||
guard isCurrent(gen) else { return }
|
||||
guard resolveDisconnect(error: error) else { return }
|
||||
guard await backoffThenRetry(generation: gen) else { return }
|
||||
continue
|
||||
}
|
||||
guard isCurrent(gen) else {
|
||||
await established.connection.close()
|
||||
return
|
||||
}
|
||||
connection = established.connection
|
||||
do {
|
||||
try await performAttachHandshake(on: established.connection)
|
||||
} catch {
|
||||
guard isCurrent(gen) else { return }
|
||||
teardownConnection()
|
||||
guard resolveDisconnect(error: error) else { return }
|
||||
guard await backoffThenRetry(generation: gen) else { return }
|
||||
continue
|
||||
}
|
||||
guard isCurrent(gen) else { return }
|
||||
becomeConnected(established, generation: gen)
|
||||
let streamError = await pumpFrames(from: established.connection, generation: gen)
|
||||
guard isCurrent(gen) else { return }
|
||||
teardownConnection()
|
||||
guard resolveDisconnect(error: streamError) else { return }
|
||||
guard await backoffThenRetry(generation: gen) else { return }
|
||||
}
|
||||
}
|
||||
|
||||
/// Connects through the internal ping hook when the transport supports it
|
||||
/// (URLSessionTermTransport does; FakeTransport does not — T-iOS-9).
|
||||
private func establishConnection() async throws
|
||||
-> (connection: TransportConnection, pinger: (any ConnectionPinger)?) {
|
||||
if let pingable = transport as? any PingableTermTransport {
|
||||
let (connection, pinger) = try await pingable.connectPingable(to: endpoint)
|
||||
return (connection, pinger)
|
||||
}
|
||||
return (try await transport.connect(to: endpoint), nil)
|
||||
}
|
||||
|
||||
/// First-frame rule: attach, then last-known dims (a shared PTY is sized
|
||||
/// by the latest writer), then the queued frames in submission order.
|
||||
/// A frame whose send throws stays queued for the next attach.
|
||||
private func performAttachHandshake(on connection: TransportConnection) async throws {
|
||||
let cwd = currentSessionId == nil ? requestedCwd : nil
|
||||
try await connection.send(MessageCodec.encode(.attach(sessionId: currentSessionId, cwd: cwd)))
|
||||
if let dims = lastKnownDims {
|
||||
try await connection.send(MessageCodec.encode(.resize(cols: dims.cols, rows: dims.rows)))
|
||||
}
|
||||
while let queued = pendingOutbound.first {
|
||||
try await connection.send(MessageCodec.encode(queued))
|
||||
pendingOutbound = Array(pendingOutbound.dropFirst())
|
||||
}
|
||||
isAttachReady = true
|
||||
}
|
||||
|
||||
private func becomeConnected(
|
||||
_ established: (connection: TransportConnection, pinger: (any ConnectionPinger)?),
|
||||
generation gen: Int
|
||||
) {
|
||||
(reconnect, _) = reconnect.reduce(.connected) // back-off ladder resets
|
||||
reconnectAttempt = 0
|
||||
if let pinger = established.pinger {
|
||||
startPing(pinger, connection: established.connection, generation: gen)
|
||||
}
|
||||
emit(.connection(.connected))
|
||||
}
|
||||
|
||||
/// Delivers server frames in arrival order until the stream ends.
|
||||
/// Returns the stream error (nil = clean finish).
|
||||
private func pumpFrames(
|
||||
from connection: TransportConnection, generation gen: Int
|
||||
) async -> (any Error)? {
|
||||
do {
|
||||
for try await frame in connection.frames {
|
||||
guard isCurrent(gen) else { return nil }
|
||||
handleFrame(frame)
|
||||
}
|
||||
return nil
|
||||
} catch {
|
||||
return error
|
||||
}
|
||||
}
|
||||
|
||||
private func teardownConnection() {
|
||||
connection = nil
|
||||
isAttachReady = false
|
||||
stopPing()
|
||||
// The away window opens at the FIRST loss; later failed retries keep it.
|
||||
disconnectedAt = disconnectedAt ?? Date()
|
||||
}
|
||||
|
||||
/// Classifies how a connection attempt/stream ended. Returns true when the
|
||||
/// lifecycle should enter backoff; terminal classifications emit their
|
||||
/// final event and finish the events stream.
|
||||
private func resolveDisconnect(error: (any Error)?) -> Bool {
|
||||
if isClosedByClient { return false }
|
||||
if hasExited {
|
||||
emit(.connection(.closed)) // session over — deliberate conclusion
|
||||
eventsContinuation.finish()
|
||||
return false
|
||||
}
|
||||
if let error, isReplayTooLarge(error) {
|
||||
hasFailedTerminally = true
|
||||
emit(.connection(.failed(.replayTooLarge))) // NEVER backoff (plan §1)
|
||||
eventsContinuation.finish()
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
/// One backoff rung: schedule per the reducer, announce, sleep on the
|
||||
/// injected clock, then ask for connectNow. False = stop the lifecycle
|
||||
/// (cancelled by close()/foregrounded restart, or superseded).
|
||||
private func backoffThenRetry(generation gen: Int) async -> Bool {
|
||||
let (afterDisconnect, effect) = reconnect.reduce(.disconnected)
|
||||
reconnect = afterDisconnect
|
||||
guard case .scheduleRetry(let delay) = effect else { return false }
|
||||
reconnectAttempt += 1
|
||||
emit(.connection(.reconnecting(attempt: reconnectAttempt, next: delay)))
|
||||
do {
|
||||
try await clock.sleep(for: delay, tolerance: nil)
|
||||
} catch {
|
||||
return false // CancellationError: close() or a lifecycle restart
|
||||
}
|
||||
guard isCurrent(gen) else { return false }
|
||||
let (afterTimer, fireEffect) = reconnect.reduce(.retryTimerFired)
|
||||
reconnect = afterTimer
|
||||
guard case .connectNow = fireEffect else { return false }
|
||||
return true
|
||||
}
|
||||
|
||||
// MARK: - Inbound frames (untrusted)
|
||||
|
||||
/// Every frame goes through `MessageCodec.decodeServer`; nil → dropped and
|
||||
/// counted, never a crash (plan §7 T-iOS-10 安全注).
|
||||
private func handleFrame(_ frame: String) {
|
||||
guard let message = MessageCodec.decodeServer(frame) else {
|
||||
droppedServerFrameCount += 1
|
||||
return
|
||||
}
|
||||
switch message {
|
||||
case .attached(let sessionId):
|
||||
currentSessionId = sessionId // ALWAYS adopt the server-issued id
|
||||
emit(.adopted(sessionId: sessionId))
|
||||
scheduleAwayDigestIfNeeded(for: sessionId)
|
||||
case .output(let data):
|
||||
emit(.output(data))
|
||||
case .exit(let code, let reason):
|
||||
hasExited = true // terminal: no reconnect once the stream ends
|
||||
emit(.exited(code: code, reason: reason))
|
||||
case .status(_, let detail, let pending, let gate):
|
||||
applyGateFrame(pending: pending, gate: gate, detail: detail)
|
||||
case .telemetry(let telemetry):
|
||||
emit(.telemetry(telemetry))
|
||||
}
|
||||
}
|
||||
|
||||
/// Folds a status frame into `GateTracker`; emits `.gate` only on change
|
||||
/// (rising edge mints a new epoch, falling edge emits nil — T-iOS-6).
|
||||
private func applyGateFrame(pending: Bool, gate: GateKind?, detail: String?) {
|
||||
let next = gateTracker.reduce(pending: pending, gate: gate, detail: detail)
|
||||
defer { gateTracker = next }
|
||||
guard next.current != gateTracker.current else { return }
|
||||
if let held = next.current { latestEmittedGateEpoch = held.epoch }
|
||||
emit(.gate(next.current))
|
||||
}
|
||||
|
||||
// MARK: - Away digest
|
||||
|
||||
/// Consumes the pending away window (set at disconnect) exactly once per
|
||||
/// reconnect: fetch the timeline off the frame loop and emit ONE digest.
|
||||
/// The initial attach has no away window, so it never digests.
|
||||
private func scheduleAwayDigestIfNeeded(for sessionId: UUID) {
|
||||
guard let since = disconnectedAt else { return }
|
||||
disconnectedAt = nil
|
||||
let source = eventsSource
|
||||
digestTask = Task { [weak self] in
|
||||
do {
|
||||
let timeline = try await source(sessionId)
|
||||
let digest = AwayDigest.reduce(
|
||||
events: timeline, since: since, limit: Self.digestRecentLimit
|
||||
)
|
||||
await self?.emitDigest(digest)
|
||||
} catch {
|
||||
await self?.recordDigestFetchFailure()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func emitDigest(_ digest: AwayDigest) {
|
||||
emit(.digest(digest))
|
||||
}
|
||||
|
||||
private func recordDigestFetchFailure() {
|
||||
digestFetchFailureCount += 1 // fetch failure = no digest, never a crash
|
||||
}
|
||||
|
||||
// MARK: - Keep-alive (T-iOS-9 internal hook + T-iOS-5 scheduler)
|
||||
|
||||
/// `URLSessionWebSocketTask` has no automatic ping (plan §1): run the
|
||||
/// PingScheduler against this connection's pinger; on `connectionLost`,
|
||||
/// close the half-dead connection so the pump ends and the normal backoff
|
||||
/// reconnect takes over — never "looks connected but dead".
|
||||
private func startPing(
|
||||
_ pinger: any ConnectionPinger, connection: TransportConnection, generation gen: Int
|
||||
) {
|
||||
let scheduler = PingScheduler()
|
||||
let clock = self.clock
|
||||
pingTask = Task { [weak self] in
|
||||
let outcome = await scheduler.run(clock: clock) { await pinger.ping() }
|
||||
guard outcome == .connectionLost else { return }
|
||||
await self?.handlePingLoss(connection: connection, generation: gen)
|
||||
}
|
||||
}
|
||||
|
||||
private func handlePingLoss(connection: TransportConnection, generation gen: Int) async {
|
||||
guard isCurrent(gen), !isClosedByClient else { return }
|
||||
await connection.close() // finishes frames → pump ends → backoff path
|
||||
}
|
||||
|
||||
private func stopPing() {
|
||||
pingTask?.cancel()
|
||||
pingTask = nil
|
||||
}
|
||||
|
||||
// MARK: - Helpers
|
||||
|
||||
private func transmit(_ message: ClientMessage, over connection: TransportConnection) async {
|
||||
do {
|
||||
try await connection.send(MessageCodec.encode(message))
|
||||
} catch {
|
||||
// The connection is dying; the stream-end path owns the reconnect.
|
||||
droppedOutboundFrameCount += 1
|
||||
}
|
||||
}
|
||||
|
||||
private func isGateDecision(_ message: ClientMessage) -> Bool {
|
||||
switch message {
|
||||
case .approve, .reject: return true
|
||||
case .attach, .input, .resize: return false
|
||||
}
|
||||
}
|
||||
|
||||
private func isCurrent(_ gen: Int) -> Bool {
|
||||
generation == gen
|
||||
}
|
||||
|
||||
private func isReplayTooLarge(_ error: any Error) -> Bool {
|
||||
(error as? TermTransportError) == .replayTooLarge
|
||||
}
|
||||
|
||||
private func emit(_ event: SessionEvent) {
|
||||
eventsContinuation.yield(event)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import Foundation
|
||||
import WireProtocol
|
||||
|
||||
/// Events `SessionEngine` publishes to the UI (frozen contract, plan §3.2).
|
||||
/// The engine's `events` stream is the UI's ONLY outlet: a single consumer
|
||||
/// (the ViewModel, T-iOS-11) iterates it on the MainActor and feeds SwiftTerm,
|
||||
/// banners, and gate UI from it.
|
||||
public enum SessionEvent: Sendable, Equatable {
|
||||
/// Connection lifecycle. `.failed` is a NON-retryable terminal state — the
|
||||
/// engine will not reconnect (plan §3.2: `.replayTooLarge` must never enter
|
||||
/// the backoff loop).
|
||||
case connection(ConnectionState)
|
||||
/// The server confirmed the attach. ALWAYS adopt the server-issued id —
|
||||
/// attaching with an unknown UUID yields a brand-new session and a NEW id
|
||||
/// (src/session/manager.ts:166-173).
|
||||
case adopted(sessionId: UUID)
|
||||
/// Opaque terminal bytes; ring-buffer replay and live stream share this
|
||||
/// shape. Feed to SwiftTerm verbatim (the @MainActor hop is the VM's job).
|
||||
case output(String)
|
||||
/// The shell exited. `code == WireConstants.spawnFailedExitCode` (-1) means
|
||||
/// the spawn never succeeded (M4). Either way the session is over — the
|
||||
/// engine stops reconnecting and concludes with `.connection(.closed)`.
|
||||
case exited(code: Int, reason: String?)
|
||||
/// The held permission gate changed; `nil` = gate lifted (GateTracker,
|
||||
/// T-iOS-6). Carries the epoch the UI must hand back when deciding.
|
||||
case gate(GateState?)
|
||||
/// Latest statusLine telemetry broadcast (B2).
|
||||
case telemetry(StatusTelemetry)
|
||||
/// "What happened while I was away" — emitted exactly ONCE after each
|
||||
/// completed reconnect, reduced from the injected `eventsSource` over the
|
||||
/// events since the disconnect moment (plan §3.2).
|
||||
case digest(AwayDigest)
|
||||
}
|
||||
|
||||
/// Connection lifecycle states (plan §3.2 comment on `SessionEvent.connection`).
|
||||
public enum ConnectionState: Sendable, Equatable {
|
||||
/// `open()` was called; the first connect is being established.
|
||||
case connecting
|
||||
/// Live: transport up, attach handshake done, frames flowing.
|
||||
case connected
|
||||
/// The transport dropped; retry number `attempt` fires after `next`.
|
||||
/// Ladder mirrors the web client via `ReconnectMachine` (1s→…→30s cap).
|
||||
case reconnecting(attempt: Int, next: Duration)
|
||||
/// Ended deliberately: client `close()` (detach — the server-side PTY keeps
|
||||
/// running) or the session exited.
|
||||
case closed
|
||||
/// NON-retryable terminal failure — the engine stopped for good; the UI
|
||||
/// shows actionable copy instead of a spinner.
|
||||
case failed(FailureReason)
|
||||
}
|
||||
|
||||
/// Why the engine gave up (plan §3.2 "不可重试终态").
|
||||
public enum FailureReason: Sendable, Equatable {
|
||||
/// The single-frame ring-buffer replay exceeded
|
||||
/// `Tunables.maxWSMessageBytes` (EMSGSIZE(40), T-iOS-2 spike): reconnecting
|
||||
/// would deterministically hit the same wall forever, so the engine never
|
||||
/// enters backoff. UI copy: lower the server's SCROLLBACK_BYTES or raise
|
||||
/// the client cap (plan §3.2.1 coupling warning).
|
||||
case replayTooLarge
|
||||
}
|
||||
@@ -0,0 +1,381 @@
|
||||
import Darwin
|
||||
import Foundation
|
||||
import WireProtocol
|
||||
|
||||
// MARK: - Typed transport errors
|
||||
|
||||
/// Transport-level errors with contract meaning for the engine (T-iOS-9).
|
||||
///
|
||||
/// Deliberately a SessionCore type, NOT WireProtocol: the frozen contract only
|
||||
/// says "stream throw = transport error" (`TransportConnection.frames` doc);
|
||||
/// WHICH error is a SessionCore concern — `SessionEngine` (T-iOS-10) lives in
|
||||
/// the same module and classifies these.
|
||||
///
|
||||
/// Connect-time failures are NOT wrapped: `connect(to:)` rethrows the
|
||||
/// underlying URLSession error verbatim, because the pairing probe
|
||||
/// (`PairingError.classify`, T-iOS-8) inspects POSIX/NSURLError codes to map
|
||||
/// `localNetworkDenied` / `atsBlocked` / `tlsFailure` — wrapping would break
|
||||
/// that taxonomy.
|
||||
public enum TermTransportError: Error, Equatable, Sendable {
|
||||
/// `receive()` failed with NSPOSIXErrorDomain EMSGSIZE(40) "Message too
|
||||
/// long" (ENOBUFS(55) accepted as fallback — T-iOS-2 spike measured; it is
|
||||
/// NOT a 1009 clean close on iOS): the single-frame ring-buffer replay
|
||||
/// exceeded `maximumMessageSize`. NON-RETRYABLE — the engine must surface
|
||||
/// `connection(.failed(.replayTooLarge))` and never feed this into the
|
||||
/// backoff loop (plan §1 / §3.2.1 coupling warning), otherwise the retry
|
||||
/// is deterministic and infinite.
|
||||
case replayTooLarge
|
||||
/// `send` on a connection that already terminated (client `close()`,
|
||||
/// server close frame, or transport error). Mirrors
|
||||
/// `FakeTransportError.sendAfterClose` (TestSupport).
|
||||
case sendAfterClose
|
||||
}
|
||||
|
||||
// MARK: - Internal ping extension point (T-iOS-9 design decision)
|
||||
|
||||
/// INTERNAL, non-frozen ping hook.
|
||||
///
|
||||
/// Why this shape: the frozen `TransportConnection` (WireProtocol §3.1) is
|
||||
/// exactly `{frames, send, close}` — no ping — and the frozen `TermTransport`
|
||||
/// protocol must not change. But `PingScheduler` (T-iOS-5) needs a per-
|
||||
/// connection `sendPing` the engine can call. Ruled-out alternatives:
|
||||
/// a ping closure on `TransportConnection` (frozen-contract change → T-iOS-3),
|
||||
/// and a transport-level registry keyed by connection (streams aren't
|
||||
/// hashable; leak-prone). Chosen: transports that can ping conform to this
|
||||
/// SessionCore-internal protocol; the engine downcasts its
|
||||
/// `any TermTransport` and, when supported, opens the connection through
|
||||
/// `connectPingable(to:)`, receiving the frozen `TransportConnection` PLUS a
|
||||
/// per-connection `ConnectionPinger`. `FakeTransport` (TestSupport) does not
|
||||
/// conform → engine tests drive `PingScheduler` through its injected closure
|
||||
/// instead. No frozen contract is touched.
|
||||
protocol PingableTermTransport: TermTransport {
|
||||
/// Like `connect(to:)`, but also hands back the ping hook for that same
|
||||
/// connection (for `PingScheduler.run(clock:sendPing:)` wiring).
|
||||
func connectPingable(to endpoint: HostEndpoint) async throws
|
||||
-> (connection: TransportConnection, pinger: any ConnectionPinger)
|
||||
}
|
||||
|
||||
/// One live connection's ping capability (see `PingableTermTransport`).
|
||||
protocol ConnectionPinger: Sendable {
|
||||
/// Send one WS ping; `true` iff the pong arrived in time (`false` = miss,
|
||||
/// counted by `PingScheduler` against `Tunables.pongMissLimit`).
|
||||
func ping() async -> Bool
|
||||
}
|
||||
|
||||
// MARK: - Transport
|
||||
|
||||
/// Production `TermTransport` over `URLSessionWebSocketTask` (T-iOS-9).
|
||||
///
|
||||
/// Known platform pitfalls this type exists to fence (plan §1, spike-verified):
|
||||
/// - Origin: stamped from `HostEndpoint.originHeader` ONLY (single source,
|
||||
/// §5.1 — any string-assembled origin in this file is a review CRITICAL).
|
||||
/// - `maximumMessageSize` defaults to 1 MiB; ring-buffer replay is one full
|
||||
/// snapshot frame → every task gets `Tunables.maxWSMessageBytes` (16 MiB).
|
||||
/// - `receive()` delivers ONE message per call → the loop re-arms forever.
|
||||
/// - No automatic ping → `ConnectionPinger` (driven by `PingScheduler`).
|
||||
public struct URLSessionTermTransport: TermTransport {
|
||||
/// `URLSessionWebSocketTask.maximumMessageSize` applied to every task.
|
||||
private let maxMessageBytes: Int
|
||||
|
||||
public init() {
|
||||
self.init(maxMessageBytes: Tunables.maxWSMessageBytes)
|
||||
}
|
||||
|
||||
/// Internal TEST seam: a shrunken cap makes the oversize→EMSGSIZE path
|
||||
/// deterministic without 16 MiB fixtures. Production code paths always go
|
||||
/// through `init()` and the frozen `Tunables.maxWSMessageBytes`.
|
||||
init(maxMessageBytes: Int) {
|
||||
self.maxMessageBytes = maxMessageBytes
|
||||
}
|
||||
|
||||
public func connect(to endpoint: HostEndpoint) async throws -> TransportConnection {
|
||||
try await openConnection(to: endpoint).transportConnection()
|
||||
}
|
||||
|
||||
/// Internal: concrete-typed connect (tests assert task configuration;
|
||||
/// `connectPingable` builds on it).
|
||||
func openConnection(to endpoint: HostEndpoint) async throws -> WSConnection {
|
||||
try await WSConnection.open(endpoint: endpoint, maxMessageBytes: maxMessageBytes)
|
||||
}
|
||||
}
|
||||
|
||||
extension URLSessionTermTransport: PingableTermTransport {
|
||||
func connectPingable(to endpoint: HostEndpoint) async throws
|
||||
-> (connection: TransportConnection, pinger: any ConnectionPinger) {
|
||||
let connection = try await openConnection(to: endpoint)
|
||||
return (connection.transportConnection(), connection)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Connection
|
||||
|
||||
/// One live WS connection: owns the mutable runtime handles (session, task,
|
||||
/// receive loop) and hides them behind the frozen `TransportConnection`
|
||||
/// capability struct.
|
||||
///
|
||||
/// Concurrency note: this is an NSLock-disciplined `@unchecked Sendable`
|
||||
/// class, not an actor, because `URLSessionWebSocketDelegate` callbacks are
|
||||
/// synchronous and ordering-sensitive (didOpen must resolve the handshake
|
||||
/// continuation before didComplete can) — hopping them through actor `Task`s
|
||||
/// would lose that ordering. Every mutable field is only touched under
|
||||
/// `lock`; the URLSession handles themselves are thread-safe.
|
||||
final class WSConnection: NSObject, @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var handshakeContinuation: CheckedContinuation<Void, any Error>?
|
||||
private var isClosedByClient = false
|
||||
private var didObserveServerCloseFrame = false
|
||||
private var isTerminated = false
|
||||
private var droppedBinaryFrameCount = 0
|
||||
|
||||
/// Set exactly once in `configure` (before any concurrent access) — IUO
|
||||
/// because `URLSession(configuration:delegate:…)` needs `self` post-init.
|
||||
private(set) var task: URLSessionWebSocketTask!
|
||||
private var session: URLSession!
|
||||
private var receiveLoop: Task<Void, Never>?
|
||||
private let frames: AsyncThrowingStream<String, any Error>
|
||||
private let framesContinuation: AsyncThrowingStream<String, any Error>.Continuation
|
||||
|
||||
private override init() {
|
||||
(frames, framesContinuation) = AsyncThrowingStream<String, any Error>.makeStream()
|
||||
super.init()
|
||||
}
|
||||
|
||||
/// Connect: build session+task, resume, await the delegate-driven
|
||||
/// handshake (didOpen / didCompleteWithError — no receive-error guessing),
|
||||
/// then start the re-arming receive loop.
|
||||
static func open(endpoint: HostEndpoint, maxMessageBytes: Int) async throws -> WSConnection {
|
||||
let connection = WSConnection()
|
||||
connection.configure(endpoint: endpoint, maxMessageBytes: maxMessageBytes)
|
||||
try await connection.performHandshake()
|
||||
connection.startReceiveLoop()
|
||||
return connection
|
||||
}
|
||||
|
||||
/// Wrap as the frozen boundary struct. Called exactly once per connection
|
||||
/// (`AsyncThrowingStream` is single-consumer).
|
||||
func transportConnection() -> TransportConnection {
|
||||
TransportConnection(
|
||||
frames: frames,
|
||||
send: { [self] in try await send($0) },
|
||||
close: { [self] in await close() }
|
||||
)
|
||||
}
|
||||
|
||||
// MARK: Setup
|
||||
|
||||
private func configure(endpoint: HostEndpoint, maxMessageBytes: Int) {
|
||||
// Origin: SINGLE source of truth = endpoint.originHeader (plan §5.1).
|
||||
var request = URLRequest(url: endpoint.wsURL)
|
||||
request.setValue(endpoint.originHeader, forHTTPHeaderField: "Origin")
|
||||
let session = URLSession(configuration: .ephemeral, delegate: self, delegateQueue: nil)
|
||||
let task = session.webSocketTask(with: request)
|
||||
task.maximumMessageSize = maxMessageBytes
|
||||
self.session = session
|
||||
self.task = task
|
||||
}
|
||||
|
||||
private func performHandshake() async throws {
|
||||
try await withCheckedThrowingContinuation {
|
||||
(continuation: CheckedContinuation<Void, any Error>) in
|
||||
lock.withLock { handshakeContinuation = continuation }
|
||||
task.resume()
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: Frozen-contract operations
|
||||
|
||||
/// Send one client text frame; explicit `.sendAfterClose` once the
|
||||
/// connection terminated (client close, server close, or error).
|
||||
func send(_ frame: String) async throws {
|
||||
let isSendable = lock.withLock { !isClosedByClient && !isTerminated }
|
||||
guard isSendable else { throw TermTransportError.sendAfterClose }
|
||||
try await task.send(.string(frame))
|
||||
}
|
||||
|
||||
/// Client detach: the server-side PTY keeps running (frozen-contract doc).
|
||||
/// Idempotent; finishes the frames stream cleanly.
|
||||
func close() async {
|
||||
lock.withLock {
|
||||
isClosedByClient = true
|
||||
isTerminated = true
|
||||
}
|
||||
task.cancel(with: .normalClosure, reason: nil)
|
||||
framesContinuation.finish()
|
||||
session.finishTasksAndInvalidate()
|
||||
}
|
||||
|
||||
// MARK: Receive loop
|
||||
|
||||
/// `URLSessionWebSocketTask.receive()` delivers exactly ONE message per
|
||||
/// call (plan §1 pitfall) — forgetting to re-arm silently stalls the
|
||||
/// stream, so this loops until the task terminates. The Task holds `self`
|
||||
/// strongly only while running; every termination path also invalidates
|
||||
/// the session (which drops its strong delegate reference to `self`).
|
||||
private func startReceiveLoop() {
|
||||
receiveLoop = Task { [self] in
|
||||
while true {
|
||||
do {
|
||||
deliver(try await task.receive())
|
||||
} catch {
|
||||
finishFrames(afterReceiveFailure: error)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func deliver(_ message: URLSessionWebSocketTask.Message) {
|
||||
switch message {
|
||||
case .string(let text):
|
||||
framesContinuation.yield(text)
|
||||
case .data:
|
||||
recordDroppedBinaryFrame()
|
||||
@unknown default:
|
||||
recordDroppedBinaryFrame()
|
||||
}
|
||||
}
|
||||
|
||||
/// The server only ever sends text frames (src/protocol.ts), but it is an
|
||||
/// untrusted input source (standards §4): binary/unknown frames are
|
||||
/// dropped and counted — never fatal, receiving continues.
|
||||
private func recordDroppedBinaryFrame() {
|
||||
lock.withLock { droppedBinaryFrameCount += 1 }
|
||||
}
|
||||
|
||||
/// Dropped-frame counter (test/diagnostic visibility).
|
||||
var droppedBinaryFrames: Int {
|
||||
lock.withLock { droppedBinaryFrameCount }
|
||||
}
|
||||
|
||||
// MARK: Termination classification
|
||||
|
||||
/// Classify how the connection ended (finish/throw are idempotent — the
|
||||
/// first termination path wins, later calls are no-ops):
|
||||
/// - clean (client `close()`, server close frame observed by the delegate,
|
||||
/// or a processed close frame per `task.closeCode`) → stream FINISHES;
|
||||
/// - EMSGSIZE(40)/ENOBUFS(55) → typed `.replayTooLarge` (non-retryable);
|
||||
/// - anything else → stream THROWS the underlying error (retryable class).
|
||||
private func finishFrames(afterReceiveFailure error: any Error) {
|
||||
let wasClosedCleanly = lock.withLock {
|
||||
isClosedByClient || didObserveServerCloseFrame
|
||||
} || task.closeCode != .invalid
|
||||
markTerminated()
|
||||
if wasClosedCleanly {
|
||||
framesContinuation.finish()
|
||||
} else if Self.isMessageTooLong(error) {
|
||||
framesContinuation.finish(throwing: TermTransportError.replayTooLarge)
|
||||
} else {
|
||||
framesContinuation.finish(throwing: error)
|
||||
}
|
||||
session.finishTasksAndInvalidate()
|
||||
}
|
||||
|
||||
private func markTerminated() {
|
||||
lock.withLock { isTerminated = true }
|
||||
}
|
||||
|
||||
private func takeHandshakeContinuation() -> CheckedContinuation<Void, any Error>? {
|
||||
lock.withLock {
|
||||
defer { handshakeContinuation = nil }
|
||||
return handshakeContinuation
|
||||
}
|
||||
}
|
||||
|
||||
/// T-iOS-2 spike (measured on a real server): exceeding
|
||||
/// `maximumMessageSize` fails `receive()` with NSPOSIXErrorDomain
|
||||
/// EMSGSIZE(40) "Message too long" — not a 1009 close. ENOBUFS(55) is the
|
||||
/// documented fallback for the same classification (plan §7 T-iOS-9
|
||||
/// errata). Checked at the top level and one underlying level.
|
||||
private static func isMessageTooLong(_ error: any Error) -> Bool {
|
||||
func matches(_ ns: NSError) -> Bool {
|
||||
ns.domain == NSPOSIXErrorDomain
|
||||
&& (ns.code == Int(EMSGSIZE) || ns.code == Int(ENOBUFS))
|
||||
}
|
||||
let ns = error as NSError
|
||||
if matches(ns) { return true }
|
||||
guard let underlying = ns.userInfo[NSUnderlyingErrorKey] as? NSError else {
|
||||
return false
|
||||
}
|
||||
return matches(underlying)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - URLSession delegate (drives state — not receive-error guessing)
|
||||
|
||||
extension WSConnection: URLSessionWebSocketDelegate {
|
||||
func urlSession(
|
||||
_ session: URLSession,
|
||||
webSocketTask: URLSessionWebSocketTask,
|
||||
didOpenWithProtocol protocol: String?
|
||||
) {
|
||||
takeHandshakeContinuation()?.resume()
|
||||
}
|
||||
|
||||
/// Server close frame processed → CLEAN close: the stream FINISHES
|
||||
/// (distinguishable from the transport-error THROW path).
|
||||
func urlSession(
|
||||
_ session: URLSession,
|
||||
webSocketTask: URLSessionWebSocketTask,
|
||||
didCloseWith closeCode: URLSessionWebSocketTask.CloseCode,
|
||||
reason: Data?
|
||||
) {
|
||||
lock.withLock {
|
||||
didObserveServerCloseFrame = true
|
||||
isTerminated = true
|
||||
}
|
||||
framesContinuation.finish()
|
||||
session.finishTasksAndInvalidate()
|
||||
}
|
||||
|
||||
func urlSession(
|
||||
_ session: URLSession,
|
||||
task: URLSessionTask,
|
||||
didCompleteWithError error: (any Error)?
|
||||
) {
|
||||
if let handshake = takeHandshakeContinuation() {
|
||||
// Handshake never opened. Rethrow the UNDERLYING error verbatim —
|
||||
// PairingError.classify (T-iOS-8) depends on its POSIX/NSURLError
|
||||
// identity (see TermTransportError doc).
|
||||
handshake.resume(throwing: error ?? URLError(.cannotConnectToHost))
|
||||
session.finishTasksAndInvalidate()
|
||||
return
|
||||
}
|
||||
// Post-open completion: the receive loop normally observes the failure
|
||||
// first; this is the belt-and-braces path (idempotent finish).
|
||||
guard let error else {
|
||||
markTerminated()
|
||||
framesContinuation.finish()
|
||||
session.finishTasksAndInvalidate()
|
||||
return
|
||||
}
|
||||
finishFrames(afterReceiveFailure: error)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Ping (internal hook implementation)
|
||||
|
||||
extension WSConnection: ConnectionPinger {
|
||||
/// One WS ping; `true` iff the pong arrived before the deadline.
|
||||
///
|
||||
/// Deadline REUSES `Tunables.pingInterval` (no new constant — adding one
|
||||
/// would be a T-iOS-3 contract change): a pong that has not arrived by the
|
||||
/// time the NEXT ping would be due is by definition a miss;
|
||||
/// `PingScheduler` tolerates `Tunables.pongMissLimit - 1` consecutive
|
||||
/// misses before declaring the connection lost. Built on a first-value
|
||||
/// AsyncStream race (not a task group) so a black-holed
|
||||
/// `pongReceiveHandler` can never hang the caller.
|
||||
func ping() async -> Bool {
|
||||
let (verdicts, continuation) = AsyncStream<Bool>.makeStream()
|
||||
task.sendPing { error in continuation.yield(error == nil) }
|
||||
let deadline = Task {
|
||||
try? await Task.sleep(for: Tunables.pingInterval)
|
||||
continuation.yield(false)
|
||||
}
|
||||
defer {
|
||||
deadline.cancel()
|
||||
continuation.finish()
|
||||
}
|
||||
var iterator = verdicts.makeAsyncIterator()
|
||||
return await iterator.next() ?? false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,353 @@
|
||||
#if os(macOS)
|
||||
import CryptoKit
|
||||
import Foundation
|
||||
import Network
|
||||
|
||||
/// T-iOS-9 test infra (owned by `URLSessionTermTransportTests`): a minimal
|
||||
/// SCRIPTED WebSocket server over raw TCP on 127.0.0.1 with an ephemeral port.
|
||||
/// macOS-only by design — the SessionCore suite runs via `swift test` on
|
||||
/// macOS (plan §7 T-iOS-9); real-Node-server coverage is T-iOS-16.
|
||||
///
|
||||
/// Deliberately raw TCP + hand-rolled RFC 6455 framing instead of
|
||||
/// `NWProtocolWebSocket`, because the tests must:
|
||||
/// (a) read the HTTP upgrade request verbatim (Origin header assertion —
|
||||
/// `NWProtocolWebSocket` hides the request headers),
|
||||
/// (b) inject binary frames, oversize frames, close frames, and abrupt TCP
|
||||
/// termination — exactly the failure modes T-iOS-9 must classify.
|
||||
///
|
||||
/// Test infra is inherently stateful I/O plumbing; all mutable state is
|
||||
/// NSLock-protected and the type is `@unchecked Sendable` under that
|
||||
/// discipline (mirrors the production adapter's documented approach).
|
||||
final class ScriptedWSServer: @unchecked Sendable {
|
||||
|
||||
/// One recorded HTTP upgrade request.
|
||||
struct Upgrade: Sendable {
|
||||
/// e.g. `GET /term HTTP/1.1`.
|
||||
let requestLine: String
|
||||
/// Header names lowercased; values trimmed.
|
||||
let headers: [String: String]
|
||||
}
|
||||
|
||||
/// RFC 6455 wire constants (named — no magic numbers).
|
||||
private enum Wire {
|
||||
static let finBit: UInt8 = 0x80
|
||||
static let opcodeMask: UInt8 = 0x0F
|
||||
static let maskBit: UInt8 = 0x80
|
||||
static let payloadLenMask: UInt8 = 0x7F
|
||||
static let len16Marker = 126
|
||||
static let len64Marker = 127
|
||||
static let maxTinyPayload = 125
|
||||
static let maskKeyLength = 4
|
||||
static let opcodeText: UInt8 = 0x1
|
||||
static let opcodeBinary: UInt8 = 0x2
|
||||
static let opcodeClose: UInt8 = 0x8
|
||||
static let opcodePing: UInt8 = 0x9
|
||||
static let opcodePong: UInt8 = 0xA
|
||||
/// RFC 6455 §1.3 Sec-WebSocket-Accept magic GUID.
|
||||
static let acceptGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
|
||||
static let headerTerminator: [UInt8] = [0x0D, 0x0A, 0x0D, 0x0A] // \r\n\r\n
|
||||
static let receiveChunkLimit = 1 << 16
|
||||
}
|
||||
|
||||
/// Upgrade requests, one per accepted client handshake (buffered stream —
|
||||
/// yields before consumption are never lost). Single consumer per test.
|
||||
let upgrades: AsyncStream<Upgrade>
|
||||
/// Text frames received FROM the client, unmasked, in arrival order.
|
||||
let clientTextFrames: AsyncStream<String>
|
||||
|
||||
private let upgradesContinuation: AsyncStream<Upgrade>.Continuation
|
||||
private let clientTextContinuation: AsyncStream<String>.Continuation
|
||||
private let queue = DispatchQueue(label: "ScriptedWSServer")
|
||||
private let lock = NSLock()
|
||||
private var listener: NWListener?
|
||||
private var connection: NWConnection?
|
||||
private var rxBuffer = [UInt8]()
|
||||
private var isHandshakeDone = false
|
||||
private var startContinuation: CheckedContinuation<UInt16, any Error>?
|
||||
private var stopContinuation: CheckedContinuation<Void, Never>?
|
||||
|
||||
init() {
|
||||
(upgrades, upgradesContinuation) = AsyncStream<Upgrade>.makeStream()
|
||||
(clientTextFrames, clientTextContinuation) = AsyncStream<String>.makeStream()
|
||||
}
|
||||
|
||||
// MARK: - Lifecycle
|
||||
|
||||
/// Bind 127.0.0.1 on a system-assigned port; returns the port once ready.
|
||||
func start() async throws -> UInt16 {
|
||||
let parameters = NWParameters.tcp
|
||||
parameters.requiredLocalEndpoint = NWEndpoint.hostPort(
|
||||
host: NWEndpoint.Host("127.0.0.1"), port: .any
|
||||
)
|
||||
let listener = try NWListener(using: parameters)
|
||||
lock.withLock { self.listener = listener }
|
||||
listener.newConnectionHandler = { [weak self] in self?.adopt(connection: $0) }
|
||||
listener.stateUpdateHandler = { [weak self] in self?.handleListenerState($0) }
|
||||
return try await withCheckedThrowingContinuation { continuation in
|
||||
lock.withLock { startContinuation = continuation }
|
||||
listener.start(queue: queue)
|
||||
}
|
||||
}
|
||||
|
||||
/// Fire-and-forget teardown (tests: `defer { server.stop() }`).
|
||||
func stop() {
|
||||
lock.withLock { listener }?.cancel()
|
||||
lock.withLock { connection }?.forceCancel()
|
||||
finishStreams()
|
||||
}
|
||||
|
||||
/// Teardown that waits for the listener to actually release the port
|
||||
/// (needed by the connect-to-dead-port test to avoid a race).
|
||||
func stopAndWait() async {
|
||||
guard let listener = lock.withLock({ listener }) else { return }
|
||||
await withCheckedContinuation { (continuation: CheckedContinuation<Void, Never>) in
|
||||
lock.withLock { stopContinuation = continuation }
|
||||
listener.cancel()
|
||||
}
|
||||
lock.withLock { connection }?.forceCancel()
|
||||
finishStreams()
|
||||
}
|
||||
|
||||
// MARK: - Scripted server actions
|
||||
|
||||
/// Send one text frame to the client.
|
||||
func send(text: String) {
|
||||
sendToCurrentConnection(opcode: Wire.opcodeText, payload: [UInt8](text.utf8))
|
||||
}
|
||||
|
||||
/// Send one binary frame (the real server never does — untrusted-input test).
|
||||
func send(binary payload: [UInt8]) {
|
||||
sendToCurrentConnection(opcode: Wire.opcodeBinary, payload: payload)
|
||||
}
|
||||
|
||||
/// Send a WS close frame with the given status code (clean close path).
|
||||
func sendClose(code: UInt16) {
|
||||
let payload = [UInt8(code >> 8), UInt8(code & 0xFF)]
|
||||
sendToCurrentConnection(opcode: Wire.opcodeClose, payload: payload)
|
||||
}
|
||||
|
||||
/// Abrupt TCP termination WITHOUT a WS close frame (transport-error path).
|
||||
func abort() {
|
||||
lock.withLock { connection }?.forceCancel()
|
||||
}
|
||||
|
||||
// MARK: - Connection handling
|
||||
|
||||
private func adopt(connection newConnection: NWConnection) {
|
||||
lock.withLock {
|
||||
connection = newConnection
|
||||
rxBuffer = []
|
||||
isHandshakeDone = false
|
||||
}
|
||||
newConnection.start(queue: queue)
|
||||
receiveNext(on: newConnection)
|
||||
}
|
||||
|
||||
private func receiveNext(on connection: NWConnection) {
|
||||
connection.receive(
|
||||
minimumIncompleteLength: 1, maximumLength: Wire.receiveChunkLimit
|
||||
) { [weak self] data, _, isComplete, error in
|
||||
guard let self else { return }
|
||||
if let data, !data.isEmpty {
|
||||
self.ingest(bytes: [UInt8](data))
|
||||
}
|
||||
guard error == nil, !isComplete else { return }
|
||||
guard let current = self.lock.withLock({ self.connection }) else { return }
|
||||
self.receiveNext(on: current)
|
||||
}
|
||||
}
|
||||
|
||||
private func ingest(bytes: [UInt8]) {
|
||||
let current: NWConnection? = lock.withLock {
|
||||
rxBuffer += bytes
|
||||
return connection
|
||||
}
|
||||
guard let current else { return }
|
||||
tryCompleteHandshake(on: current)
|
||||
drainClientFrames(on: current)
|
||||
}
|
||||
|
||||
private func handleListenerState(_ state: NWListener.State) {
|
||||
switch state {
|
||||
case .ready:
|
||||
let port = lock.withLock { listener?.port?.rawValue }
|
||||
takeStartContinuation()?.resume(returning: port ?? 0)
|
||||
case .failed(let error):
|
||||
takeStartContinuation()?.resume(throwing: error)
|
||||
case .cancelled:
|
||||
takeStopContinuation()?.resume()
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Handshake
|
||||
|
||||
private func tryCompleteHandshake(on connection: NWConnection) {
|
||||
let requestBytes: [UInt8]? = lock.withLock {
|
||||
guard !isHandshakeDone, let end = Self.headerEndIndex(in: rxBuffer) else {
|
||||
return nil
|
||||
}
|
||||
let request = Array(rxBuffer[..<end])
|
||||
rxBuffer.removeFirst(end + Wire.headerTerminator.count)
|
||||
isHandshakeDone = true
|
||||
return request
|
||||
}
|
||||
guard let requestBytes else { return }
|
||||
let upgrade = Self.parseUpgrade(String(decoding: requestBytes, as: UTF8.self))
|
||||
connection.send(
|
||||
content: Self.handshakeResponse(for: upgrade),
|
||||
completion: .contentProcessed { _ in }
|
||||
)
|
||||
upgradesContinuation.yield(upgrade)
|
||||
}
|
||||
|
||||
private static func headerEndIndex(in bytes: [UInt8]) -> Int? {
|
||||
let terminator = Wire.headerTerminator
|
||||
guard bytes.count >= terminator.count else { return nil }
|
||||
for start in 0...(bytes.count - terminator.count)
|
||||
where Array(bytes[start..<(start + terminator.count)]) == terminator {
|
||||
return start
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
private static func parseUpgrade(_ text: String) -> Upgrade {
|
||||
let lines = text.components(separatedBy: "\r\n")
|
||||
var headers: [String: String] = [:]
|
||||
for line in lines.dropFirst() {
|
||||
guard let colon = line.firstIndex(of: ":") else { continue }
|
||||
let name = line[..<colon].trimmingCharacters(in: .whitespaces).lowercased()
|
||||
let value = line[line.index(after: colon)...].trimmingCharacters(in: .whitespaces)
|
||||
headers[name] = value
|
||||
}
|
||||
return Upgrade(requestLine: lines.first ?? "", headers: headers)
|
||||
}
|
||||
|
||||
private static func handshakeResponse(for upgrade: Upgrade) -> Data {
|
||||
let key = upgrade.headers["sec-websocket-key"] ?? ""
|
||||
let digest = Insecure.SHA1.hash(data: Data((key + Wire.acceptGUID).utf8))
|
||||
let accept = Data(digest).base64EncodedString()
|
||||
let response = "HTTP/1.1 101 Switching Protocols\r\n"
|
||||
+ "Upgrade: websocket\r\n"
|
||||
+ "Connection: Upgrade\r\n"
|
||||
+ "Sec-WebSocket-Accept: \(accept)\r\n\r\n"
|
||||
return Data(response.utf8)
|
||||
}
|
||||
|
||||
// MARK: - Client frame parsing
|
||||
|
||||
private func drainClientFrames(on connection: NWConnection) {
|
||||
while true {
|
||||
let frame: (opcode: UInt8, payload: [UInt8])? = lock.withLock {
|
||||
guard isHandshakeDone, let parsed = Self.parseFrame(rxBuffer) else {
|
||||
return nil
|
||||
}
|
||||
rxBuffer.removeFirst(parsed.consumed)
|
||||
return (parsed.opcode, parsed.payload)
|
||||
}
|
||||
guard let frame else { return }
|
||||
dispatch(frame: frame, on: connection)
|
||||
}
|
||||
}
|
||||
|
||||
private func dispatch(frame: (opcode: UInt8, payload: [UInt8]), on connection: NWConnection) {
|
||||
switch frame.opcode {
|
||||
case Wire.opcodeText:
|
||||
clientTextContinuation.yield(String(decoding: frame.payload, as: UTF8.self))
|
||||
case Wire.opcodePing:
|
||||
// Echo the payload back as a pong (RFC 6455 §5.5.3).
|
||||
rawSend(opcode: Wire.opcodePong, payload: frame.payload, on: connection)
|
||||
case Wire.opcodeClose:
|
||||
rawSend(opcode: Wire.opcodeClose, payload: frame.payload, on: connection)
|
||||
connection.cancel()
|
||||
default:
|
||||
break // binary/pong/continuation from the client — irrelevant here
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse one complete (possibly masked) client frame; nil while incomplete.
|
||||
private static func parseFrame(
|
||||
_ bytes: [UInt8]
|
||||
) -> (opcode: UInt8, payload: [UInt8], consumed: Int)? {
|
||||
guard bytes.count >= 2 else { return nil }
|
||||
let opcode = bytes[0] & Wire.opcodeMask
|
||||
let isMasked = bytes[1] & Wire.maskBit != 0
|
||||
var length = Int(bytes[1] & Wire.payloadLenMask)
|
||||
var offset = 2
|
||||
if length == Wire.len16Marker {
|
||||
guard bytes.count >= offset + 2 else { return nil }
|
||||
length = Int(bytes[offset]) << 8 | Int(bytes[offset + 1])
|
||||
offset += 2
|
||||
} else if length == Wire.len64Marker {
|
||||
guard bytes.count >= offset + 8 else { return nil }
|
||||
length = bytes[offset..<(offset + 8)].reduce(0) { ($0 << 8) | Int($1) }
|
||||
offset += 8
|
||||
}
|
||||
let maskCount = isMasked ? Wire.maskKeyLength : 0
|
||||
guard bytes.count >= offset + maskCount + length else { return nil }
|
||||
var payload = Array(bytes[(offset + maskCount)..<(offset + maskCount + length)])
|
||||
if isMasked {
|
||||
let mask = Array(bytes[offset..<(offset + maskCount)])
|
||||
for index in payload.indices {
|
||||
payload[index] ^= mask[index % Wire.maskKeyLength]
|
||||
}
|
||||
}
|
||||
return (opcode, payload, offset + maskCount + length)
|
||||
}
|
||||
|
||||
// MARK: - Server frame encoding (server→client frames are unmasked)
|
||||
|
||||
private func sendToCurrentConnection(opcode: UInt8, payload: [UInt8]) {
|
||||
guard let connection = lock.withLock({ connection }) else { return }
|
||||
rawSend(opcode: opcode, payload: payload, on: connection)
|
||||
}
|
||||
|
||||
private func rawSend(opcode: UInt8, payload: [UInt8], on connection: NWConnection) {
|
||||
connection.send(
|
||||
content: Self.encodeServerFrame(opcode: opcode, payload: payload),
|
||||
completion: .contentProcessed { _ in }
|
||||
)
|
||||
}
|
||||
|
||||
private static func encodeServerFrame(opcode: UInt8, payload: [UInt8]) -> Data {
|
||||
var frame: [UInt8] = [Wire.finBit | opcode]
|
||||
let count = payload.count
|
||||
if count <= Wire.maxTinyPayload {
|
||||
frame.append(UInt8(count))
|
||||
} else if count <= Int(UInt16.max) {
|
||||
frame.append(UInt8(Wire.len16Marker))
|
||||
frame.append(UInt8((count >> 8) & 0xFF))
|
||||
frame.append(UInt8(count & 0xFF))
|
||||
} else {
|
||||
frame.append(UInt8(Wire.len64Marker))
|
||||
for shift in stride(from: 56, through: 0, by: -8) {
|
||||
frame.append(UInt8((count >> shift) & 0xFF))
|
||||
}
|
||||
}
|
||||
frame += payload
|
||||
return Data(frame)
|
||||
}
|
||||
|
||||
// MARK: - Continuation bookkeeping
|
||||
|
||||
private func takeStartContinuation() -> CheckedContinuation<UInt16, any Error>? {
|
||||
lock.withLock {
|
||||
defer { startContinuation = nil }
|
||||
return startContinuation
|
||||
}
|
||||
}
|
||||
|
||||
private func takeStopContinuation() -> CheckedContinuation<Void, Never>? {
|
||||
lock.withLock {
|
||||
defer { stopContinuation = nil }
|
||||
return stopContinuation
|
||||
}
|
||||
}
|
||||
|
||||
private func finishStreams() {
|
||||
upgradesContinuation.finish()
|
||||
clientTextContinuation.finish()
|
||||
}
|
||||
}
|
||||
#endif
|
||||
@@ -0,0 +1,616 @@
|
||||
import Foundation
|
||||
import Testing
|
||||
import TestSupport
|
||||
import WireProtocol
|
||||
@testable import SessionCore
|
||||
|
||||
/// T-iOS-10 · SessionEngine actor (plan §3.2 / §7).
|
||||
/// Connection lifecycle, attach-first framing, reconnect backoff, gate
|
||||
/// sequencing, away digest, and terminal failure states — all against
|
||||
/// TestSupport.FakeTransport + FakeClock: zero real waits, zero network.
|
||||
@Suite("SessionEngine")
|
||||
struct SessionEngineTests {
|
||||
// MARK: - Server-side frame fixtures (untrusted wire input, hand-built JSON)
|
||||
|
||||
private enum ServerFrames {
|
||||
static func attached(_ id: UUID) -> String {
|
||||
#"{"type":"attached","sessionId":"\#(id.uuidString.lowercased())"}"#
|
||||
}
|
||||
|
||||
static func output(_ data: String) -> String {
|
||||
#"{"type":"output","data":"\#(data)"}"#
|
||||
}
|
||||
|
||||
static func exit(code: Int, reason: String) -> String {
|
||||
#"{"type":"exit","code":\#(code),"reason":"\#(reason)"}"#
|
||||
}
|
||||
|
||||
static func status(pending: Bool, gate: String? = nil, detail: String? = nil) -> String {
|
||||
var fields = ["\"type\":\"status\"", "\"status\":\"working\"", "\"pending\":\(pending)"]
|
||||
if let gate { fields.append("\"gate\":\"\(gate)\"") }
|
||||
if let detail { fields.append("\"detail\":\"\(detail)\"") }
|
||||
return "{\(fields.joined(separator: ","))}"
|
||||
}
|
||||
}
|
||||
|
||||
/// Records `eventsSource` calls and serves a scripted timeline response.
|
||||
private actor TimelineRecorder {
|
||||
private(set) var requestedIds: [UUID] = []
|
||||
private let response: [TimelineEvent]
|
||||
|
||||
init(response: [TimelineEvent]) {
|
||||
self.response = response
|
||||
}
|
||||
|
||||
func fetch(_ id: UUID) -> [TimelineEvent] {
|
||||
requestedIds.append(id)
|
||||
return response
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Harness
|
||||
|
||||
/// One engine + fakes + a single events iterator (the stream's one consumer).
|
||||
private final class Harness {
|
||||
let transport = FakeTransport()
|
||||
let clock = FakeClock()
|
||||
let engine: SessionEngine
|
||||
private var iterator: AsyncStream<SessionEvent>.AsyncIterator
|
||||
|
||||
init(
|
||||
eventsSource: @escaping @Sendable (UUID) async throws -> [TimelineEvent] = { _ in [] }
|
||||
) throws {
|
||||
let baseURL = try #require(URL(string: "http://192.168.1.5:3000"))
|
||||
let endpoint = try #require(HostEndpoint(baseURL: baseURL))
|
||||
engine = SessionEngine(
|
||||
transport: transport, clock: clock, endpoint: endpoint,
|
||||
eventsSource: eventsSource
|
||||
)
|
||||
iterator = engine.events.makeAsyncIterator()
|
||||
}
|
||||
|
||||
func next() async -> SessionEvent? {
|
||||
await iterator.next()
|
||||
}
|
||||
|
||||
/// Standard preamble: open → .connecting → .connected, then the server
|
||||
/// confirms the attach and the engine adopts `id`.
|
||||
func openAndAdopt(sessionId: UUID? = nil, adopting id: UUID) async {
|
||||
await engine.open(sessionId: sessionId, cwd: nil)
|
||||
#expect(await next() == .connection(.connecting))
|
||||
#expect(await next() == .connection(.connected))
|
||||
await transport.emit(frame: ServerFrames.attached(id))
|
||||
#expect(await next() == .adopted(sessionId: id))
|
||||
}
|
||||
|
||||
/// Frames the client sent on connection `index` (nil if none exists).
|
||||
func frames(onConnection index: Int) async -> [String]? {
|
||||
let byConnection = await transport.sentFramesByConnection
|
||||
guard byConnection.indices.contains(index) else { return nil }
|
||||
return byConnection[index]
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Attach-first framing
|
||||
|
||||
@Test("open() sends attach as the very first frame; pre-open sends queue behind it in order")
|
||||
func attachIsFirstFrameAndEarlierSendsQueueBehindIt() async throws {
|
||||
// Arrange
|
||||
let harness = try Harness()
|
||||
// Act: user input arrives BEFORE open (must never precede attach).
|
||||
await harness.engine.send(.input(data: "queued-before-open"))
|
||||
await harness.engine.open(sessionId: nil, cwd: nil)
|
||||
#expect(await harness.next() == .connection(.connecting))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
|
||||
// Assert: first frame is attach(null), queued input follows, no resize
|
||||
// (no dims are known yet — server spawned 80x24, src/server.ts:714-717).
|
||||
let frames = try #require(await harness.frames(onConnection: 0))
|
||||
#expect(frames == [
|
||||
MessageCodec.encode(.attach(sessionId: nil, cwd: nil)),
|
||||
MessageCodec.encode(.input(data: "queued-before-open")),
|
||||
])
|
||||
}
|
||||
|
||||
@Test("open() forwards a valid absolute cwd and drops an invalid relative one")
|
||||
func openForwardsValidCwdAndDropsInvalidCwd() async throws {
|
||||
// Arrange / Act: valid absolute cwd.
|
||||
let withCwd = try Harness()
|
||||
await withCwd.engine.open(sessionId: nil, cwd: "/Users/dev/proj")
|
||||
#expect(await withCwd.next() == .connection(.connecting))
|
||||
#expect(await withCwd.next() == .connection(.connected))
|
||||
|
||||
// Assert
|
||||
let cwdFrames = try #require(await withCwd.frames(onConnection: 0))
|
||||
#expect(cwdFrames.first == MessageCodec.encode(.attach(sessionId: nil, cwd: "/Users/dev/proj")))
|
||||
|
||||
// Act: relative cwd would make the server DROP the whole attach frame
|
||||
// (src/protocol.ts:142-149) — the engine must strip it, not send it.
|
||||
let withBadCwd = try Harness()
|
||||
await withBadCwd.engine.open(sessionId: nil, cwd: "relative/path")
|
||||
#expect(await withBadCwd.next() == .connection(.connecting))
|
||||
#expect(await withBadCwd.next() == .connection(.connected))
|
||||
|
||||
// Assert
|
||||
let strippedFrames = try #require(await withBadCwd.frames(onConnection: 0))
|
||||
#expect(strippedFrames.first == MessageCodec.encode(.attach(sessionId: nil, cwd: nil)))
|
||||
}
|
||||
|
||||
@Test("open() with a non-v4 UUID attaches null instead of a frame the server would discard")
|
||||
func openWithNonV4SessionIdAttachesNull() async throws {
|
||||
// Arrange: version nibble '1' — fails the server's SESSION_ID_RE, so the
|
||||
// server would silently drop the attach and the connection would hang.
|
||||
let nonV4 = try #require(UUID(uuidString: "12345678-1234-1234-1234-123456789abc"))
|
||||
let harness = try Harness()
|
||||
|
||||
// Act
|
||||
await harness.engine.open(sessionId: nonV4, cwd: nil)
|
||||
#expect(await harness.next() == .connection(.connecting))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
|
||||
// Assert
|
||||
let frames = try #require(await harness.frames(onConnection: 0))
|
||||
#expect(frames.first == MessageCodec.encode(.attach(sessionId: nil, cwd: nil)))
|
||||
}
|
||||
|
||||
// MARK: - Adoption
|
||||
|
||||
@Test("attached frame emits adopted; a server-issued NEW id replaces the requested one on re-attach")
|
||||
func adoptsServerIssuedIdAndReattachesWithIt() async throws {
|
||||
// Arrange: ask for an id the server does not know — it creates a fresh
|
||||
// session and returns a NEW id (src/session/manager.ts:166-173).
|
||||
let requestedId = UUID()
|
||||
let serverIssuedId = UUID()
|
||||
let harness = try Harness()
|
||||
|
||||
// Act
|
||||
await harness.engine.open(sessionId: requestedId, cwd: nil)
|
||||
#expect(await harness.next() == .connection(.connecting))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
await harness.transport.emit(frame: ServerFrames.attached(serverIssuedId))
|
||||
|
||||
// Assert: adopted carries the SERVER's id.
|
||||
#expect(await harness.next() == .adopted(sessionId: serverIssuedId))
|
||||
let firstFrames = try #require(await harness.frames(onConnection: 0))
|
||||
#expect(firstFrames.first == MessageCodec.encode(.attach(sessionId: requestedId, cwd: nil)))
|
||||
|
||||
// Act: drop the transport, let the backoff timer fire.
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
harness.clock.advance(by: .seconds(1))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
|
||||
// Assert: the re-attach uses the ADOPTED id, not the requested one.
|
||||
let secondFrames = try #require(await harness.frames(onConnection: 1))
|
||||
#expect(secondFrames.first == MessageCodec.encode(.attach(sessionId: serverIssuedId, cwd: nil)))
|
||||
}
|
||||
|
||||
// MARK: - Output ordering
|
||||
|
||||
@Test("output frames are delivered in order with none dropped")
|
||||
func outputFramesDeliverInOrder() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act: replay-then-live shape — three frames in a row.
|
||||
await harness.transport.emit(frame: ServerFrames.output("replay"))
|
||||
await harness.transport.emit(frame: ServerFrames.output("live-1"))
|
||||
await harness.transport.emit(frame: ServerFrames.output("live-2"))
|
||||
|
||||
// Assert
|
||||
#expect(await harness.next() == .output("replay"))
|
||||
#expect(await harness.next() == .output("live-1"))
|
||||
#expect(await harness.next() == .output("live-2"))
|
||||
}
|
||||
|
||||
// MARK: - Exit is terminal
|
||||
|
||||
@Test("exit(code:-1) emits exited and never reconnects (spawn failure is terminal)")
|
||||
func exitMinusOneIsTerminalWithoutReconnect() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act: M4 path — server sends exit(-1, reason) then closes the WS.
|
||||
await harness.transport.emit(frame: ServerFrames.exit(code: -1, reason: "spawn failed"))
|
||||
await harness.transport.finishFrames()
|
||||
|
||||
// Assert: exited, then a deliberate closed — and the stream FINISHES
|
||||
// (no .reconnecting can ever follow).
|
||||
#expect(await harness.next() == .exited(code: -1, reason: "spawn failed"))
|
||||
#expect(await harness.next() == .connection(.closed))
|
||||
#expect(await harness.next() == nil)
|
||||
#expect(harness.clock.pendingSleeperCount == 0)
|
||||
#expect(await harness.transport.connectAttempts.count == 1)
|
||||
|
||||
// Assert: the engine is inert after exit — sends are dropped, counted.
|
||||
await harness.engine.send(.input(data: "too late"))
|
||||
#expect(await harness.engine.droppedOutboundFrameCount == 1)
|
||||
}
|
||||
|
||||
// MARK: - Reconnect
|
||||
|
||||
@Test("transport error → reconnecting event; clock advance auto-reconnects and re-attaches the SAME session id")
|
||||
func disconnectSchedulesBackoffAndReattachesSameId() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act: the wire dies; input typed while offline must queue.
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
await harness.engine.send(.input(data: "typed-offline"))
|
||||
harness.clock.advance(by: .seconds(1))
|
||||
|
||||
// Assert: reconnected; attach FIRST (same id), offline input flushed behind it.
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
let secondFrames = try #require(await harness.frames(onConnection: 1))
|
||||
#expect(secondFrames == [
|
||||
MessageCodec.encode(.attach(sessionId: sessionId, cwd: nil)),
|
||||
MessageCodec.encode(.input(data: "typed-offline")),
|
||||
])
|
||||
}
|
||||
|
||||
@Test("repeated connect failures walk the backoff ladder with rising attempt counts")
|
||||
func repeatedConnectFailuresWalkBackoffLadder() async throws {
|
||||
// Arrange: the first two connect attempts fail outright.
|
||||
let harness = try Harness()
|
||||
await harness.transport.scriptConnectFailure()
|
||||
await harness.transport.scriptConnectFailure()
|
||||
|
||||
// Act / Assert: 1s then 2s rungs (mirrors public/terminal-session.ts).
|
||||
await harness.engine.open(sessionId: nil, cwd: nil)
|
||||
#expect(await harness.next() == .connection(.connecting))
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
harness.clock.advance(by: .seconds(1))
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 2, next: .seconds(2))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
harness.clock.advance(by: .seconds(2))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
#expect(await harness.transport.connectAttempts.count == 3)
|
||||
}
|
||||
|
||||
// MARK: - Foregrounding (latest-writer-wins sizing)
|
||||
|
||||
@Test("notifyForegrounded while connected sends only a resize — no reconnect")
|
||||
func notifyForegroundedWhileConnectedSendsResizeOnly() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act
|
||||
await harness.engine.notifyForegrounded(dims: (cols: 120, rows: 40))
|
||||
|
||||
// Assert
|
||||
let frames = try #require(await harness.frames(onConnection: 0))
|
||||
#expect(frames.last == MessageCodec.encode(.resize(cols: 120, rows: 40)))
|
||||
#expect(await harness.transport.connectAttempts.count == 1)
|
||||
}
|
||||
|
||||
@Test("notifyForegrounded while disconnected reconnects immediately and re-sends dims after attach")
|
||||
func notifyForegroundedWhileDisconnectedReconnectsNowWithResize() async throws {
|
||||
// Arrange: connected, then the wire dies and the engine parks in backoff.
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
|
||||
// Act: foreground — connect NOW, no clock advance at all.
|
||||
await harness.engine.notifyForegrounded(dims: (cols: 100, rows: 30))
|
||||
|
||||
// Assert: reconnected without the timer; attach first, then the dims
|
||||
// (latest-writer-wins — this device reclaims full-screen).
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
let secondFrames = try #require(await harness.frames(onConnection: 1))
|
||||
#expect(secondFrames == [
|
||||
MessageCodec.encode(.attach(sessionId: sessionId, cwd: nil)),
|
||||
MessageCodec.encode(.resize(cols: 100, rows: 30)),
|
||||
])
|
||||
#expect(harness.clock.pendingSleeperCount == 0)
|
||||
}
|
||||
|
||||
@Test("dims learned from send(.resize) are replayed right after attach on reconnect")
|
||||
func lastKnownDimsFromSendResizeReplayOnReattach() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
await harness.engine.send(.resize(cols: 90, rows: 30))
|
||||
|
||||
// Act: invalid dims must be dropped, not stored (server would discard).
|
||||
await harness.engine.send(.resize(cols: 0, rows: 30))
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
harness.clock.advance(by: .seconds(1))
|
||||
|
||||
// Assert
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
let secondFrames = try #require(await harness.frames(onConnection: 1))
|
||||
#expect(secondFrames == [
|
||||
MessageCodec.encode(.attach(sessionId: sessionId, cwd: nil)),
|
||||
MessageCodec.encode(.resize(cols: 90, rows: 30)),
|
||||
])
|
||||
#expect(await harness.engine.droppedOutboundFrameCount == 1)
|
||||
}
|
||||
|
||||
// MARK: - Gate sequencing
|
||||
|
||||
@Test("gate rising edge → gate(epoch 1); falling edge → gate(nil); stale approve dropped via canDecide")
|
||||
func gateSequencingAndStaleApproveDropped() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
let approveFrame = MessageCodec.encode(.approve(mode: nil))
|
||||
|
||||
// Act: approve BEFORE any gate ever rose → dropped.
|
||||
await harness.engine.send(.approve(mode: nil))
|
||||
#expect(await harness.engine.droppedDecisionCount == 1)
|
||||
|
||||
// Act: rising edge mints epoch 1.
|
||||
await harness.transport.emit(frame: ServerFrames.status(pending: true, gate: "tool", detail: "Bash"))
|
||||
#expect(await harness.next() == .gate(GateState(kind: .tool, detail: "Bash", epoch: 1)))
|
||||
|
||||
// Act: approve while the gate is held → sent.
|
||||
await harness.engine.send(.approve(mode: nil))
|
||||
#expect(await harness.transport.sentFrames.filter { $0 == approveFrame }.count == 1)
|
||||
|
||||
// Act: falling edge lifts the gate.
|
||||
await harness.transport.emit(frame: ServerFrames.status(pending: false))
|
||||
#expect(await harness.next() == .gate(nil))
|
||||
|
||||
// Assert: post-resolution approve/reject are STALE — never sent
|
||||
// (GateTracker.canDecide, 防误批新 gate).
|
||||
await harness.engine.send(.approve(mode: nil))
|
||||
await harness.engine.send(.reject)
|
||||
#expect(await harness.transport.sentFrames.filter { $0 == approveFrame }.count == 1)
|
||||
#expect(await harness.transport.sentFrames.contains(MessageCodec.encode(.reject)) == false)
|
||||
#expect(await harness.engine.droppedDecisionCount == 3)
|
||||
}
|
||||
|
||||
@Test("gate decisions are never queued while disconnected — dropped, not replayed")
|
||||
func gateDecisionWhileDisconnectedIsDroppedNotQueued() async throws {
|
||||
// Arrange: a tool gate is held, then the wire dies.
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
await harness.transport.emit(frame: ServerFrames.status(pending: true, gate: "tool"))
|
||||
#expect(await harness.next() == .gate(GateState(kind: .tool, detail: nil, epoch: 1)))
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
|
||||
// Act: a decision tapped while offline must NOT be queued for later —
|
||||
// by reconnect time the gate may be a DIFFERENT one.
|
||||
await harness.engine.send(.approve(mode: nil))
|
||||
harness.clock.advance(by: .seconds(1))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
|
||||
// Assert: re-attach flushed no approve.
|
||||
let secondFrames = try #require(await harness.frames(onConnection: 1))
|
||||
#expect(secondFrames == [MessageCodec.encode(.attach(sessionId: sessionId, cwd: nil))])
|
||||
#expect(await harness.engine.droppedDecisionCount == 1)
|
||||
}
|
||||
|
||||
// MARK: - Away digest
|
||||
|
||||
@Test("reconnect emits exactly one digest, reduced from eventsSource since the disconnect")
|
||||
func reconnectEmitsExactlyOneDigest() async throws {
|
||||
// Arrange: one away-window event (far future `at`) and one ancient
|
||||
// event (at: 0) that the `since` filter must exclude.
|
||||
let sessionId = UUID()
|
||||
let awayEvent = TimelineEvent(at: Int.max / 2, class: "tool", toolName: "Bash", label: "ran Bash")
|
||||
let staleEvent = TimelineEvent(at: 0, class: "done", toolName: nil, label: "finished")
|
||||
let recorder = TimelineRecorder(response: [awayEvent, staleEvent])
|
||||
let harness = try Harness(eventsSource: { await recorder.fetch($0) })
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act: disconnect → backoff → reconnect → server confirms the attach.
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
harness.clock.advance(by: .seconds(1))
|
||||
#expect(await harness.next() == .connection(.connected))
|
||||
await harness.transport.emit(frame: ServerFrames.attached(sessionId))
|
||||
#expect(await harness.next() == .adopted(sessionId: sessionId))
|
||||
|
||||
// Assert: exactly one digest; the stale event is filtered by `since`.
|
||||
let expected = AwayDigest(
|
||||
toolRuns: 1, waitingCount: 0, sawDone: false, sawStuck: false,
|
||||
recent: [awayEvent]
|
||||
)
|
||||
#expect(await harness.next() == .digest(expected))
|
||||
#expect(await recorder.requestedIds == [sessionId])
|
||||
|
||||
// Assert: no second digest sneaks in ahead of live traffic.
|
||||
await harness.transport.emit(frame: ServerFrames.output("after-digest"))
|
||||
#expect(await harness.next() == .output("after-digest"))
|
||||
}
|
||||
|
||||
@Test("initial connect emits no digest — there was no away window")
|
||||
func initialConnectEmitsNoDigest() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let recorder = TimelineRecorder(response: [])
|
||||
let harness = try Harness(eventsSource: { await recorder.fetch($0) })
|
||||
|
||||
// Act
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
await harness.transport.emit(frame: ServerFrames.output("first"))
|
||||
|
||||
// Assert: the event right after adoption is live output, not a digest,
|
||||
// and the events source was never consulted.
|
||||
#expect(await harness.next() == .output("first"))
|
||||
#expect(await recorder.requestedIds.isEmpty)
|
||||
}
|
||||
|
||||
// MARK: - Non-retryable failure
|
||||
|
||||
@Test("replayTooLarge → connection(.failed(.replayTooLarge)) and reconnection STOPS")
|
||||
func replayTooLargeIsTerminalAndNeverBacksOff() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act: the transport classifies EMSGSIZE as the typed error (T-iOS-9).
|
||||
await harness.transport.emitError(TermTransportError.replayTooLarge)
|
||||
|
||||
// Assert: explicit terminal failure, stream finished, NO backoff —
|
||||
// retrying would deterministically fail forever (plan §1).
|
||||
#expect(await harness.next() == .connection(.failed(.replayTooLarge)))
|
||||
#expect(await harness.next() == nil)
|
||||
#expect(harness.clock.pendingSleeperCount == 0)
|
||||
harness.clock.advance(by: .seconds(60))
|
||||
#expect(await harness.transport.connectAttempts.count == 1)
|
||||
}
|
||||
|
||||
// MARK: - Close
|
||||
|
||||
@Test("close() closes the transport, finishes the events stream, and leaks no tasks")
|
||||
func closeClosesTransportAndFinishesEvents() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act
|
||||
await harness.engine.close()
|
||||
|
||||
// Assert: deliberate detach (PTY keeps running server-side).
|
||||
#expect(await harness.transport.closeCallCount == 1)
|
||||
#expect(await harness.next() == .connection(.closed))
|
||||
await confirmation("events stream finishes after close") { streamFinished in
|
||||
while await harness.next() != nil {}
|
||||
streamFinished()
|
||||
}
|
||||
#expect(harness.clock.pendingSleeperCount == 0)
|
||||
|
||||
// Assert: close is idempotent.
|
||||
await harness.engine.close()
|
||||
#expect(await harness.transport.closeCallCount == 1)
|
||||
}
|
||||
|
||||
@Test("close() during backoff cancels the retry timer — no reconnect ever fires")
|
||||
func closeDuringBackoffCancelsRetryTimer() async throws {
|
||||
// Arrange: parked in the 1s backoff sleep.
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
await harness.transport.emitError(FakeTransportError.scriptedConnectFailure)
|
||||
#expect(await harness.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
await harness.clock.waitForSleepers(count: 1)
|
||||
|
||||
// Act
|
||||
await harness.engine.close()
|
||||
|
||||
// Assert: sleeper cancelled, stream done, and time passing changes nothing.
|
||||
#expect(await harness.next() == .connection(.closed))
|
||||
#expect(await harness.next() == nil)
|
||||
#expect(harness.clock.pendingSleeperCount == 0)
|
||||
harness.clock.advance(by: .seconds(60))
|
||||
#expect(await harness.transport.connectAttempts.count == 1)
|
||||
}
|
||||
|
||||
// MARK: - Untrusted input
|
||||
|
||||
@Test("malformed server frames are dropped and counted; the stream keeps flowing")
|
||||
func malformedFramesAreDroppedAndCounted() async throws {
|
||||
// Arrange
|
||||
let sessionId = UUID()
|
||||
let harness = try Harness()
|
||||
await harness.openAndAdopt(adopting: sessionId)
|
||||
|
||||
// Act: garbage, unknown type, wrong-typed required field — then a
|
||||
// valid frame that must still arrive.
|
||||
await harness.transport.emit(frame: "not json at all")
|
||||
await harness.transport.emit(frame: #"{"type":"mystery"}"#)
|
||||
await harness.transport.emit(frame: #"{"type":"output","data":42}"#)
|
||||
await harness.transport.emit(frame: ServerFrames.output("still alive"))
|
||||
|
||||
// Assert
|
||||
#expect(await harness.next() == .output("still alive"))
|
||||
#expect(await harness.engine.droppedServerFrameCount == 3)
|
||||
}
|
||||
|
||||
// MARK: - Ping wiring (internal hook, T-iOS-9 integration)
|
||||
|
||||
/// Pops scripted pong verdicts; answers `true` once the script runs out.
|
||||
private actor PongVerdicts {
|
||||
private var verdicts: [Bool]
|
||||
|
||||
init(_ verdicts: [Bool]) {
|
||||
self.verdicts = verdicts
|
||||
}
|
||||
|
||||
func pop() -> Bool {
|
||||
verdicts.isEmpty ? true : verdicts.removeFirst()
|
||||
}
|
||||
}
|
||||
|
||||
private struct ScriptedPinger: ConnectionPinger {
|
||||
let verdicts: PongVerdicts
|
||||
|
||||
func ping() async -> Bool {
|
||||
await verdicts.pop()
|
||||
}
|
||||
}
|
||||
|
||||
/// FakeTransport that also exposes the SessionCore-internal ping hook.
|
||||
private struct PingableFakeTransport: PingableTermTransport {
|
||||
let base: FakeTransport
|
||||
let pinger: ScriptedPinger
|
||||
|
||||
func connect(to endpoint: HostEndpoint) async throws -> TransportConnection {
|
||||
try await base.connect(to: endpoint)
|
||||
}
|
||||
|
||||
func connectPingable(to endpoint: HostEndpoint) async throws
|
||||
-> (connection: TransportConnection, pinger: any ConnectionPinger) {
|
||||
(try await base.connect(to: endpoint), pinger)
|
||||
}
|
||||
}
|
||||
|
||||
@Test("pongMissLimit consecutive missed pongs tear the connection down into the reconnect path")
|
||||
func missedPongsTriggerReconnect() async throws {
|
||||
// Arrange: an engine on a PINGABLE transport whose pongs never arrive.
|
||||
let base = FakeTransport()
|
||||
let clock = FakeClock()
|
||||
let verdicts = PongVerdicts([false, false])
|
||||
let transport = PingableFakeTransport(base: base, pinger: ScriptedPinger(verdicts: verdicts))
|
||||
let baseURL = try #require(URL(string: "http://192.168.1.5:3000"))
|
||||
let endpoint = try #require(HostEndpoint(baseURL: baseURL))
|
||||
let engine = SessionEngine(
|
||||
transport: transport, clock: clock, endpoint: endpoint,
|
||||
eventsSource: { _ in [] }
|
||||
)
|
||||
var iterator = engine.events.makeAsyncIterator()
|
||||
|
||||
// Act: connect; the ping scheduler parks on the injected clock.
|
||||
await engine.open(sessionId: nil, cwd: nil)
|
||||
#expect(await iterator.next() == .connection(.connecting))
|
||||
#expect(await iterator.next() == .connection(.connected))
|
||||
await clock.waitForSleepers(count: 1)
|
||||
// Miss #1 — tolerated (pongMissLimit = 2); scheduler re-parks.
|
||||
clock.advance(by: Tunables.pingInterval)
|
||||
await clock.waitForSleepers(count: 1)
|
||||
// Miss #2 — connection declared lost; engine must close it and back off.
|
||||
clock.advance(by: Tunables.pingInterval)
|
||||
|
||||
// Assert: the half-dead connection was torn down (close → stream finish)
|
||||
// and the normal reconnect path took over. Never "looks connected but dead".
|
||||
#expect(await iterator.next() == .connection(.reconnecting(attempt: 1, next: .seconds(1))))
|
||||
#expect(await base.closeCallCount == 1)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,258 @@
|
||||
#if os(macOS)
|
||||
import Darwin
|
||||
import Foundation
|
||||
import Testing
|
||||
import WireProtocol
|
||||
@testable import SessionCore
|
||||
|
||||
/// T-iOS-9 · `URLSessionTermTransport` against an in-process scripted WS
|
||||
/// server (`ScriptedWSServer`: raw-TCP loopback, ephemeral port — macOS-only
|
||||
/// test infra per plan §7 T-iOS-9; the real-Node-server pass is T-iOS-16).
|
||||
///
|
||||
/// These are real-I/O loopback tests: there are NO pacing sleeps — every wait
|
||||
/// is continuation/stream-driven, and `withTimeout` is a fail-fast bound, not
|
||||
/// a synchronization primitive.
|
||||
@Suite("URLSessionTermTransport", .timeLimit(.minutes(3)))
|
||||
struct URLSessionTermTransportTests {
|
||||
|
||||
/// Named test constants (no magic numbers).
|
||||
private enum TestTunables {
|
||||
/// Fail-fast upper bound for any single loopback wait.
|
||||
static let ioTimeout: Duration = .seconds(15)
|
||||
/// Frame count for the re-arm/ordering test (RED list: 100).
|
||||
static let orderedFrameCount = 100
|
||||
/// Internal transport seam value: shrunken `maximumMessageSize` so the
|
||||
/// oversize→EMSGSIZE path is deterministic without 16 MiB fixtures.
|
||||
static let tinyMessageCap = 64 * 1024
|
||||
/// 2 × `tinyMessageCap` — guaranteed over the shrunken cap.
|
||||
static let oversizePayloadBytes = 128 * 1024
|
||||
/// RFC 6455 normal-closure status code.
|
||||
static let normalCloseCode: UInt16 = 1000
|
||||
/// Arbitrary non-text payload for the binary-frame drop test.
|
||||
static let binaryPayload: [UInt8] = [0xDE, 0xAD, 0xBE, 0xEF]
|
||||
}
|
||||
|
||||
private struct TimedOut: Error {}
|
||||
|
||||
// MARK: - Helpers
|
||||
|
||||
private static func withTimeout<T: Sendable>(
|
||||
_ timeout: Duration = TestTunables.ioTimeout,
|
||||
_ operation: @escaping @Sendable () async throws -> T
|
||||
) async throws -> T {
|
||||
try await withThrowingTaskGroup(of: T.self) { group in
|
||||
group.addTask { try await operation() }
|
||||
group.addTask {
|
||||
try await Task.sleep(for: timeout)
|
||||
throw TimedOut()
|
||||
}
|
||||
guard let first = try await group.next() else { throw TimedOut() }
|
||||
group.cancelAll()
|
||||
return first
|
||||
}
|
||||
}
|
||||
|
||||
private static func makeEndpoint(port: UInt16) throws -> HostEndpoint {
|
||||
let url = try #require(URL(string: "http://127.0.0.1:\(port)"))
|
||||
return try #require(HostEndpoint(baseURL: url))
|
||||
}
|
||||
|
||||
private static func startServer() async throws
|
||||
-> (server: ScriptedWSServer, endpoint: HostEndpoint) {
|
||||
let server = ScriptedWSServer()
|
||||
let port = try await server.start()
|
||||
return (server, try makeEndpoint(port: port))
|
||||
}
|
||||
|
||||
private static func firstUpgrade(
|
||||
_ server: ScriptedWSServer
|
||||
) async throws -> ScriptedWSServer.Upgrade {
|
||||
try await withTimeout {
|
||||
var iterator = server.upgrades.makeAsyncIterator()
|
||||
guard let upgrade = await iterator.next() else { throw TimedOut() }
|
||||
return upgrade
|
||||
}
|
||||
}
|
||||
|
||||
private static func collect(
|
||||
_ count: Int, from frames: AsyncThrowingStream<String, any Error>
|
||||
) async throws -> [String] {
|
||||
var collected: [String] = []
|
||||
for try await frame in frames {
|
||||
collected.append(frame)
|
||||
if collected.count == count { return collected }
|
||||
}
|
||||
return collected
|
||||
}
|
||||
|
||||
// MARK: - RED list (plan §7 T-iOS-9 Steps)
|
||||
|
||||
@Test("upgrade request carries Origin == endpoint.originHeader (single source, §5.1)")
|
||||
func upgradeRequestCarriesEndpointOriginHeader() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
|
||||
let connection = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
let upgrade = try await Self.firstUpgrade(server)
|
||||
|
||||
#expect(upgrade.headers["origin"] == endpoint.originHeader)
|
||||
#expect(upgrade.requestLine.hasPrefix("GET \(WireConstants.wsPath) "))
|
||||
await connection.close()
|
||||
}
|
||||
|
||||
@Test("connected task uses Tunables.maxWSMessageBytes, not the 1 MiB platform default")
|
||||
func connectedTaskUsesFrozenMaxMessageSize() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
|
||||
let connection = try await URLSessionTermTransport().openConnection(to: endpoint)
|
||||
|
||||
#expect(connection.task.maximumMessageSize == Tunables.maxWSMessageBytes)
|
||||
await connection.close()
|
||||
}
|
||||
|
||||
@Test("receive loop re-arms continuously: 100 frames arrive, in order")
|
||||
func receiveLoopRearmsAcross100OrderedFrames() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let connection = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
let expected = (0..<TestTunables.orderedFrameCount).map { "frame-\($0)" }
|
||||
|
||||
for frame in expected { server.send(text: frame) }
|
||||
let frames = connection.frames
|
||||
let received = try await Self.withTimeout {
|
||||
try await Self.collect(TestTunables.orderedFrameCount, from: frames)
|
||||
}
|
||||
|
||||
#expect(received == expected)
|
||||
await connection.close()
|
||||
}
|
||||
|
||||
@Test("server close frame → frames stream finishes cleanly (no throw)")
|
||||
func serverCloseFrameFinishesStream() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let connection = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
|
||||
server.sendClose(code: TestTunables.normalCloseCode)
|
||||
let frames = connection.frames
|
||||
let leftover = try await Self.withTimeout {
|
||||
var collected: [String] = []
|
||||
for try await frame in frames { collected.append(frame) }
|
||||
return collected
|
||||
}
|
||||
|
||||
#expect(leftover.isEmpty)
|
||||
}
|
||||
|
||||
@Test("abrupt TCP kill → frames stream throws (distinguishable; NOT replayTooLarge)")
|
||||
func abruptTerminationThrowsTransportError() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let connection = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
|
||||
server.abort()
|
||||
let frames = connection.frames
|
||||
var thrown: (any Error)?
|
||||
do {
|
||||
try await Self.withTimeout { for try await _ in frames {} }
|
||||
} catch {
|
||||
thrown = error
|
||||
}
|
||||
|
||||
let failure = try #require(thrown, "stream must throw after abrupt termination")
|
||||
#expect(!(failure is TimedOut), "stream neither finished nor threw in time")
|
||||
#expect((failure as? TermTransportError) != .replayTooLarge)
|
||||
}
|
||||
|
||||
@Test("oversize frame → typed TermTransportError.replayTooLarge (EMSGSIZE 40 / ENOBUFS 55)")
|
||||
func oversizeFrameThrowsTypedReplayTooLarge() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let transport = URLSessionTermTransport(maxMessageBytes: TestTunables.tinyMessageCap)
|
||||
let connection = try await transport.connect(to: endpoint)
|
||||
|
||||
server.send(text: String(repeating: "x", count: TestTunables.oversizePayloadBytes))
|
||||
let frames = connection.frames
|
||||
|
||||
await #expect(throws: TermTransportError.replayTooLarge) {
|
||||
try await Self.withTimeout { for try await _ in frames {} }
|
||||
}
|
||||
}
|
||||
|
||||
@Test("send after close() → explicit TermTransportError.sendAfterClose, no crash")
|
||||
func sendAfterCloseThrowsExplicitError() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let connection = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
|
||||
await connection.close()
|
||||
|
||||
await #expect(throws: TermTransportError.sendAfterClose) {
|
||||
try await connection.send(MessageCodec.encode(.input(data: "x")))
|
||||
}
|
||||
}
|
||||
|
||||
@Test("binary frame is dropped (counted) and receiving continues")
|
||||
func binaryFrameDroppedAndReceivingContinues() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let wsConnection = try await URLSessionTermTransport().openConnection(to: endpoint)
|
||||
let connection = wsConnection.transportConnection()
|
||||
|
||||
server.send(binary: TestTunables.binaryPayload)
|
||||
server.send(text: "still-alive")
|
||||
let frames = connection.frames
|
||||
let received = try await Self.withTimeout {
|
||||
try await Self.collect(1, from: frames)
|
||||
}
|
||||
|
||||
#expect(received == ["still-alive"])
|
||||
#expect(wsConnection.droppedBinaryFrames == 1)
|
||||
await connection.close()
|
||||
}
|
||||
|
||||
@Test("send() delivers the client text frame to the server verbatim")
|
||||
func sendDeliversTextFrameToServer() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
let connection = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
let frame = MessageCodec.encode(.attach(sessionId: nil, cwd: nil))
|
||||
|
||||
try await connection.send(frame)
|
||||
let received = try await Self.withTimeout {
|
||||
var iterator = server.clientTextFrames.makeAsyncIterator()
|
||||
guard let first = await iterator.next() else { throw TimedOut() }
|
||||
return first
|
||||
}
|
||||
|
||||
#expect(received == frame)
|
||||
await connection.close()
|
||||
}
|
||||
|
||||
@Test("internal ping hook (non-frozen extension point): sendPing resolves pong")
|
||||
func internalPingHookResolvesPong() async throws {
|
||||
let (server, endpoint) = try await Self.startServer()
|
||||
defer { server.stop() }
|
||||
|
||||
let (connection, pinger) =
|
||||
try await URLSessionTermTransport().connectPingable(to: endpoint)
|
||||
let isPongReceived = try await Self.withTimeout { await pinger.ping() }
|
||||
|
||||
#expect(isPongReceived)
|
||||
await connection.close()
|
||||
}
|
||||
|
||||
@Test("connect to a dead port throws (handshake failure path, no hang)")
|
||||
func connectToDeadPortThrows() async throws {
|
||||
let server = ScriptedWSServer()
|
||||
let port = try await server.start()
|
||||
await server.stopAndWait()
|
||||
let endpoint = try Self.makeEndpoint(port: port)
|
||||
|
||||
await #expect(throws: (any Error).self) {
|
||||
_ = try await URLSessionTermTransport().connect(to: endpoint)
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
Reference in New Issue
Block a user