Skip to content

feat(sdk-swift): add fleet terminal sessions #1501

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
khaliqgant merged 4 commits into from
Aug 13, 2026
Merged

feat(sdk-swift): add fleet terminal sessions #1501

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9cb2061
feat(sdk-swift): add fleet terminal sessions
Aug 13, 2026
a834304
fix(terminal): preserve safe operation boundaries
Aug 13, 2026
f8bc8c3
fix(terminal): harden SDK reconnect and discovery
Aug 13, 2026
810bf0e
fix(sdk-swift): close failed terminal handshakes
Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions crates/broker/src/runtime/event_loop.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>,
pub(super) deadline: Instant,
}

Expand Down
61 changes: 61 additions & 0 deletions crates/broker/src/runtime/fleet.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
},
);
Expand Down Expand Up @@ -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,
Comment thread
khaliqgant marked this conversation as resolved.
},
);
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(
Expand Down Expand Up @@ -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, .. }
Expand Down
35 changes: 29 additions & 6 deletions crates/broker/src/runtime/maintenance.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>)> =
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);
Expand Down
121 changes: 76 additions & 45 deletions crates/broker/src/runtime/worker_events.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, TerminalSession>,
delivery_states: &HashMap<WorkerName, InboundDeliveryState>,
session_id: &str,
) -> Option<InboundDeliveryMode> {
terminal_sessions
.get(session_id)
.and_then(|session| delivery_states.get(&session.agent))
.map(|state| state.mode)
.or(Some(InboundDeliveryMode::AutoInject))
) -> (Option<InboundDeliveryMode>, Option<String>) {
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(
Expand Down Expand Up @@ -201,11 +209,11 @@ mod terminal_ready_delivery_mode_tests {
let delivery_states: HashMap<WorkerName, InboundDeliveryState> = 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"
);
Expand All @@ -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<String, TerminalSession> = HashMap::new();
let delivery_states: HashMap<WorkerName, InboundDeliveryState> = 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));
}
}

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand All @@ -1107,19 +1120,20 @@ 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(),
code: "snapshot_failed".into(),
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 {
Expand All @@ -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),
Expand All @@ -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 { .. });
Expand Down Expand Up @@ -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.
Expand Down
Loading
Loading