diff --git a/CHANGELOG.md b/CHANGELOG.md index fe9d860a3..874bfb2a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `@agent-relay/session` provides Relayhistory-backed cross-harness session continuity with immutable ownership, steering attribution, native Claude resume, and portable journal injection for other handoffs. +- The Swift SDK now provides broker-backed fleet terminal sessions through `AgentClient.terminals`, including live node discovery, view/drive modes, authoritative snapshots, bounded reconnect, input acknowledgements, compare-and-set delivery-mode restoration, and structured close outcomes. + +### Fixed + +- Fleet terminal snapshot refresh failures stay scoped to their correlated request instead of closing a healthy streaming session. + +- Fleet terminal readiness reports the broker's delivery-mode revision so native drive clients preserve concurrent changes across reconnect and close. + +- Swift terminal sessions reject further input after an acknowledgement becomes uncertain, preventing duplicate keystrokes after timeout or reconnect. + ## [11.5.6] - 2026-08-13 ### Fixed diff --git a/crates/broker/src/runtime/event_loop.rs b/crates/broker/src/runtime/event_loop.rs index de76bde2b..38ef9eae9 100644 --- a/crates/broker/src/runtime/event_loop.rs +++ b/crates/broker/src/runtime/event_loop.rs @@ -293,6 +293,8 @@ pub(super) struct TerminalSession { pub(super) struct TerminalSnapshotRequest { pub(super) session_id: String, + /// Present for an explicit refresh; `None` means initial session readiness. + pub(super) client_request_id: Option, pub(super) deadline: Instant, } diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 1d5827a9a..df6f50ef3 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -235,6 +235,7 @@ impl BrokerRuntime { request_id.clone(), TerminalSnapshotRequest { session_id: session_id.clone(), + client_request_id: None, deadline: Instant::now() + TERMINAL_SNAPSHOT_TIMEOUT, }, ); @@ -433,6 +434,65 @@ impl BrokerRuntime { } } } + TerminalControlEvent::Message(TerminalFromCloud::Snapshot { + session_id, + request_id: client_request_id, + }) => { + let Some(session) = self.terminal_sessions.get(&session_id).cloned() else { + self.send_terminal(TerminalToCloud::Error { + session_id, + code: "session_not_found".into(), + message: "terminal session is not active".into(), + request_id: Some(client_request_id), + }); + return; + }; + if !session.ready { + self.send_terminal(TerminalToCloud::Error { + session_id, + code: "session_not_ready".into(), + message: "terminal snapshot is not ready".into(), + request_id: Some(client_request_id), + }); + return; + } + if self + .terminal_snapshot_requests + .values() + .any(|pending| pending.session_id == session_id) + { + self.send_terminal(TerminalToCloud::Error { + session_id, + code: "snapshot_in_flight".into(), + message: "a terminal snapshot is already in flight".into(), + request_id: Some(client_request_id), + }); + return; + } + let worker_request_id = format!("terminal_snapshot_{}", Uuid::new_v4().simple()); + self.terminal_snapshot_requests.insert( + worker_request_id.clone(), + TerminalSnapshotRequest { + session_id: session_id.clone(), + client_request_id: Some(client_request_id.clone()), + deadline: Instant::now() + TERMINAL_SNAPSHOT_TIMEOUT, + }, + ); + if let Err(error) = self.workers.try_send_to_worker( + session.agent.as_str(), + "snapshot_pty", + Some(RequestId::new(worker_request_id.clone())), + json!({ "format": "ansi" }), + ) { + self.terminal_snapshot_requests.remove(&worker_request_id); + self.send_terminal(TerminalToCloud::Error { + session_id, + code: "snapshot_failed".into(), + message: error.to_string(), + request_id: Some(client_request_id), + }); + } + } TerminalControlEvent::Message(TerminalFromCloud::Close { session_id }) => { if let Some(session) = self.terminal_sessions.remove(&session_id) { release_terminal_resize_ownership( @@ -569,6 +629,7 @@ impl BrokerRuntime { pub(super) fn send_terminal(&mut self, message: TerminalToCloud) { let session_id = match &message { TerminalToCloud::Ready { session_id, .. } + | TerminalToCloud::Snapshot { session_id, .. } | TerminalToCloud::Output { session_id, .. } | TerminalToCloud::InputAck { session_id, .. } | TerminalToCloud::Error { session_id, .. } diff --git a/crates/broker/src/runtime/maintenance.rs b/crates/broker/src/runtime/maintenance.rs index 61195b652..1f168e617 100644 --- a/crates/broker/src/runtime/maintenance.rs +++ b/crates/broker/src/runtime/maintenance.rs @@ -40,13 +40,36 @@ impl BrokerRuntime { // A worker can disappear before answering `snapshot_pty`. Bound these // terminal-only RPCs so their sessions cannot remain live forever. - let expired_terminal_snapshots: Vec<(String, String)> = terminal_snapshot_requests - .iter() - .filter(|(_, pending)| pending.deadline <= now) - .map(|(request_id, pending)| (request_id.clone(), pending.session_id.clone())) - .collect(); - for (request_id, session_id) in expired_terminal_snapshots { + let expired_terminal_snapshots: Vec<(String, String, Option)> = + terminal_snapshot_requests + .iter() + .filter(|(_, pending)| pending.deadline <= now) + .map(|(request_id, pending)| { + ( + request_id.clone(), + pending.session_id.clone(), + pending.client_request_id.clone(), + ) + }) + .collect(); + for (request_id, session_id, client_request_id) in expired_terminal_snapshots { terminal_snapshot_requests.remove(&request_id); + if let Some(client_request_id) = client_request_id { + if terminal_sessions.contains_key(&session_id) + && !try_send_terminal( + terminal_control_tx, + TerminalToCloud::Error { + session_id: session_id.clone(), + code: "snapshot_timeout".into(), + message: "terminal snapshot timed out".into(), + request_id: Some(client_request_id), + }, + ) + { + tracing::warn!(target = "relay_broker::terminal", session_id = %session_id, "terminal queue full or closed while reporting refresh snapshot timeout"); + } + continue; + } if let Some(session) = terminal_sessions.remove(&session_id) { release_terminal_resize_ownership(resize_owners, &session.agent, &session_id); terminal_input_requests.retain(|_, pending| pending.session_id != session_id); diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 292aee508..70073e796 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -14,16 +14,24 @@ const TERMINAL_PENDING_OUTPUT_MAX_BYTES: usize = 1024 * 1024; /// keep its own inferred default, which can be wrong, e.g. `manual_flush` for /// `--node drive`). The broker's logical default for a worker with no state /// entry is [`InboundDeliveryMode::AutoInject`], so fall back to it explicitly. -fn resolve_ready_delivery_mode( +fn resolve_ready_delivery_state( terminal_sessions: &HashMap, delivery_states: &HashMap, session_id: &str, -) -> Option { - terminal_sessions - .get(session_id) - .and_then(|session| delivery_states.get(&session.agent)) - .map(|state| state.mode) - .or(Some(InboundDeliveryMode::AutoInject)) +) -> (Option, Option) { + let Some(session) = terminal_sessions.get(session_id) else { + return (None, None); + }; + let state = delivery_states.get(&session.agent); + ( + Some(state.map(|value| value.mode).unwrap_or_default()), + Some( + state + .map(|value| value.revision) + .unwrap_or_default() + .to_string(), + ), + ) } fn publish_terminal_output( @@ -201,11 +209,11 @@ mod terminal_ready_delivery_mode_tests { let delivery_states: HashMap = HashMap::new(); let resolved = - resolve_ready_delivery_mode(&terminal_sessions, &delivery_states, "session-a"); + resolve_ready_delivery_state(&terminal_sessions, &delivery_states, "session-a"); assert_eq!( resolved, - Some(InboundDeliveryMode::AutoInject), + (Some(InboundDeliveryMode::AutoInject), Some("0".into())), "a fresh PTY with no delivery_states entry must still advertise the broker's \ logical default instead of omitting the mode" ); @@ -227,20 +235,23 @@ mod terminal_ready_delivery_mode_tests { ); let resolved = - resolve_ready_delivery_mode(&terminal_sessions, &delivery_states, "session-b"); + resolve_ready_delivery_state(&terminal_sessions, &delivery_states, "session-b"); - assert_eq!(resolved, Some(InboundDeliveryMode::ManualFlush)); + assert_eq!( + resolved, + (Some(InboundDeliveryMode::ManualFlush), Some("3".into())) + ); } #[test] - fn returns_auto_inject_for_an_unknown_session_id() { + fn returns_no_delivery_state_for_an_unknown_session_id() { let terminal_sessions: HashMap = HashMap::new(); let delivery_states: HashMap = HashMap::new(); let resolved = - resolve_ready_delivery_mode(&terminal_sessions, &delivery_states, "no-such-session"); + resolve_ready_delivery_state(&terminal_sessions, &delivery_states, "no-such-session"); - assert_eq!(resolved, Some(InboundDeliveryMode::AutoInject)); + assert_eq!(resolved, (None, None)); } } @@ -1036,11 +1047,11 @@ impl BrokerRuntime { .and_then(Value::as_str) .and_then(|request_id| terminal_input_requests.remove(request_id)) .map(|request| request.session_id); - let terminal_session_id = value + let terminal_snapshot_request = value .get("request_id") .and_then(Value::as_str) .and_then(|request_id| terminal_snapshot_requests.remove(request_id)) - .map(|request| request.session_id); + .map(|request| (request.session_id, request.client_request_id)); if let Some(session_id) = terminal_input_session_id { if !terminal_sessions.contains_key(&session_id) { return; @@ -1098,7 +1109,9 @@ impl BrokerRuntime { "terminal output queue is full", ); } - } else if let Some(session_id) = terminal_session_id { + } else if let Some((session_id, client_request_id)) = + terminal_snapshot_request + { if !terminal_sessions.contains_key(&session_id) { return; } @@ -1107,11 +1120,12 @@ impl BrokerRuntime { // current delivery mode in the Ready frame, giving // the client an authoritative initial mode rather // than an inferred guess. - let session_delivery_mode = resolve_ready_delivery_mode( - terminal_sessions, - delivery_states, - &session_id, - ); + let (session_delivery_mode, session_delivery_revision) = + resolve_ready_delivery_state( + terminal_sessions, + delivery_states, + &session_id, + ); let message = if msg_type != "snapshot_response" { TerminalToCloud::Error { session_id: session_id.clone(), @@ -1119,7 +1133,7 @@ impl BrokerRuntime { message: "terminal snapshot returned an unexpected worker response" .into(), - request_id: None, + request_id: client_request_id.clone(), } } else if let Some(error) = payload.get("error") { TerminalToCloud::Error { @@ -1134,7 +1148,7 @@ impl BrokerRuntime { .and_then(Value::as_str) .unwrap_or("terminal snapshot failed") .to_string(), - request_id: None, + request_id: client_request_id.clone(), } } else if let (Some(screen), Some(rows), Some(cols)) = ( payload.get("screen").and_then(Value::as_str), @@ -1147,23 +1161,38 @@ impl BrokerRuntime { .and_then(Value::as_u64) .and_then(|value| u16::try_from(value).ok()), ) { - TerminalToCloud::Ready { - session_id: session_id.clone(), - screen: screen.to_string(), - rows, - cols, - offset: payload - .get("offset") - .and_then(Value::as_u64) - .unwrap_or(0), - delivery_mode: session_delivery_mode, + if let Some(request_id) = client_request_id.clone() { + TerminalToCloud::Snapshot { + session_id: session_id.clone(), + request_id, + screen: screen.to_string(), + rows, + cols, + offset: payload + .get("offset") + .and_then(Value::as_u64) + .unwrap_or(0), + } + } else { + TerminalToCloud::Ready { + session_id: session_id.clone(), + screen: screen.to_string(), + rows, + cols, + offset: payload + .get("offset") + .and_then(Value::as_u64) + .unwrap_or(0), + delivery_mode: session_delivery_mode, + delivery_revision: session_delivery_revision, + } } } else { TerminalToCloud::Error { session_id: session_id.clone(), code: "snapshot_failed".into(), message: "terminal snapshot response was malformed".into(), - request_id: None, + request_id: client_request_id.clone(), } }; let snapshot_ready = matches!(&message, TerminalToCloud::Ready { .. }); @@ -1215,16 +1244,18 @@ impl BrokerRuntime { break; } } - } else if let Some((code, message)) = snapshot_failure { - end_terminal_session( - terminal_control_tx, - terminal_sessions, - terminal_snapshot_requests, - terminal_input_requests, - &session_id, - &code, - &message, - ); + } else if client_request_id.is_none() { + if let Some((code, message)) = snapshot_failure { + end_terminal_session( + terminal_control_tx, + terminal_sessions, + terminal_snapshot_requests, + terminal_input_requests, + &session_id, + &code, + &message, + ); + } } } else { // Generic worker request/response dispatch. diff --git a/crates/broker/src/terminal_control.rs b/crates/broker/src/terminal_control.rs index ea8d5d82a..e3c7da75f 100644 --- a/crates/broker/src/terminal_control.rs +++ b/crates/broker/src/terminal_control.rs @@ -52,6 +52,14 @@ pub(crate) enum TerminalFromCloud { rows: u16, cols: u16, }, + /// Request a fresh authoritative ANSI snapshot for an existing session. + /// The request id is echoed in the reply so clients can coalesce repaint + /// repairs without confusing a late response for a newer capture. + #[serde(rename = "terminal.snapshot")] + Snapshot { + session_id: String, + request_id: String, + }, #[serde(rename = "terminal.close")] Close { session_id: String }, /// Request the broker flip the inbound delivery mode for the session's @@ -90,6 +98,19 @@ pub(crate) enum TerminalToCloud { /// time (older broker or headless worker). #[serde(skip_serializing_if = "Option::is_none")] delivery_mode: Option, + /// Monotonic broker revision paired with `delivery_mode` for + /// compare-and-set restoration during structured close. + #[serde(skip_serializing_if = "Option::is_none")] + delivery_revision: Option, + }, + #[serde(rename = "terminal.snapshot")] + Snapshot { + session_id: String, + request_id: String, + screen: String, + rows: u16, + cols: u16, + offset: u64, }, #[serde(rename = "terminal.output")] Output { @@ -338,6 +359,7 @@ mod tests { cols: 80, offset: 3, delivery_mode: None, + delivery_revision: None, }) .unwrap(); assert_eq!(ready["type"], "terminal.ready"); @@ -350,9 +372,23 @@ mod tests { cols: 80, offset: 3, delivery_mode: Some(InboundDeliveryMode::ManualFlush), + delivery_revision: Some("7".into()), }) .unwrap(); assert_eq!(ready_with_mode["delivery_mode"], "manual_flush"); + assert_eq!(ready_with_mode["delivery_revision"], "7"); + let snapshot = serde_json::to_value(TerminalToCloud::Snapshot { + session_id: "s".into(), + request_id: "snapshot-1".into(), + screen: "screen".into(), + rows: 24, + cols: 80, + offset: 4, + }) + .unwrap(); + assert_eq!(snapshot["type"], "terminal.snapshot"); + assert_eq!(snapshot["request_id"], "snapshot-1"); + assert_eq!(snapshot["offset"], 4); let ack = serde_json::to_value(TerminalToCloud::InputAck { session_id: "s".into(), bytes_written: 1, diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift b/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift index 1838447cb..a7d1dea6f 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/AgentRelayClient.swift @@ -351,6 +351,10 @@ public final class AgentClient: @unchecked Sendable { /// Fleet nodes: list/get/bind/unbind. public var nodes: RelayNodes { RelayNodes(core: core) } + /// Broker-backed fleet terminal sessions over the authenticated hosted + /// transport. No workspace key or broker credential is exposed. + public var terminals: RelayTerminals { RelayTerminals(core: core, rest: rest) } + /// Message triggers: list/create/update/delete. public var triggers: RelayTriggers { RelayTriggers(core: core) } @@ -1185,6 +1189,11 @@ actor HostedParticipantCore { return value } + static func normalizeTerminalAgent(_ value: String) -> String { + stripSigil(value.trimmingCharacters(in: .whitespacesAndNewlines)) + .trimmingCharacters(in: .whitespacesAndNewlines) + } + static func normalizeChannel(_ value: String) -> String { stripSigil(value).trimmingCharacters(in: .whitespacesAndNewlines) } diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacadeTypes.swift b/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacadeTypes.swift index 503bb2f24..27bf0165e 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacadeTypes.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacadeTypes.swift @@ -263,6 +263,19 @@ public struct RelayNode: Sendable, Equatable { public let lastHeartbeatAt: String? } +public struct RelayNodeAgentBinding: Sendable, Equatable { + public let id: String + public let agentId: String + public let agentName: String + public let nodeId: String + public let nodeName: String + public let nodeKind: String + public let nodeRole: String + public let status: String + public let sessionRef: String? + public let priority: Int +} + public struct RelayListNodesOptions: Sendable, Equatable { public let capability: String? public let name: String? diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacades.swift b/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacades.swift index dcbdf1dd1..5696c75e9 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacades.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/RelayFacades.swift @@ -187,6 +187,10 @@ public struct RelayNodes: Sendable { try await core.nodeGet(name: name) } + public func agents(_ name: String) async throws -> [RelayNodeAgentBinding] { + try await core.nodeAgents(name: name) + } + public func bind(_ name: String, agent: String) async throws { try await core.nodeBind(name: name, agent: agent) } @@ -571,6 +575,69 @@ extension HostedParticipantCore { } } + func nodeAgents(name: String) async throws -> [RelayNodeAgentBinding] { + let relay = try relayCast() + return try await run { + let bindings = try await relay.nodes.listAgents(name) + return bindings.map { RelayNodeAgentBinding($0) } + } + } + + func terminalTarget(agent name: String) async throws -> RelayTerminalTarget { + let cleanName = Self.normalizeTerminalAgent(name) + guard !cleanName.isEmpty else { + throw RelayError.protocolError( + code: "invalid_terminal_target", + message: "Terminal agent name cannot be empty", + retryable: false + ) + } + + let relay = try relayCast() + return try await run { + let liveNodes = try await relay.nodes.list(Relaycast.NodeListQuery()) + .filter(\.live) + .sorted { $0.name < $1.name } + let nodes = relay.nodes + var remaining = liveNodes.makeIterator() + var candidates: [Relaycast.NodeAgentBinding] = [] + try await withThrowingTaskGroup(of: [Relaycast.NodeAgentBinding].self) { group in + for _ in 0.. $1.priority } + return $0.nodeName < $1.nodeName + }).first else { + throw RelayError.protocolError( + code: "terminal_target_unavailable", + message: "No live fleet node is hosting agent '\(cleanName)'", + retryable: true + ) + } + return RelayTerminalTarget( + nodeId: binding.nodeId, + nodeName: binding.nodeName, + agentId: binding.agentId, + agentName: binding.agentName, + sessionRef: binding.sessionRef + ) + } + } + func nodeBind(name: String, agent: String) async throws { let relay = try relayCast() try await run { _ = try await relay.nodes.bindAgent(name, request: Relaycast.BindAgentToNodeRequest(agentName: Self.stripSigil(agent))) } diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/RelayRestClient.swift b/packages/sdk-swift/Sources/AgentRelaySDK/RelayRestClient.swift index c53d79682..a71d1ac04 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/RelayRestClient.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/RelayRestClient.swift @@ -111,6 +111,20 @@ struct DmMessageRow: Decodable, Sendable { } } +struct RelayTerminalTicket: Decodable, Sendable { + let sessionId: String + let terminalUrl: String + let resumeToken: String + let expiresAt: String? + + enum CodingKeys: String, CodingKey { + case sessionId = "session_id" + case terminalUrl = "terminal_url" + case resumeToken = "resume_token" + case expiresAt = "expires_at" + } +} + // MARK: - RelayRestClient /// Internal REST client for the hosted API endpoints the facade serves @@ -284,6 +298,20 @@ struct RelayRestClient: Sendable { return Self.sortedOldestFirst(events) } + // MARK: - Fleet terminals + + func createTerminalSession( + node: String, + agent: String, + mode: RelayTerminalMode + ) async throws -> RelayTerminalTicket { + let cleanAgent = HostedParticipantCore.normalizeTerminalAgent(agent) + return try await post( + "/v1/nodes/\(Self.encodePathSegment(node))/terminal/sessions", + body: TerminalSessionRequestBody(agent: cleanAgent, mode: mode) + ) + } + // MARK: - Generic transport /// `requestTimeout`, when set, becomes the `URLRequest.timeoutInterval` of @@ -390,6 +418,11 @@ struct RelayRestClient: Sendable { let input: JSONValue } + private struct TerminalSessionRequestBody: Encodable { + let agent: String + let mode: RelayTerminalMode + } + private static func decodeEnvelope(data: Data, statusCode: Int, path: String) throws -> T { let envelope: RestEnvelope do { diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/RelayTerminal.swift b/packages/sdk-swift/Sources/AgentRelaySDK/RelayTerminal.swift new file mode 100644 index 000000000..eb17d3206 --- /dev/null +++ b/packages/sdk-swift/Sources/AgentRelaySDK/RelayTerminal.swift @@ -0,0 +1,1020 @@ +import Foundation + +public enum RelayTerminalMode: String, Codable, Hashable, Sendable { + case view + case drive + case passthrough +} + +public enum RelayTerminalDeliveryMode: String, Codable, Equatable, Sendable { + case autoInject = "auto_inject" + case manualFlush = "manual_flush" +} + +public struct RelayTerminalSnapshot: Equatable, Sendable { + public let data: Data + public let rows: Int + public let columns: Int + public let offset: UInt64 + + public init(data: Data, rows: Int, columns: Int, offset: UInt64) { + self.data = data + self.rows = rows + self.columns = columns + self.offset = offset + } +} + +public struct RelayTerminalFailure: Error, Equatable, LocalizedError, Sendable { + public let code: String + public let message: String + + public init(code: String, message: String) { + self.code = code + self.message = message + } + + public var errorDescription: String? { message } +} + +public enum RelayTerminalEvent: Equatable, Sendable { + case ready( + snapshot: RelayTerminalSnapshot, + deliveryMode: RelayTerminalDeliveryMode?, + deliveryRevision: String? + ) + case output(data: Data, offset: UInt64?) + case inputAcknowledged(bytesWritten: Int) + case failure(RelayTerminalFailure) + case closed(code: String?, message: String?) +} + +public enum RelayTerminalRestorationStatus: Equatable, Sendable { + case notRequired + case restored + case skippedConcurrentChange + case unconfirmed(RelayTerminalFailure) +} + +public struct RelayTerminalCloseOutcome: Equatable, Sendable { + public let alreadyClosed: Bool + public let remoteCloseConfirmed: Bool + public let restoration: RelayTerminalRestorationStatus + + public init( + alreadyClosed: Bool, + remoteCloseConfirmed: Bool, + restoration: RelayTerminalRestorationStatus + ) { + self.alreadyClosed = alreadyClosed + self.remoteCloseConfirmed = remoteCloseConfirmed + self.restoration = restoration + } +} + +public struct RelayTerminalTarget: Equatable, Sendable { + public let nodeId: String + public let nodeName: String + public let agentId: String + public let agentName: String + public let sessionRef: String? + + public init( + nodeId: String, + nodeName: String, + agentId: String, + agentName: String, + sessionRef: String? + ) { + self.nodeId = nodeId + self.nodeName = nodeName + self.agentId = agentId + self.agentName = agentName + self.sessionRef = sessionRef + } +} + +/// Fleet terminal entry point available on every hosted `AgentClient`. +/// Authentication uses that client's scoped participant credential. +public struct RelayTerminals: Sendable { + let core: HostedParticipantCore + let rest: RelayRestClient + + public func resolve(agent: String) async throws -> RelayTerminalTarget { + try await core.terminalTarget(agent: agent) + } + + /// Resolve the live fleet node hosting `agent`, then open a terminal there. + /// Applications do not need to reproduce fleet placement discovery. + public func open( + agent: String, + mode: RelayTerminalMode + ) async throws -> RelayTerminalSession { + let target = try await resolve(agent: agent) + return try await open(node: target.nodeName, agent: target.agentName, mode: mode) + } + + public func open( + node: String, + agent: String, + mode: RelayTerminalMode + ) async throws -> RelayTerminalSession { + let cleanNode = node.trimmingCharacters(in: .whitespacesAndNewlines) + let cleanAgent = HostedParticipantCore.normalizeTerminalAgent(agent) + guard !cleanNode.isEmpty, !cleanAgent.isEmpty else { + throw RelayError.protocolError( + code: "invalid_terminal_target", + message: "Terminal node and agent names cannot be empty", + retryable: false + ) + } + let ticket = try await rest.createTerminalSession( + node: cleanNode, + agent: cleanAgent, + mode: mode + ) + return try await RelayTerminalSession.open( + ticket: ticket, + node: cleanNode, + agent: cleanAgent, + mode: mode, + baseURL: rest.baseURL, + urlSession: rest.session + ) + } +} + +/// A broker-backed terminal session. The SDK owns the ticket, websocket, +/// reconnect, input ordering, authoritative snapshots, and delivery-mode +/// restoration; applications provide only a renderer and user intent. +public final class RelayTerminalSession: @unchecked Sendable { + public let id: String + public let node: String + public let agent: String + public let mode: RelayTerminalMode + public let initialSnapshot: RelayTerminalSnapshot + + private let core: RelayTerminalCore + + private init( + id: String, + node: String, + agent: String, + mode: RelayTerminalMode, + initialSnapshot: RelayTerminalSnapshot, + core: RelayTerminalCore + ) { + self.id = id + self.node = node + self.agent = agent + self.mode = mode + self.initialSnapshot = initialSnapshot + self.core = core + } + + static func open( + ticket: RelayTerminalTicket, + node: String, + agent: String, + mode: RelayTerminalMode, + baseURL: URL, + urlSession: URLSession + ) async throws -> RelayTerminalSession { + let core = try RelayTerminalCore( + ticket: ticket, + mode: mode, + baseURL: baseURL, + urlSession: urlSession + ) + let ready = try await core.connect() + if mode != .view { + try await core.acquireDrive(previous: ready) + } + return RelayTerminalSession( + id: ticket.sessionId, + node: node, + agent: agent, + mode: mode, + initialSnapshot: ready.snapshot, + core: core + ) + } + + public func events() async -> AsyncThrowingStream { + await core.events() + } + + public func snapshot() async throws -> RelayTerminalSnapshot { + try await core.snapshot() + } + + public func sendInput(_ data: Data) async throws { + guard mode != .view else { + throw RelayTerminalFailure(code: "read_only", message: "View terminal sessions cannot send input") + } + try await core.sendInput(data) + } + + public func resize(rows: Int, columns: Int) async throws { + guard mode != .view else { + throw RelayTerminalFailure(code: "read_only", message: "View terminal sessions cannot resize the PTY") + } + guard rows > 0, rows <= Int(UInt16.max), columns > 0, columns <= Int(UInt16.max) else { + throw RelayTerminalFailure(code: "invalid_dimensions", message: "Terminal dimensions must be positive 16-bit integers") + } + try await core.resize(rows: rows, columns: columns) + } + + public func close() async -> RelayTerminalCloseOutcome { + await core.close() + } +} + +private struct RelayTerminalReady: Sendable { + let snapshot: RelayTerminalSnapshot + let deliveryMode: RelayTerminalDeliveryMode? + let deliveryRevision: String? +} + +private struct RelayTerminalWireFrame: Decodable { + let type: String + let sessionId: String? + let requestId: String? + let screen: String? + let chunk: String? + let rows: Int? + let columns: Int? + let offset: UInt64? + let bytesWritten: Int? + let code: String? + let message: String? + let deliveryMode: RelayTerminalDeliveryMode? + let mode: RelayTerminalDeliveryMode? + let deliveryRevision: String? + let revision: String? + let matched: Bool? + + enum CodingKeys: String, CodingKey { + case type, screen, chunk, rows, offset, code, message, mode, revision, matched + case sessionId = "session_id" + case requestId = "request_id" + case columns = "cols" + case bytesWritten = "bytes_written" + case deliveryMode = "delivery_mode" + case deliveryRevision = "delivery_revision" + } + + var snapshot: RelayTerminalSnapshot? { + guard let screen, let rows, let columns else { return nil } + return RelayTerminalSnapshot( + data: Data(screen.utf8), + rows: rows, + columns: columns, + offset: offset ?? 0 + ) + } +} + +private struct RelayTerminalInputFrame: Encodable { + let type = "terminal.input" + let sessionId: String + let dataBase64: String + + enum CodingKeys: String, CodingKey { + case type + case sessionId = "session_id" + case dataBase64 = "data_base64" + } +} + +private struct RelayTerminalResizeFrame: Encodable { + let type = "terminal.resize" + let sessionId: String + let rows: Int + let columns: Int + + enum CodingKeys: String, CodingKey { + case type, rows + case sessionId = "session_id" + case columns = "cols" + } +} + +private struct RelayTerminalSnapshotFrame: Encodable { + let type = "terminal.snapshot" + let sessionId: String + let requestId: String + + enum CodingKeys: String, CodingKey { + case type + case sessionId = "session_id" + case requestId = "request_id" + } +} + +private struct RelayTerminalSetDeliveryModeFrame: Encodable { + let type = "terminal.set_delivery_mode" + let sessionId: String + let requestId: String + let mode: RelayTerminalDeliveryMode + let expectedMode: RelayTerminalDeliveryMode? + let expectedRevision: String? + + enum CodingKeys: String, CodingKey { + case type, mode + case sessionId = "session_id" + case requestId = "request_id" + case expectedMode = "expected_mode" + case expectedRevision = "expected_revision" + } +} + +private struct RelayTerminalCloseFrame: Encodable { + let type = "terminal.close" + let sessionId: String + + enum CodingKeys: String, CodingKey { + case type + case sessionId = "session_id" + } +} + +private struct RelayTerminalDeliveryModeResult: Sendable { + let mode: RelayTerminalDeliveryMode + let matched: Bool + let revision: String +} + +private actor RelayTerminalCore { + private let ticket: RelayTerminalTicket + private let mode: RelayTerminalMode + private let baseURL: URL + private let urlSession: URLSession + private let encoder = JSONEncoder() + private let decoder = JSONDecoder() + private let stream: AsyncThrowingStream + private let continuation: AsyncThrowingStream.Continuation + + private var socket: URLSessionWebSocketTask? + private var receiveTask: Task? + private var closed = false + private var closeReported = false + private var inputContinuation: CheckedContinuation? + private var inputTimeoutTask: Task? + private var inputUncertainty: RelayTerminalFailure? + private var snapshotContinuations: [String: CheckedContinuation] = [:] + private var snapshotTimeoutTasks: [String: Task] = [:] + private var deliveryContinuations: [String: CheckedContinuation] = [:] + private var deliveryTimeoutTasks: [String: Task] = [:] + private var closeContinuation: CheckedContinuation? + private var closeTimeoutTask: Task? + private var closeInProgress = false + private var closeWaiters: [CheckedContinuation] = [] + private var completedCloseOutcome: RelayTerminalCloseOutcome? + private var priorDeliveryMode: RelayTerminalDeliveryMode? + private var assertedDeliveryMode: RelayTerminalDeliveryMode? + private var assertedDeliveryRevision: String? + + init( + ticket: RelayTerminalTicket, + mode: RelayTerminalMode, + baseURL: URL, + urlSession: URLSession + ) throws { + guard let terminalURL = URL(string: ticket.terminalUrl) else { + throw RelayError.protocolError( + code: "invalid_terminal_url", + message: "Terminal session response contained an invalid URL", + retryable: false + ) + } + guard Self.sameOrigin(terminalURL, baseURL) else { + throw RelayError.protocolError( + code: "invalid_terminal_origin", + message: "Terminal session URL did not match the configured Relay origin", + retryable: false + ) + } + self.ticket = ticket + self.mode = mode + self.baseURL = baseURL + self.urlSession = urlSession + var captured: AsyncThrowingStream.Continuation! + stream = AsyncThrowingStream(bufferingPolicy: .bufferingNewest(256)) { captured = $0 } + continuation = captured + } + + func events() -> AsyncThrowingStream { stream } + + func connect() async throws -> RelayTerminalReady { + let ready: RelayTerminalReady + do { + ready = try await openSocket(url: try websocketURL(ticket.terminalUrl)) + } catch let failure as RelayTerminalFailure { + socket?.cancel(with: .goingAway, reason: nil) + socket = nil + throw failure + } catch let error as RelayError { + socket?.cancel(with: .goingAway, reason: nil) + socket = nil + throw error + } catch { + socket?.cancel(with: .goingAway, reason: nil) + socket = nil + throw RelayTerminalFailure( + code: "node_unreachable", + message: "Terminal transport could not connect to the fleet node" + ) + } + yield(.ready( + snapshot: ready.snapshot, + deliveryMode: ready.deliveryMode, + deliveryRevision: ready.deliveryRevision + )) + startReceiveLoop() + return ready + } + + func acquireDrive(previous: RelayTerminalReady) async throws { + guard let previousMode = previous.deliveryMode, + let previousRevision = previous.deliveryRevision else { + _ = await close() + throw RelayTerminalFailure( + code: "delivery_mode_unavailable", + message: "The broker did not provide delivery-mode state required for safe drive attach" + ) + } + priorDeliveryMode = previousMode + let result: RelayTerminalDeliveryModeResult + do { + result = try await setDeliveryMode( + .autoInject, + expectedMode: previousMode, + expectedRevision: previousRevision + ) + } catch { + _ = await close() + throw error + } + guard result.matched else { + _ = await close() + throw RelayTerminalFailure( + code: "delivery_mode_conflict", + message: "The terminal delivery mode changed while drive attach was opening" + ) + } + assertedDeliveryMode = result.mode + assertedDeliveryRevision = result.revision + } + + func sendInput(_ data: Data) async throws { + try requireOpen() + if let inputUncertainty { throw inputUncertainty } + guard inputContinuation == nil else { + throw RelayTerminalFailure( + code: "input_backpressure", + message: "A terminal input write is already awaiting acknowledgement" + ) + } + let frame = RelayTerminalInputFrame( + sessionId: ticket.sessionId, + dataBase64: data.base64EncodedString() + ) + try await withCheckedThrowingContinuation { continuation in + inputContinuation = continuation + inputTimeoutTask = operationTimeoutTask { [weak self] in + await self?.markInputUncertain(RelayTerminalFailure( + code: "input_timeout", + message: "Terminal input acknowledgement timed out" + )) + } + sendRequest(frame) { [weak self] error in + guard let error else { return } + Task { await self?.markInputUncertain(error) } + } + } + } + + func resize(rows: Int, columns: Int) async throws { + try requireOpen() + try await send( + RelayTerminalResizeFrame( + sessionId: ticket.sessionId, + rows: rows, + columns: columns + ) + ) + } + + func snapshot() async throws -> RelayTerminalSnapshot { + try requireOpen() + let requestID = "snapshot_\(UUID().uuidString.lowercased())" + let frame = RelayTerminalSnapshotFrame( + sessionId: ticket.sessionId, + requestId: requestID + ) + return try await withCheckedThrowingContinuation { continuation in + snapshotContinuations[requestID] = continuation + snapshotTimeoutTasks[requestID] = operationTimeoutTask { [weak self] in + await self?.failSnapshot( + requestID: requestID, + error: RelayTerminalFailure( + code: "snapshot_timeout", + message: "Terminal snapshot request timed out" + ) + ) + } + sendRequest(frame) { [weak self] error in + guard let error else { return } + Task { await self?.failSnapshot(requestID: requestID, error: error) } + } + } + } + + func close() async -> RelayTerminalCloseOutcome { + if closeInProgress { + return await withCheckedContinuation { continuation in + closeWaiters.append(continuation) + } + } + if let completedCloseOutcome { + return RelayTerminalCloseOutcome( + alreadyClosed: true, + remoteCloseConfirmed: completedCloseOutcome.remoteCloseConfirmed, + restoration: completedCloseOutcome.restoration + ) + } + guard !closed else { + return RelayTerminalCloseOutcome( + alreadyClosed: true, + remoteCloseConfirmed: closeReported, + restoration: mode == .view ? .notRequired : .unconfirmed( + RelayTerminalFailure(code: "already_closed", message: "Terminal session was already closed") + ) + ) + } + + closeInProgress = true + let outcome = await performClose() + completedCloseOutcome = outcome + closeInProgress = false + let waiters = closeWaiters + closeWaiters.removeAll() + waiters.forEach { $0.resume(returning: outcome) } + return outcome + } + + private func performClose() async -> RelayTerminalCloseOutcome { + let restoration: RelayTerminalRestorationStatus + if mode == .view { + restoration = .notRequired + } else if let priorDeliveryMode, + let assertedDeliveryMode, + let assertedDeliveryRevision { + do { + let result = try await setDeliveryMode( + priorDeliveryMode, + expectedMode: assertedDeliveryMode, + expectedRevision: assertedDeliveryRevision + ) + restoration = result.matched ? .restored : .skippedConcurrentChange + } catch { + restoration = .unconfirmed(Self.failure(error)) + } + } else { + restoration = .unconfirmed( + RelayTerminalFailure( + code: "delivery_mode_unavailable", + message: "The drive session did not retain enough state to confirm restoration" + ) + ) + } + + let remoteCloseConfirmed = await requestCloseConfirmation() + finish() + return RelayTerminalCloseOutcome( + alreadyClosed: false, + remoteCloseConfirmed: remoteCloseConfirmed, + restoration: restoration + ) + } + + private func setDeliveryMode( + _ mode: RelayTerminalDeliveryMode, + expectedMode: RelayTerminalDeliveryMode, + expectedRevision: String + ) async throws -> RelayTerminalDeliveryModeResult { + try requireOpen() + let requestID = "mode_\(UUID().uuidString.lowercased())" + let frame = RelayTerminalSetDeliveryModeFrame( + sessionId: ticket.sessionId, + requestId: requestID, + mode: mode, + expectedMode: expectedMode, + expectedRevision: expectedRevision + ) + return try await withCheckedThrowingContinuation { continuation in + deliveryContinuations[requestID] = continuation + deliveryTimeoutTasks[requestID] = operationTimeoutTask { [weak self] in + await self?.failDeliveryMode( + requestID: requestID, + error: RelayTerminalFailure( + code: "delivery_mode_timeout", + message: "Terminal delivery-mode request timed out" + ) + ) + } + sendRequest(frame) { [weak self] error in + guard let error else { return } + Task { await self?.failDeliveryMode(requestID: requestID, error: error) } + } + } + } + + private func startReceiveLoop() { + receiveTask?.cancel() + receiveTask = Task { [weak self] in + await self?.receiveLoop() + } + } + + private func receiveLoop() async { + var reconnectAttempt = 0 + while !closed { + guard let socket else { return } + do { + let message = try await socket.receive() + try await handle(message) + } catch is CancellationError { + return + } catch { + guard !closed else { return } + if inputContinuation != nil { + markInputUncertain(Self.failure(error), closeSession: false) + } + failPending(Self.failure(error)) + guard reconnectAttempt < 5 else { + finish(throwing: RelayTerminalFailure( + code: "terminal_reconnect_failed", + message: "Terminal transport could not reconnect" + )) + return + } + let delay = min(0.5 * pow(2.0, Double(reconnectAttempt)), 8) + reconnectAttempt += 1 + do { + try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000)) + let ready = try await openSocket(url: try resumeURL()) + try refreshDriveAssertion(after: ready) + if inputUncertainty != nil { + Task { [weak self] in + _ = await self?.close() + } + } else { + yield(.ready( + snapshot: ready.snapshot, + deliveryMode: ready.deliveryMode, + deliveryRevision: ready.deliveryRevision + )) + } + reconnectAttempt = 0 + } catch is CancellationError { + return + } catch let failure as RelayTerminalFailure where failure.code == "delivery_mode_conflict" { + finish(throwing: failure) + return + } catch { + continue + } + } + } + } + + /// A resumed lane reports the broker's current delivery-mode revision in + /// its fresh Ready frame. Accept a drive resume only while the asserted + /// mode remains in force, then advance the CAS revision used at close. + /// If an operator changed the mode during the disconnect, ending this + /// session preserves that concurrent change instead of overwriting it. + private func refreshDriveAssertion(after ready: RelayTerminalReady) throws { + guard mode != .view else { return } + guard ready.deliveryMode == .autoInject, + let revision = ready.deliveryRevision else { + assertedDeliveryMode = nil + assertedDeliveryRevision = nil + throw RelayTerminalFailure( + code: "delivery_mode_conflict", + message: "The terminal delivery mode changed while the drive session was reconnecting" + ) + } + assertedDeliveryMode = .autoInject + assertedDeliveryRevision = revision + } + + private func openSocket(url: URL) async throws -> RelayTerminalReady { + let task = urlSession.webSocketTask(with: url) + socket = task + task.resume() + let frame = try decode(try await task.receive()) + guard frame.type == "terminal.ready", + frame.sessionId == ticket.sessionId, + let snapshot = frame.snapshot else { + task.cancel(with: .protocolError, reason: nil) + throw RelayTerminalFailure( + code: "terminal_handshake_failed", + message: "Terminal transport did not provide a valid ready frame" + ) + } + return RelayTerminalReady( + snapshot: snapshot, + deliveryMode: frame.deliveryMode, + deliveryRevision: frame.deliveryRevision + ) + } + + private func handle(_ message: URLSessionWebSocketTask.Message) async throws { + let frame = try decode(message) + guard frame.sessionId == ticket.sessionId else { return } + switch frame.type { + case "terminal.ready": + guard let snapshot = frame.snapshot else { return } + yield(.ready( + snapshot: snapshot, + deliveryMode: frame.deliveryMode, + deliveryRevision: frame.deliveryRevision + )) + case "terminal.snapshot": + guard let requestID = frame.requestId, + let snapshot = frame.snapshot, + let pending = snapshotContinuations.removeValue(forKey: requestID) else { return } + snapshotTimeoutTasks.removeValue(forKey: requestID)?.cancel() + pending.resume(returning: snapshot) + case "terminal.output": + guard let chunk = frame.chunk else { return } + yield(.output(data: Data(chunk.utf8), offset: frame.offset)) + case "terminal.input_ack": + let pending = inputContinuation + inputContinuation = nil + inputTimeoutTask?.cancel() + inputTimeoutTask = nil + pending?.resume() + yield(.inputAcknowledged(bytesWritten: frame.bytesWritten ?? 0)) + case "terminal.delivery_mode": + guard let requestID = frame.requestId, + let pending = deliveryContinuations.removeValue(forKey: requestID), + let mode = frame.mode, + let revision = frame.revision else { return } + deliveryTimeoutTasks.removeValue(forKey: requestID)?.cancel() + pending.resume(returning: RelayTerminalDeliveryModeResult( + mode: mode, + matched: frame.matched ?? true, + revision: revision + )) + case "terminal.error": + let failure = RelayTerminalFailure( + code: frame.code ?? "terminal_error", + message: frame.message ?? "Terminal operation failed" + ) + if let requestID = frame.requestId { + snapshotTimeoutTasks.removeValue(forKey: requestID)?.cancel() + deliveryTimeoutTasks.removeValue(forKey: requestID)?.cancel() + snapshotContinuations.removeValue(forKey: requestID)?.resume(throwing: failure) + deliveryContinuations.removeValue(forKey: requestID)?.resume(throwing: failure) + } else if let pending = inputContinuation { + inputContinuation = nil + inputTimeoutTask?.cancel() + inputTimeoutTask = nil + pending.resume(throwing: failure) + } + yield(.failure(failure)) + case "terminal.closed": + closeReported = true + let closeWaiter = closeContinuation + closeContinuation = nil + closeTimeoutTask?.cancel() + closeTimeoutTask = nil + closeWaiter?.resume(returning: true) + yield(.closed(code: frame.code, message: frame.message)) + finish() + default: + return + } + } + + private func send(_ frame: T) async throws { + try requireOpen() + guard let socket else { + throw RelayTerminalFailure(code: "not_connected", message: "Terminal transport is not connected") + } + let data = try encoder.encode(frame) + guard let string = String(data: data, encoding: .utf8) else { + throw RelayError.encodingFailed("Could not encode terminal frame") + } + try await socket.send(.string(string)) + } + + private func sendRequest(_ frame: T, completion: @escaping @Sendable (Error?) -> Void) { + guard !closed, let socket else { + completion(RelayTerminalFailure(code: "not_connected", message: "Terminal transport is not connected")) + return + } + do { + let data = try encoder.encode(frame) + guard let string = String(data: data, encoding: .utf8) else { + completion(RelayError.encodingFailed("Could not encode terminal frame")) + return + } + socket.send(.string(string), completionHandler: completion) + } catch { + completion(error) + } + } + + private func decode(_ message: URLSessionWebSocketTask.Message) throws -> RelayTerminalWireFrame { + let data: Data + switch message { + case .data(let value): data = value + case .string(let value): data = Data(value.utf8) + @unknown default: + throw RelayTerminalFailure(code: "invalid_frame", message: "Terminal transport returned an unknown frame type") + } + do { + return try decoder.decode(RelayTerminalWireFrame.self, from: data) + } catch { + throw RelayTerminalFailure(code: "invalid_frame", message: "Terminal transport returned malformed data") + } + } + + private func requireOpen() throws { + guard !closed else { + throw RelayTerminalFailure(code: "closed", message: "Terminal session is closed") + } + } + + private func failInput(_ error: Error) -> Bool { + let pending = inputContinuation + inputContinuation = nil + inputTimeoutTask?.cancel() + inputTimeoutTask = nil + pending?.resume(throwing: error) + return pending != nil + } + + private func markInputUncertain(_ error: Error, closeSession: Bool = true) { + guard inputUncertainty == nil else { return } + guard inputContinuation != nil else { return } + let source = Self.failure(error) + let failure = RelayTerminalFailure( + code: "input_result_uncertain", + message: "Terminal input result became uncertain (\(source.message)); the SDK is closing this session to prevent duplicate keystrokes." + ) + inputUncertainty = failure + _ = failInput(failure) + yield(.failure(failure)) + if closeSession { + Task { [weak self] in + _ = await self?.close() + } + } + } + + private func failSnapshot(requestID: String, error: Error) { + snapshotTimeoutTasks.removeValue(forKey: requestID)?.cancel() + snapshotContinuations.removeValue(forKey: requestID)?.resume(throwing: error) + } + + private func failDeliveryMode(requestID: String, error: Error) { + deliveryTimeoutTasks.removeValue(forKey: requestID)?.cancel() + deliveryContinuations.removeValue(forKey: requestID)?.resume(throwing: error) + } + + private func failPending(_ error: Error) { + _ = failInput(error) + let snapshots = snapshotContinuations.values + snapshotContinuations.removeAll() + snapshotTimeoutTasks.values.forEach { $0.cancel() } + snapshotTimeoutTasks.removeAll() + snapshots.forEach { $0.resume(throwing: error) } + let deliveries = deliveryContinuations.values + deliveryContinuations.removeAll() + deliveryTimeoutTasks.values.forEach { $0.cancel() } + deliveryTimeoutTasks.removeAll() + deliveries.forEach { $0.resume(throwing: error) } + } + + private func yield(_ event: RelayTerminalEvent) { + if case .dropped = continuation.yield(event) { + finish(throwing: RelayTerminalFailure( + code: "output_backpressure", + message: "Terminal output exceeded the SDK event buffer" + )) + } + } + + private func finish(throwing error: Error? = nil) { + guard !closed else { return } + closed = true + receiveTask?.cancel() + receiveTask = nil + socket?.cancel(with: .normalClosure, reason: nil) + socket = nil + let failure = error ?? RelayTerminalFailure(code: "closed", message: "Terminal session closed") + failPending(failure) + let closeWaiter = closeContinuation + closeContinuation = nil + closeTimeoutTask?.cancel() + closeTimeoutTask = nil + closeWaiter?.resume(returning: closeReported) + if let error { + continuation.finish(throwing: error) + } else { + continuation.finish() + } + } + + private func websocketURL(_ value: String) throws -> URL { + guard var components = URLComponents(string: value) else { + throw RelayTerminalFailure(code: "invalid_terminal_url", message: "Terminal session URL is invalid") + } + switch components.scheme?.lowercased() { + case "https": components.scheme = "wss" + case "http": components.scheme = "ws" + case "wss", "ws": break + default: + throw RelayTerminalFailure(code: "invalid_terminal_url", message: "Terminal session URL must use HTTP or WebSocket transport") + } + guard let url = components.url else { + throw RelayTerminalFailure(code: "invalid_terminal_url", message: "Terminal session URL is invalid") + } + return url + } + + private func resumeURL() throws -> URL { + guard var components = URLComponents(string: ticket.terminalUrl) else { + throw RelayTerminalFailure(code: "invalid_terminal_url", message: "Terminal session URL is invalid") + } + var items = (components.queryItems ?? []).filter { $0.name != "ticket" } + items.append(URLQueryItem(name: "session_id", value: ticket.sessionId)) + items.append(URLQueryItem(name: "resume", value: ticket.resumeToken)) + components.queryItems = items + guard let value = components.string else { + throw RelayTerminalFailure(code: "invalid_terminal_url", message: "Terminal resume URL is invalid") + } + return try websocketURL(value) + } + + private func requestCloseConfirmation() async -> Bool { + guard !closed, socket != nil else { return closeReported } + return await withCheckedContinuation { continuation in + closeContinuation = continuation + closeTimeoutTask = Task { [weak self] in + try? await Task.sleep(nanoseconds: 2_000_000_000) + guard !Task.isCancelled else { return } + await self?.finishCloseWait(confirmed: false) + } + sendRequest(RelayTerminalCloseFrame(sessionId: ticket.sessionId)) { [weak self] error in + guard error != nil else { return } + Task { await self?.finishCloseWait(confirmed: false) } + } + } + } + + private func finishCloseWait(confirmed: Bool) { + let pending = closeContinuation + closeContinuation = nil + closeTimeoutTask?.cancel() + closeTimeoutTask = nil + pending?.resume(returning: confirmed) + } + + private func operationTimeoutTask( + _ action: @escaping @Sendable () async -> Void + ) -> Task { + Task { + try? await Task.sleep(nanoseconds: 10_000_000_000) + guard !Task.isCancelled else { return } + await action() + } + } + + private static func sameOrigin(_ lhs: URL, _ rhs: URL) -> Bool { + func defaultPort(_ url: URL) -> Int? { + if let port = url.port { return port } + switch url.scheme?.lowercased() { + case "https", "wss": return 443 + case "http", "ws": return 80 + default: return nil + } + } + return lhs.host?.lowercased() == rhs.host?.lowercased() + && defaultPort(lhs) == defaultPort(rhs) + && ((lhs.scheme == "https" || lhs.scheme == "wss") == (rhs.scheme == "https" || rhs.scheme == "wss")) + } + + private static func failure(_ error: Error) -> RelayTerminalFailure { + if let failure = error as? RelayTerminalFailure { return failure } + if case RelayError.protocolError(let code, let message, _) = error { + return RelayTerminalFailure(code: code, message: message) + } + return RelayTerminalFailure(code: "terminal_error", message: error.localizedDescription) + } +} diff --git a/packages/sdk-swift/Sources/AgentRelaySDK/RelaycastTranslate.swift b/packages/sdk-swift/Sources/AgentRelaySDK/RelaycastTranslate.swift index 9942569ec..5c8333c33 100644 --- a/packages/sdk-swift/Sources/AgentRelaySDK/RelaycastTranslate.swift +++ b/packages/sdk-swift/Sources/AgentRelaySDK/RelaycastTranslate.swift @@ -246,6 +246,23 @@ extension RelayNode { } } +extension RelayNodeAgentBinding { + init(_ binding: Relaycast.NodeAgentBinding) { + self.init( + id: binding.id, + agentId: binding.agentId, + agentName: binding.agentName, + nodeId: binding.nodeId, + nodeName: binding.nodeName, + nodeKind: binding.nodeKind, + nodeRole: binding.nodeRole, + status: binding.status, + sessionRef: binding.sessionRef, + priority: binding.priority + ) + } +} + extension RelayTrigger { init(_ trigger: Relaycast.Trigger) { self.init( diff --git a/packages/sdk-swift/Tests/AgentRelaySDKTests/RelayRestTests.swift b/packages/sdk-swift/Tests/AgentRelaySDKTests/RelayRestTests.swift index 7a8779cc5..cbacddc75 100644 --- a/packages/sdk-swift/Tests/AgentRelaySDKTests/RelayRestTests.swift +++ b/packages/sdk-swift/Tests/AgentRelaySDKTests/RelayRestTests.swift @@ -84,6 +84,33 @@ final class RelayRestTests: XCTestCase { } } + func testTerminalTicketUsesScopedParticipantCredentialAndTypedBody() async throws { + StubURLProtocol.store.enqueue(StubResponse(statusCode: 201, json: """ + {"ok":true,"data":{ + "session_id":"term_1", + "terminal_url":"https://stub.test/v1/nodes/sf-mini/terminal/connect?ticket=tt_live_redacted", + "resume_token":"tr_live_redacted", + "expires_at":"2026-08-13T14:00:00Z" + }} + """)) + + let ticket = try await client.createTerminalSession( + node: "sf mini/primary", + agent: " @chief ", + mode: .drive + ) + + XCTAssertEqual(ticket.sessionId, "term_1") + XCTAssertEqual(ticket.resumeToken, "tr_live_redacted") + let request = try XCTUnwrap(StubURLProtocol.store.recordedRequests().first) + XCTAssertEqual(request.url?.path, "/v1/nodes/sf mini/primary/terminal/sessions") + XCTAssertEqual(request.method, "POST") + XCTAssertEqual(request.headers["Authorization"], "Bearer at_test_token") + let body = try XCTUnwrap(request.body) + let object = try XCTUnwrap(JSONSerialization.jsonObject(with: body) as? [String: String]) + XCTAssertEqual(object, ["agent": "chief", "mode": "drive"]) + } + func testRetries429ThenSucceedsWithSecondRequest() async throws { // First attempt is rate limited (Retry-After: 0 keeps the test fast); // the retry succeeds and the caller only observes the success.