diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md new file mode 100644 index 000000000..2afa4cd3d --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md @@ -0,0 +1,39 @@ +# Trajectory: Fix Swift AsyncStream continuation leaks for issue 1266 + +> **Status:** ✅ Completed +> **Confidence:** 95% +> **Started:** July 15, 2026 at 06:39 AM +> **Completed:** July 15, 2026 at 06:53 AM + +--- + +## Summary + +Fixed Swift SDK AsyncStream continuation leaks with cancellation cleanup, bounded buffers, lazy channel streams, disconnect generation fencing, and 129 passing regression tests + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Use token-keyed bounded AsyncStream registries in both Swift SDKs + +- **Chose:** Use token-keyed bounded AsyncStream registries in both Swift SDKs +- **Reasoning:** Stable UUIDs plus termination tombstones close cancellation-before-registration races; bufferingNewest(256) bounds event/message queues, bufferingNewest(1) bounds state, and lazy channel streams avoid join-only queues. + +### Fence Swift stream registration with disconnect generations + +- **Chose:** Fence Swift stream registration with disconnect generations +- **Reasoning:** Claude review identified tombstone growth and post-disconnect registration resurrection. Cancel-aware registration tasks eliminate termination tombstones, while a lock-protected generation invalidated at disconnect rejects actor tasks scheduled by older stream epochs. + +--- + +## Chapters + +### 1. Work + +_Agent: default_ + +- Use token-keyed bounded AsyncStream registries in both Swift SDKs: Use token-keyed bounded AsyncStream registries in both Swift SDKs +- Fence Swift stream registration with disconnect generations: Fence Swift stream registration with disconnect generations diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json new file mode 100644 index 000000000..3a9283dc7 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json @@ -0,0 +1,65 @@ +{ + "id": "traj_bwj7uf3ksgz6", + "version": 1, + "task": { + "title": "Fix Swift AsyncStream continuation leaks for issue 1266" + }, + "status": "completed", + "startedAt": "2026-07-15T10:39:35.461Z", + "completedAt": "2026-07-15T10:53:40.411Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-07-15T10:45:05.763Z" + } + ], + "chapters": [ + { + "id": "chap_vvjmcd71cc1k", + "title": "Work", + "agentName": "default", + "startedAt": "2026-07-15T10:45:05.763Z", + "endedAt": "2026-07-15T10:53:40.411Z", + "events": [ + { + "ts": 1784112305764, + "type": "decision", + "content": "Use token-keyed bounded AsyncStream registries in both Swift SDKs: Use token-keyed bounded AsyncStream registries in both Swift SDKs", + "raw": { + "question": "Use token-keyed bounded AsyncStream registries in both Swift SDKs", + "chosen": "Use token-keyed bounded AsyncStream registries in both Swift SDKs", + "alternatives": [], + "reasoning": "Stable UUIDs plus termination tombstones close cancellation-before-registration races; bufferingNewest(256) bounds event/message queues, bufferingNewest(1) bounds state, and lazy channel streams avoid join-only queues." + }, + "significance": "high" + }, + { + "ts": 1784112604320, + "type": "decision", + "content": "Fence Swift stream registration with disconnect generations: Fence Swift stream registration with disconnect generations", + "raw": { + "question": "Fence Swift stream registration with disconnect generations", + "chosen": "Fence Swift stream registration with disconnect generations", + "alternatives": [], + "reasoning": "Claude review identified tombstone growth and post-disconnect registration resurrection. Cancel-aware registration tasks eliminate termination tombstones, while a lock-protected generation invalidated at disconnect rejects actor tasks scheduled by older stream epochs." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Fixed Swift SDK AsyncStream continuation leaks with cancellation cleanup, bounded buffers, lazy channel streams, disconnect generation fencing, and 129 passing regression tests", + "approach": "Standard approach", + "confidence": 0.95 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "bd1531aaea1bdb442c6c6c02498dc032cd0aecf6", + "endRef": "bd1531aaea1bdb442c6c6c02498dc032cd0aecf6" + } +} diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md new file mode 100644 index 000000000..6afeb62e9 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md @@ -0,0 +1,33 @@ +# Trajectory: Resolve renewed PR 1268 conflicts after broker Swift parity merge + +> **Status:** ✅ Completed +> **Confidence:** 98% +> **Started:** July 15, 2026 at 07:30 AM +> **Completed:** July 15, 2026 at 07:31 AM + +--- + +## Summary + +Merged current main into PR 1268, composed broker parity additions with stream lifecycle changes, and passed all 151 Swift tests + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Compose broker parity APIs with stream lifecycle registries + +- **Chose:** Compose broker parity APIs with stream lifecycle registries +- **Reasoning:** Current main added independent decoding/path helpers and control/observability tests; PR 1268 added UUID continuation registries and lifecycle tests. Keeping both preserves all behavior, with connection-state fan-out continuing through registry values. + +--- + +## Chapters + +### 1. Work + +_Agent: default_ + +- Composed broker parity APIs with stream lifecycle registries while preserving connection-state fan-out. diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json new file mode 100644 index 000000000..65ad594c0 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json @@ -0,0 +1,53 @@ +{ + "id": "traj_nt7clb37dtr7", + "version": 1, + "task": { + "title": "Resolve renewed PR 1268 conflicts after broker Swift parity merge" + }, + "status": "completed", + "startedAt": "2026-07-15T11:30:37.003Z", + "completedAt": "2026-07-15T11:31:36.846Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-07-15T11:31:36.445Z" + } + ], + "chapters": [ + { + "id": "chap_47tubwavnk4u", + "title": "Work", + "agentName": "default", + "startedAt": "2026-07-15T11:31:36.445Z", + "endedAt": "2026-07-15T11:31:36.846Z", + "events": [ + { + "ts": 1784115096446, + "type": "decision", + "content": "Compose broker parity APIs with stream lifecycle registries: Compose broker parity APIs with stream lifecycle registries", + "raw": { + "question": "Compose broker parity APIs with stream lifecycle registries", + "chosen": "Compose broker parity APIs with stream lifecycle registries", + "alternatives": [], + "reasoning": "Current main added independent decoding/path helpers and control/observability tests; PR 1268 added UUID continuation registries and lifecycle tests. Keeping both preserves all behavior, with connection-state fan-out continuing through registry values." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Merged current main into PR 1268, composed broker parity additions with stream lifecycle changes, and passed all 151 Swift tests", + "approach": "Standard approach", + "confidence": 0.98 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "38e00002cfba37694c8d0a9bd45d3c748d29c605", + "endRef": "38e00002cfba37694c8d0a9bd45d3c748d29c605" + } +} diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md new file mode 100644 index 000000000..17c4a4ef1 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md @@ -0,0 +1,33 @@ +# Trajectory: Resolve PR 1268 merge conflicts with current main + +> **Status:** ✅ Completed +> **Confidence:** 98% +> **Started:** July 15, 2026 at 07:01 AM +> **Completed:** July 15, 2026 at 07:02 AM + +--- + +## Summary + +Merged current origin/main into PR 1268, retained all Unreleased patch changelog entries, and passed all 129 Swift tests + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Keep all Unreleased patch changelog entries + +- **Chose:** Keep all Unreleased patch changelog entries +- **Reasoning:** The only merge conflict combined independent user-visible fixes from PR 1268 and current main, all at patch impact, so no entry should be dropped. + +--- + +## Chapters + +### 1. Work + +_Agent: default_ + +- Keep all Unreleased patch changelog entries: Keep all Unreleased patch changelog entries diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/trajectory.json new file mode 100644 index 000000000..c7aaf024d --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/trajectory.json @@ -0,0 +1,53 @@ +{ + "id": "traj_u94plhvjk6ew", + "version": 1, + "task": { + "title": "Resolve PR 1268 merge conflicts with current main" + }, + "status": "completed", + "startedAt": "2026-07-15T11:01:37.458Z", + "completedAt": "2026-07-15T11:02:30.385Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-07-15T11:02:29.972Z" + } + ], + "chapters": [ + { + "id": "chap_rb8pg5ihhb6g", + "title": "Work", + "agentName": "default", + "startedAt": "2026-07-15T11:02:29.972Z", + "endedAt": "2026-07-15T11:02:30.385Z", + "events": [ + { + "ts": 1784113349973, + "type": "decision", + "content": "Keep all Unreleased patch changelog entries: Keep all Unreleased patch changelog entries", + "raw": { + "question": "Keep all Unreleased patch changelog entries", + "chosen": "Keep all Unreleased patch changelog entries", + "alternatives": [], + "reasoning": "The only merge conflict combined independent user-visible fixes from PR 1268 and current main, all at patch impact, so no entry should be dropped." + }, + "significance": "high" + } + ] + } + ], + "retrospective": { + "summary": "Merged current origin/main into PR 1268, retained all Unreleased patch changelog entries, and passed all 129 Swift tests", + "approach": "Standard approach", + "confidence": 0.98 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "049ff657924e52309956da08695bbd5d57990739", + "endRef": "049ff657924e52309956da08695bbd5d57990739" + } +} diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md new file mode 100644 index 000000000..6f565850c --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md @@ -0,0 +1,35 @@ +# Trajectory: Address PR 1268 review feedback + +> **Status:** ✅ Completed +> **Task:** PR-1268 +> **Confidence:** 96% +> **Started:** July 15, 2026 at 03:06 PM +> **Completed:** July 15, 2026 at 03:11 PM + +--- + +## Summary + +Addressed PR 1268 review feedback by synchronizing requested channel stream registration before subscribe, documenting the ordering contract, bounding lifecycle tests with throwing deadlines, removing obsolete broker subscription state, applying Swift registry cleanups, and correcting duplicated trajectory prose. All 151 Swift tests pass. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Coordinate requested channel stream registration with subscribe + +- **Chose:** Coordinate requested channel stream registration with subscribe +- **Reasoning:** Preserves lossless subscription startup for consumers that request events first while keeping join-only subscriptions free of unused event buffers; documented the required ordering in API docs and quick start. + +--- + +## Chapters + +### 1. Work + +_Agent: default_ + +- Coordinate requested channel stream registration with subscribe: Coordinate requested channel stream registration with subscribe +- Addressed every valid new review thread: registration ordering, bounded test waits, obsolete broker state, Swift dictionary cleanups, and duplicated trajectory prose. Two Dictionary.Values suggestions remain intentionally unapplied because finishAll mutates registry values; the trajectory empty-commit warning misreads one conflict-resolution record as the whole PR. diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json new file mode 100644 index 000000000..88335bf3f --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json @@ -0,0 +1,67 @@ +{ + "id": "traj_ym4mjwfa0kmx", + "version": 1, + "task": { + "title": "Address PR 1268 review feedback", + "source": { + "system": "plain", + "id": "PR-1268" + } + }, + "status": "completed", + "startedAt": "2026-07-15T19:06:53.525Z", + "completedAt": "2026-07-15T19:11:25.090Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-07-15T19:11:18.474Z" + } + ], + "chapters": [ + { + "id": "chap_dd5ahsddm9v4", + "title": "Work", + "agentName": "default", + "startedAt": "2026-07-15T19:11:18.474Z", + "endedAt": "2026-07-15T19:11:25.090Z", + "events": [ + { + "ts": 1784142678475, + "type": "decision", + "content": "Coordinate requested channel stream registration with subscribe: Coordinate requested channel stream registration with subscribe", + "raw": { + "question": "Coordinate requested channel stream registration with subscribe", + "chosen": "Coordinate requested channel stream registration with subscribe", + "alternatives": [], + "reasoning": "Preserves lossless subscription startup for consumers that request events first while keeping join-only subscriptions free of unused event buffers; documented the required ordering in API docs and quick start." + }, + "significance": "high" + }, + { + "ts": 1784142678946, + "type": "reflection", + "content": "Addressed every valid new review thread: registration ordering, bounded test waits, obsolete broker state, Swift dictionary cleanups, and duplicated trajectory prose. Two Dictionary.Values suggestions remain intentionally unapplied because finishAll mutates registry values; the trajectory empty-commit warning misreads one conflict-resolution record as the whole PR.", + "raw": { + "confidence": 0.94 + }, + "significance": "high", + "tags": ["confidence:0.94"] + } + ] + } + ], + "retrospective": { + "summary": "Addressed PR 1268 review feedback by synchronizing requested channel stream registration before subscribe, documenting the ordering contract, bounding lifecycle tests with throwing deadlines, removing obsolete broker subscription state, applying Swift registry cleanups, and correcting duplicated trajectory prose. All 151 Swift tests pass.", + "approach": "Standard approach", + "confidence": 0.96 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "de5c3fd53e953523a02c60223dd50d32e7c3902b", + "endRef": "de5c3fd53e953523a02c60223dd50d32e7c3902b" + } +} diff --git a/CHANGELOG.md b/CHANGELOG.md index 2214e5d2e..2f2c0a516 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- `AgentRelaySDK` and `AgentRelayBrokerSDK` (Swift) release cancelled async-stream consumers, bound event buffering, and avoid creating channel event queues for join-only subscriptions, preventing reconnect-driven memory growth. - `agent-relay-broker` retries the Codex model-detection spawn (`codex debug models`) on `ExecutableFileBusy` (`ETXTBSY`), so a concurrent `fork`/`exec` race under load no longer aborts detection and spuriously falls back away from the requested model. - `agent-relay-broker` bounds each initial Relaycast startup handshake attempt and retries on timeout with backoff (per-attempt deadline scaled by the configured workspace count), so a stalled backend connection no longer hangs `agent-relay node up`/`init` until an external supervisor kills the broker (which surfaced as an opaque "broker exited with code null during initial handshake"). Returned errors are surfaced immediately rather than replayed. Tunable via `AGENT_RELAY_HANDSHAKE_TIMEOUT_MS` and `AGENT_RELAY_HANDSHAKE_ATTEMPTS`. diff --git a/packages/sdk-swift/README.md b/packages/sdk-swift/README.md index 59bc87420..1d464a181 100644 --- a/packages/sdk-swift/README.md +++ b/packages/sdk-swift/README.md @@ -62,10 +62,11 @@ let registration = try await relay.registerOrRotate(name: "swift-agent") let agent = registration.asClient() let channel = agent.channel("general") +let events = channel.events try await channel.subscribe() try await channel.post("Hello from Swift") -for await event in channel.events { +for await event in events { print("\(event.from): \(event.body)") } ``` diff --git a/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift b/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift index 77a21f0be..fa99d4712 100644 --- a/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift +++ b/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift @@ -36,7 +36,72 @@ public struct AgentRegistration: Sendable { } } +private struct ContinuationRegistry { + typealias Continuation = AsyncStream.Continuation + + private(set) var active: [UUID: Continuation] = [:] + + var count: Int { active.count } + var continuations: Dictionary.Values { active.values } + + mutating func register(_ continuation: Continuation, id: UUID) { + active[id] = continuation + } + + mutating func unregister(id: UUID) { + active.removeValue(forKey: id) + } + + mutating func finishAll() { + for continuation in active.values { continuation.finish() } + active.removeAll() + } +} + +final class StreamLifecycle: @unchecked Sendable { + private let lock = NSLock() + private var generation: UInt64 = 0 + + func snapshot() -> UInt64 { + lock.lock() + defer { lock.unlock() } + return generation + } + + func advance() { + lock.lock() + generation &+= 1 + lock.unlock() + } +} + +private final class StreamRegistrationTasks: @unchecked Sendable { + private let lock = NSLock() + private var tasks: [UUID: Task] = [:] + + func insert(_ task: Task, id: UUID) { + lock.lock() + tasks[id] = task + lock.unlock() + } + + func remove(id: UUID) { + lock.lock() + tasks.removeValue(forKey: id) + lock.unlock() + } + + func takeAll() -> [Task] { + lock.lock() + defer { lock.unlock() } + let pending = Array(tasks.values) + tasks.removeAll() + return pending + } +} + actor BrokerCore { + nonisolated let streamLifecycle = StreamLifecycle() let apiKey: String let transport: any RelayTransportClient let http: any RelayHTTPClient @@ -44,10 +109,10 @@ actor BrokerCore { let decoder = JSONDecoder() private var routerTask: Task? - private var channelContinuations: [String: [AsyncStream.Continuation]] = [:] - private var brokerEventContinuations: [AsyncStream.Continuation] = [] - private var inboundMessageContinuations: [AsyncStream.Continuation] = [] - private var connectionStateContinuations: [AsyncStream.Continuation] = [] + private var channelContinuations: [String: ContinuationRegistry] = [:] + private var brokerEventContinuations = ContinuationRegistry() + private var inboundMessageContinuations = ContinuationRegistry() + private var connectionStateContinuations = ContinuationRegistry() init(apiKey: String, transport: any RelayTransportClient, http: any RelayHTTPClient) { self.apiKey = apiKey @@ -73,20 +138,64 @@ actor BrokerCore { notifyConnectionState(.connected) } - func registerChannelContinuation(_ continuation: AsyncStream.Continuation, for channel: String) { - channelContinuations[channel, default: []].append(continuation) + func registerChannelContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64, for channel: String) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + channelContinuations[channel, default: ContinuationRegistry()].register(continuation, id: id) + } + + func unregisterChannelContinuation(id: UUID, for channel: String) { + guard var registry = channelContinuations[channel] else { return } + registry.unregister(id: id) + channelContinuations[channel] = registry.count > 0 ? registry : nil + } + + func registerBrokerEventContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + brokerEventContinuations.register(continuation, id: id) + } + + func unregisterBrokerEventContinuation(id: UUID) { + brokerEventContinuations.unregister(id: id) + } + + func registerInboundMessageContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + inboundMessageContinuations.register(continuation, id: id) } - func registerBrokerEventContinuation(_ continuation: AsyncStream.Continuation) { - brokerEventContinuations.append(continuation) + func unregisterInboundMessageContinuation(id: UUID) { + inboundMessageContinuations.unregister(id: id) } - func registerInboundMessageContinuation(_ continuation: AsyncStream.Continuation) { - inboundMessageContinuations.append(continuation) + func registerConnectionStateContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + connectionStateContinuations.register(continuation, id: id) } - func registerConnectionStateContinuation(_ continuation: AsyncStream.Continuation) { - connectionStateContinuations.append(continuation) + func unregisterConnectionStateContinuation(id: UUID) { + connectionStateContinuations.unregister(id: id) + } + + func continuationCounts() -> (channels: Int, channelRegistries: Int, brokerEvents: Int, inbound: Int, connectionState: Int) { + ( + channels: channelContinuations.values.reduce(0) { $0 + $1.count }, + channelRegistries: channelContinuations.count, + brokerEvents: brokerEventContinuations.count, + inbound: inboundMessageContinuations.count, + connectionState: connectionStateContinuations.count + ) } func sendChannelPost(channel: String, text: String) async throws { @@ -201,20 +310,20 @@ actor BrokerCore { } func disconnect() async { + // Invalidate registrations scheduled before this disconnect, including + // tasks that have not reached the actor yet. + streamLifecycle.advance() routerTask?.cancel() routerTask = nil await transport.disconnect() notifyConnectionState(.disconnected) - for continuation in brokerEventContinuations { continuation.finish() } - brokerEventContinuations.removeAll() - for continuation in inboundMessageContinuations { continuation.finish() } - inboundMessageContinuations.removeAll() - for continuations in channelContinuations.values { - for continuation in continuations { continuation.finish() } + brokerEventContinuations.finishAll() + inboundMessageContinuations.finishAll() + for key in channelContinuations.keys { + channelContinuations[key]?.finishAll() } channelContinuations.removeAll() - for continuation in connectionStateContinuations { continuation.finish() } - connectionStateContinuations.removeAll() + connectionStateContinuations.finishAll() } private func sendMessageHTTP(_ payload: SendMessagePayload) async throws { @@ -250,8 +359,8 @@ actor BrokerCore { return value.addingPercentEncoding(withAllowedCharacters: allowed) ?? value } - private func notifyConnectionState(_ state: ConnectionStateChange) { - for continuation in connectionStateContinuations { + func notifyConnectionState(_ state: ConnectionStateChange) { + for continuation in connectionStateContinuations.continuations { continuation.yield(state) } } @@ -272,7 +381,7 @@ actor BrokerCore { } private func routeInboundMessage(_ message: InboundMessage) { - for continuation in inboundMessageContinuations { + for continuation in inboundMessageContinuations.continuations { continuation.yield(message) } if case .event(let event) = message { @@ -280,20 +389,22 @@ actor BrokerCore { } } - private func routeBrokerEvent(_ event: BrokerEvent, alreadyYieldedInbound: Bool = false) { + func routeBrokerEvent(_ event: BrokerEvent, alreadyYieldedInbound: Bool = false) { if !alreadyYieldedInbound { - for continuation in inboundMessageContinuations { + for continuation in inboundMessageContinuations.continuations { continuation.yield(.event(event)) } } - for continuation in brokerEventContinuations { + for continuation in brokerEventContinuations.continuations { continuation.yield(event) } if case .relayInbound(let relayEvent) = event { let message = RelayChannelEvent(from: relayEvent.from, body: relayEvent.body, threadId: relayEvent.threadId) - for continuation in channelContinuations[relayEvent.target] ?? [] { - continuation.yield(message) + if let continuations = channelContinuations[relayEvent.target]?.continuations { + for continuation in continuations { + continuation.yield(message) + } } } } @@ -409,6 +520,12 @@ public final class AgentRelayBrokerClient: @unchecked Sendable { } } + init(core: BrokerCore, apiKey: String, baseURL: URL) { + self.core = core + self.apiKey = apiKey + self.baseURL = baseURL + } + public func channel(_ name: String) -> Channel { Channel(name: name, core: core) } @@ -554,21 +671,54 @@ public final class AgentRelayBrokerClient: @unchecked Sendable { await core.disconnect() } + /// Broker events. If the consumer falls behind, the oldest event is + /// dropped so that at most the newest 256 events are buffered. public var brokerEvents: AsyncStream { - AsyncStream { continuation in - Task { await core.registerBrokerEventContinuation(continuation) } + let id = UUID() + let core = self.core + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(256)) { continuation in + let registrationTask = Task { + await core.registerBrokerEventContinuation(continuation, id: id, generation: generation) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + Task { await core.unregisterBrokerEventContinuation(id: id) } + } } } + /// Inbound messages. If the consumer falls behind, the oldest message is + /// dropped so that at most the newest 256 messages are buffered. public var inboundMessages: AsyncStream { - AsyncStream { continuation in - Task { await core.registerInboundMessageContinuation(continuation) } + let id = UUID() + let core = self.core + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(256)) { continuation in + let registrationTask = Task { + await core.registerInboundMessageContinuation(continuation, id: id, generation: generation) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + Task { await core.unregisterInboundMessageContinuation(id: id) } + } } } + /// Connection changes are latest-state-only: an unread state is replaced + /// when a newer state arrives, so at most one value is buffered. public var connectionState: AsyncStream { - AsyncStream { continuation in - Task { await core.registerConnectionStateContinuation(continuation) } + let id = UUID() + let core = self.core + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in + let registrationTask = Task { + await core.registerConnectionStateContinuation(continuation, id: id, generation: generation) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + Task { await core.unregisterConnectionStateContinuation(id: id) } + } } } @@ -587,24 +737,41 @@ public typealias AgentRelayClient = AgentRelayBrokerClient public final class Channel: @unchecked Sendable { public let name: String private let core: BrokerCore - private let continuationRef: AsyncStream.Continuation? - private let subscriptionLock = NSLock() - private var subscribed = false - public let events: AsyncStream + private let streamRegistrations = StreamRegistrationTasks() + + /// Channel delivery is registered only when the stream is requested. + /// Slow consumers retain at most the newest 256 events. + /// + /// Request this stream before calling ``subscribe()`` when events emitted + /// immediately after subscription must be retained. `subscribe()` waits + /// for every stream requested up to that point to finish registering. + public var events: AsyncStream { + let id = UUID() + let core = self.core + let name = self.name + let streamRegistrations = self.streamRegistrations + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(256)) { continuation in + let registrationTask = Task { + await core.registerChannelContinuation(continuation, id: id, generation: generation, for: name) + } + streamRegistrations.insert(registrationTask, id: id) + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + streamRegistrations.remove(id: id) + Task { await core.unregisterChannelContinuation(id: id, for: name) } + } + } + } init(name: String, core: BrokerCore) { self.name = name self.core = core - var continuation: AsyncStream.Continuation? - self.events = AsyncStream { incoming in - continuation = incoming - } - self.continuationRef = continuation } public func subscribe() async throws { - if markSubscribed(), let continuationRef { - await core.registerChannelContinuation(continuationRef, for: name) + for registrationTask in streamRegistrations.takeAll() { + await registrationTask.value } try await core.ensureConnected() } @@ -612,14 +779,6 @@ public final class Channel: @unchecked Sendable { public func post(_ text: String) async throws { try await core.sendChannelPost(channel: name, text: text) } - - private func markSubscribed() -> Bool { - subscriptionLock.lock() - defer { subscriptionLock.unlock() } - guard !subscribed else { return false } - subscribed = true - return true - } } public final class AgentClient: @unchecked Sendable { diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift b/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift index e4776a578..1838447cb 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift @@ -476,21 +476,54 @@ public final class AgentClient: @unchecked Sendable { try await rest.dmHistory(with: agent, limit: limit, before: before) } + /// Realtime events. If the consumer falls behind, the oldest event is + /// dropped so that at most the newest 256 events are buffered. public var events: AsyncStream { - AsyncStream { continuation in - Task { await core.registerEventContinuation(continuation) } + let id = UUID() + let core = self.core + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(256)) { continuation in + let registrationTask = Task { + await core.registerEventContinuation(continuation, id: id, generation: generation) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + Task { await core.unregisterEventContinuation(id: id) } + } } } + /// Inbound messages. If the consumer falls behind, the oldest message is + /// dropped so that at most the newest 256 messages are buffered. public var inboundMessages: AsyncStream { - AsyncStream { continuation in - Task { await core.registerInboundMessageContinuation(continuation) } + let id = UUID() + let core = self.core + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(256)) { continuation in + let registrationTask = Task { + await core.registerInboundMessageContinuation(continuation, id: id, generation: generation) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + Task { await core.unregisterInboundMessageContinuation(id: id) } + } } } + /// Connection changes are latest-state-only: an unread state is replaced + /// when a newer state arrives, so at most one value is buffered. public var connectionState: AsyncStream { - AsyncStream { continuation in - Task { await core.registerConnectionStateContinuation(continuation) } + let id = UUID() + let core = self.core + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(1)) { continuation in + let registrationTask = Task { + await core.registerConnectionStateContinuation(continuation, id: id, generation: generation) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + Task { await core.unregisterConnectionStateContinuation(id: id) } + } } } } @@ -498,22 +531,41 @@ public final class AgentClient: @unchecked Sendable { public final class RelayChannel: @unchecked Sendable { public let name: String private let core: HostedParticipantCore - private let continuationRef: AsyncStream.Continuation? - public let events: AsyncStream + private let streamRegistrations = StreamRegistrationTasks() + + /// Channel delivery is registered only when the stream is requested. + /// Slow consumers retain at most the newest 256 events. + /// + /// Request this stream before calling ``subscribe()`` when events emitted + /// immediately after subscription must be retained. `subscribe()` waits + /// for every stream requested up to that point to finish registering. + public var events: AsyncStream { + let id = UUID() + let core = self.core + let name = self.name + let streamRegistrations = self.streamRegistrations + let generation = core.streamLifecycle.snapshot() + return AsyncStream(bufferingPolicy: .bufferingNewest(256)) { continuation in + let registrationTask = Task { + await core.registerChannelContinuation(continuation, id: id, generation: generation, for: name) + } + streamRegistrations.insert(registrationTask, id: id) + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + streamRegistrations.remove(id: id) + Task { await core.unregisterChannelContinuation(id: id, for: name) } + } + } + } init(name: String, core: HostedParticipantCore) { self.name = name self.core = core - var continuation: AsyncStream.Continuation? - self.events = AsyncStream { incoming in - continuation = incoming - } - self.continuationRef = continuation } public func subscribe() async throws { - if let continuationRef { - await core.registerChannelContinuation(continuationRef, for: name) + for registrationTask in streamRegistrations.takeAll() { + await registrationTask.value } try await core.subscribe(channel: name) } @@ -530,12 +582,77 @@ private struct RegisteredAction: Sendable { let handler: RelayActionHandler } +private struct ContinuationRegistry { + typealias Continuation = AsyncStream.Continuation + + private(set) var active: [UUID: Continuation] = [:] + + var count: Int { active.count } + var continuations: Dictionary.Values { active.values } + + mutating func register(_ continuation: Continuation, id: UUID) { + active[id] = continuation + } + + mutating func unregister(id: UUID) { + active.removeValue(forKey: id) + } + + mutating func finishAll() { + for continuation in active.values { continuation.finish() } + active.removeAll() + } +} + +final class StreamLifecycle: @unchecked Sendable { + private let lock = NSLock() + private var generation: UInt64 = 0 + + func snapshot() -> UInt64 { + lock.lock() + defer { lock.unlock() } + return generation + } + + func advance() { + lock.lock() + generation &+= 1 + lock.unlock() + } +} + +private final class StreamRegistrationTasks: @unchecked Sendable { + private let lock = NSLock() + private var tasks: [UUID: Task] = [:] + + func insert(_ task: Task, id: UUID) { + lock.lock() + tasks[id] = task + lock.unlock() + } + + func remove(id: UUID) { + lock.lock() + tasks.removeValue(forKey: id) + lock.unlock() + } + + func takeAll() -> [Task] { + lock.lock() + defer { lock.unlock() } + let pending = Array(tasks.values) + tasks.removeAll() + return pending + } +} + /// Higher-level glue kept ON TOP of the relaycast engine SDK: the /// action-dispatch loop, channel-event normalization into `RelayChannelEvent`, /// AsyncStream fan-out, and subscription bookkeeping. The realtime socket and /// HTTP calls are delegated to the wrapped `Relaycast.AgentClient` / /// `Relaycast.RelayCast`. actor HostedParticipantCore { + nonisolated let streamLifecycle = StreamLifecycle() /// How the per-agent engine is obtained. `.ready` is used by `reconnect`, /// which already has a live engine; `.deferred` builds the engine lazily via /// `relay.asAgent(token)` on first connect so that `asAgent`/configuration @@ -558,10 +675,10 @@ actor HostedParticipantCore { private var connected = false private var listenersInstalled = false private var subscribedChannels: Set = [] - private var channelContinuations: [String: [AsyncStream.Continuation]] = [:] - private var inboundMessageContinuations: [AsyncStream.Continuation] = [] - private var eventContinuations: [AsyncStream.Continuation] = [] - private var connectionStateContinuations: [AsyncStream.Continuation] = [] + private var channelContinuations: [String: ContinuationRegistry] = [:] + private var inboundMessageContinuations = ContinuationRegistry() + private var eventContinuations = ContinuationRegistry() + private var connectionStateContinuations = ContinuationRegistry() private var actionHandlers: [String: RegisteredAction] = [:] private var unsubscribeHandlers: [() -> Void] = [] @@ -640,6 +757,9 @@ actor HostedParticipantCore { } func disconnect() async { + // Invalidate registrations scheduled before this disconnect, including + // tasks that have not reached the actor yet. + streamLifecycle.advance() // Only tear down an engine that was actually built/connected; building // one here just to disconnect it would be pointless (and could throw). if let resolvedEngine { @@ -654,32 +774,75 @@ actor HostedParticipantCore { eventPump?.cancel() eventPump = nil notifyConnectionState(.disconnected) - for continuations in channelContinuations.values { - for continuation in continuations { continuation.finish() } + for key in channelContinuations.keys { + channelContinuations[key]?.finishAll() } channelContinuations.removeAll() - for continuation in inboundMessageContinuations { continuation.finish() } - inboundMessageContinuations.removeAll() - for continuation in eventContinuations { continuation.finish() } - eventContinuations.removeAll() - for continuation in connectionStateContinuations { continuation.finish() } - connectionStateContinuations.removeAll() + inboundMessageContinuations.finishAll() + eventContinuations.finishAll() + connectionStateContinuations.finishAll() + } + + func registerChannelContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64, for channel: String) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + let channel = Self.normalizeChannel(channel) + channelContinuations[channel, default: ContinuationRegistry()].register(continuation, id: id) } - func registerChannelContinuation(_ continuation: AsyncStream.Continuation, for channel: String) { - channelContinuations[Self.normalizeChannel(channel), default: []].append(continuation) + func unregisterChannelContinuation(id: UUID, for channel: String) { + let channel = Self.normalizeChannel(channel) + guard var registry = channelContinuations[channel] else { return } + registry.unregister(id: id) + channelContinuations[channel] = registry.count > 0 ? registry : nil } - func registerInboundMessageContinuation(_ continuation: AsyncStream.Continuation) { - inboundMessageContinuations.append(continuation) + func registerInboundMessageContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + inboundMessageContinuations.register(continuation, id: id) + } + + func unregisterInboundMessageContinuation(id: UUID) { + inboundMessageContinuations.unregister(id: id) + } + + func registerEventContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + eventContinuations.register(continuation, id: id) } - func registerEventContinuation(_ continuation: AsyncStream.Continuation) { - eventContinuations.append(continuation) + func unregisterEventContinuation(id: UUID) { + eventContinuations.unregister(id: id) } - func registerConnectionStateContinuation(_ continuation: AsyncStream.Continuation) { - connectionStateContinuations.append(continuation) + func registerConnectionStateContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { + guard !Task.isCancelled, generation == streamLifecycle.snapshot() else { + continuation.finish() + return + } + connectionStateContinuations.register(continuation, id: id) + } + + func unregisterConnectionStateContinuation(id: UUID) { + connectionStateContinuations.unregister(id: id) + } + + func continuationCounts() -> (channels: Int, channelRegistries: Int, inbound: Int, events: Int, connectionState: Int) { + ( + channels: channelContinuations.values.reduce(0) { $0 + $1.count }, + channelRegistries: channelContinuations.count, + inbound: inboundMessageContinuations.count, + events: eventContinuations.count, + connectionState: connectionStateContinuations.count + ) } func subscribe(channel: String) async throws { @@ -850,8 +1013,8 @@ actor HostedParticipantCore { // MARK: - Event routing (glue kept on top of the engine) - private func routeEvent(_ event: RelayEvent) { - for continuation in eventContinuations { + func routeEvent(_ event: RelayEvent) { + for continuation in eventContinuations.continuations { continuation.yield(event) } @@ -865,12 +1028,14 @@ actor HostedParticipantCore { } guard let message = channelEvent(from: event) else { return } - for continuation in inboundMessageContinuations { + for continuation in inboundMessageContinuations.continuations { continuation.yield(message) } if let channel = message.channel { - for continuation in channelContinuations[Self.normalizeChannel(channel)] ?? [] { - continuation.yield(message) + if let continuations = channelContinuations[Self.normalizeChannel(channel)]?.continuations { + for continuation in continuations { + continuation.yield(message) + } } } } @@ -1005,8 +1170,8 @@ actor HostedParticipantCore { return ["value": .string(output)] } - private func notifyConnectionState(_ state: ConnectionStateChange) { - for continuation in connectionStateContinuations { + func notifyConnectionState(_ state: ConnectionStateChange) { + for continuation in connectionStateContinuations.continuations { continuation.yield(state) } } diff --git a/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift b/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift index 404423d4a..5d9912a83 100644 --- a/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift +++ b/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift @@ -109,6 +109,36 @@ private func jsonObject(_ data: Data) throws -> [String: Any] { } final class AgentRelayBrokerSDKTests: XCTestCase { + private enum AsyncTestTimeout: Error { + case exceeded + } + + private func waitUntil(_ predicate: @escaping () async -> Bool) async throws { + for _ in 0..<2_000 { + if await predicate() { return } + try await Task.sleep(nanoseconds: 1_000_000) + } + XCTFail("Timed out waiting for async condition") + throw AsyncTestTimeout.exceeded + } + + private func withTimeout( + _ operation: @escaping @Sendable () async -> T + ) async throws -> T { + try await withThrowingTaskGroup(of: T.self) { group in + group.addTask { await operation() } + group.addTask { + try await Task.sleep(nanoseconds: 2_000_000_000) + throw AsyncTestTimeout.exceeded + } + guard let result = try await group.next() else { + throw AsyncTestTimeout.exceeded + } + group.cancelAll() + return result + } + } + func testAgentRelayBrokerClientInit() { let client = AgentRelayBrokerClient(apiKey: "rk_test_key") XCTAssertEqual(client.apiKey, "rk_test_key") @@ -236,13 +266,16 @@ final class AgentRelayBrokerSDKTests: XCTestCase { let core = BrokerCore(apiKey: "rk_test", transport: transport, http: MockRelayHTTP()) let channel = Channel(name: "ops", core: core) let recorder = BrokerChannelRecorder() + let eventStream = channel.events let readTask = Task { - for await event in channel.events { + for await event in eventStream { await recorder.append(event) } } try await channel.subscribe() + let countsAfterSubscribe = await core.continuationCounts() + XCTAssertEqual(countsAfterSubscribe.channels, 1) try await channel.subscribe() await transport.emit( """ @@ -271,6 +304,152 @@ final class AgentRelayBrokerSDKTests: XCTestCase { XCTAssertEqual(events.first?.body, "hello") } + func testCancellingPublicStreamsRemovesContinuationsAcrossEpochs() async throws { + let core = BrokerCore(apiKey: "rk_test", transport: MockRelayTransport(), http: MockRelayHTTP()) + let client = AgentRelayBrokerClient( + core: core, + apiKey: "rk_test", + baseURL: URL(string: "http://localhost:3889")! + ) + + for _ in 0..<20 { + let eventTask = Task { for await _ in client.brokerEvents {} } + let inboundTask = Task { for await _ in client.inboundMessages {} } + let stateTask = Task { for await _ in client.connectionState {} } + try await waitUntil { + let counts = await core.continuationCounts() + return counts.brokerEvents == 1 && counts.inbound == 1 && counts.connectionState == 1 + } + + eventTask.cancel() + inboundTask.cancel() + stateTask.cancel() + _ = await (eventTask.value, inboundTask.value, stateTask.value) + try await waitUntil { + let counts = await core.continuationCounts() + return counts.brokerEvents == 0 && counts.inbound == 0 && counts.connectionState == 0 + } + } + } + + func testJoinOnlyChannelDoesNotRegisterEventContinuation() async throws { + let core = BrokerCore(apiKey: "rk_test", transport: MockRelayTransport(), http: MockRelayHTTP()) + let channel = Channel(name: "ops", core: core) + + try await channel.subscribe() + try await channel.subscribe() + + let counts = await core.continuationCounts() + XCTAssertEqual(counts.channels, 0) + } + + func testTerminationBeforeRegistrationDoesNotResurrectContinuation() async throws { + let core = BrokerCore(apiKey: "rk_test", transport: MockRelayTransport(), http: MockRelayHTTP()) + let id = UUID() + var continuationRef: AsyncStream.Continuation? + _ = AsyncStream { continuationRef = $0 } + + let continuation = try XCTUnwrap(continuationRef) + let generation = core.streamLifecycle.snapshot() + await Task { + withUnsafeCurrentTask { $0?.cancel() } + await core.registerBrokerEventContinuation(continuation, id: id, generation: generation) + }.value + + let counts = await core.continuationCounts() + XCTAssertEqual(counts.brokerEvents, 0) + } + + func testRegistrationFromBeforeDisconnectCannotResurrectStream() async throws { + let core = BrokerCore(apiKey: "rk_test", transport: MockRelayTransport(), http: MockRelayHTTP()) + let id = UUID() + let staleGeneration = core.streamLifecycle.snapshot() + var continuationRef: AsyncStream.Continuation? + let stream = AsyncStream { continuationRef = $0 } + + await core.disconnect() + await core.registerBrokerEventContinuation(try XCTUnwrap(continuationRef), id: id, generation: staleGeneration) + + let first = try await withTimeout { await stream.first(where: { _ in true }) } + XCTAssertNil(first) + let counts = await core.continuationCounts() + XCTAssertEqual(counts.brokerEvents, 0) + } + + func testDisconnectIsIdempotentAndFinishesLiveStreams() async throws { + let transport = MockRelayTransport() + let core = BrokerCore(apiKey: "rk_test", transport: transport, http: MockRelayHTTP()) + let channel = Channel(name: "ops", core: core) + let task = Task { for await _ in channel.events {} } + try await waitUntil { await core.continuationCounts().channels == 1 } + + await core.disconnect() + await core.disconnect() + _ = await task.value + + let counts = await core.continuationCounts() + XCTAssertEqual(counts.channels, 0) + } + + func testDisconnectDoesNotLeaveEmptyChannelRegistries() async throws { + let core = BrokerCore(apiKey: "rk_test", transport: MockRelayTransport(), http: MockRelayHTTP()) + + for epoch in 0..<20 { + let channel = Channel(name: "ops-\(epoch)", core: core) + let task = Task { for await _ in channel.events {} } + try await waitUntil { await core.continuationCounts().channels == 1 } + + await core.disconnect() + _ = await task.value + try await waitUntil { + let counts = await core.continuationCounts() + return counts.channels == 0 && counts.channelRegistries == 0 + } + } + } + + func testEventAndConnectionStateBuffersDropOldestValues() async throws { + let core = BrokerCore(apiKey: "rk_test", transport: MockRelayTransport(), http: MockRelayHTTP()) + let client = AgentRelayBrokerClient( + core: core, + apiKey: "rk_test", + baseURL: URL(string: "http://localhost:3889")! + ) + let eventStream = client.brokerEvents + let stateStream = client.connectionState + try await waitUntil { + let counts = await core.continuationCounts() + return counts.brokerEvents == 1 && counts.connectionState == 1 + } + + for index in 0..<300 { + await core.routeBrokerEvent(.unknown(kind: "event.\(index)", rawJSON: nil)) + } + await core.notifyConnectionState(.connected) + await core.notifyConnectionState(.disconnected) + + let eventKinds = try await withTimeout { + var eventIterator = eventStream.makeAsyncIterator() + var kinds: [String] = [] + for _ in 0..<256 { + guard case .unknown(let kind, _)? = await eventIterator.next() else { break } + kinds.append(kind) + } + return kinds + } + XCTAssertEqual(eventKinds.count, 256) + XCTAssertEqual(eventKinds.first, "event.44") + XCTAssertEqual(eventKinds.last, "event.299") + + let state = try await withTimeout { + var stateIterator = stateStream.makeAsyncIterator() + return await stateIterator.next() + } + guard case .disconnected? = state else { + return XCTFail("Expected only the newest connection state") + } + } + // MARK: - Broker control & observability func testResolveAPIURLPreservesEmbeddedQuery() { diff --git a/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift b/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift index b713f62ff..3805a8dfa 100644 --- a/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift +++ b/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift @@ -9,6 +9,48 @@ import Relaycast /// we verify the facade configuration and the bridging glue that keeps /// AgentRelaySDK's public surface intact on top of relaycast. final class HostedParticipantSDKTests: XCTestCase { + private enum AsyncTestTimeout: Error { + case exceeded + } + + private func makeParticipantCore() throws -> HostedParticipantCore { + HostedParticipantCore( + engineSource: .deferred( + relayResult: Result { try Relaycast.RelayCast(options: Relaycast.RelayCastOptions(apiKey: "rk_test")) }, + token: "at_test" + ), + agentId: "a1", + agentName: "alice", + token: "at_test", + baseURL: URL(string: "https://cast.agentrelay.com")! + ) + } + + private func waitUntil(_ predicate: @escaping () async -> Bool) async throws { + for _ in 0..<2_000 { + if await predicate() { return } + try await Task.sleep(nanoseconds: 1_000_000) + } + XCTFail("Timed out waiting for async condition") + throw AsyncTestTimeout.exceeded + } + + private func withTimeout( + _ operation: @escaping @Sendable () async -> T + ) async throws -> T { + try await withThrowingTaskGroup(of: T.self) { group in + group.addTask { await operation() } + group.addTask { + try await Task.sleep(nanoseconds: 2_000_000_000) + throw AsyncTestTimeout.exceeded + } + guard let result = try await group.next() else { + throw AsyncTestTimeout.exceeded + } + group.cancelAll() + return result + } + } // MARK: - Facade configuration @@ -230,4 +272,148 @@ final class HostedParticipantSDKTests: XCTestCase { XCTAssertEqual(handle.name, "echo") await handle.unregister() } + + // MARK: - AsyncStream lifecycle + + func testCancellingPublicStreamsRemovesContinuationsAcrossEpochs() async throws { + let core = try makeParticipantCore() + let client = AgentClient(core: core, id: "a1", name: "alice", token: "at_test") + + for _ in 0..<20 { + let eventTask = Task { for await _ in client.events {} } + let inboundTask = Task { for await _ in client.inboundMessages {} } + let stateTask = Task { for await _ in client.connectionState {} } + + try await waitUntil { + let counts = await core.continuationCounts() + return counts.events == 1 && counts.inbound == 1 && counts.connectionState == 1 + } + + eventTask.cancel() + inboundTask.cancel() + stateTask.cancel() + _ = await (eventTask.value, inboundTask.value, stateTask.value) + + try await waitUntil { + let counts = await core.continuationCounts() + return counts.events == 0 && counts.inbound == 0 && counts.connectionState == 0 + } + } + } + + func testChannelDoesNotRegisterUntilEventsAreRequested() async throws { + let core = try makeParticipantCore() + let channel = RelayChannel(name: "ops", core: core) + let initialCounts = await core.continuationCounts() + XCTAssertEqual(initialCounts.channels, 0) + + let task = Task { for await _ in channel.events {} } + try await waitUntil { await core.continuationCounts().channels == 1 } + + task.cancel() + _ = await task.value + try await waitUntil { await core.continuationCounts().channels == 0 } + } + + func testTerminationBeforeRegistrationDoesNotResurrectContinuation() async throws { + let core = try makeParticipantCore() + let id = UUID() + var continuationRef: AsyncStream.Continuation? + _ = AsyncStream { continuationRef = $0 } + + let continuation = try XCTUnwrap(continuationRef) + let generation = core.streamLifecycle.snapshot() + await Task { + withUnsafeCurrentTask { $0?.cancel() } + await core.registerEventContinuation(continuation, id: id, generation: generation) + }.value + + let counts = await core.continuationCounts() + XCTAssertEqual(counts.events, 0) + } + + func testRegistrationFromBeforeDisconnectCannotResurrectStream() async throws { + let core = try makeParticipantCore() + let id = UUID() + let staleGeneration = core.streamLifecycle.snapshot() + var continuationRef: AsyncStream.Continuation? + let stream = AsyncStream { continuationRef = $0 } + + await core.disconnect() + await core.registerEventContinuation(try XCTUnwrap(continuationRef), id: id, generation: staleGeneration) + + let first = try await withTimeout { await stream.first(where: { _ in true }) } + XCTAssertNil(first) + let counts = await core.continuationCounts() + XCTAssertEqual(counts.events, 0) + } + + func testDisconnectIsIdempotentAndFinishesLiveStreams() async throws { + let core = try makeParticipantCore() + let client = AgentClient(core: core, id: "a1", name: "alice", token: "at_test") + let task = Task { for await _ in client.events {} } + try await waitUntil { await core.continuationCounts().events == 1 } + + await client.disconnect() + await client.disconnect() + _ = await task.value + + let counts = await core.continuationCounts() + XCTAssertEqual(counts.events, 0) + } + + func testDisconnectDoesNotLeaveEmptyChannelRegistries() async throws { + let core = try makeParticipantCore() + + for epoch in 0..<20 { + let channel = RelayChannel(name: "ops-\(epoch)", core: core) + let task = Task { for await _ in channel.events {} } + try await waitUntil { await core.continuationCounts().channels == 1 } + + await core.disconnect() + _ = await task.value + try await waitUntil { + let counts = await core.continuationCounts() + return counts.channels == 0 && counts.channelRegistries == 0 + } + } + } + + func testEventAndConnectionStateBuffersDropOldestValues() async throws { + let core = try makeParticipantCore() + let client = AgentClient(core: core, id: "a1", name: "alice", token: "at_test") + let eventStream = client.events + let stateStream = client.connectionState + try await waitUntil { + let counts = await core.continuationCounts() + return counts.events == 1 && counts.connectionState == 1 + } + + for index in 0..<300 { + await core.routeEvent(RelayEvent(type: "event.\(index)")) + } + await core.notifyConnectionState(.connected) + await core.notifyConnectionState(.disconnected) + + let eventTypes = try await withTimeout { + var eventIterator = eventStream.makeAsyncIterator() + var types: [String] = [] + for _ in 0..<256 { + guard let event = await eventIterator.next() else { break } + types.append(event.type) + } + return types + } + XCTAssertEqual(eventTypes.count, 256) + XCTAssertEqual(eventTypes.first, "event.44") + XCTAssertEqual(eventTypes.last, "event.299") + + let state = try await withTimeout { + var stateIterator = stateStream.makeAsyncIterator() + return await stateIterator.next() + } + guard case .disconnected? = state else { + return XCTFail("Expected only the newest connection state") + } + } }