From a2b14ab6e7ac10ec971a4a1b2e32daf9c969db58 Mon Sep 17 00:00:00 2001 From: Yaojia Wang Date: Sat, 4 Jul 2026 22:55:56 +0200 Subject: [PATCH] =?UTF-8?q?feat(ios):=20W2=20connection=20core=20=E2=80=94?= =?UTF-8?q?=20URLSessionTermTransport=20+=20SessionEngine=20actor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- docs/PROGRESS_LOG.md | 6 +- .../Sources/SessionCore/SessionEngine.swift | 480 ++++++++++++++ .../Sources/SessionCore/SessionEvent.swift | 60 ++ .../SessionCore/URLSessionTermTransport.swift | 381 +++++++++++ .../SessionCoreTests/ScriptedWSServer.swift | 353 ++++++++++ .../SessionCoreTests/SessionEngineTests.swift | 616 ++++++++++++++++++ .../URLSessionTermTransportTests.swift | 258 ++++++++ 7 files changed, 2153 insertions(+), 1 deletion(-) create mode 100644 ios/Packages/SessionCore/Sources/SessionCore/SessionEngine.swift create mode 100644 ios/Packages/SessionCore/Sources/SessionCore/SessionEvent.swift create mode 100644 ios/Packages/SessionCore/Sources/SessionCore/URLSessionTermTransport.swift create mode 100644 ios/Packages/SessionCore/Tests/SessionCoreTests/ScriptedWSServer.swift create mode 100644 ios/Packages/SessionCore/Tests/SessionCoreTests/SessionEngineTests.swift create mode 100644 ios/Packages/SessionCore/Tests/SessionCoreTests/URLSessionTermTransportTests.swift diff --git a/docs/PROGRESS_LOG.md b/docs/PROGRESS_LOG.md index 7fd980e..e629249 100644 --- a/docs/PROGRESS_LOG.md +++ b/docs/PROGRESS_LOG.md @@ -38,7 +38,11 @@ - **T-iOS-7 HostRegistry**: §3.3 契约;KeychainHostStore 经 SecItemShim 缝(swift test 打 fake;真 shim kSecUseDataProtectionKeychain+AfterFirstUnlockThisDeviceOnly,属性字典单点构造+逐字单测);InMemoryHostStore 入 Sources 供 App-VM 测试。30 测,own-Sources 覆盖 88.1%。 - **T-iOS-8 APIClient+探针**: 44 测先全绿后 **`[!] BLOCKED` 正确上报冻结契约自相矛盾**——§3.4 `runPairingProbe → Result` 中 Host 是 §3.3 HostRegistry 类型,而 APIClient 依赖边仅 WireProtocol(§1 零耦合),签名无法编译(两轮 review 均漏)。**裁定(orchestrator,三选一取 c)**: 探针职责=验证 endpoint → 返回 `Result`,Host{id,name} 由 T-iOS-12 VM 构造(id/name 本非探针所知);同时按法定路径加 `Tunables.pairingProbeTimeout`(10s,§3.2.1 新增行+绊线断言)。裁定已写回 PLAN §3.4/§3.2.1/T-iOS-8,公开包装+冒烟测试由 orchestrator 落地。Origin **iff**-G 入测试名;探针②成功后必带 Origin kill 不留孤儿;403 kill→originRejected(HTTP 侧 G 守卫也是配对必验项)。45 测,own-Sources 覆盖 98.1%。 - **W1 验证(orchestrator 亲验)**: 5 包 178 tests 全绿(59+3+41+30+45);覆盖率(own-Sources 口径)SessionCore 99.1%/HostRegistry 88.1%/APIClient 98.1%,全过 80% 门;`git status` 全部在 Owns 内零越界。**发现归 T-iOS-16**: §9 覆盖率命令的 ignore regex 不排静态链入的依赖包源码,CI 接线时须按包过滤(本次已按正确口径亲算)。 -- 待续: W2(T-iOS-9→10,同包串行)→ W3 → W4 → W5。**偏差(orchestrator 决策,W1 起生效)**: 并行 builder 不用 git worktree(Workflow worktree 合并回主树增加失败面),改为主树直跑——文件互斥 + 同包任务串行化,禁 agent 碰 git;每波末独立 verify + orchestrator commit 兜底。 +- **[x] W2 连接核心(T-iOS-9→10 串行;首次派发 T-iOS-9 agent 死于 API 连接中断——现场零文件、直接 resume 重跑,非任务失败)**: + - **T-iOS-9 URLSessionTermTransport**: 冻结 TransportConnection 不动,ping 经 SessionCore 内部(非冻结)`PingableTermTransport/ConnectionPinger` 扩展点接入;pong 时限复用 `Tunables.pingInterval`(无新常量);EMSGSIZE 归类 40 主/55 兜底→类型化 `.replayTooLarge`;connect 失败原样重抛底层错误(保住 PairingError.classify);状态 delegate 驱动,close 三态可区分;测试服务器 = 裸 TCP 手写 RFC6455 framing(可逐字验 Origin/注入 binary/超帧)。11 测,文件覆盖 97.83%,iOS triple 编译过。 + - **T-iOS-10 SessionEngine actor**: 单 Task 生命周期循环 + generation 计数防陈旧任务复活;首帧铁律 attach→lastKnownDims resize→排队帧按序 flush;终态(exit / replayTooLarge)绝不进 backoff;digest 断点原子消费恰好一次;gate 决策 canDecide 兜底、断线期决策丢弃不入队(重连后 gate 可能换代);防御:非 v4 UUID/相对 cwd 剥离(服务器会静默丢帧致挂死)。**决策**: 初始 resize 依赖首次 sizeChanged/notifyForegrounded(冻结 open 无 dims 参数,SwiftTerm 首次布局即触发,已注 doc);digest limit 传 Int.max(截断属 UI,T-iOS-14)。20 测,SessionEngine.swift 94.98%(480 行,<800 硬上限,内聚裁量已记)。 + - **W2 验证(独立 agent,5/5 PASS)**: SessionCore 72 测(41 W1 完好+31 新);四包回归全绿(**累计 209 tests**);own-Sources 覆盖 96.68%;冻结契约零改动;4 项语义抽查(Origin 单点 :169/16MiB 来自 Tunables/replayTooLarge 断重连路径逐行核/decodeServer 唯一入口 nil-drop)全数在源码落实。 +- 待续: W3(UI 四路+T-iOS-16/17)→ W4 → W5。**W3 编排决策**: 四个 UI 任务全编进同一 App target,主树并行 xcodebuild 必互踩 → [11→12→13→14] 串行链 ∥ T-iOS-16 ∥ T-iOS-17;VM 单测需 WebTermTests target,project.yml 归 T-iOS-1 Owns → orchestrator 预先增补(协调点,同 W1 manifest 先例)。**偏差(orchestrator 决策,W1 起生效)**: 并行 builder 不用 git worktree(Workflow worktree 合并回主树增加失败面),改为主树直跑——文件互斥 + 同包任务串行化,禁 agent 碰 git;每波末独立 verify + orchestrator commit 兜底。 ### ✅ iOS 客户端实施计划 — `docs/PLAN_IOS_CLIENT.md`(规划完成 — 2026-07-04,main,未提交;纯文档,零代码改动) **多 agent 编排 = 一个 ultracode `Workflow`(66 agents,7 阶段)**:5 读者并行(wire contract 逐 file:line / 功能矩阵 / relay 现状 / 模板规范 / iOS 平台联网调研)→ 3 架构师独立竞标(full-native / WKWebView-hybrid / phased)→ 3 评委多视角打分(UX/工程/安全,**phased 26.5 胜出**,native 23.5、hybrid 16.5,21 条 mustSteal 嫁接)→ 综合简报 → 写手出稿(843 行)→ **4 评审交叉验证**(对照 src/ 源码逐条核实 / iOS API 联网核查 / 安全 / 计划质量)→ 每条 finding 对抗式 verify(CRITICAL/HIGH 双票)→ fixer 应用。 diff --git a/ios/Packages/SessionCore/Sources/SessionCore/SessionEngine.swift b/ios/Packages/SessionCore/Sources/SessionCore/SessionEngine.swift new file mode 100644 index 0000000..51aeada --- /dev/null +++ b/ios/Packages/SessionCore/Sources/SessionCore/SessionEngine.swift @@ -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 + 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 + private let eventsContinuation: AsyncStream.Continuation + + // MARK: - Connection runtime state (actor-confined) + + private var reconnect: ReconnectMachine + private var connection: TransportConnection? + private var connectionTask: Task? + private var pingTask: Task? + private var digestTask: Task? + /// 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, + 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.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) + } +} diff --git a/ios/Packages/SessionCore/Sources/SessionCore/SessionEvent.swift b/ios/Packages/SessionCore/Sources/SessionCore/SessionEvent.swift new file mode 100644 index 0000000..dde7281 --- /dev/null +++ b/ios/Packages/SessionCore/Sources/SessionCore/SessionEvent.swift @@ -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 +} diff --git a/ios/Packages/SessionCore/Sources/SessionCore/URLSessionTermTransport.swift b/ios/Packages/SessionCore/Sources/SessionCore/URLSessionTermTransport.swift new file mode 100644 index 0000000..bc2e164 --- /dev/null +++ b/ios/Packages/SessionCore/Sources/SessionCore/URLSessionTermTransport.swift @@ -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? + 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? + private let frames: AsyncThrowingStream + private let framesContinuation: AsyncThrowingStream.Continuation + + private override init() { + (frames, framesContinuation) = AsyncThrowingStream.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) 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? { + 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.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 + } +} diff --git a/ios/Packages/SessionCore/Tests/SessionCoreTests/ScriptedWSServer.swift b/ios/Packages/SessionCore/Tests/SessionCoreTests/ScriptedWSServer.swift new file mode 100644 index 0000000..57caf7d --- /dev/null +++ b/ios/Packages/SessionCore/Tests/SessionCoreTests/ScriptedWSServer.swift @@ -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 + /// Text frames received FROM the client, unmasked, in arrival order. + let clientTextFrames: AsyncStream + + private let upgradesContinuation: AsyncStream.Continuation + private let clientTextContinuation: AsyncStream.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? + private var stopContinuation: CheckedContinuation? + + init() { + (upgrades, upgradesContinuation) = AsyncStream.makeStream() + (clientTextFrames, clientTextContinuation) = AsyncStream.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) 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[.. 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[.. 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? { + lock.withLock { + defer { startContinuation = nil } + return startContinuation + } + } + + private func takeStopContinuation() -> CheckedContinuation? { + lock.withLock { + defer { stopContinuation = nil } + return stopContinuation + } + } + + private func finishStreams() { + upgradesContinuation.finish() + clientTextContinuation.finish() + } +} +#endif diff --git a/ios/Packages/SessionCore/Tests/SessionCoreTests/SessionEngineTests.swift b/ios/Packages/SessionCore/Tests/SessionCoreTests/SessionEngineTests.swift new file mode 100644 index 0000000..395011b --- /dev/null +++ b/ios/Packages/SessionCore/Tests/SessionCoreTests/SessionEngineTests.swift @@ -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.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) + } +} diff --git a/ios/Packages/SessionCore/Tests/SessionCoreTests/URLSessionTermTransportTests.swift b/ios/Packages/SessionCore/Tests/SessionCoreTests/URLSessionTermTransportTests.swift new file mode 100644 index 0000000..0acfbaf --- /dev/null +++ b/ios/Packages/SessionCore/Tests/SessionCoreTests/URLSessionTermTransportTests.swift @@ -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( + _ 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 + ) 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..