From 049ff657924e52309956da08695bbd5d57990739 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 15 Jul 2026 06:54:16 -0400 Subject: [PATCH 1/5] fix Swift async stream lifecycle leaks --- .../2026-07/traj_bwj7uf3ksgz6/summary.md | 36 +++ .../2026-07/traj_bwj7uf3ksgz6/trajectory.json | 65 +++++ CHANGELOG.md | 6 +- .../AgentRelayBrokerClient.swift | 222 ++++++++++++++---- .../AgentRelaySDK/AgentRelayClient.swift | 215 +++++++++++++---- .../AgentRelayBrokerSDKTests.swift | 151 ++++++++++++ .../AgentRelaySDKTests.swift | 158 +++++++++++++ 7 files changed, 767 insertions(+), 86 deletions(-) create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json 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..d5f2cfa02 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md @@ -0,0 +1,36 @@ +# 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..55522edce --- /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" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index ef10283e6..974fc8021 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### 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. ## [10.2.0] - 2026-07-14 diff --git a/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift b/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift index 2f3706403..a2beaedd8 100644 --- a/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift +++ b/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift @@ -36,7 +36,47 @@ 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() + } +} + actor BrokerCore { + nonisolated let streamLifecycle = StreamLifecycle() let apiKey: String let transport: any RelayTransportClient let http: any RelayHTTPClient @@ -44,10 +84,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 +113,68 @@ 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) + if registry.count == 0 { + channelContinuations.removeValue(forKey: channel) + } else { + channelContinuations[channel] = registry + } + } + + 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 registerBrokerEventContinuation(_ continuation: AsyncStream.Continuation) { - brokerEventContinuations.append(continuation) + func unregisterBrokerEventContinuation(id: UUID) { + brokerEventContinuations.unregister(id: id) } - 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 registerConnectionStateContinuation(_ continuation: AsyncStream.Continuation) { - connectionStateContinuations.append(continuation) + func unregisterInboundMessageContinuation(id: UUID) { + inboundMessageContinuations.unregister(id: id) + } + + 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, 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 { @@ -123,20 +211,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 { @@ -152,8 +240,8 @@ actor BrokerCore { } } - private func notifyConnectionState(_ state: ConnectionStateChange) { - for continuation in connectionStateContinuations { + func notifyConnectionState(_ state: ConnectionStateChange) { + for continuation in connectionStateContinuations.continuations { continuation.yield(state) } } @@ -174,7 +262,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 { @@ -182,19 +270,19 @@ 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] ?? [] { + for continuation in channelContinuations[relayEvent.target]?.continuations ?? [:].values { continuation.yield(message) } } @@ -273,6 +361,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) } @@ -298,21 +392,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) } + } } } @@ -331,25 +458,34 @@ 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 + + /// Channel delivery is registered only when the stream is requested. + /// Slow consumers retain at most the newest 256 events. + public var events: AsyncStream { + let id = UUID() + let core = self.core + let name = self.name + 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) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + 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) - } + _ = markSubscribed() try await core.ensureConnected() } diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift b/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift index e4776a578..126587440 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,23 +531,31 @@ 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 + + /// Channel delivery is registered only when the stream is requested. + /// Slow consumers retain at most the newest 256 events. + public var events: AsyncStream { + let id = UUID() + let core = self.core + let name = self.name + 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) + } + continuation.onTermination = { @Sendable _ in + registrationTask.cancel() + 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) - } try await core.subscribe(channel: name) } @@ -530,12 +571,52 @@ 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() + } +} + /// 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 +639,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 +721,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 +738,79 @@ 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, for channel: String) { - channelContinuations[Self.normalizeChannel(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 + } + let channel = Self.normalizeChannel(channel) + channelContinuations[channel, default: ContinuationRegistry()].register(continuation, id: id) + } + + func unregisterChannelContinuation(id: UUID, for channel: String) { + let channel = Self.normalizeChannel(channel) + guard var registry = channelContinuations[channel] else { return } + registry.unregister(id: id) + if registry.count == 0 { + channelContinuations.removeValue(forKey: channel) + } else { + channelContinuations[channel] = registry + } } - 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 registerEventContinuation(_ continuation: AsyncStream.Continuation) { - eventContinuations.append(continuation) + func unregisterInboundMessageContinuation(id: UUID) { + inboundMessageContinuations.unregister(id: id) } - func registerConnectionStateContinuation(_ continuation: AsyncStream.Continuation) { - connectionStateContinuations.append(continuation) + 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 unregisterEventContinuation(id: UUID) { + eventContinuations.unregister(id: id) + } + + 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 +981,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,11 +996,11 @@ 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)] ?? [] { + for continuation in channelContinuations[Self.normalizeChannel(channel)]?.continuations ?? [:].values { continuation.yield(message) } } @@ -1005,8 +1136,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 1f16017a6..9238b712f 100644 --- a/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift +++ b/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift @@ -89,6 +89,14 @@ private func jsonObject(_ data: Data) throws -> [String: Any] { } final class AgentRelayBrokerSDKTests: XCTestCase { + private func waitUntil(_ predicate: @escaping () async -> Bool) async throws { + for _ in 0..<500 { + if await predicate() { return } + try await Task.sleep(nanoseconds: 1_000_000) + } + XCTFail("Timed out waiting for async condition") + } + func testAgentRelayBrokerClientInit() { let client = AgentRelayBrokerClient(apiKey: "rk_test_key") XCTAssertEqual(client.apiKey, "rk_test_key") @@ -222,6 +230,8 @@ final class AgentRelayBrokerSDKTests: XCTestCase { } } + try await waitUntil { await core.continuationCounts().channels == 1 } + try await channel.subscribe() try await channel.subscribe() await transport.emit( @@ -250,4 +260,145 @@ final class AgentRelayBrokerSDKTests: XCTestCase { XCTAssertEqual(events.count, 1) 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 = 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) + + var eventIterator = eventStream.makeAsyncIterator() + var eventKinds: [String] = [] + for _ in 0..<256 { + guard case .unknown(let kind, _)? = await eventIterator.next() else { + return XCTFail("Expected buffered unknown event") + } + eventKinds.append(kind) + } + XCTAssertEqual(eventKinds.first, "event.44") + XCTAssertEqual(eventKinds.last, "event.299") + + var stateIterator = stateStream.makeAsyncIterator() + guard case .disconnected? = await stateIterator.next() else { + return XCTFail("Expected only the newest connection state") + } + } } diff --git a/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift b/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift index b713f62ff..7d46c7ae8 100644 --- a/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift +++ b/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift @@ -10,6 +10,27 @@ import Relaycast /// AgentRelaySDK's public surface intact on top of relaycast. final class HostedParticipantSDKTests: XCTestCase { + 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..<500 { + if await predicate() { return } + try await Task.sleep(nanoseconds: 1_000_000) + } + XCTFail("Timed out waiting for async condition") + } + // MARK: - Facade configuration func testClientInitDefaultsToHostedGateway() { @@ -230,4 +251,141 @@ 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 = 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) + + var eventIterator = eventStream.makeAsyncIterator() + var eventTypes: [String] = [] + for _ in 0..<256 { + let event = await eventIterator.next() + eventTypes.append(try XCTUnwrap(event).type) + } + XCTAssertEqual(eventTypes.first, "event.44") + XCTAssertEqual(eventTypes.last, "event.299") + + var stateIterator = stateStream.makeAsyncIterator() + guard case .disconnected? = await stateIterator.next() else { + return XCTFail("Expected only the newest connection state") + } + } } From 38e00002cfba37694c8d0a9bd45d3c748d29c605 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 15 Jul 2026 11:03:58 +0000 Subject: [PATCH 2/5] style: auto-format with Prettier --- .../completed/2026-07/traj_bwj7uf3ksgz6/summary.md | 5 ++++- .../completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json | 2 +- .../completed/2026-07/traj_u94plhvjk6ew/summary.md | 4 +++- .../completed/2026-07/traj_u94plhvjk6ew/trajectory.json | 2 +- 4 files changed, 9 insertions(+), 4 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md index d5f2cfa02..2afa4cd3d 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/summary.md @@ -18,10 +18,12 @@ Fixed Swift SDK AsyncStream continuation leaks with cancellation cleanup, bounde ## 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. @@ -30,7 +32,8 @@ Fixed Swift SDK AsyncStream continuation leaks with cancellation cleanup, bounde ## Chapters ### 1. Work -*Agent: default* + +_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 index 55522edce..3a9283dc7 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_bwj7uf3ksgz6/trajectory.json @@ -62,4 +62,4 @@ "startRef": "bd1531aaea1bdb442c6c6c02498dc032cd0aecf6", "endRef": "bd1531aaea1bdb442c6c6c02498dc032cd0aecf6" } -} \ No newline at end of file +} diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md index 7eb9f92d1..17c4a4ef1 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/summary.md @@ -18,6 +18,7 @@ Merged current origin/main into PR 1268, retained all Unreleased patch changelog ## 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. @@ -26,6 +27,7 @@ Merged current origin/main into PR 1268, retained all Unreleased patch changelog ## Chapters ### 1. Work -*Agent: default* + +_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 index 2387ca04d..c7aaf024d 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_u94plhvjk6ew/trajectory.json @@ -50,4 +50,4 @@ "startRef": "049ff657924e52309956da08695bbd5d57990739", "endRef": "049ff657924e52309956da08695bbd5d57990739" } -} \ No newline at end of file +} From de5c3fd53e953523a02c60223dd50d32e7c3902b Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 15 Jul 2026 11:32:57 +0000 Subject: [PATCH 3/5] style: auto-format with Prettier --- .../completed/2026-07/traj_nt7clb37dtr7/summary.md | 4 +++- .../completed/2026-07/traj_nt7clb37dtr7/trajectory.json | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md index 8ca5b050d..9be6058dd 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md @@ -18,6 +18,7 @@ Merged current main into PR 1268, composed broker parity additions with stream l ## 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. @@ -26,6 +27,7 @@ Merged current main into PR 1268, composed broker parity additions with stream l ## Chapters ### 1. Work -*Agent: default* + +_Agent: default_ - Compose broker parity APIs with stream lifecycle registries: Compose broker parity APIs with stream lifecycle registries diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json index 1f8f4eb1d..65ad594c0 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/trajectory.json @@ -50,4 +50,4 @@ "startRef": "38e00002cfba37694c8d0a9bd45d3c748d29c605", "endRef": "38e00002cfba37694c8d0a9bd45d3c748d29c605" } -} \ No newline at end of file +} From ad259d4f3cb639cb5814db474dea612786d31754 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 15 Jul 2026 15:11:46 -0400 Subject: [PATCH 4/5] fix(swift): address stream lifecycle review feedback --- .../2026-07/traj_nt7clb37dtr7/summary.md | 2 +- .../2026-07/traj_ym4mjwfa0kmx/summary.md | 33 +++++++++ .../2026-07/traj_ym4mjwfa0kmx/trajectory.json | 69 +++++++++++++++++++ packages/sdk-swift/README.md | 3 +- .../AgentRelayBrokerClient.swift | 59 +++++++++++----- .../AgentRelaySDK/AgentRelayClient.swift | 48 +++++++++++-- .../AgentRelayBrokerSDKTests.swift | 54 +++++++++++---- .../AgentRelaySDKTests.swift | 46 ++++++++++--- 8 files changed, 265 insertions(+), 49 deletions(-) create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md create mode 100644 .agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md index 9be6058dd..6afeb62e9 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_nt7clb37dtr7/summary.md @@ -30,4 +30,4 @@ Merged current main into PR 1268, composed broker parity additions with stream l _Agent: default_ -- Compose broker parity APIs with stream lifecycle registries: Compose broker parity APIs with stream lifecycle registries +- Composed broker parity APIs with stream lifecycle registries while preserving connection-state fan-out. 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..8d10f88da --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md @@ -0,0 +1,33 @@ +# 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..76aba3c63 --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json @@ -0,0 +1,69 @@ +{ + "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" + } +} \ No newline at end of file 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 c571155ab..fa99d4712 100644 --- a/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift +++ b/packages/sdk-swift/Sources/AgentRelayBrokerSDK/AgentRelayBrokerClient.swift @@ -75,6 +75,31 @@ final class StreamLifecycle: @unchecked Sendable { } } +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 @@ -124,11 +149,7 @@ actor BrokerCore { func unregisterChannelContinuation(id: UUID, for channel: String) { guard var registry = channelContinuations[channel] else { return } registry.unregister(id: id) - if registry.count == 0 { - channelContinuations.removeValue(forKey: channel) - } else { - channelContinuations[channel] = registry - } + channelContinuations[channel] = registry.count > 0 ? registry : nil } func registerBrokerEventContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { @@ -380,8 +401,10 @@ actor BrokerCore { if case .relayInbound(let relayEvent) = event { let message = RelayChannelEvent(from: relayEvent.from, body: relayEvent.body, threadId: relayEvent.threadId) - for continuation in channelContinuations[relayEvent.target]?.continuations ?? [:].values { - continuation.yield(message) + if let continuations = channelContinuations[relayEvent.target]?.continuations { + for continuation in continuations { + continuation.yield(message) + } } } } @@ -714,22 +737,28 @@ public typealias AgentRelayClient = AgentRelayBrokerClient public final class Channel: @unchecked Sendable { public let name: String private let core: BrokerCore - private let subscriptionLock = NSLock() - private var subscribed = false + 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) } } } @@ -741,21 +770,15 @@ public final class Channel: @unchecked Sendable { } public func subscribe() async throws { - _ = markSubscribed() + for registrationTask in streamRegistrations.takeAll() { + await registrationTask.value + } try await core.ensureConnected() } 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 126587440..1838447cb 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift @@ -531,20 +531,28 @@ public final class AgentClient: @unchecked Sendable { public final class RelayChannel: @unchecked Sendable { public let name: String private let core: HostedParticipantCore + 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) } } } @@ -556,6 +564,9 @@ public final class RelayChannel: @unchecked Sendable { } public func subscribe() async throws { + for registrationTask in streamRegistrations.takeAll() { + await registrationTask.value + } try await core.subscribe(channel: name) } @@ -610,6 +621,31 @@ final class StreamLifecycle: @unchecked Sendable { } } +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 @@ -760,11 +796,7 @@ actor HostedParticipantCore { let channel = Self.normalizeChannel(channel) guard var registry = channelContinuations[channel] else { return } registry.unregister(id: id) - if registry.count == 0 { - channelContinuations.removeValue(forKey: channel) - } else { - channelContinuations[channel] = registry - } + channelContinuations[channel] = registry.count > 0 ? registry : nil } func registerInboundMessageContinuation(_ continuation: AsyncStream.Continuation, id: UUID, generation: UInt64) { @@ -1000,8 +1032,10 @@ actor HostedParticipantCore { continuation.yield(message) } if let channel = message.channel { - for continuation in channelContinuations[Self.normalizeChannel(channel)]?.continuations ?? [:].values { - continuation.yield(message) + if let continuations = channelContinuations[Self.normalizeChannel(channel)]?.continuations { + for continuation in continuations { + continuation.yield(message) + } } } } diff --git a/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift b/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift index 8cdac5d8e..5d9912a83 100644 --- a/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift +++ b/packages/sdk-swift/Tests/AgentRelayBrokerSDKTests/AgentRelayBrokerSDKTests.swift @@ -109,12 +109,34 @@ 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..<500 { + 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() { @@ -244,15 +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 waitUntil { await core.continuationCounts().channels == 1 } - try await channel.subscribe() + let countsAfterSubscribe = await core.continuationCounts() + XCTAssertEqual(countsAfterSubscribe.channels, 1) try await channel.subscribe() await transport.emit( """ @@ -347,7 +370,7 @@ final class AgentRelayBrokerSDKTests: XCTestCase { await core.disconnect() await core.registerBrokerEventContinuation(try XCTUnwrap(continuationRef), id: id, generation: staleGeneration) - let first = await stream.first(where: { _ in true }) + let first = try await withTimeout { await stream.first(where: { _ in true }) } XCTAssertNil(first) let counts = await core.continuationCounts() XCTAssertEqual(counts.brokerEvents, 0) @@ -405,19 +428,24 @@ final class AgentRelayBrokerSDKTests: XCTestCase { await core.notifyConnectionState(.connected) await core.notifyConnectionState(.disconnected) - var eventIterator = eventStream.makeAsyncIterator() - var eventKinds: [String] = [] - for _ in 0..<256 { - guard case .unknown(let kind, _)? = await eventIterator.next() else { - return XCTFail("Expected buffered unknown event") + 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) } - eventKinds.append(kind) + return kinds } + XCTAssertEqual(eventKinds.count, 256) XCTAssertEqual(eventKinds.first, "event.44") XCTAssertEqual(eventKinds.last, "event.299") - var stateIterator = stateStream.makeAsyncIterator() - guard case .disconnected? = await stateIterator.next() else { + 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") } } diff --git a/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift b/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift index 7d46c7ae8..3805a8dfa 100644 --- a/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift +++ b/packages/sdk-swift/Tests/AgentRelaySDKTests/AgentRelaySDKTests.swift @@ -9,6 +9,9 @@ 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( @@ -24,11 +27,29 @@ final class HostedParticipantSDKTests: XCTestCase { } private func waitUntil(_ predicate: @escaping () async -> Bool) async throws { - for _ in 0..<500 { + 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 @@ -321,7 +342,7 @@ final class HostedParticipantSDKTests: XCTestCase { await core.disconnect() await core.registerEventContinuation(try XCTUnwrap(continuationRef), id: id, generation: staleGeneration) - let first = await stream.first(where: { _ in true }) + let first = try await withTimeout { await stream.first(where: { _ in true }) } XCTAssertNil(first) let counts = await core.continuationCounts() XCTAssertEqual(counts.events, 0) @@ -374,17 +395,24 @@ final class HostedParticipantSDKTests: XCTestCase { await core.notifyConnectionState(.connected) await core.notifyConnectionState(.disconnected) - var eventIterator = eventStream.makeAsyncIterator() - var eventTypes: [String] = [] - for _ in 0..<256 { - let event = await eventIterator.next() - eventTypes.append(try XCTUnwrap(event).type) + 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") - var stateIterator = stateStream.makeAsyncIterator() - guard case .disconnected? = await stateIterator.next() else { + 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") } } From 818c6cf81afc9536eab2bac6d69c024c3e209f06 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 15 Jul 2026 19:12:52 +0000 Subject: [PATCH 5/5] style: auto-format with Prettier --- .../completed/2026-07/traj_ym4mjwfa0kmx/summary.md | 4 +++- .../completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json | 6 ++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md index 8d10f88da..6f565850c 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md +++ b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/summary.md @@ -19,6 +19,7 @@ Addressed PR 1268 review feedback by synchronizing requested channel stream regi ## 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. @@ -27,7 +28,8 @@ Addressed PR 1268 review feedback by synchronizing requested channel stream regi ## Chapters ### 1. Work -*Agent: default* + +_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 index 76aba3c63..88335bf3f 100644 --- a/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json +++ b/.agentworkforce/trajectories/completed/2026-07/traj_ym4mjwfa0kmx/trajectory.json @@ -46,9 +46,7 @@ "confidence": 0.94 }, "significance": "high", - "tags": [ - "confidence:0.94" - ] + "tags": ["confidence:0.94"] } ] } @@ -66,4 +64,4 @@ "startRef": "de5c3fd53e953523a02c60223dd50d32e7c3902b", "endRef": "de5c3fd53e953523a02c60223dd50d32e7c3902b" } -} \ No newline at end of file +}