refactor: register terminal backends and reuse shared workflows
Extract multiplexer dialect and resumable terminal engine registries while preserving session and checkpoint compatibility. Share local-network and theme policies, add regression coverage, and update discovery actions to Cancel and Rescan with localized Telnet copy.
This commit is contained in:
1 parent
5a74809da8
commit
9da4b350cf
39 files changed
+2120
-838
No files matched your search
@@ -0,0 +1,194 @@
|
||||
import ETBootstrap
|
||||
import ETSession
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
actor EternalTerminalEngine: ResumableTerminalEngine {
|
||||
// Keep the original flat payload keys: existing encrypted recovery files outlive app upgrades.
|
||||
struct Payload: Codable, Sendable {
|
||||
let clientID: String
|
||||
let passkey: Data
|
||||
let checkpoint: ETSessionCheckpoint
|
||||
let integration: RemoteShellIntegrationBootstrap?
|
||||
let commandFilter: RemoteCommandEventFilter?
|
||||
let integratedShell: ResumableTerminalShellState.IntegratedShell?
|
||||
let pendingIntegrationCommand: Data?
|
||||
|
||||
var shellState: ResumableTerminalShellState {
|
||||
.init(
|
||||
integration: integration, commandFilter: commandFilter, integratedShell: integratedShell,
|
||||
pendingIntegrationCommand: pendingIntegrationCommand)
|
||||
}
|
||||
}
|
||||
|
||||
private let receive: EventHandler
|
||||
private var session: ETTerminalSession?
|
||||
private var credentials: ETCredentials?
|
||||
private var bridge: TerminalTCPBridge?
|
||||
private var reader: Task<Void, Never>?
|
||||
private var observer: Task<Void, Never>?
|
||||
private var closed = false
|
||||
|
||||
init(receive: @escaping EventHandler) {
|
||||
self.receive = receive
|
||||
}
|
||||
|
||||
func prepare(_ context: ResumableTerminalEngineContext) async throws -> ResumableTerminalEnginePreparation {
|
||||
let payload = try context.checkpoint.map(Self.decodeCheckpoint)
|
||||
let socket = try await context.route()
|
||||
guard !closed else {
|
||||
socket.close()
|
||||
throw CancellationError()
|
||||
}
|
||||
let bridge = try TerminalTCPBridge(open: context.route, initial: socket)
|
||||
self.bridge = bridge
|
||||
if let payload {
|
||||
credentials = ETCredentials(clientID: payload.clientID, passkey: payload.passkey)
|
||||
session = try ETTerminalSession(
|
||||
host: "127.0.0.1", port: bridge.port, clientID: payload.clientID,
|
||||
passkey: payload.passkey, checkpoint: payload.checkpoint)
|
||||
return .init(shell: payload.shellState)
|
||||
}
|
||||
let executor = BootstrapExecutor(connection: context.bootstrap)
|
||||
do {
|
||||
credentials = try await ETBootstrap().run(using: executor)
|
||||
} catch {
|
||||
if let original = await executor.failure { throw original }
|
||||
throw TerminalTransportError.bootstrapFailed
|
||||
}
|
||||
try checkOpen()
|
||||
guard let credentials else { throw TerminalTransportError.bootstrapFailed }
|
||||
session = try ETTerminalSession(
|
||||
host: "127.0.0.1", port: bridge.port, clientID: credentials.clientID, passkey: credentials.passkey,
|
||||
environmentVariables: Dictionary(
|
||||
context.environment.map { ($0.variable, $0.value) }, uniquingKeysWith: { _, last in last }))
|
||||
var result = ResumableTerminalEnginePreparation(startupInput: try context.startup.eternalInput)
|
||||
if context.shellIntegrationEnabled && context.startup.plan == nil {
|
||||
let integration = RemoteShellIntegrationBootstrap()
|
||||
result.shell.integration = integration
|
||||
result.shell.commandFilter = RemoteCommandEventFilter(token: integration.token)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
static func decodeCheckpoint(_ payload: Data) throws -> Payload {
|
||||
try JSONDecoder().decode(Payload.self, from: payload)
|
||||
}
|
||||
|
||||
func start(columns: UInt16, rows: UInt16) async throws {
|
||||
try checkOpen()
|
||||
guard let session else { throw TerminalTransportError.sessionFailed }
|
||||
startReaders(output: session.output, states: session.stateChanges)
|
||||
try await session.connect(initialDimensions: .init(rows: Int(rows), columns: Int(columns)))
|
||||
try checkOpen()
|
||||
}
|
||||
|
||||
func startReaders(output: AsyncStream<Data>, states: AsyncStream<ETConnectionState>) {
|
||||
reader = Task { [weak self] in
|
||||
for await data in output {
|
||||
guard let self, !Task.isCancelled else { return }
|
||||
await self.deliver(data)
|
||||
}
|
||||
}
|
||||
observer = Task { [weak self] in
|
||||
for await state in states {
|
||||
guard let self, !Task.isCancelled else { return }
|
||||
await self.receiveState(state)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func deliver(_ data: Data) async {
|
||||
guard !closed else { return }
|
||||
await receive(.output(data))
|
||||
}
|
||||
|
||||
private func receiveState(_ state: ETConnectionState) async {
|
||||
guard !closed else { return }
|
||||
switch state {
|
||||
case .connected: await receive(.connected)
|
||||
case .reconnecting, .disconnected: await receive(.recovering)
|
||||
case .failed: await receive(.failed)
|
||||
case .sessionEnded:
|
||||
// Vendor output finishes first, but our consumer may still be draining it.
|
||||
await reader?.value
|
||||
guard !closed else { return }
|
||||
await receive(.ended)
|
||||
case .closed, .idle, .bootstrapping, .connecting: break
|
||||
}
|
||||
}
|
||||
|
||||
func write(_ data: Data) async throws {
|
||||
try checkOpen()
|
||||
try await session?.send(data)
|
||||
}
|
||||
|
||||
func resize(columns: UInt16, rows: UInt16) async throws {
|
||||
try checkOpen()
|
||||
try await session?.resize(rows: Int(rows), cols: Int(columns))
|
||||
}
|
||||
|
||||
func checkpoint(shell: @Sendable () async -> ResumableTerminalShellState) async throws -> Data {
|
||||
try checkOpen()
|
||||
guard let session, let credentials else { throw TerminalTransportError.invalidCheckpoint }
|
||||
let checkpoint = try await session.prepareForApplicationBackground()
|
||||
let shell = await shell()
|
||||
return try JSONEncoder().encode(
|
||||
Payload(
|
||||
clientID: credentials.clientID, passkey: credentials.passkey, checkpoint: checkpoint,
|
||||
integration: shell.integration, commandFilter: shell.commandFilter,
|
||||
integratedShell: shell.integratedShell, pendingIntegrationCommand: shell.pendingIntegrationCommand))
|
||||
}
|
||||
|
||||
func resume() async throws {
|
||||
try checkOpen()
|
||||
await session?.resumeFromApplicationBackground()
|
||||
}
|
||||
|
||||
func prepareRetry() async {
|
||||
await bridge?.retry()
|
||||
}
|
||||
|
||||
func retry() async throws {
|
||||
try checkOpen()
|
||||
await session?.notifyNetworkPathChanged()
|
||||
}
|
||||
|
||||
func networkAvailable() async {
|
||||
await session?.notifyNetworkPathChanged()
|
||||
}
|
||||
|
||||
func close() async {
|
||||
guard !closed else { return }
|
||||
closed = true
|
||||
reader?.cancel()
|
||||
observer?.cancel()
|
||||
await session?.close()
|
||||
await bridge?.close()
|
||||
session = nil
|
||||
bridge = nil
|
||||
credentials = nil
|
||||
}
|
||||
|
||||
private func checkOpen() throws {
|
||||
try Task.checkCancellation()
|
||||
guard !closed else { throw CancellationError() }
|
||||
}
|
||||
}
|
||||
|
||||
private actor BootstrapExecutor: ETBootstrapExecutor {
|
||||
let connection: SSHConnection
|
||||
private(set) var failure: (any Error)?
|
||||
init(connection: SSHConnection) { self.connection = connection }
|
||||
func run(command: String) async throws -> String {
|
||||
do {
|
||||
let result = try await connection.execute(
|
||||
command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 65_536, onOutput: nil)
|
||||
return result.stdout + result.stderr
|
||||
} catch {
|
||||
failure = error
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
import Foundation
|
||||
import MoshBootstrap
|
||||
import MoshCore
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
actor MoshTerminalEngine: ResumableTerminalEngine {
|
||||
private let receive: EventHandler
|
||||
private var session: MoshClientSession?
|
||||
private var reader: Task<Void, Never>?
|
||||
private var beforeConnect: ConnectionWakeAction?
|
||||
private var rollbackPID: Int32?
|
||||
private var closed = false
|
||||
|
||||
init(receive: @escaping EventHandler) {
|
||||
self.receive = receive
|
||||
}
|
||||
|
||||
func prepare(_ context: ResumableTerminalEngineContext) async throws -> ResumableTerminalEnginePreparation {
|
||||
beforeConnect = context.request.configuration.beforeConnect
|
||||
if let payload = context.checkpoint {
|
||||
let restored = try await MoshClientSession.restore(from: Self.decodeCheckpoint(payload))
|
||||
guard !closed else {
|
||||
await restored.stop()
|
||||
throw CancellationError()
|
||||
}
|
||||
session = restored
|
||||
} else {
|
||||
let range = context.options.moshUDPPortRange.replacingOccurrences(of: "-", with: ":")
|
||||
var command = "mosh-server new -s -c 256"
|
||||
if !range.isEmpty { command += " -p \(range)" }
|
||||
command += try context.startup.moshArguments
|
||||
let variables = context.environment.map {
|
||||
"'\($0.variable)=\($0.value.replacingOccurrences(of: "'", with: "'\\''"))'"
|
||||
}
|
||||
command = (["env"] + variables + ["LANG=C.UTF-8", "LC_ALL=C.UTF-8", command]).joined(separator: " ")
|
||||
let result = try await context.bootstrap.execute(
|
||||
command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 65_536, onOutput: nil)
|
||||
let info: MoshServerConnectInfo
|
||||
do {
|
||||
info = try MoshServerOutputParser.parse(result.stdout + result.stderr)
|
||||
} catch {
|
||||
throw TerminalTransportError.bootstrapFailed
|
||||
}
|
||||
rollbackPID = info.serverPID
|
||||
try checkOpen()
|
||||
session = MoshClientSession(
|
||||
endpoint: .init(host: context.request.configuration.host, port: info.port, keyBase64_22: info.key))
|
||||
}
|
||||
return .init()
|
||||
}
|
||||
|
||||
static func decodeCheckpoint(_ payload: Data) throws -> MoshSnapshot {
|
||||
try JSONDecoder().decode(MoshSnapshot.self, from: payload)
|
||||
}
|
||||
|
||||
func start(columns: UInt16, rows: UInt16) async throws {
|
||||
try checkOpen()
|
||||
guard let session else { throw TerminalTransportError.sessionFailed }
|
||||
try await session.start()
|
||||
try await session.enqueue(.resize(cols: Int32(columns), rows: Int32(rows)))
|
||||
// An open UDP socket is not proof that the server is reachable.
|
||||
let deadline = ContinuousClock.now.advanced(by: .seconds(20))
|
||||
while true {
|
||||
try checkOpen()
|
||||
let operations = await session.drainHostOps()
|
||||
if !operations.isEmpty {
|
||||
await deliver(operations)
|
||||
break
|
||||
}
|
||||
let state = await session.state
|
||||
if case .failed = state { throw TerminalTransportError.sessionFailed }
|
||||
if state == .stopped {
|
||||
rollbackPID = nil
|
||||
_ = await receivePoll(state: state, operations: session.drainHostOps())
|
||||
return
|
||||
}
|
||||
guard ContinuousClock.now < deadline else { throw TerminalTransportError.unreachable }
|
||||
try await Task.sleep(for: .milliseconds(20))
|
||||
}
|
||||
rollbackPID = nil
|
||||
await receive(.connected)
|
||||
try checkOpen()
|
||||
reader = Task { [weak self] in
|
||||
while !Task.isCancelled {
|
||||
guard let self else { return }
|
||||
// Sample state before draining so the final poll includes trailing output.
|
||||
let state = await session.state
|
||||
guard await self.receivePoll(state: state, operations: session.drainHostOps()) else { return }
|
||||
try? await Task.sleep(for: .milliseconds(16))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func receivePoll(state: MoshSessionState, operations: [MoshHostOp]) async -> Bool {
|
||||
guard !closed else { return false }
|
||||
await deliver(operations)
|
||||
switch state {
|
||||
case .stopped:
|
||||
await receive(.ended)
|
||||
return false
|
||||
case .failed:
|
||||
await receive(.failed)
|
||||
return false
|
||||
case .idle, .starting, .running, .suspending, .suspended, .stopping:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
private func deliver(_ operations: [MoshHostOp]) async {
|
||||
guard !closed else { return }
|
||||
if !operations.isEmpty { await receive(.activity) }
|
||||
for operation in operations {
|
||||
if case .hostBytes(let data) = operation { await receive(.output(data)) }
|
||||
}
|
||||
}
|
||||
|
||||
func write(_ data: Data) async throws {
|
||||
try checkOpen()
|
||||
try await session?.enqueue(.keystrokes(data))
|
||||
}
|
||||
|
||||
func resize(columns: UInt16, rows: UInt16) async throws {
|
||||
try checkOpen()
|
||||
try await session?.enqueue(.resize(cols: Int32(columns), rows: Int32(rows)))
|
||||
}
|
||||
|
||||
func checkpoint(shell: @Sendable () async -> ResumableTerminalShellState) async throws -> Data {
|
||||
try checkOpen()
|
||||
guard let session else { throw TerminalTransportError.invalidCheckpoint }
|
||||
return try JSONEncoder().encode(await session.prepareForApplicationBackground())
|
||||
}
|
||||
|
||||
func beforeResume() async throws {
|
||||
await beforeConnect?()
|
||||
try checkOpen()
|
||||
}
|
||||
|
||||
func resume() async throws {
|
||||
try checkOpen()
|
||||
try await session?.resumeFromApplicationBackground()
|
||||
}
|
||||
|
||||
func retry() async throws {
|
||||
try await beforeResume()
|
||||
_ = try await session?.prepareForApplicationBackground()
|
||||
try await session?.resumeFromApplicationBackground()
|
||||
}
|
||||
|
||||
func rollback(using bootstrap: SSHConnection) async {
|
||||
guard let pid = rollbackPID, pid > 1 else { return }
|
||||
rollbackPID = nil
|
||||
// Undo only the server launched by this attempt, even if the caller was cancelled.
|
||||
await Task.detached {
|
||||
_ = try? await bootstrap.execute(
|
||||
"[ \"$(ps -p \(pid) -o comm=)\" = mosh-server ] && kill -TERM \(pid)",
|
||||
workingDirectory: nil, stdin: nil, timeout: 3, outputLimit: 1024, onOutput: nil)
|
||||
}.value
|
||||
}
|
||||
|
||||
func close() async {
|
||||
guard !closed else { return }
|
||||
closed = true
|
||||
reader?.cancel()
|
||||
beforeConnect = nil
|
||||
await session?.stop()
|
||||
session = nil
|
||||
}
|
||||
|
||||
private func checkOpen() throws {
|
||||
try Task.checkCancellation()
|
||||
guard !closed else { throw CancellationError() }
|
||||
}
|
||||
}
|
||||
@@ -24,16 +24,19 @@ public nonisolated final class RemoteConnectionCoordinator: Sendable {
|
||||
}
|
||||
private let keyboardInteractive: SSHKeyboardInteractiveHandler?
|
||||
private let maxSessionsPolicy: SSHMaxSessionsPolicy
|
||||
private let transportEngines: ResumableTerminalEngineRegistry
|
||||
let logger: Logger
|
||||
|
||||
public init(
|
||||
keyboardInteractive: SSHKeyboardInteractiveHandler? = nil,
|
||||
logger: Logger = ApplicationLogger.disabled("termind.remote-connection"),
|
||||
maxSessionsPolicy: SSHMaxSessionsPolicy = SSHMaxSessionsPolicy()
|
||||
maxSessionsPolicy: SSHMaxSessionsPolicy = SSHMaxSessionsPolicy(),
|
||||
transportEngines: ResumableTerminalEngineRegistry = .live
|
||||
) {
|
||||
self.keyboardInteractive = keyboardInteractive
|
||||
self.logger = logger
|
||||
self.maxSessionsPolicy = maxSessionsPolicy
|
||||
self.transportEngines = transportEngines
|
||||
}
|
||||
|
||||
public func updateSSHPreferences(_ preferences: SSHPreferences) {
|
||||
@@ -45,7 +48,7 @@ public nonisolated final class RemoteConnectionCoordinator: Sendable {
|
||||
{
|
||||
RemoteConnectionCoordinator(
|
||||
keyboardInteractive: handler, logger: logger,
|
||||
maxSessionsPolicy: maxSessionsPolicy)
|
||||
maxSessionsPolicy: maxSessionsPolicy, transportEngines: transportEngines)
|
||||
}
|
||||
public func makePTYChannel(
|
||||
output: @escaping SSHPTYChannel.OutputHandler,
|
||||
@@ -55,7 +58,7 @@ public nonisolated final class RemoteConnectionCoordinator: Sendable {
|
||||
SSHPTYChannel(
|
||||
output: output, exit: exit, shellIntegration: shellIntegration, keyboardInteractive: keyboardInteractive,
|
||||
agentForwardingFactory: agentFactory,
|
||||
logger: logger, maxSessionsPolicy: maxSessionsPolicy
|
||||
logger: logger, maxSessionsPolicy: maxSessionsPolicy, transportEngines: transportEngines
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ nonisolated enum RemoteHerdrCommands {
|
||||
}
|
||||
|
||||
static var list: String {
|
||||
RemoteMultiplexerKind.herdr.discovery + "\n\"$exe\" session list --json"
|
||||
RemoteMultiplexerCommands.discovery(.herdr) + "\n\"$exe\" session list --json"
|
||||
}
|
||||
|
||||
static func parseList(_ text: String) throws -> [RemoteMultiplexerCandidate] {
|
||||
@@ -61,3 +61,96 @@ nonisolated enum RemoteHerdrCommands {
|
||||
"""
|
||||
}
|
||||
}
|
||||
|
||||
nonisolated struct RemoteHerdrDialect: RemoteSocketMultiplexerDialect {
|
||||
let id = SessionPersistenceBackendID.herdr
|
||||
let paneVariable: String? = "HERDR_PANE_ID"
|
||||
var list: String { RemoteHerdrCommands.list }
|
||||
|
||||
func validateVersion(_ output: String) throws {
|
||||
let version = try RemoteMultiplexerVersion(output, backend: id)
|
||||
// Preserve both supported release lines; 0.8.0 and 0.8.1 lack the lifecycle fixes.
|
||||
guard
|
||||
version.major > 0 || version.minor > 8 || (version.minor == 8 && version.patch >= 2)
|
||||
|| (version.minor == 7 && version.patch >= 5)
|
||||
else { throw RemoteSessionFailure.unsupported }
|
||||
}
|
||||
func validPane(_ value: String) -> Bool {
|
||||
!value.isEmpty && value.utf8.count <= 128
|
||||
&& value.utf8.allSatisfy {
|
||||
(48...57).contains($0) || (65...90).contains($0) || (97...122).contains($0)
|
||||
|| [45, 58, 95].contains($0)
|
||||
}
|
||||
}
|
||||
func parseList(_ text: String) throws -> [RemoteMultiplexerCandidate] { try RemoteHerdrCommands.parseList(text) }
|
||||
func create(name: String, token: UUID) -> String { RemoteHerdrCommands.create(name: name) }
|
||||
func attach(name: String) -> String { "\"$exe\" --session \(name) client" }
|
||||
func stop(name: String) -> String { "\"$exe\" session stop \(name) --json" }
|
||||
func validRoute(_ identity: RemoteMultiplexerIdentity, name: String) -> Bool {
|
||||
identity.socket == identity.root + "/herdr.sock"
|
||||
&& (name == "default"
|
||||
? identity.root.hasSuffix("/herdr") : identity.root.hasSuffix("/herdr/sessions/" + name))
|
||||
}
|
||||
func routing(_ identity: RemoteMultiplexerIdentity) throws -> String {
|
||||
let root = identity.root
|
||||
let config: String
|
||||
if root.hasSuffix("/herdr") {
|
||||
config = String(root.dropLast("/herdr".count))
|
||||
} else if let range = root.range(of: "/herdr/sessions/", options: .backwards),
|
||||
RemoteMultiplexerCommands.validName(String(root[range.upperBound...]))
|
||||
{
|
||||
config = String(root[..<range.lowerBound])
|
||||
} else {
|
||||
throw RemoteSessionFailure.invalidResponse
|
||||
}
|
||||
guard RemoteMultiplexerCommands.validPath(config) else { throw RemoteSessionFailure.invalidResponse }
|
||||
return "export XDG_CONFIG_HOME=\(try RemoteMultiplexerCommands.quote(config))"
|
||||
}
|
||||
func metadataFields(_ candidate: RemoteMultiplexerCandidate) throws -> String {
|
||||
"directory=$(directory_fingerprint \(try RemoteMultiplexerCommands.quote(candidate.root))) || exit 79"
|
||||
}
|
||||
func validateTermination(_ reference: RemoteSessionReference, identity: RemoteMultiplexerIdentity) throws {
|
||||
guard reference.session.name != "default", identity.directoryFingerprint != nil else {
|
||||
throw RemoteSessionFailure.ownershipMismatch
|
||||
}
|
||||
}
|
||||
func terminationCleanup(_ reference: RemoteSessionReference, identity: RemoteMultiplexerIdentity) throws -> String {
|
||||
"""
|
||||
private_state || exit 79
|
||||
[ "$(cat "$meta/owner")" = '\(reference.ownershipToken!.uuidString)' ] || exit 79
|
||||
[ "$(cat "$meta/identity")" = \(try RemoteMultiplexerCommands.quote(identity.fingerprint)) ] || exit 79
|
||||
[ "$(directory_fingerprint \(try RemoteMultiplexerCommands.quote(identity.root)))" = '\(identity.directoryFingerprint!)' ] || exit 79
|
||||
[ ! -e \(try RemoteMultiplexerCommands.quote(identity.socket)) ] || exit 79
|
||||
"$exe" session delete \(try RemoteMultiplexerCommands.quote(reference.session.name)) --json >/dev/null || exit 78
|
||||
[ ! -e \(try RemoteMultiplexerCommands.quote(identity.root)) ] || exit 79
|
||||
"""
|
||||
}
|
||||
func runtimeCommands(name: String) -> (String, String) {
|
||||
("\"$exe\" session list --json", "\"$exe\" --session \(name) pane list")
|
||||
}
|
||||
func activePane(native: String, focus: String, reference: RemoteSessionReference) throws -> String? {
|
||||
let sessions = try RemoteHerdrCommands.parseList(native)
|
||||
guard sessions.count == 1, sessions.first?.name == reference.session.name else { return nil }
|
||||
struct Response: Decodable {
|
||||
struct Result: Decodable {
|
||||
struct Pane: Decodable {
|
||||
let paneID: String
|
||||
enum CodingKeys: String, CodingKey { case paneID = "pane_id" }
|
||||
}
|
||||
let panes: [Pane]
|
||||
}
|
||||
let result: Result
|
||||
}
|
||||
let response = try JSONDecoder().decode(Response.self, from: Data(focus.utf8))
|
||||
// Herdr's pane-current is global, not a per-client focus API.
|
||||
guard response.result.panes.count == 1, let pane = response.result.panes.first?.paneID, validPane(pane) else {
|
||||
return nil
|
||||
}
|
||||
// Global pane state cannot bind an attached client's session/focus, including endpoint switching.
|
||||
// Journals remain useful for lifecycle events, but active-context operations stay disabled.
|
||||
return nil
|
||||
}
|
||||
func writeDraft(name: String, pane: String, key: String) -> String {
|
||||
"\"$exe\" --session \(name) pane send-text \(pane) \(key)"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
/// The registration boundary for session commands, runtime inspection, and shell pane identity.
|
||||
nonisolated protocol RemoteMultiplexerDialect: Sendable {
|
||||
var id: SessionPersistenceBackendID { get }
|
||||
var capabilities: RemoteSessionCapabilities { get }
|
||||
var versionArgument: String { get }
|
||||
var paneVariable: String? { get }
|
||||
func validateVersion(_ output: String) throws
|
||||
func validPane(_ value: String) -> Bool
|
||||
func backend(execute: @escaping RemoteSessionCommandExecutor) -> any RemoteSessionBackend
|
||||
func runtime(
|
||||
reference: RemoteSessionReference?, execute: @escaping RemoteSessionCommandExecutor, charset: RemoteCharset
|
||||
) throws -> any RemoteMultiplexerRuntime
|
||||
func helper(_ reference: RemoteSessionReference) throws -> String?
|
||||
}
|
||||
|
||||
nonisolated protocol RemoteMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
func draftTarget(
|
||||
context: TerminalShellContext, terminal: String, replacing: Bool
|
||||
) async throws -> RemoteMultiplexerDraftTarget
|
||||
}
|
||||
|
||||
nonisolated enum RemoteMultiplexerRegistry {
|
||||
static let dialects: [any RemoteMultiplexerDialect] = [
|
||||
RemoteTmuxDialect(), RemoteZellijDialect(), RemoteZmxDialect(), RemoteHerdrDialect(),
|
||||
]
|
||||
private static let registrations = Dictionary(uniqueKeysWithValues: dialects.map { ($0.id, $0) })
|
||||
|
||||
static func dialect(_ id: SessionPersistenceBackendID) -> (any RemoteMultiplexerDialect)? {
|
||||
registrations[id]
|
||||
}
|
||||
|
||||
static func socketDialect(_ id: SessionPersistenceBackendID) -> (any RemoteSocketMultiplexerDialect)? {
|
||||
dialect(id) as? any RemoteSocketMultiplexerDialect
|
||||
}
|
||||
}
|
||||
|
||||
extension RemoteMultiplexerDialect {
|
||||
nonisolated func probe(execute: RemoteSessionCommandExecutor) async throws -> RemoteSessionCapabilities {
|
||||
let result = try await execute(RemoteMultiplexerCommands.discovery(id) + "\n\"$exe\" \(versionArgument)")
|
||||
try RemoteMultiplexerBackend.check(result)
|
||||
try validateVersion(result.stdout)
|
||||
return capabilities
|
||||
}
|
||||
}
|
||||
|
||||
/// Only native differences live here; the socket backend retains framing and ownership orchestration.
|
||||
nonisolated protocol RemoteSocketMultiplexerDialect: RemoteMultiplexerDialect {
|
||||
var list: String { get }
|
||||
func parseList(_ text: String) throws -> [RemoteMultiplexerCandidate]
|
||||
// Creation receives a validated managed name. Attach/stop/runtimeCommands/writeDraft receive quoted arguments.
|
||||
func create(name: String, token: UUID) -> String
|
||||
func attach(name: String) -> String
|
||||
func stop(name: String) -> String
|
||||
func routing(_ identity: RemoteMultiplexerIdentity) throws -> String
|
||||
func validRoute(_ identity: RemoteMultiplexerIdentity, name: String) -> Bool
|
||||
func validateIdentity(_ identity: RemoteMultiplexerIdentity) throws
|
||||
func identityGuard(_ identity: RemoteMultiplexerIdentity, name: String) throws -> String
|
||||
func ownershipGuard(_ identity: RemoteMultiplexerIdentity, name: String, token: UUID) throws -> String
|
||||
func metadataOwnershipGuard(_ candidate: RemoteMultiplexerCandidate) throws -> String
|
||||
func metadataFields(_ candidate: RemoteMultiplexerCandidate) throws -> String
|
||||
func prepareAttachment() -> String
|
||||
func validateTermination(_ reference: RemoteSessionReference, identity: RemoteMultiplexerIdentity) throws
|
||||
func terminationCleanup(_ reference: RemoteSessionReference, identity: RemoteMultiplexerIdentity) throws -> String
|
||||
func runtimeCommands(name: String) -> (String, String)
|
||||
func activePane(native: String, focus: String, reference: RemoteSessionReference) throws -> String?
|
||||
func writeDraft(name: String, pane: String, key: String) -> String
|
||||
var restoreShellEnvironment: String { get }
|
||||
}
|
||||
|
||||
extension RemoteSocketMultiplexerDialect {
|
||||
nonisolated var versionArgument: String { "--version" }
|
||||
nonisolated var capabilities: RemoteSessionCapabilities { .init(shellIntegration: true) }
|
||||
nonisolated func backend(execute: @escaping RemoteSessionCommandExecutor) -> any RemoteSessionBackend {
|
||||
RemoteMultiplexerBackend(dialect: self, execute: execute)
|
||||
}
|
||||
nonisolated func runtime(
|
||||
reference: RemoteSessionReference?, execute: @escaping RemoteSessionCommandExecutor, charset: RemoteCharset
|
||||
) throws -> any RemoteMultiplexerRuntime {
|
||||
guard let reference else { throw TerminalRuntimeError.unsupported }
|
||||
return RemoteSocketMultiplexerRuntime(reference: reference, execute: execute, charset: charset)
|
||||
}
|
||||
nonisolated func helper(_ reference: RemoteSessionReference) throws -> String? {
|
||||
try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: id).helper
|
||||
}
|
||||
nonisolated func validateIdentity(_ identity: RemoteMultiplexerIdentity) throws {
|
||||
guard !identity.isolated else { throw RemoteSessionFailure.invalidResponse }
|
||||
}
|
||||
nonisolated func identityGuard(_ identity: RemoteMultiplexerIdentity, name: String) throws -> String { "" }
|
||||
nonisolated func ownershipGuard(_ identity: RemoteMultiplexerIdentity, name: String, token: UUID) throws -> String {
|
||||
""
|
||||
}
|
||||
nonisolated func metadataOwnershipGuard(_ candidate: RemoteMultiplexerCandidate) throws -> String { "" }
|
||||
nonisolated func metadataFields(_ candidate: RemoteMultiplexerCandidate) throws -> String { "" }
|
||||
nonisolated func prepareAttachment() -> String { "" }
|
||||
nonisolated func validateTermination(_ reference: RemoteSessionReference, identity: RemoteMultiplexerIdentity)
|
||||
throws
|
||||
{}
|
||||
nonisolated func terminationCleanup(_ reference: RemoteSessionReference, identity: RemoteMultiplexerIdentity) throws
|
||||
-> String
|
||||
{
|
||||
""
|
||||
}
|
||||
nonisolated var restoreShellEnvironment: String { "" }
|
||||
}
|
||||
|
||||
nonisolated struct RemoteMultiplexerVersion {
|
||||
let major: Int
|
||||
let minor: Int
|
||||
let patch: Int
|
||||
|
||||
init(_ output: String, backend: SessionPersistenceBackendID) throws {
|
||||
let fields = output.split(whereSeparator: \.isWhitespace)
|
||||
guard fields.count >= 2, fields[0].lowercased() == backend.rawValue else {
|
||||
throw RemoteSessionFailure.invalidResponse
|
||||
}
|
||||
let components = fields[1].split(separator: ".")
|
||||
guard components.count == 3, let major = Int(components[0]), let minor = Int(components[1]),
|
||||
let patch = Int(components[2]), major >= 0, minor >= 0, patch >= 0
|
||||
else { throw RemoteSessionFailure.unsupported }
|
||||
self.major = major
|
||||
self.minor = minor
|
||||
self.patch = patch
|
||||
}
|
||||
}
|
||||
@@ -21,37 +21,32 @@ public nonisolated struct RemoteMultiplexerRuntimeAdapter: TerminalRuntimeAdapte
|
||||
self.charset = charset
|
||||
}
|
||||
|
||||
public func inspect(terminal: String) async throws -> TerminalRuntimeUpdate {
|
||||
if let attachment, attachment.backend != .tmux {
|
||||
return try await RemoteSocketMultiplexerRuntime(
|
||||
reference: attachment, execute: execute, charset: charset
|
||||
).inspect(terminal: terminal)
|
||||
private var dialect: any RemoteMultiplexerDialect {
|
||||
get throws {
|
||||
guard let dialect = RemoteMultiplexerRegistry.dialect(attachment?.backend ?? .tmux) else {
|
||||
throw TerminalRuntimeError.unsupported
|
||||
}
|
||||
return dialect
|
||||
}
|
||||
return try await RemoteTmuxRuntimeDialect(execute: execute, charset: charset).inspect(terminal: terminal)
|
||||
}
|
||||
|
||||
public func inspect(terminal: String) async throws -> TerminalRuntimeUpdate {
|
||||
try await dialect.runtime(reference: attachment, execute: execute, charset: charset).inspect(terminal: terminal)
|
||||
}
|
||||
|
||||
static func helper(for attachment: RemoteSessionReference?) throws -> String? {
|
||||
guard let attachment, attachment.backend != .tmux else { return nil }
|
||||
guard let kind = RemoteMultiplexerKind(rawValue: attachment.backend.rawValue) else {
|
||||
guard let attachment else { return nil }
|
||||
guard let dialect = RemoteMultiplexerRegistry.dialect(attachment.backend) else {
|
||||
throw TerminalRuntimeError.unsupported
|
||||
}
|
||||
return try RemoteMultiplexerIdentity.decode(attachment.session.identity, kind: kind).helper
|
||||
return try dialect.helper(attachment)
|
||||
}
|
||||
|
||||
func draftTarget(
|
||||
context: TerminalShellContext, terminal: String, replacing: Bool
|
||||
) async throws -> RemoteMultiplexerDraftTarget {
|
||||
if let attachment, attachment.backend != .tmux {
|
||||
return try await RemoteSocketMultiplexerRuntime(
|
||||
reference: attachment, execute: execute, charset: charset
|
||||
).draftTarget(context: context, terminal: terminal, replacing: replacing)
|
||||
}
|
||||
let update = try await inspect(terminal: terminal)
|
||||
guard update.active == context.identity,
|
||||
update.contexts.first(where: { $0.identity == context.identity }) == context
|
||||
else { throw TerminalRuntimeError.staleContext }
|
||||
return try RemoteTmuxRuntimeDialect.draftTarget(
|
||||
context: context, terminal: terminal, clientIdentity: update.clientIdentity, replacing: replacing)
|
||||
try await dialect.runtime(reference: attachment, execute: execute, charset: charset)
|
||||
.draftTarget(context: context, terminal: terminal, replacing: replacing)
|
||||
}
|
||||
|
||||
func refreshSize(of client: RemoteTmuxClientIdentity) async throws {
|
||||
@@ -59,7 +54,7 @@ public nonisolated struct RemoteMultiplexerRuntimeAdapter: TerminalRuntimeAdapte
|
||||
}
|
||||
}
|
||||
|
||||
nonisolated struct RemoteSocketMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
nonisolated struct RemoteSocketMultiplexerRuntime: RemoteMultiplexerRuntime {
|
||||
let reference: RemoteSessionReference
|
||||
let execute: RemoteSessionCommandExecutor
|
||||
var charset: RemoteCharset = .utf8
|
||||
@@ -74,24 +69,23 @@ nonisolated struct RemoteSocketMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
let focus: String
|
||||
}
|
||||
|
||||
private var identity: RemoteMultiplexerIdentity {
|
||||
private var dialect: any RemoteSocketMultiplexerDialect {
|
||||
get throws {
|
||||
guard let kind = RemoteMultiplexerKind(rawValue: reference.backend.rawValue) else {
|
||||
guard let dialect = RemoteMultiplexerRegistry.socketDialect(reference.backend) else {
|
||||
throw TerminalRuntimeError.unsupported
|
||||
}
|
||||
return try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
return dialect
|
||||
}
|
||||
}
|
||||
|
||||
private var identity: RemoteMultiplexerIdentity {
|
||||
get throws {
|
||||
try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: dialect.id)
|
||||
}
|
||||
}
|
||||
|
||||
private func nativeCommands() throws -> (String, String) {
|
||||
let name = try RemoteMultiplexerCommands.quote(reference.session.name)
|
||||
switch try identity.kind {
|
||||
case .tmux: throw TerminalRuntimeError.unsupported
|
||||
case .zellij:
|
||||
return ("\"$exe\" list-sessions --short --no-formatting", "\"$exe\" --session \(name) action list-clients")
|
||||
case .zmx: return ("\"$exe\" list", "printf 'shell'")
|
||||
case .herdr: return ("\"$exe\" session list --json", "\"$exe\" --session \(name) pane list")
|
||||
}
|
||||
try dialect.runtimeCommands(name: RemoteMultiplexerCommands.quote(reference.session.name))
|
||||
}
|
||||
|
||||
private func clientGuard(terminal: String) throws -> String {
|
||||
@@ -188,8 +182,9 @@ nonisolated struct RemoteSocketMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
_ text: String, reference: RemoteSessionReference, charset: RemoteCharset
|
||||
) throws -> Snapshot {
|
||||
guard text.utf8.count <= 1024 * 1024,
|
||||
let kind = RemoteMultiplexerKind(rawValue: reference.backend.rawValue)
|
||||
let dialect = RemoteMultiplexerRegistry.socketDialect(reference.backend)
|
||||
else { throw TerminalRuntimeError.invalidResponse }
|
||||
let kind = dialect.id
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
let lines = text.components(separatedBy: "\n")
|
||||
guard lines.count >= 5, lines[0].hasPrefix("C|"), UInt64(lines[0].dropFirst(2)) != nil,
|
||||
@@ -197,7 +192,7 @@ nonisolated struct RemoteSocketMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
else { throw TerminalRuntimeError.invalidResponse }
|
||||
let native = try RemoteMultiplexerCommands.unhex(String(lines[1].dropFirst(2)), limit: 512 * 1024)
|
||||
let focus = try RemoteMultiplexerCommands.unhex(String(lines[2].dropFirst(2)), limit: 512 * 1024)
|
||||
let activePane = try activePane(kind: kind, native: native, focus: focus, reference: reference)
|
||||
let activePane = try dialect.activePane(native: native, focus: focus, reference: reference)
|
||||
var contexts: [TerminalShellContext] = []
|
||||
var events: [TerminalContextEvent] = []
|
||||
var active: TerminalContextIdentity?
|
||||
@@ -233,55 +228,6 @@ nonisolated struct RemoteSocketMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
native: native, focus: focus)
|
||||
}
|
||||
|
||||
private static func activePane(
|
||||
kind: RemoteMultiplexerKind, native: String, focus: String, reference: RemoteSessionReference
|
||||
) throws -> String? {
|
||||
switch kind {
|
||||
case .tmux: throw TerminalRuntimeError.unsupported
|
||||
case .zellij:
|
||||
// A switched client cannot be bound using its original argv. Require a unique live session too.
|
||||
guard native == reference.session.name else { return nil }
|
||||
let clients = focus.split(separator: "\n")
|
||||
guard clients.first == "CLIENT_ID ZELLIJ_PANE_ID RUNNING_COMMAND", clients.count == 2 else { return nil }
|
||||
let fields = clients[1].split(whereSeparator: \.isWhitespace)
|
||||
guard fields.count >= 3, UInt64(fields[0]) != nil, fields[1].hasPrefix("terminal_") else { return nil }
|
||||
let pane = String(fields[1].dropFirst("terminal_".count))
|
||||
return RemoteShellContextRecord.validPane(pane, backend: "zellij") ? pane : nil
|
||||
case .zmx:
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
let root = identity.root.utf8.map { String(format: "%02x", $0) }.joined()
|
||||
let sessions = try RemoteZmxCommands.parseList("ROOT|\(root)\n" + native + "\nEND\n")
|
||||
guard sessions.count == 1, let session = sessions.first,
|
||||
session.name == reference.session.name, session.process == identity.process,
|
||||
session.created == identity.created, session.clients == 1
|
||||
else { return nil }
|
||||
return "shell"
|
||||
case .herdr:
|
||||
let sessions = try RemoteHerdrCommands.parseList(native)
|
||||
guard sessions.count == 1, sessions.first?.name == reference.session.name else { return nil }
|
||||
struct Response: Decodable {
|
||||
struct Result: Decodable {
|
||||
struct Pane: Decodable {
|
||||
let paneID: String
|
||||
enum CodingKeys: String, CodingKey { case paneID = "pane_id" }
|
||||
}
|
||||
let panes: [Pane]
|
||||
}
|
||||
let result: Result
|
||||
}
|
||||
let response = try JSONDecoder().decode(Response.self, from: Data(focus.utf8))
|
||||
// Herdr's pane-current is global, not a per-client focus API.
|
||||
guard response.result.panes.count == 1, let pane = response.result.panes.first?.paneID,
|
||||
RemoteShellContextRecord.validPane(pane, backend: "herdr")
|
||||
else { return nil }
|
||||
// The supported CLI only exposes global pane state, not the attached client's
|
||||
// session/focus binding (including endpoint switching). Do not infer it from ps.
|
||||
// Journals remain useful for lifecycle events; active-context operations stay disabled.
|
||||
_ = pane
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func draftTarget(
|
||||
context: TerminalShellContext, terminal: String, replacing: Bool
|
||||
) async throws -> RemoteMultiplexerDraftTarget {
|
||||
@@ -301,13 +247,7 @@ nonisolated struct RemoteSocketMultiplexerRuntime: TerminalRuntimeAdapter {
|
||||
let key = try RemoteMultiplexerCommands.quote(replacing ? "\u{1B}[777;2~" : "\u{1B}[777;1~")
|
||||
let name = try RemoteMultiplexerCommands.quote(reference.session.name)
|
||||
let pane = try RemoteMultiplexerCommands.quote(context.identity.pane)
|
||||
let write: String
|
||||
switch try identity.kind {
|
||||
case .tmux: throw TerminalRuntimeError.unsupported
|
||||
case .zellij: write = "\"$exe\" --session \(name) action write-chars --pane-id \(pane) \(key)"
|
||||
case .zmx: write = "\"$exe\" send \(name) \(key)"
|
||||
case .herdr: write = "\"$exe\" --session \(name) pane send-text \(pane) \(key)"
|
||||
}
|
||||
let write = try dialect.writeDraft(name: name, pane: pane, key: key)
|
||||
return .init(
|
||||
root: guardCommand + "\nhelper=\(try RemoteMultiplexerCommands.quote(helper))",
|
||||
trigger: """
|
||||
|
||||
@@ -6,19 +6,19 @@ import TermindDomain
|
||||
public nonisolated struct RemoteMultiplexerService: RemoteSessionBackend {
|
||||
public typealias Executor = RemoteSessionCommandExecutor
|
||||
public let id: SessionPersistenceBackendID
|
||||
private let execute: Executor
|
||||
private let implementation: (any RemoteSessionBackend)?
|
||||
|
||||
public init(backend: SessionPersistenceBackendID = .tmux, execute: @escaping Executor) {
|
||||
id = backend
|
||||
self.execute = execute
|
||||
implementation = RemoteMultiplexerRegistry.dialect(backend)?.backend(execute: execute)
|
||||
}
|
||||
|
||||
private var backend: RemoteMultiplexerBackend {
|
||||
private var backend: any RemoteSessionBackend {
|
||||
get throws {
|
||||
guard let kind = RemoteMultiplexerKind(rawValue: id.rawValue) else {
|
||||
guard let implementation else {
|
||||
throw RemoteSessionFailure.unsupported
|
||||
}
|
||||
return .init(kind: kind, execute: execute)
|
||||
return implementation
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3,56 +3,17 @@ import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
nonisolated enum RemoteMultiplexerKind: String, Codable, Sendable {
|
||||
case tmux, zellij, zmx, herdr
|
||||
|
||||
var id: SessionPersistenceBackendID { .init(rawValue: rawValue) }
|
||||
|
||||
func validateVersion(_ output: String) throws {
|
||||
if self == .tmux { return try RemoteTmuxCommands.validateVersion(output) }
|
||||
let fields = output.split(whereSeparator: \.isWhitespace)
|
||||
guard fields.count >= 2, fields[0].lowercased() == rawValue else {
|
||||
throw RemoteSessionFailure.invalidResponse
|
||||
}
|
||||
let components = fields[1].split(separator: ".")
|
||||
guard components.count == 3, let major = Int(components[0]), let minor = Int(components[1]),
|
||||
let patch = Int(components[2]), major >= 0, minor >= 0, patch >= 0
|
||||
else { throw RemoteSessionFailure.unsupported }
|
||||
let supported: Bool
|
||||
// Keep vvterm's supported baselines, not guessed theoretical minimums.
|
||||
// zmx 0.7 adds ownership labels; Zellij 0.44.3 and Herdr 0.8.2 include lifecycle fixes.
|
||||
switch self {
|
||||
case .tmux: supported = false
|
||||
case .zellij: supported = major > 0 || minor > 44 || (minor == 44 && patch >= 3)
|
||||
case .zmx: supported = major > 0 || minor >= 7
|
||||
case .herdr:
|
||||
supported = major > 0 || minor > 8 || (minor == 8 && patch >= 2) || (minor == 7 && patch >= 5)
|
||||
}
|
||||
guard supported else { throw RemoteSessionFailure.unsupported }
|
||||
}
|
||||
|
||||
var discovery: String {
|
||||
"""
|
||||
export LC_ALL=C
|
||||
exe=$(command -v \(rawValue) 2>/dev/null)
|
||||
case "$exe" in /*) ;; *) exe= ;; esac
|
||||
if [ -z "$exe" ]; then
|
||||
for dir in /opt/homebrew/bin /usr/local/bin /usr/bin /snap/bin "$HOME/.local/bin" "$HOME/.cargo/bin"; do
|
||||
if [ -f "$dir/\(rawValue)" ] && [ -x "$dir/\(rawValue)" ]; then exe="$dir/\(rawValue)"; break; fi
|
||||
done
|
||||
fi
|
||||
[ -n "$exe" ] && [ -x "$exe" ] || exit 77
|
||||
unset ZELLIJ ZELLIJ_PANE_ID ZELLIJ_SESSION_NAME ZMX_SESSION ZMX_SESSION_PREFIX
|
||||
unset HERDR_SESSION HERDR_SOCKET_PATH HERDR_CLIENT_SOCKET_PATH HERDR_PANE_ID HERDR_PANE_RUNTIME_ID
|
||||
unset HERDR_ACTIVE_PANE_ID HERDR_ACTIVE_TAB_ID HERDR_ACTIVE_WORKSPACE_ID HERDR_ACTIVE_PANE_CWD
|
||||
unset HERDR_WORKSPACE_ID HERDR_TAB_ID HERDR_STARTUP_CWD HERDR_BIN_PATH
|
||||
"""
|
||||
}
|
||||
}
|
||||
|
||||
nonisolated struct RemoteMultiplexerIdentity: Codable, Equatable, Sendable {
|
||||
var version = 1
|
||||
let kind: RemoteMultiplexerKind
|
||||
// Keep the legacy raw-string discriminator, independently of the public backend ID's Codable shape.
|
||||
private let kindValue: String
|
||||
var kind: SessionPersistenceBackendID { .init(rawValue: kindValue) }
|
||||
|
||||
private enum CodingKeys: String, CodingKey {
|
||||
case version, root, socket, fingerprint, process, created, owner, helper, isolated
|
||||
case directoryFingerprint, daemonProcess, daemonStarted
|
||||
case kindValue = "kind"
|
||||
}
|
||||
let root: String
|
||||
let socket: String
|
||||
let fingerprint: String
|
||||
@@ -65,12 +26,36 @@ nonisolated struct RemoteMultiplexerIdentity: Codable, Equatable, Sendable {
|
||||
var daemonProcess: String? = nil
|
||||
var daemonStarted: String? = nil
|
||||
|
||||
static func decode(_ data: Data, kind: RemoteMultiplexerKind) throws -> Self {
|
||||
init(
|
||||
kind: SessionPersistenceBackendID, root: String, socket: String, fingerprint: String,
|
||||
process: String?, created: String?, owner: UUID?, helper: String?, isolated: Bool,
|
||||
directoryFingerprint: String? = nil, daemonProcess: String? = nil, daemonStarted: String? = nil
|
||||
) {
|
||||
kindValue = kind.rawValue
|
||||
self.root = root
|
||||
self.socket = socket
|
||||
self.fingerprint = fingerprint
|
||||
self.process = process
|
||||
self.created = created
|
||||
self.owner = owner
|
||||
self.helper = helper
|
||||
self.isolated = isolated
|
||||
self.directoryFingerprint = directoryFingerprint
|
||||
self.daemonProcess = daemonProcess
|
||||
self.daemonStarted = daemonStarted
|
||||
}
|
||||
|
||||
static func decode(_ data: Data, kind: SessionPersistenceBackendID) throws -> Self {
|
||||
let value: Self
|
||||
do { value = try JSONDecoder().decode(Self.self, from: data) } catch {
|
||||
throw RemoteSessionFailure.invalidResponse
|
||||
}
|
||||
guard value.version == 1, value.kind == kind, kind != .tmux else {
|
||||
guard RemoteMultiplexerRegistry.dialect(value.kind) != nil else {
|
||||
throw RemoteSessionFailure.invalidResponse
|
||||
}
|
||||
guard value.version == 1, value.kind == kind,
|
||||
let dialect = RemoteMultiplexerRegistry.socketDialect(kind)
|
||||
else {
|
||||
throw RemoteSessionFailure.unsupported
|
||||
}
|
||||
guard RemoteMultiplexerCommands.validPath(value.root), RemoteMultiplexerCommands.validPath(value.socket),
|
||||
@@ -80,17 +65,12 @@ nonisolated struct RemoteMultiplexerIdentity: Codable, Equatable, Sendable {
|
||||
value.helper.map(RemoteMultiplexerCommands.validHelper) ?? true,
|
||||
value.process.map({ UInt64($0).map { $0 > 1 } ?? false }) ?? true,
|
||||
value.created.map({ UInt64($0) != nil }) ?? true,
|
||||
kind != .zmx || (value.process != nil && value.created != nil),
|
||||
kind != .zmx
|
||||
|| (value.daemonProcess.flatMap(UInt64.init).map { $0 > 1 } == true
|
||||
&& value.daemonStarted?.isEmpty == false
|
||||
&& value.daemonStarted!.utf8.allSatisfy { (32...126).contains($0) }),
|
||||
value.directoryFingerprint.map({
|
||||
$0.split(separator: ":").count == 2
|
||||
&& $0.utf8.allSatisfy { (48...57).contains($0) || $0 == 58 }
|
||||
}) ?? true,
|
||||
kind == .zellij || !value.isolated
|
||||
}) ?? true
|
||||
else { throw RemoteSessionFailure.invalidResponse }
|
||||
try dialect.validateIdentity(value)
|
||||
return value
|
||||
}
|
||||
}
|
||||
@@ -106,6 +86,24 @@ nonisolated struct RemoteMultiplexerCandidate {
|
||||
}
|
||||
|
||||
nonisolated enum RemoteMultiplexerCommands {
|
||||
static func discovery(_ backend: SessionPersistenceBackendID) -> String {
|
||||
"""
|
||||
export LC_ALL=C
|
||||
exe=$(command -v \(backend.rawValue) 2>/dev/null)
|
||||
case "$exe" in /*) ;; *) exe= ;; esac
|
||||
if [ -z "$exe" ]; then
|
||||
for dir in /opt/homebrew/bin /usr/local/bin /usr/bin /snap/bin "$HOME/.local/bin" "$HOME/.cargo/bin"; do
|
||||
if [ -f "$dir/\(backend.rawValue)" ] && [ -x "$dir/\(backend.rawValue)" ]; then exe="$dir/\(backend.rawValue)"; break; fi
|
||||
done
|
||||
fi
|
||||
[ -n "$exe" ] && [ -x "$exe" ] || exit 77
|
||||
unset ZELLIJ ZELLIJ_PANE_ID ZELLIJ_SESSION_NAME ZMX_SESSION ZMX_SESSION_PREFIX
|
||||
unset HERDR_SESSION HERDR_SOCKET_PATH HERDR_CLIENT_SOCKET_PATH HERDR_PANE_ID HERDR_PANE_RUNTIME_ID
|
||||
unset HERDR_ACTIVE_PANE_ID HERDR_ACTIVE_TAB_ID HERDR_ACTIVE_WORKSPACE_ID HERDR_ACTIVE_PANE_CWD
|
||||
unset HERDR_WORKSPACE_ID HERDR_TAB_ID HERDR_STARTUP_CWD HERDR_BIN_PATH
|
||||
"""
|
||||
}
|
||||
|
||||
static func quote(_ value: String) throws -> String {
|
||||
guard !value.contains("\0") else { throw RemoteSessionFailure.invalidResponse }
|
||||
return "'" + value.replacingOccurrences(of: "'", with: "'\\''") + "'"
|
||||
@@ -187,7 +185,7 @@ nonisolated enum RemoteMultiplexerCommands {
|
||||
}
|
||||
"""#
|
||||
|
||||
static func stateRoot(_ kind: RemoteMultiplexerKind, create: Bool) -> String {
|
||||
static func stateRoot(_ kind: SessionPersistenceBackendID, create: Bool) -> String {
|
||||
"""
|
||||
state="/tmp/termind-sessions-$(id -u)"
|
||||
\(create ? "private_dir \"$state\" || exit 78" : "")
|
||||
@@ -208,33 +206,15 @@ nonisolated enum RemoteMultiplexerCommands {
|
||||
}
|
||||
|
||||
static func routing(_ identity: RemoteMultiplexerIdentity) throws -> String {
|
||||
switch identity.kind {
|
||||
case .tmux: throw RemoteSessionFailure.unsupported
|
||||
case .zellij:
|
||||
return """
|
||||
export ZELLIJ_SOCKET_DIR=\(try quote(identity.root))
|
||||
\(identity.isolated ? "export XDG_CACHE_HOME=\(try quote(identity.root + "/cache"))" : "")
|
||||
"""
|
||||
case .zmx: return "export ZMX_DIR=\(try quote(identity.root))"
|
||||
case .herdr:
|
||||
let root = identity.root
|
||||
let config: String
|
||||
if root.hasSuffix("/herdr") {
|
||||
config = String(root.dropLast("/herdr".count))
|
||||
} else if let range = root.range(of: "/herdr/sessions/", options: .backwards),
|
||||
validName(String(root[range.upperBound...]))
|
||||
{
|
||||
config = String(root[..<range.lowerBound])
|
||||
} else {
|
||||
throw RemoteSessionFailure.invalidResponse
|
||||
}
|
||||
guard validPath(config) else { throw RemoteSessionFailure.invalidResponse }
|
||||
return "export XDG_CONFIG_HOME=\(try quote(config))"
|
||||
guard let dialect = RemoteMultiplexerRegistry.socketDialect(identity.kind) else {
|
||||
throw RemoteSessionFailure.unsupported
|
||||
}
|
||||
return try dialect.routing(identity)
|
||||
}
|
||||
|
||||
static func guardIdentity(_ reference: RemoteSessionReference) throws -> String {
|
||||
guard let kind = RemoteMultiplexerKind(rawValue: reference.backend.rawValue),
|
||||
let kind = reference.backend
|
||||
guard let dialect = RemoteMultiplexerRegistry.socketDialect(kind),
|
||||
reference.version == 1, validName(reference.session.name)
|
||||
else { throw RemoteSessionFailure.unsupported }
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
@@ -244,12 +224,7 @@ nonisolated enum RemoteMultiplexerCommands {
|
||||
}
|
||||
let owned =
|
||||
try reference.ownershipToken.map { token in
|
||||
let labelGuard =
|
||||
kind == .zmx
|
||||
? """
|
||||
[ "$("$exe" get \(try quote(reference.session.name)) termind_owner)" = '\(token.uuidString)' ] || exit 79
|
||||
[ "$(cat "$meta/native" 2>/dev/null)" = '\(identity.process!):\(identity.created!)' ] || exit 79
|
||||
""" : ""
|
||||
let labelGuard = try dialect.ownershipGuard(identity, name: reference.session.name, token: token)
|
||||
return """
|
||||
\(stateRoot(kind, create: false))
|
||||
meta="$state/\(reference.session.name)"
|
||||
@@ -261,51 +236,32 @@ nonisolated enum RemoteMultiplexerCommands {
|
||||
} ?? ""
|
||||
return """
|
||||
\(utilities)
|
||||
\(kind.discovery)
|
||||
\(discovery(kind))
|
||||
\(try routing(identity))
|
||||
[ "$(fingerprint \(try quote(identity.socket)))" = \(try quote(identity.fingerprint)) ] || exit 79
|
||||
\(kind == .zmx ? """
|
||||
[ "$(\(RemoteZmxCommands.nativeIdentity(name: reference.session.name)))" = '\(identity.process!):\(identity.created!)' ] || exit 79
|
||||
[ "$(ps -p '\(identity.process!)' -o ppid= | tr -d ' ')" = '\(identity.daemonProcess!)' ] || exit 79
|
||||
[ "$(process_start '\(identity.daemonProcess!)')" = \(try quote(identity.daemonStarted!)) ] || exit 79
|
||||
""" : "")
|
||||
\(try dialect.identityGuard(identity, name: reference.session.name))
|
||||
\(owned)
|
||||
"""
|
||||
}
|
||||
|
||||
static func validateRoute(_ identity: RemoteMultiplexerIdentity, name: String) throws {
|
||||
let valid: Bool
|
||||
switch identity.kind {
|
||||
case .tmux: valid = false
|
||||
case .zmx: valid = identity.socket == identity.root + "/" + name
|
||||
case .zellij: valid = identity.socket.hasSuffix("/" + name)
|
||||
case .herdr:
|
||||
valid =
|
||||
identity.socket == identity.root + "/herdr.sock"
|
||||
&& (name == "default"
|
||||
? identity.root.hasSuffix("/herdr") : identity.root.hasSuffix("/herdr/sessions/" + name))
|
||||
}
|
||||
guard valid else { throw RemoteSessionFailure.invalidResponse }
|
||||
guard let dialect = RemoteMultiplexerRegistry.socketDialect(identity.kind),
|
||||
dialect.validRoute(identity, name: name)
|
||||
else { throw RemoteSessionFailure.invalidResponse }
|
||||
_ = try routing(identity)
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared ownership and framing policy; native command dialects remain backend-owned.
|
||||
nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
let kind: RemoteMultiplexerKind
|
||||
nonisolated struct RemoteMultiplexerBackend: RemoteSessionBackend {
|
||||
let dialect: any RemoteSocketMultiplexerDialect
|
||||
var kind: SessionPersistenceBackendID { dialect.id }
|
||||
var id: SessionPersistenceBackendID { dialect.id }
|
||||
let execute: RemoteSessionCommandExecutor
|
||||
|
||||
var capabilities: RemoteSessionCapabilities {
|
||||
.init(windowCount: kind == .tmux, clientCount: kind == .tmux || kind == .zmx, shellIntegration: true)
|
||||
}
|
||||
var capabilities: RemoteSessionCapabilities { dialect.capabilities }
|
||||
|
||||
func probe() async throws -> RemoteSessionCapabilities {
|
||||
let versionFlag = kind == .tmux ? "-V" : kind == .zmx ? "version" : "--version"
|
||||
let result = try await execute(kind.discovery + "\n\"$exe\" \(versionFlag)")
|
||||
try Self.check(result)
|
||||
try kind.validateVersion(result.stdout)
|
||||
return capabilities
|
||||
}
|
||||
func probe() async throws -> RemoteSessionCapabilities { try await dialect.probe(execute: execute) }
|
||||
|
||||
static func check(_ result: RemoteSessionCommandResult) throws {
|
||||
guard !result.truncated, result.stdout.utf8.count <= 1024 * 1024 else {
|
||||
@@ -322,12 +278,7 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
}
|
||||
|
||||
func validate(_ reference: RemoteSessionReference) throws {
|
||||
guard reference.version == 1, reference.backend == kind.id else { throw RemoteSessionFailure.unsupported }
|
||||
if kind == .tmux {
|
||||
_ = try RemoteTmuxCommands.attach(reference)
|
||||
return
|
||||
}
|
||||
guard reference.version == 1, reference.backend == kind.id,
|
||||
guard reference.version == 1, reference.backend == kind,
|
||||
RemoteMultiplexerCommands.validName(reference.session.name)
|
||||
else { throw RemoteSessionFailure.unsupported }
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
@@ -340,14 +291,6 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
}
|
||||
|
||||
func matches(_ reference: RemoteSessionReference, session: RemoteSessionDescriptor) -> Bool {
|
||||
if kind == .tmux {
|
||||
guard (try? validate(reference)) != nil, reference.session.id == session.id,
|
||||
let saved = try? TmuxSessionIdentity.decode(reference.session.identity),
|
||||
let current = try? TmuxSessionIdentity.decode(session.identity)
|
||||
else { return false }
|
||||
return saved.created == current.created && saved.serverPID == current.serverPID
|
||||
&& (reference.ownershipToken.map { current.owner == $0.uuidString } ?? true)
|
||||
}
|
||||
guard (try? validate(reference)) != nil, reference.session.id == session.id,
|
||||
let saved = try? RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind),
|
||||
let current = try? RemoteMultiplexerIdentity.decode(session.identity, kind: kind)
|
||||
@@ -361,22 +304,9 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
}
|
||||
|
||||
func sessions() async throws -> [RemoteSessionDescriptor] {
|
||||
let command: String
|
||||
switch kind {
|
||||
case .tmux: command = RemoteTmuxCommands.list
|
||||
case .zellij: command = RemoteZellijCommands.list
|
||||
case .zmx: command = RemoteZmxCommands.list
|
||||
case .herdr: command = RemoteHerdrCommands.list
|
||||
}
|
||||
let response = try await execute(command)
|
||||
let response = try await execute(dialect.list)
|
||||
try Self.check(response)
|
||||
let candidates: [RemoteMultiplexerCandidate]
|
||||
switch kind {
|
||||
case .tmux: return try RemoteTmuxCommands.parseSessions(response.stdout)
|
||||
case .zellij: candidates = try RemoteZellijCommands.parseList(response.stdout)
|
||||
case .zmx: candidates = try RemoteZmxCommands.parseList(response.stdout)
|
||||
case .herdr: candidates = try RemoteHerdrCommands.parseList(response.stdout)
|
||||
}
|
||||
let candidates = try dialect.parseList(response.stdout)
|
||||
guard candidates.count <= 256, Set(candidates.map { "\($0.isolated):\($0.name)" }).count == candidates.count
|
||||
else { throw RemoteSessionFailure.invalidResponse }
|
||||
guard !candidates.isEmpty else { return [] }
|
||||
@@ -404,29 +334,14 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
if private_state && [ "$(cat "$meta/identity" 2>/dev/null)" = "$fp" ]; then
|
||||
owner=$(cat "$meta/owner") || exit 78
|
||||
helper=$(cat "$meta/helper") || exit 78
|
||||
\(kind == .zmx ? """
|
||||
export ZMX_DIR=\(try RemoteMultiplexerCommands.quote(candidate.root))
|
||||
if [ "$("$exe" get '\(candidate.name)' termind_owner)" != "$owner" ] \
|
||||
|| [ "$(cat "$meta/native" 2>/dev/null)" != '\(candidate.process!):\(candidate.created!)' ]; then
|
||||
owner=-; helper=-
|
||||
fi
|
||||
""" : "")
|
||||
\(try dialect.metadataOwnershipGuard(candidate))
|
||||
fi
|
||||
\(kind == .herdr ? "directory=$(directory_fingerprint \(try RemoteMultiplexerCommands.quote(candidate.root))) || exit 79" : "")
|
||||
\(kind == .zmx ? """
|
||||
# zmx's list PID is the PTY child, not the daemon. Bind its parent separately.
|
||||
daemon=$(ps -p '\(candidate.process!)' -o ppid= | tr -d ' ') || exit 79
|
||||
comm=$(ps -p "$daemon" -o comm=)
|
||||
[ "${comm##*/}" = zmx ] || exit 79
|
||||
born=$(process_start "$daemon")
|
||||
[ -n "$born" ] || exit 79
|
||||
born=$(hex "$born")
|
||||
""" : "")
|
||||
\(try dialect.metadataFields(candidate))
|
||||
printf 'F|\(index)|%s|%s|%s|%s|%s|%s|%s\\n' "$(hex "$socket")" "$fp" "$owner" "$helper" "$directory" "$daemon" "$born"
|
||||
"""
|
||||
}.joined(separator: "\n")
|
||||
let metadata = try await execute(
|
||||
kind.discovery + "\n" + RemoteMultiplexerCommands.utilities + "\n"
|
||||
RemoteMultiplexerCommands.discovery(kind) + "\n" + RemoteMultiplexerCommands.utilities + "\n"
|
||||
+ RemoteMultiplexerCommands.stateRoot(kind, create: false)
|
||||
+ "\n" + commands)
|
||||
try Self.check(metadata)
|
||||
@@ -460,18 +375,6 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
func create(
|
||||
token: UUID, startupCommand: String?, shellIntegration: Bool, environment: [HostEnvironmentVariable]
|
||||
) async throws -> RemoteSessionDescriptor {
|
||||
if kind == .tmux {
|
||||
let result = try await execute(
|
||||
RemoteTmuxCommands.create(
|
||||
token: token, startupCommand: startupCommand, shellIntegration: shellIntegration,
|
||||
environment: environment))
|
||||
try Self.check(result)
|
||||
let confirmed = try RemoteTmuxCommands.parseSessions(result.stdout)
|
||||
guard confirmed.count == 1, let session = confirmed.first,
|
||||
try TmuxSessionIdentity.decode(session.identity).owner == token.uuidString
|
||||
else { throw RemoteSessionFailure.ownershipMismatch }
|
||||
return session
|
||||
}
|
||||
let name = RemoteMultiplexerCommands.managedName(token)
|
||||
if let existing = try await sessions().first(where: { $0.name == name }) {
|
||||
let identity = try RemoteMultiplexerIdentity.decode(existing.identity, kind: kind)
|
||||
@@ -479,7 +382,7 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
// An interrupted response may leave a confirmed incarnation without its ready marker.
|
||||
// Only finish that same incarnation; never replay a startup command or adopt by name.
|
||||
let reference = RemoteSessionReference(
|
||||
backend: kind.id, session: existing, hostID: UUID(), endpoint: "", username: "", ownershipToken: token)
|
||||
backend: kind, session: existing, hostID: UUID(), endpoint: "", username: "", ownershipToken: token)
|
||||
let completion = try await execute(
|
||||
"""
|
||||
\(try RemoteMultiplexerCommands.guardIdentity(reference))
|
||||
@@ -495,13 +398,7 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
guard completion.stdout == "TERMIND_CREATED\n" else { throw RemoteSessionFailure.invalidResponse }
|
||||
return existing
|
||||
}
|
||||
let native: String
|
||||
switch kind {
|
||||
case .tmux: throw RemoteSessionFailure.unsupported
|
||||
case .zellij: native = RemoteZellijCommands.create(name: name)
|
||||
case .zmx: native = RemoteZmxCommands.create(name: name, token: token)
|
||||
case .herdr: native = RemoteHerdrCommands.create(name: name)
|
||||
}
|
||||
let native = dialect.create(name: name, token: token)
|
||||
let requireStartup = startupCommand?.isEmpty == false
|
||||
let shell =
|
||||
try environment.last(where: { $0.variable == "SHELL" }).map {
|
||||
@@ -510,7 +407,7 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
let result = try await execute(
|
||||
"""
|
||||
export TERMIND_SHELL_LOCALE_SET="${LC_ALL+x}" TERMIND_SHELL_LOCALE="${LC_ALL-}"
|
||||
\(kind.discovery)
|
||||
\(RemoteMultiplexerCommands.discovery(kind))
|
||||
\(RemoteMultiplexerCommands.utilities)
|
||||
\(RemoteMultiplexerCommands.stateRoot(kind, create: true))
|
||||
umask 077
|
||||
@@ -570,30 +467,10 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
|
||||
func launch(_ reference: RemoteSessionReference, created: Bool) throws -> RemoteSessionLaunch {
|
||||
try validate(reference)
|
||||
if kind == .tmux {
|
||||
return .init(
|
||||
command: try RemoteTmuxCommands.attach(reference), attachment: reference, created: created)
|
||||
}
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
let name = try RemoteMultiplexerCommands.quote(reference.session.name)
|
||||
let attach: String
|
||||
switch kind {
|
||||
case .tmux: throw RemoteSessionFailure.unsupported
|
||||
case .zellij: attach = "\"$exe\" attach \(name)"
|
||||
case .zmx:
|
||||
// attach is an upsert. If the daemon disappears after validation, its replacement must exit,
|
||||
// not become an unrelated interactive shell (or replay the original startup command).
|
||||
attach = "\"$exe\" attach \(name) /bin/sh -c 'exit 79'"
|
||||
case .herdr: attach = "\"$exe\" --session \(name) client"
|
||||
}
|
||||
let noResurrection =
|
||||
kind == .zellij
|
||||
? """
|
||||
# An attach command can otherwise resurrect a cached session after a disconnect race.
|
||||
cache="/tmp/termind-zellij-attach-cache-$(id -u)"
|
||||
private_dir "$cache" || exit 79
|
||||
export XDG_CACHE_HOME="$cache"
|
||||
""" : ""
|
||||
let attach = dialect.attach(name: name)
|
||||
let noResurrection = dialect.prepareAttachment()
|
||||
let bind: String
|
||||
if let helper = identity.helper {
|
||||
bind = """
|
||||
@@ -613,27 +490,11 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
|
||||
func terminate(_ reference: RemoteSessionReference) async throws {
|
||||
try validate(reference)
|
||||
guard reference.isManaged, kind != .herdr || reference.session.name != "default" else {
|
||||
throw RemoteSessionFailure.ownershipMismatch
|
||||
}
|
||||
if kind == .tmux {
|
||||
let result = try await execute(RemoteTmuxCommands.terminate(reference))
|
||||
try Self.check(result)
|
||||
guard result.stdout == "TERMIND_ENDED\n" else { throw RemoteSessionFailure.ownershipMismatch }
|
||||
return
|
||||
}
|
||||
guard reference.isManaged else { throw RemoteSessionFailure.ownershipMismatch }
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: kind)
|
||||
if kind == .herdr, identity.directoryFingerprint == nil {
|
||||
throw RemoteSessionFailure.ownershipMismatch
|
||||
}
|
||||
try dialect.validateTermination(reference, identity: identity)
|
||||
let name = try RemoteMultiplexerCommands.quote(reference.session.name)
|
||||
let stop: String
|
||||
switch kind {
|
||||
case .tmux: throw RemoteSessionFailure.unsupported
|
||||
case .zellij: stop = "\"$exe\" kill-session \(name)"
|
||||
case .zmx: stop = "\"$exe\" kill \(name) --force"
|
||||
case .herdr: stop = "\"$exe\" session stop \(name) --json"
|
||||
}
|
||||
let stop = dialect.stop(name: name)
|
||||
let result = try await execute(
|
||||
"""
|
||||
\(try RemoteMultiplexerCommands.guardIdentity(reference))
|
||||
@@ -643,15 +504,7 @@ nonisolated struct RemoteMultiplexerBackend: Sendable {
|
||||
attempt=$((attempt + 1)); [ "$attempt" -lt 50 ] || exit 79
|
||||
sleep 0.1
|
||||
done
|
||||
\(kind == .herdr ? """
|
||||
private_state || exit 79
|
||||
[ "$(cat "$meta/owner")" = '\(reference.ownershipToken!.uuidString)' ] || exit 79
|
||||
[ "$(cat "$meta/identity")" = \(try RemoteMultiplexerCommands.quote(identity.fingerprint)) ] || exit 79
|
||||
[ "$(directory_fingerprint \(try RemoteMultiplexerCommands.quote(identity.root)))" = '\(identity.directoryFingerprint!)' ] || exit 79
|
||||
[ ! -e \(try RemoteMultiplexerCommands.quote(identity.socket)) ] || exit 79
|
||||
"$exe" session delete \(name) --json >/dev/null || exit 78
|
||||
[ ! -e \(try RemoteMultiplexerCommands.quote(identity.root)) ] || exit 79
|
||||
""" : "")
|
||||
\(try dialect.terminationCleanup(reference, identity: identity))
|
||||
printf 'TERMIND_ENDED\\n'
|
||||
""")
|
||||
try Self.check(result)
|
||||
|
||||
@@ -5,9 +5,10 @@ extension RemoteSessionBackendRegistry {
|
||||
public nonisolated static var live: Self {
|
||||
Self(
|
||||
Dictionary(
|
||||
uniqueKeysWithValues: SessionPersistenceBackendID.supported.map {
|
||||
id -> (SessionPersistenceBackendID, Factory) in
|
||||
(id, { execute in RemoteMultiplexerService(backend: id, execute: execute) })
|
||||
uniqueKeysWithValues: RemoteMultiplexerRegistry.dialects.map {
|
||||
dialect -> (SessionPersistenceBackendID, Factory) in
|
||||
let id = dialect.id
|
||||
return (id, { execute in RemoteMultiplexerService(backend: id, execute: execute) })
|
||||
}))
|
||||
}
|
||||
}
|
||||
@@ -77,13 +77,7 @@ nonisolated enum RemoteShellContextJournal {
|
||||
"""#
|
||||
|
||||
static func initialization(shell: String, token: String, backend: String? = nil) -> String {
|
||||
let paneVariable: String?
|
||||
switch backend {
|
||||
case "zellij": paneVariable = "ZELLIJ_PANE_ID"
|
||||
case "herdr": paneVariable = "HERDR_PANE_ID"
|
||||
case "zmx": paneVariable = nil
|
||||
default: paneVariable = "TMUX_PANE"
|
||||
}
|
||||
let paneVariable = RemoteMultiplexerRegistry.dialect(.init(rawValue: backend ?? "tmux"))?.paneVariable
|
||||
let tag = backend == nil ? "M1" : "M2"
|
||||
let suffix = backend.map { "|\($0)" } ?? ""
|
||||
if shell == "fish" {
|
||||
@@ -156,8 +150,7 @@ nonisolated struct RemoteShellContextRecord {
|
||||
let modern = meta.count == 9 && meta.first == "M2"
|
||||
guard legacy || modern else { throw TerminalRuntimeError.invalidResponse }
|
||||
let backend = modern ? meta[8] : (meta[2] == "shell" ? "shell" : "tmux")
|
||||
guard ["shell", "tmux", "zellij", "zmx", "herdr"].contains(backend),
|
||||
UUID(uuidString: meta[1]) != nil, UInt64(meta[6]) != nil,
|
||||
guard UUID(uuidString: meta[1]) != nil, UInt64(meta[6]) != nil,
|
||||
Self.validPane(meta[2], backend: backend),
|
||||
meta[3].hasPrefix("ctx."),
|
||||
meta[3].utf8.allSatisfy({
|
||||
@@ -237,18 +230,8 @@ nonisolated struct RemoteShellContextRecord {
|
||||
}
|
||||
|
||||
static func validPane(_ value: String, backend: String) -> Bool {
|
||||
switch backend {
|
||||
case "shell", "zmx": return value == "shell"
|
||||
case "tmux": return validPane(value)
|
||||
case "zellij": return !value.isEmpty && value.utf8.allSatisfy { (48...57).contains($0) }
|
||||
case "herdr":
|
||||
return !value.isEmpty && value.utf8.count <= 128
|
||||
&& value.utf8.allSatisfy {
|
||||
(48...57).contains($0) || (65...90).contains($0) || (97...122).contains($0)
|
||||
|| [45, 58, 95].contains($0)
|
||||
}
|
||||
default: return false
|
||||
}
|
||||
if backend == "shell" { return value == "shell" }
|
||||
return RemoteMultiplexerRegistry.dialect(.init(rawValue: backend))?.validPane(value) ?? false
|
||||
}
|
||||
|
||||
func context(server: String, session: String, backend: String) -> TerminalShellContext {
|
||||
|
||||
@@ -62,14 +62,7 @@ nonisolated enum RemoteShellStartup {
|
||||
fi
|
||||
unset TERMIND_SHELL_LOCALE_SET TERMIND_SHELL_LOCALE
|
||||
""")
|
||||
\(backend == "zellij" ? """
|
||||
if [ "${TERMIND_SHELL_CACHE_HOME_SET:-}" = x ]; then
|
||||
export XDG_CACHE_HOME="$TERMIND_SHELL_CACHE_HOME"
|
||||
else
|
||||
unset XDG_CACHE_HOME
|
||||
fi
|
||||
unset TERMIND_SHELL_CACHE_HOME_SET TERMIND_SHELL_CACHE_HOME
|
||||
""" : "")
|
||||
\(backend.flatMap { RemoteMultiplexerRegistry.socketDialect(.init(rawValue: $0)) }?.restoreShellEnvironment ?? "")
|
||||
\(try RemoteMultiplexerCommands.environment(environment))
|
||||
SHELL=${TERMIND_TMUX_SHELL:-${SHELL:-/bin/sh}}
|
||||
export SHELL
|
||||
|
||||
@@ -19,7 +19,7 @@ nonisolated enum RemoteTerminalDraft {
|
||||
let root: String
|
||||
let trigger: String
|
||||
if let target {
|
||||
guard RemoteMultiplexerKind(rawValue: identity.backend) != nil else {
|
||||
guard RemoteMultiplexerRegistry.dialect(.init(rawValue: identity.backend)) != nil else {
|
||||
throw TerminalRuntimeError.staleContext
|
||||
}
|
||||
root = target.root
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
nonisolated struct RemoteTmuxDialect: RemoteMultiplexerDialect {
|
||||
let id = SessionPersistenceBackendID.tmux
|
||||
let capabilities = RemoteSessionCapabilities(windowCount: true, clientCount: true, shellIntegration: true)
|
||||
let versionArgument = "-V"
|
||||
let paneVariable: String? = "TMUX_PANE"
|
||||
|
||||
func validateVersion(_ output: String) throws { try RemoteTmuxCommands.validateVersion(output) }
|
||||
func validPane(_ value: String) -> Bool { RemoteShellContextRecord.validPane(value) }
|
||||
func backend(execute: @escaping RemoteSessionCommandExecutor) -> any RemoteSessionBackend {
|
||||
RemoteTmuxBackend(execute: execute)
|
||||
}
|
||||
func runtime(
|
||||
reference: RemoteSessionReference?, execute: @escaping RemoteSessionCommandExecutor, charset: RemoteCharset
|
||||
) throws -> any RemoteMultiplexerRuntime {
|
||||
RemoteTmuxRuntimeDialect(execute: execute, charset: charset)
|
||||
}
|
||||
func helper(_ reference: RemoteSessionReference) throws -> String? { nil }
|
||||
}
|
||||
|
||||
private nonisolated struct RemoteTmuxBackend: RemoteSessionBackend {
|
||||
let execute: RemoteSessionCommandExecutor
|
||||
private let dialect = RemoteTmuxDialect()
|
||||
var id: SessionPersistenceBackendID { dialect.id }
|
||||
var capabilities: RemoteSessionCapabilities { dialect.capabilities }
|
||||
|
||||
func probe() async throws -> RemoteSessionCapabilities { try await dialect.probe(execute: execute) }
|
||||
func sessions() async throws -> [RemoteSessionDescriptor] {
|
||||
let response = try await execute(RemoteTmuxCommands.list)
|
||||
try RemoteMultiplexerBackend.check(response)
|
||||
return try RemoteTmuxCommands.parseSessions(response.stdout)
|
||||
}
|
||||
func validate(_ reference: RemoteSessionReference) throws {
|
||||
guard reference.version == 1, reference.backend == id else { throw RemoteSessionFailure.unsupported }
|
||||
_ = try RemoteTmuxCommands.attach(reference)
|
||||
}
|
||||
func matches(_ reference: RemoteSessionReference, session: RemoteSessionDescriptor) -> Bool {
|
||||
guard (try? validate(reference)) != nil, reference.session.id == session.id,
|
||||
let saved = try? TmuxSessionIdentity.decode(reference.session.identity),
|
||||
let current = try? TmuxSessionIdentity.decode(session.identity)
|
||||
else { return false }
|
||||
return saved.created == current.created && saved.serverPID == current.serverPID
|
||||
&& (reference.ownershipToken.map { current.owner == $0.uuidString } ?? true)
|
||||
}
|
||||
func create(
|
||||
token: UUID, startupCommand: String?, shellIntegration: Bool, environment: [HostEnvironmentVariable]
|
||||
) async throws -> RemoteSessionDescriptor {
|
||||
let result = try await execute(
|
||||
RemoteTmuxCommands.create(
|
||||
token: token, startupCommand: startupCommand, shellIntegration: shellIntegration,
|
||||
environment: environment))
|
||||
try RemoteMultiplexerBackend.check(result)
|
||||
let confirmed = try RemoteTmuxCommands.parseSessions(result.stdout)
|
||||
guard confirmed.count == 1, let session = confirmed.first,
|
||||
try TmuxSessionIdentity.decode(session.identity).owner == token.uuidString
|
||||
else { throw RemoteSessionFailure.ownershipMismatch }
|
||||
return session
|
||||
}
|
||||
func launch(_ reference: RemoteSessionReference, created: Bool) throws -> RemoteSessionLaunch {
|
||||
try validate(reference)
|
||||
return .init(command: try RemoteTmuxCommands.attach(reference), attachment: reference, created: created)
|
||||
}
|
||||
func terminate(_ reference: RemoteSessionReference) async throws {
|
||||
try validate(reference)
|
||||
guard reference.isManaged else { throw RemoteSessionFailure.ownershipMismatch }
|
||||
let result = try await execute(RemoteTmuxCommands.terminate(reference))
|
||||
try RemoteMultiplexerBackend.check(result)
|
||||
guard result.stdout == "TERMIND_ENDED\n" else { throw RemoteSessionFailure.ownershipMismatch }
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@ import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
nonisolated struct RemoteTmuxRuntimeDialect: TerminalRuntimeAdapter {
|
||||
nonisolated struct RemoteTmuxRuntimeDialect: RemoteMultiplexerRuntime {
|
||||
private let execute: RemoteSessionCommandExecutor
|
||||
private let charset: RemoteCharset
|
||||
init(execute: @escaping RemoteSessionCommandExecutor, charset: RemoteCharset = .utf8) {
|
||||
@@ -16,6 +16,17 @@ nonisolated struct RemoteTmuxRuntimeDialect: TerminalRuntimeAdapter {
|
||||
return try Self.parse(result.stdout, terminal: terminal, charset: charset)
|
||||
}
|
||||
|
||||
func draftTarget(
|
||||
context: TerminalShellContext, terminal: String, replacing: Bool
|
||||
) async throws -> RemoteMultiplexerDraftTarget {
|
||||
let update = try await inspect(terminal: terminal)
|
||||
guard update.active == context.identity,
|
||||
update.contexts.first(where: { $0.identity == context.identity }) == context
|
||||
else { throw TerminalRuntimeError.staleContext }
|
||||
return try Self.draftTarget(
|
||||
context: context, terminal: terminal, clientIdentity: update.clientIdentity, replacing: replacing)
|
||||
}
|
||||
|
||||
func refreshSize(of client: RemoteTmuxClientIdentity) async throws {
|
||||
let result = try await execute(Self.resizeCommand(client: client))
|
||||
guard result.exitCode == 0, !result.truncated, result.stdout == "TERMIND_RESIZED\n" else {
|
||||
|
||||
@@ -11,7 +11,7 @@ nonisolated enum RemoteZellijCommands {
|
||||
|
||||
static var list: String {
|
||||
"""
|
||||
\(RemoteMultiplexerKind.zellij.discovery)
|
||||
\(RemoteMultiplexerCommands.discovery(.zellij))
|
||||
\(RemoteMultiplexerCommands.utilities)
|
||||
\(userRoot)
|
||||
user_root=$root
|
||||
@@ -83,3 +83,69 @@ nonisolated enum RemoteZellijCommands {
|
||||
"""
|
||||
}
|
||||
}
|
||||
|
||||
nonisolated struct RemoteZellijDialect: RemoteSocketMultiplexerDialect {
|
||||
let id = SessionPersistenceBackendID.zellij
|
||||
let paneVariable: String? = "ZELLIJ_PANE_ID"
|
||||
var list: String { RemoteZellijCommands.list }
|
||||
|
||||
func validateVersion(_ output: String) throws {
|
||||
let version = try RemoteMultiplexerVersion(output, backend: id)
|
||||
guard version.major > 0 || version.minor > 44 || (version.minor == 44 && version.patch >= 3) else {
|
||||
throw RemoteSessionFailure.unsupported
|
||||
}
|
||||
}
|
||||
func validPane(_ value: String) -> Bool {
|
||||
!value.isEmpty && value.utf8.allSatisfy { (48...57).contains($0) }
|
||||
}
|
||||
func parseList(_ text: String) throws -> [RemoteMultiplexerCandidate] {
|
||||
try RemoteZellijCommands.parseList(text)
|
||||
}
|
||||
func create(name: String, token: UUID) -> String { RemoteZellijCommands.create(name: name) }
|
||||
func attach(name: String) -> String { "\"$exe\" attach \(name)" }
|
||||
func stop(name: String) -> String { "\"$exe\" kill-session \(name)" }
|
||||
func validateIdentity(_ identity: RemoteMultiplexerIdentity) throws {}
|
||||
func validRoute(_ identity: RemoteMultiplexerIdentity, name: String) -> Bool {
|
||||
identity.socket.hasSuffix("/" + name)
|
||||
}
|
||||
func routing(_ identity: RemoteMultiplexerIdentity) throws -> String {
|
||||
"""
|
||||
export ZELLIJ_SOCKET_DIR=\(try RemoteMultiplexerCommands.quote(identity.root))
|
||||
\(identity.isolated ? "export XDG_CACHE_HOME=\(try RemoteMultiplexerCommands.quote(identity.root + "/cache"))" : "")
|
||||
"""
|
||||
}
|
||||
func prepareAttachment() -> String {
|
||||
"""
|
||||
# An attach command can otherwise resurrect a cached session after a disconnect race.
|
||||
cache="/tmp/termind-zellij-attach-cache-$(id -u)"
|
||||
private_dir "$cache" || exit 79
|
||||
export XDG_CACHE_HOME="$cache"
|
||||
"""
|
||||
}
|
||||
var restoreShellEnvironment: String {
|
||||
"""
|
||||
if [ "${TERMIND_SHELL_CACHE_HOME_SET:-}" = x ]; then
|
||||
export XDG_CACHE_HOME="$TERMIND_SHELL_CACHE_HOME"
|
||||
else
|
||||
unset XDG_CACHE_HOME
|
||||
fi
|
||||
unset TERMIND_SHELL_CACHE_HOME_SET TERMIND_SHELL_CACHE_HOME
|
||||
"""
|
||||
}
|
||||
func runtimeCommands(name: String) -> (String, String) {
|
||||
("\"$exe\" list-sessions --short --no-formatting", "\"$exe\" --session \(name) action list-clients")
|
||||
}
|
||||
func activePane(native: String, focus: String, reference: RemoteSessionReference) throws -> String? {
|
||||
// A switched client cannot be bound using its original argv. Require a unique live session too.
|
||||
guard native == reference.session.name else { return nil }
|
||||
let clients = focus.split(separator: "\n")
|
||||
guard clients.first == "CLIENT_ID ZELLIJ_PANE_ID RUNNING_COMMAND", clients.count == 2 else { return nil }
|
||||
let fields = clients[1].split(whereSeparator: \.isWhitespace)
|
||||
guard fields.count >= 3, UInt64(fields[0]) != nil, fields[1].hasPrefix("terminal_") else { return nil }
|
||||
let pane = String(fields[1].dropFirst("terminal_".count))
|
||||
return validPane(pane) ? pane : nil
|
||||
}
|
||||
func writeDraft(name: String, pane: String, key: String) -> String {
|
||||
"\"$exe\" --session \(name) action write-chars --pane-id \(pane) \(key)"
|
||||
}
|
||||
}
|
||||
@@ -11,7 +11,7 @@ nonisolated enum RemoteZmxCommands {
|
||||
|
||||
static var list: String {
|
||||
"""
|
||||
\(RemoteMultiplexerKind.zmx.discovery)
|
||||
\(RemoteMultiplexerCommands.discovery(.zmx))
|
||||
\(RemoteMultiplexerCommands.utilities)
|
||||
\(root)
|
||||
printf 'ROOT|%s\\n' "$(hex "$root")"
|
||||
@@ -76,3 +76,82 @@ nonisolated enum RemoteZmxCommands {
|
||||
"""
|
||||
}
|
||||
}
|
||||
|
||||
nonisolated struct RemoteZmxDialect: RemoteSocketMultiplexerDialect {
|
||||
let id = SessionPersistenceBackendID.zmx
|
||||
let paneVariable: String? = nil
|
||||
let versionArgument = "version"
|
||||
let capabilities = RemoteSessionCapabilities(clientCount: true, shellIntegration: true)
|
||||
var list: String { RemoteZmxCommands.list }
|
||||
|
||||
func validateVersion(_ output: String) throws {
|
||||
let version = try RemoteMultiplexerVersion(output, backend: id)
|
||||
guard version.major > 0 || version.minor >= 7 else { throw RemoteSessionFailure.unsupported }
|
||||
}
|
||||
func validPane(_ value: String) -> Bool { value == "shell" }
|
||||
func parseList(_ text: String) throws -> [RemoteMultiplexerCandidate] { try RemoteZmxCommands.parseList(text) }
|
||||
func create(name: String, token: UUID) -> String { RemoteZmxCommands.create(name: name, token: token) }
|
||||
func attach(name: String) -> String {
|
||||
// attach is an upsert. A daemon lost after validation must not become an unrelated shell.
|
||||
"\"$exe\" attach \(name) /bin/sh -c 'exit 79'"
|
||||
}
|
||||
func stop(name: String) -> String { "\"$exe\" kill \(name) --force" }
|
||||
func validRoute(_ identity: RemoteMultiplexerIdentity, name: String) -> Bool {
|
||||
identity.socket == identity.root + "/" + name
|
||||
}
|
||||
func routing(_ identity: RemoteMultiplexerIdentity) throws -> String {
|
||||
"export ZMX_DIR=\(try RemoteMultiplexerCommands.quote(identity.root))"
|
||||
}
|
||||
func validateIdentity(_ identity: RemoteMultiplexerIdentity) throws {
|
||||
guard !identity.isolated, identity.process != nil, identity.created != nil,
|
||||
identity.daemonProcess.flatMap(UInt64.init).map({ $0 > 1 }) == true,
|
||||
identity.daemonStarted?.isEmpty == false,
|
||||
identity.daemonStarted!.utf8.allSatisfy({ (32...126).contains($0) })
|
||||
else { throw RemoteSessionFailure.invalidResponse }
|
||||
}
|
||||
func identityGuard(_ identity: RemoteMultiplexerIdentity, name: String) throws -> String {
|
||||
"""
|
||||
[ "$(\(RemoteZmxCommands.nativeIdentity(name: name)))" = '\(identity.process!):\(identity.created!)' ] || exit 79
|
||||
[ "$(ps -p '\(identity.process!)' -o ppid= | tr -d ' ')" = '\(identity.daemonProcess!)' ] || exit 79
|
||||
[ "$(process_start '\(identity.daemonProcess!)')" = \(try RemoteMultiplexerCommands.quote(identity.daemonStarted!)) ] || exit 79
|
||||
"""
|
||||
}
|
||||
func ownershipGuard(_ identity: RemoteMultiplexerIdentity, name: String, token: UUID) throws -> String {
|
||||
"""
|
||||
[ "$("$exe" get \(try RemoteMultiplexerCommands.quote(name)) termind_owner)" = '\(token.uuidString)' ] || exit 79
|
||||
[ "$(cat "$meta/native" 2>/dev/null)" = '\(identity.process!):\(identity.created!)' ] || exit 79
|
||||
"""
|
||||
}
|
||||
func metadataOwnershipGuard(_ candidate: RemoteMultiplexerCandidate) throws -> String {
|
||||
"""
|
||||
export ZMX_DIR=\(try RemoteMultiplexerCommands.quote(candidate.root))
|
||||
if [ "$("$exe" get '\(candidate.name)' termind_owner)" != "$owner" ] \
|
||||
|| [ "$(cat "$meta/native" 2>/dev/null)" != '\(candidate.process!):\(candidate.created!)' ]; then
|
||||
owner=-; helper=-
|
||||
fi
|
||||
"""
|
||||
}
|
||||
func metadataFields(_ candidate: RemoteMultiplexerCandidate) throws -> String {
|
||||
"""
|
||||
# zmx's list PID is the PTY child, not the daemon. Bind its parent separately.
|
||||
daemon=$(ps -p '\(candidate.process!)' -o ppid= | tr -d ' ') || exit 79
|
||||
comm=$(ps -p "$daemon" -o comm=)
|
||||
[ "${comm##*/}" = zmx ] || exit 79
|
||||
born=$(process_start "$daemon")
|
||||
[ -n "$born" ] || exit 79
|
||||
born=$(hex "$born")
|
||||
"""
|
||||
}
|
||||
func runtimeCommands(name: String) -> (String, String) { ("\"$exe\" list", "printf 'shell'") }
|
||||
func activePane(native: String, focus: String, reference: RemoteSessionReference) throws -> String? {
|
||||
let identity = try RemoteMultiplexerIdentity.decode(reference.session.identity, kind: id)
|
||||
let root = identity.root.utf8.map { String(format: "%02x", $0) }.joined()
|
||||
let sessions = try RemoteZmxCommands.parseList("ROOT|\(root)\n" + native + "\nEND\n")
|
||||
guard sessions.count == 1, let session = sessions.first,
|
||||
session.name == reference.session.name, session.process == identity.process,
|
||||
session.created == identity.created, session.clients == 1
|
||||
else { return nil }
|
||||
return "shell"
|
||||
}
|
||||
func writeDraft(name: String, pane: String, key: String) -> String { "\"$exe\" send \(name) \(key)" }
|
||||
}
|
||||
@@ -1,8 +1,4 @@
|
||||
import ETBootstrap
|
||||
import ETSession
|
||||
import Foundation
|
||||
import MoshBootstrap
|
||||
import MoshCore
|
||||
import Network
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
@@ -12,53 +8,34 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
public typealias StateHandler = @Sendable (ResumableTerminalState) -> Void
|
||||
|
||||
private let transport: HostTerminalProtocol
|
||||
private let registry: ResumableTerminalEngineRegistry
|
||||
private let output: SSHPTYChannel.OutputHandler
|
||||
private let stateChanged: StateHandler
|
||||
private let integrationChanged: SSHPTYChannel.ShellIntegrationHandler
|
||||
private var mosh: MoshClientSession?
|
||||
private var eternal: ETTerminalSession?
|
||||
private var bridge: TerminalTCPBridge?
|
||||
private var reader: Task<Void, Never>?
|
||||
private var observer: Task<Void, Never>?
|
||||
private var engine: (any ResumableTerminalEngine)?
|
||||
private var maintenance: Task<Void, Never>?
|
||||
private var pathMonitor: NWPathMonitor?
|
||||
private var beforeConnect: ConnectionWakeAction?
|
||||
private var closed = false
|
||||
private var frozen = false
|
||||
private var state: ResumableTerminalState = .connecting
|
||||
private var credentials: ETCredentials?
|
||||
private var integration: RemoteShellIntegrationBootstrap?
|
||||
private var commandFilter: RemoteCommandEventFilter?
|
||||
private var integratedShell: IntegratedShell?
|
||||
private var integratedShell: ResumableTerminalShellState.IntegratedShell?
|
||||
private var pendingCheckpoint: TerminalTransportCheckpoint?
|
||||
private var pendingIntegrationCommand: Data?
|
||||
private var sendsInFlight = 0
|
||||
private var sendWaiters: [CheckedContinuation<Void, Never>] = []
|
||||
private var viewport: (columns: UInt16, rows: UInt16) = (80, 24)
|
||||
|
||||
private struct EternalPayload: Codable {
|
||||
let clientID: String
|
||||
let passkey: Data
|
||||
let checkpoint: ETSessionCheckpoint
|
||||
let integration: RemoteShellIntegrationBootstrap?
|
||||
let commandFilter: RemoteCommandEventFilter?
|
||||
let integratedShell: IntegratedShell?
|
||||
let pendingIntegrationCommand: Data?
|
||||
}
|
||||
|
||||
private struct IntegratedShell: Codable {
|
||||
let shell: String
|
||||
let systemInfo: RemoteSystemInfo?
|
||||
let homeDirectory: String?
|
||||
}
|
||||
|
||||
public init(
|
||||
transport: HostTerminalProtocol, checkpoint: TerminalTransportCheckpoint? = nil,
|
||||
output: @escaping SSHPTYChannel.OutputHandler,
|
||||
stateChanged: @escaping StateHandler,
|
||||
shellIntegration: @escaping SSHPTYChannel.ShellIntegrationHandler
|
||||
shellIntegration: @escaping SSHPTYChannel.ShellIntegrationHandler,
|
||||
registry: ResumableTerminalEngineRegistry = .live
|
||||
) {
|
||||
self.transport = transport
|
||||
self.registry = registry
|
||||
self.pendingCheckpoint = checkpoint
|
||||
self.output = output
|
||||
self.stateChanged = stateChanged
|
||||
@@ -73,6 +50,7 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
startupPreparation: SSHPTYChannel.StartupPreparation? = nil
|
||||
) async throws {
|
||||
guard !closed, options.isValid, transport.isResumable else { throw CancellationError() }
|
||||
guard engine == nil else { throw TerminalTransportError.sessionFailed }
|
||||
viewport = (columns, rows)
|
||||
guard options.available(isTailscaleManaged: request.configuration.network != nil).contains(transport) else {
|
||||
throw TerminalTransportError.unsupportedRoute
|
||||
@@ -83,16 +61,19 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
{
|
||||
throw TerminalTransportError.invalidCheckpoint
|
||||
}
|
||||
beforeConnect = request.configuration.beforeConnect
|
||||
try Task.checkCancellation()
|
||||
await request.configuration.beforeConnect?()
|
||||
try Task.checkCancellation()
|
||||
publish(.connecting)
|
||||
// Reject an unregistered engine before wake, route, bootstrap or startup side effects.
|
||||
let engine = try registry.make(transport) { [weak self] event in
|
||||
await self?.receiveEngineEvent(event)
|
||||
}
|
||||
self.engine = engine
|
||||
sendsInFlight += 1
|
||||
defer { finishSend() }
|
||||
let bootstrap = SSHConnection()
|
||||
var rollbackMoshPID: Int32?
|
||||
do {
|
||||
try checkOpen()
|
||||
await request.configuration.beforeConnect?()
|
||||
try checkOpen()
|
||||
publish(.connecting)
|
||||
if pendingCheckpoint == nil {
|
||||
try await bootstrap.connect(using: request, progress: progress, connectionFailure: nil)
|
||||
try checkOpen()
|
||||
@@ -101,95 +82,31 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
restoring: pendingCheckpoint != nil, preparation: startupPreparation,
|
||||
execute: { command in try await TerminalTransportRoute.execute(command, on: bootstrap) })
|
||||
try checkOpen()
|
||||
if let checkpoint = pendingCheckpoint {
|
||||
try await restore(checkpoint, route: route)
|
||||
} else {
|
||||
if transport == .mosh {
|
||||
let range = options.moshUDPPortRange.replacingOccurrences(of: "-", with: ":")
|
||||
var command = "mosh-server new -s -c 256"
|
||||
if !range.isEmpty { command += " -p \(range)" }
|
||||
command += try startup.moshArguments
|
||||
let variables = environment.map {
|
||||
"'\($0.variable)=\($0.value.replacingOccurrences(of: "'", with: "'\\''"))'"
|
||||
}
|
||||
command = (["env"] + variables + ["LANG=C.UTF-8", "LC_ALL=C.UTF-8", command])
|
||||
.joined(separator: " ")
|
||||
let result = try await bootstrap.execute(
|
||||
command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 65_536, onOutput: nil)
|
||||
let info: MoshServerConnectInfo
|
||||
do {
|
||||
info = try MoshServerOutputParser.parse(result.stdout + result.stderr)
|
||||
} catch {
|
||||
throw TerminalTransportError.bootstrapFailed
|
||||
}
|
||||
rollbackMoshPID = info.serverPID
|
||||
mosh = MoshClientSession(
|
||||
endpoint: .init(
|
||||
host: request.configuration.host, port: info.port, keyBase64_22: info.key))
|
||||
} else {
|
||||
let bridge = try TerminalTCPBridge(open: route, initial: await route())
|
||||
self.bridge = bridge
|
||||
let executor = BootstrapExecutor(connection: bootstrap)
|
||||
do {
|
||||
credentials = try await ETBootstrap().run(using: executor)
|
||||
} catch {
|
||||
if let original = await executor.failure { throw original }
|
||||
throw TerminalTransportError.bootstrapFailed
|
||||
}
|
||||
try checkOpen()
|
||||
guard let credentials else { throw TerminalTransportError.bootstrapFailed }
|
||||
eternal = try ETTerminalSession(
|
||||
host: "127.0.0.1", port: bridge.port,
|
||||
clientID: credentials.clientID, passkey: credentials.passkey,
|
||||
environmentVariables: Dictionary(
|
||||
environment.map { ($0.variable, $0.value) },
|
||||
uniquingKeysWith: { _, last in last }))
|
||||
if shellIntegrationEnabled && startup.plan == nil {
|
||||
let integration = RemoteShellIntegrationBootstrap()
|
||||
self.integration = integration
|
||||
commandFilter = RemoteCommandEventFilter(token: integration.token)
|
||||
}
|
||||
let preparation = try await engine.prepare(
|
||||
.init(
|
||||
request: request, options: options, environment: environment,
|
||||
shellIntegrationEnabled: shellIntegrationEnabled, checkpoint: pendingCheckpoint?.payload,
|
||||
startup: startup, bootstrap: bootstrap, route: route))
|
||||
try checkOpen()
|
||||
integration = preparation.shell.integration
|
||||
commandFilter = preparation.shell.commandFilter
|
||||
integratedShell = preparation.shell.integratedShell
|
||||
pendingIntegrationCommand = preparation.shell.pendingIntegrationCommand
|
||||
if pendingCheckpoint != nil {
|
||||
integration?.resetDeadlinesAfterRestore()
|
||||
if let shell = integratedShell {
|
||||
integrationChanged(
|
||||
.available(shell: shell.shell, systemInfo: shell.systemInfo, homeDirectory: shell.homeDirectory)
|
||||
)
|
||||
}
|
||||
}
|
||||
try await engine.start(columns: columns, rows: rows)
|
||||
try checkOpen()
|
||||
if let mosh {
|
||||
try await mosh.start()
|
||||
try await mosh.enqueue(.resize(cols: Int32(columns), rows: Int32(rows)))
|
||||
// An open UDP socket is not proof that the server is reachable.
|
||||
let deadline = ContinuousClock.now.advanced(by: .seconds(20))
|
||||
while true {
|
||||
try checkOpen()
|
||||
let operations = await mosh.drainHostOps()
|
||||
if !operations.isEmpty {
|
||||
deliver(operations)
|
||||
break
|
||||
}
|
||||
let state = await mosh.state
|
||||
if case .failed = state { throw TerminalTransportError.sessionFailed }
|
||||
if state == .stopped {
|
||||
rollbackMoshPID = nil
|
||||
_ = await receiveMoshPoll(state: state, operations: mosh.drainHostOps())
|
||||
await bootstrap.disconnect()
|
||||
return
|
||||
}
|
||||
guard ContinuousClock.now < deadline else { throw TerminalTransportError.unreachable }
|
||||
try await Task.sleep(for: .milliseconds(20))
|
||||
}
|
||||
rollbackMoshPID = nil
|
||||
publish(.connected)
|
||||
startMoshReader(mosh)
|
||||
} else if let eternal {
|
||||
startEternalReaders(eternal)
|
||||
try await eternal.connect(initialDimensions: .init(rows: Int(rows), columns: Int(columns)))
|
||||
try checkOpen()
|
||||
if let input = try startup.eternalInput {
|
||||
try await eternal.send(input)
|
||||
}
|
||||
if pendingCheckpoint == nil, let integration { try await eternal.send(integration.startCommand) }
|
||||
if let command = pendingIntegrationCommand {
|
||||
pendingIntegrationCommand = nil
|
||||
try await eternal.send(command)
|
||||
}
|
||||
if let input = preparation.startupInput { try await engine.write(input) }
|
||||
if pendingCheckpoint == nil, let integration { try await engine.write(integration.startCommand) }
|
||||
if let command = pendingIntegrationCommand {
|
||||
pendingIntegrationCommand = nil
|
||||
try await engine.write(command)
|
||||
}
|
||||
await bootstrap.disconnect()
|
||||
pendingCheckpoint = nil
|
||||
@@ -201,52 +118,20 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
await bootstrap.disconnect()
|
||||
return
|
||||
}
|
||||
if let pid = rollbackMoshPID, pid > 1 {
|
||||
// Only undo the server this attempt just launched; never kill a user's other sessions.
|
||||
await Task.detached {
|
||||
_ = try? await bootstrap.execute(
|
||||
"[ \"$(ps -p \(pid) -o comm=)\" = mosh-server ] && kill -TERM \(pid)",
|
||||
workingDirectory: nil, stdin: nil, timeout: 3, outputLimit: 1024, onOutput: nil)
|
||||
}.value
|
||||
}
|
||||
await engine.rollback(using: bootstrap)
|
||||
await bootstrap.disconnect()
|
||||
await close()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private func restore(_ checkpoint: TerminalTransportCheckpoint, route: @escaping RouteFactory) async throws {
|
||||
if transport == .mosh {
|
||||
let snapshot = try JSONDecoder().decode(MoshSnapshot.self, from: checkpoint.payload)
|
||||
mosh = try await MoshClientSession.restore(from: snapshot)
|
||||
} else {
|
||||
let payload = try JSONDecoder().decode(EternalPayload.self, from: checkpoint.payload)
|
||||
credentials = ETCredentials(clientID: payload.clientID, passkey: payload.passkey)
|
||||
let bridge = try TerminalTCPBridge(open: route, initial: await route())
|
||||
self.bridge = bridge
|
||||
eternal = try ETTerminalSession(
|
||||
host: "127.0.0.1", port: bridge.port, clientID: payload.clientID,
|
||||
passkey: payload.passkey, checkpoint: payload.checkpoint)
|
||||
integration = payload.integration
|
||||
integration?.resetDeadlinesAfterRestore()
|
||||
commandFilter = payload.commandFilter
|
||||
integratedShell = payload.integratedShell
|
||||
pendingIntegrationCommand = payload.pendingIntegrationCommand
|
||||
if let shell = integratedShell {
|
||||
integrationChanged(
|
||||
.available(shell: shell.shell, systemInfo: shell.systemInfo, homeDirectory: shell.homeDirectory))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public func write(_ data: Data) async throws {
|
||||
try checkOpen()
|
||||
guard !frozen else { throw TerminalTransportError.suspended }
|
||||
integration?.noteUserInput()
|
||||
sendsInFlight += 1
|
||||
defer { finishSend() }
|
||||
if let mosh { try await mosh.enqueue(.keystrokes(data)) }
|
||||
if let eternal { try await eternal.send(data) }
|
||||
try await engine?.write(data)
|
||||
}
|
||||
|
||||
public func resize(columns: UInt16, rows: UInt16) async throws {
|
||||
@@ -255,31 +140,27 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
guard !frozen else { return }
|
||||
sendsInFlight += 1
|
||||
defer { finishSend() }
|
||||
if let mosh { try await mosh.enqueue(.resize(cols: Int32(columns), rows: Int32(rows))) }
|
||||
if let eternal { try await eternal.resize(rows: Int(rows), cols: Int(columns)) }
|
||||
try await engine?.resize(columns: columns, rows: rows)
|
||||
}
|
||||
|
||||
private var shellState: ResumableTerminalShellState {
|
||||
.init(
|
||||
integration: integration, commandFilter: commandFilter, integratedShell: integratedShell,
|
||||
pendingIntegrationCommand: pendingIntegrationCommand)
|
||||
}
|
||||
|
||||
public func suspend() async throws -> TerminalTransportCheckpoint {
|
||||
try checkOpen()
|
||||
guard state == .connected || state == .recovering else { throw TerminalTransportError.sessionFailed }
|
||||
guard state == .connected || state == .recovering, let engine else {
|
||||
throw TerminalTransportError.sessionFailed
|
||||
}
|
||||
frozen = true
|
||||
if sendsInFlight > 0 {
|
||||
await withCheckedContinuation { sendWaiters.append($0) }
|
||||
}
|
||||
try checkOpen()
|
||||
let payload: Data
|
||||
if let mosh {
|
||||
payload = try JSONEncoder().encode(await mosh.prepareForApplicationBackground())
|
||||
} else if let eternal, let credentials {
|
||||
let checkpoint = try await eternal.prepareForApplicationBackground()
|
||||
payload = try JSONEncoder().encode(
|
||||
EternalPayload(
|
||||
clientID: credentials.clientID, passkey: credentials.passkey,
|
||||
checkpoint: checkpoint, integration: integration, commandFilter: commandFilter,
|
||||
integratedShell: integratedShell, pendingIntegrationCommand: pendingIntegrationCommand))
|
||||
} else {
|
||||
throw TerminalTransportError.invalidCheckpoint
|
||||
}
|
||||
// Read current shell state when the engine builds its checkpoint, after protocol preparation.
|
||||
let payload = try await engine.checkpoint { await self.shellState }
|
||||
try checkOpen()
|
||||
publish(.suspended)
|
||||
return TerminalTransportCheckpoint(transport: transport, payload: payload)
|
||||
@@ -287,14 +168,11 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
|
||||
public func resume() async throws {
|
||||
try checkOpen()
|
||||
if mosh != nil {
|
||||
await beforeConnect?()
|
||||
try checkOpen()
|
||||
}
|
||||
try await engine?.beforeResume()
|
||||
try checkOpen()
|
||||
frozen = false
|
||||
publish(.recovering)
|
||||
if let mosh { try await mosh.resumeFromApplicationBackground() }
|
||||
if let eternal { await eternal.resumeFromApplicationBackground() }
|
||||
try await engine?.resume()
|
||||
try checkOpen()
|
||||
integration?.resetDeadlinesAfterRestore()
|
||||
try await resize(columns: viewport.columns, rows: viewport.rows)
|
||||
@@ -302,46 +180,45 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
pendingIntegrationCommand = nil
|
||||
sendsInFlight += 1
|
||||
defer { finishSend() }
|
||||
try await eternal?.send(command)
|
||||
try await engine?.write(command)
|
||||
}
|
||||
}
|
||||
|
||||
public func retryNow() async throws {
|
||||
try checkOpen()
|
||||
await bridge?.retry()
|
||||
await engine?.prepareRetry()
|
||||
if frozen {
|
||||
try await resume()
|
||||
return
|
||||
}
|
||||
if let eternal { await eternal.notifyNetworkPathChanged() }
|
||||
if let mosh {
|
||||
await beforeConnect?()
|
||||
try checkOpen()
|
||||
_ = try await mosh.prepareForApplicationBackground()
|
||||
try await mosh.resumeFromApplicationBackground()
|
||||
}
|
||||
try await engine?.retry()
|
||||
}
|
||||
|
||||
public func close() async {
|
||||
guard !closed else { return }
|
||||
closed = true
|
||||
reader?.cancel()
|
||||
observer?.cancel()
|
||||
maintenance?.cancel()
|
||||
pathMonitor?.cancel()
|
||||
pathMonitor = nil
|
||||
beforeConnect = nil
|
||||
if let mosh { await mosh.stop() }
|
||||
if let eternal { await eternal.close() }
|
||||
await bridge?.close()
|
||||
mosh = nil
|
||||
eternal = nil
|
||||
bridge = nil
|
||||
credentials = nil
|
||||
await engine?.close()
|
||||
engine = nil
|
||||
pendingCheckpoint = nil
|
||||
if state != .ended { publish(.closed) }
|
||||
}
|
||||
|
||||
func receiveEngineEvent(_ event: ResumableTerminalEngineEvent) async {
|
||||
guard !closed else { return }
|
||||
switch event {
|
||||
case .output(let data): await receive(data)
|
||||
case .activity, .connected:
|
||||
if !frozen { publish(.connected) }
|
||||
case .recovering:
|
||||
if !frozen { publish(.recovering) }
|
||||
case .failed: publish(.failed)
|
||||
case .ended: await finishRemoteExit()
|
||||
}
|
||||
}
|
||||
|
||||
private func checkOpen() throws {
|
||||
try Task.checkCancellation()
|
||||
guard !closed else { throw CancellationError() }
|
||||
@@ -368,77 +245,6 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
await close()
|
||||
}
|
||||
|
||||
private func startMoshReader(_ session: MoshClientSession) {
|
||||
reader = Task { [weak self] in
|
||||
while !Task.isCancelled {
|
||||
guard let self else { return }
|
||||
// Read terminal state before draining, so the final poll cannot miss trailing output.
|
||||
let state = await session.state
|
||||
guard await self.receiveMoshPoll(state: state, operations: session.drainHostOps()) else { return }
|
||||
try? await Task.sleep(for: .milliseconds(16))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func receiveMoshPoll(state: MoshSessionState, operations: [MoshHostOp]) async -> Bool {
|
||||
guard !closed else { return false }
|
||||
deliver(operations)
|
||||
switch state {
|
||||
case .stopped:
|
||||
// Remote shutdown reaches this state only with swift-mosh's shutdown-marker fix (PR #2).
|
||||
await finishRemoteExit()
|
||||
return false
|
||||
case .failed:
|
||||
publish(.failed)
|
||||
return false
|
||||
case .idle, .starting, .running, .suspending, .suspended, .stopping:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
private func deliver(_ operations: [MoshHostOp]) {
|
||||
guard !closed else { return }
|
||||
if !operations.isEmpty, !frozen { publish(.connected) }
|
||||
for operation in operations {
|
||||
if case .hostBytes(let data) = operation { output(data) }
|
||||
}
|
||||
}
|
||||
|
||||
private func startEternalReaders(_ session: ETTerminalSession) {
|
||||
startEternalReaders(output: session.output, states: session.stateChanges)
|
||||
}
|
||||
|
||||
func startEternalReaders(output: AsyncStream<Data>, states: AsyncStream<ETConnectionState>) {
|
||||
reader = Task { [weak self] in
|
||||
for await data in output {
|
||||
guard let self, !Task.isCancelled else { return }
|
||||
await self.receive(data)
|
||||
}
|
||||
}
|
||||
observer = Task { [weak self] in
|
||||
for await state in states {
|
||||
guard let self, !Task.isCancelled else { return }
|
||||
await self.receiveEternalState(state)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func receiveEternalState(_ state: ETConnectionState) async {
|
||||
guard !closed else { return }
|
||||
switch state {
|
||||
case .connected:
|
||||
if !frozen { publish(.connected) }
|
||||
case .reconnecting, .disconnected:
|
||||
if !frozen { publish(.recovering) }
|
||||
case .failed: publish(.failed)
|
||||
case .sessionEnded:
|
||||
// The vendor finishes output before emitting this state, but our consumer may still be draining it.
|
||||
await reader?.value
|
||||
await finishRemoteExit()
|
||||
case .closed, .idle, .bootstrapping, .connecting: break
|
||||
}
|
||||
}
|
||||
|
||||
private func receive(_ data: Data) async {
|
||||
guard !closed else { return }
|
||||
if var integration {
|
||||
@@ -446,7 +252,7 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
self.integration = update.isComplete ? nil : integration
|
||||
await apply(update)
|
||||
} else {
|
||||
deliverEternalOutput(data)
|
||||
deliverIntegratedOutput(data)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -456,7 +262,7 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
pendingIntegrationCommand = command
|
||||
} else {
|
||||
sendsInFlight += 1
|
||||
do { try await eternal?.send(command) } catch {
|
||||
do { try await engine?.write(command) } catch {
|
||||
if !closed { publish(.failed) }
|
||||
}
|
||||
finishSend()
|
||||
@@ -464,17 +270,17 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
}
|
||||
guard !closed else { return }
|
||||
if update.integrationInstalled {
|
||||
let shell = IntegratedShell(
|
||||
let shell = ResumableTerminalShellState.IntegratedShell(
|
||||
shell: update.shell ?? "unknown", systemInfo: update.systemInfo, homeDirectory: update.homeDirectory)
|
||||
integratedShell = shell
|
||||
integrationChanged(
|
||||
.available(shell: shell.shell, systemInfo: shell.systemInfo, homeDirectory: shell.homeDirectory))
|
||||
}
|
||||
if update.homeResolutionUnavailable { integrationChanged(.homeResolutionUnavailable) }
|
||||
deliverEternalOutput(update.visibleOutput ?? Data())
|
||||
deliverIntegratedOutput(update.visibleOutput ?? Data())
|
||||
}
|
||||
|
||||
private func deliverEternalOutput(_ data: Data) {
|
||||
private func deliverIntegratedOutput(_ data: Data) {
|
||||
guard !closed else { return }
|
||||
var visible = data
|
||||
if var commandFilter {
|
||||
@@ -524,22 +330,6 @@ public actor ResumableTerminalClient: ResumableTerminalSession {
|
||||
guard !closed, !frozen, state == .connected || state == .recovering else { return }
|
||||
if !available { publish(.recovering) }
|
||||
// Never bootstrap another shell: both engines retain their own protocol state.
|
||||
if available { await eternal?.notifyNetworkPathChanged() }
|
||||
}
|
||||
}
|
||||
|
||||
private actor BootstrapExecutor: ETBootstrapExecutor {
|
||||
let connection: SSHConnection
|
||||
private(set) var failure: (any Error)?
|
||||
init(connection: SSHConnection) { self.connection = connection }
|
||||
func run(command: String) async throws -> String {
|
||||
do {
|
||||
let result = try await connection.execute(
|
||||
command, workingDirectory: nil, stdin: nil, timeout: 20, outputLimit: 65_536, onOutput: nil)
|
||||
return result.stdout + result.stderr
|
||||
} catch {
|
||||
failure = error
|
||||
throw error
|
||||
}
|
||||
if available { await engine?.networkAvailable() }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
import Foundation
|
||||
import TermindApplication
|
||||
import TermindDomain
|
||||
|
||||
nonisolated enum ResumableTerminalEngineEvent: Sendable {
|
||||
case output(Data)
|
||||
case activity, connected, recovering, failed, ended
|
||||
}
|
||||
|
||||
nonisolated struct ResumableTerminalShellState: Sendable {
|
||||
nonisolated struct IntegratedShell: Codable, Sendable {
|
||||
let shell: String
|
||||
let systemInfo: RemoteSystemInfo?
|
||||
let homeDirectory: String?
|
||||
}
|
||||
|
||||
var integration: RemoteShellIntegrationBootstrap?
|
||||
var commandFilter: RemoteCommandEventFilter?
|
||||
var integratedShell: IntegratedShell?
|
||||
var pendingIntegrationCommand: Data?
|
||||
}
|
||||
|
||||
nonisolated struct ResumableTerminalEnginePreparation: Sendable {
|
||||
var shell = ResumableTerminalShellState()
|
||||
var startupInput: Data?
|
||||
}
|
||||
|
||||
nonisolated struct ResumableTerminalEngineContext: Sendable {
|
||||
let request: SSHConnectionRequest
|
||||
let options: HostTerminalTransports
|
||||
let environment: [HostEnvironmentVariable]
|
||||
let shellIntegrationEnabled: Bool
|
||||
let checkpoint: Data?
|
||||
let startup: ResumableTerminalStartup
|
||||
let bootstrap: SSHConnection
|
||||
let route: ResumableTerminalClient.RouteFactory
|
||||
}
|
||||
|
||||
/// Engines own protocol resources, not application lifecycle or checkpoint eligibility.
|
||||
/// Every event and integration write passes through the client's shared serialization barrier.
|
||||
nonisolated protocol ResumableTerminalEngine: Sendable {
|
||||
typealias EventHandler = @Sendable (ResumableTerminalEngineEvent) async -> Void
|
||||
|
||||
func prepare(_ context: ResumableTerminalEngineContext) async throws -> ResumableTerminalEnginePreparation
|
||||
func start(columns: UInt16, rows: UInt16) async throws
|
||||
func write(_ data: Data) async throws
|
||||
func resize(columns: UInt16, rows: UInt16) async throws
|
||||
func checkpoint(shell: @Sendable () async -> ResumableTerminalShellState) async throws -> Data
|
||||
func beforeResume() async throws
|
||||
func resume() async throws
|
||||
func prepareRetry() async
|
||||
func retry() async throws
|
||||
func networkAvailable() async
|
||||
func rollback(using bootstrap: SSHConnection) async
|
||||
func close() async
|
||||
}
|
||||
|
||||
extension ResumableTerminalEngine {
|
||||
func beforeResume() async throws {}
|
||||
func prepareRetry() async {}
|
||||
func networkAvailable() async {}
|
||||
func rollback(using bootstrap: SSHConnection) async {}
|
||||
}
|
||||
|
||||
/// Registrations are immutable and local to a coordinator; no process-global overrides.
|
||||
public nonisolated struct ResumableTerminalEngineRegistry: Sendable {
|
||||
typealias Factory = @Sendable (@escaping ResumableTerminalEngine.EventHandler) -> any ResumableTerminalEngine
|
||||
private let factories: [HostTerminalProtocol: Factory]
|
||||
|
||||
init(factories: [HostTerminalProtocol: Factory]) {
|
||||
self.factories = factories
|
||||
}
|
||||
|
||||
public static let live = Self(factories: [
|
||||
.mosh: { MoshTerminalEngine(receive: $0) },
|
||||
.eternalTerminal: { EternalTerminalEngine(receive: $0) },
|
||||
])
|
||||
|
||||
func make(
|
||||
_ transport: HostTerminalProtocol, receive: @escaping ResumableTerminalEngine.EventHandler
|
||||
) throws -> any ResumableTerminalEngine {
|
||||
guard let factory = factories[transport] else { throw TerminalTransportError.unsupportedRoute }
|
||||
return factory(receive)
|
||||
}
|
||||
}
|
||||
@@ -739,6 +739,7 @@ public actor SSHPTYChannel {
|
||||
private var startupPreparation: StartupPreparation?
|
||||
private let transportOutput: OutputHandler
|
||||
private let transportIntegration: ShellIntegrationHandler
|
||||
private let transportEngines: ResumableTerminalEngineRegistry
|
||||
private var resumable: ResumableTerminalClient?
|
||||
private var transportOptions = HostTerminalTransports()
|
||||
private var transportKind: HostTerminalProtocol = .ssh
|
||||
@@ -753,10 +754,12 @@ public actor SSHPTYChannel {
|
||||
keyboardInteractive: SSHKeyboardInteractiveHandler? = nil,
|
||||
agentForwardingFactory: SSHAgentForwardingFactory? = nil,
|
||||
logger: Logging.Logger = ApplicationLogger.disabled("termind.ssh-pty-channel"),
|
||||
maxSessionsPolicy: SSHMaxSessionsPolicy = SSHMaxSessionsPolicy()
|
||||
maxSessionsPolicy: SSHMaxSessionsPolicy = SSHMaxSessionsPolicy(),
|
||||
transportEngines: ResumableTerminalEngineRegistry = .live
|
||||
) {
|
||||
transportOutput = output
|
||||
transportIntegration = shellIntegration
|
||||
self.transportEngines = transportEngines
|
||||
self.keyboardInteractive = keyboardInteractive
|
||||
self.agentForwardingFactory = agentForwardingFactory
|
||||
worker = SSHPTYChannelWorker(
|
||||
@@ -779,7 +782,7 @@ public actor SSHPTYChannel {
|
||||
guard let transportRoute else { throw TerminalTransportError.unsupportedRoute }
|
||||
let client = ResumableTerminalClient(
|
||||
transport: transportKind, checkpoint: transportCheckpoint, output: transportOutput,
|
||||
stateChanged: transportState, shellIntegration: transportIntegration)
|
||||
stateChanged: transportState, shellIntegration: transportIntegration, registry: transportEngines)
|
||||
resumable = client
|
||||
try await client.connect(
|
||||
request: request, options: transportOptions, environment: environmentVariables,
|
||||
|
||||
Reference in new issue
Block a user