From ff50fd407a916ab8d8919f4667e0f3a592ac3059 Mon Sep 17 00:00:00 2001 From: Wang Date: Tue, 1 Sep 2026 15:21:34 +0800 Subject: [PATCH 01/14] fix(desktop): issue live peer routes for shared sessions Generated-by: Codex --- .../runtime-host-local-remote-access.test.ts | 94 +++++++++++- .../runtime-host-profile-service.test.ts | 45 +++++- apps/desktop/src/main/runtime-host-boot.ts | 3 + apps/desktop/src/main/runtime-host-client.ts | 5 + .../main/runtime-host-local-remote-access.ts | 30 +++- .../src/main/runtime-host-profile-service.ts | 48 ++++-- native/runtime-host-peer/src/engine.rs | 145 ++++++++---------- .../src/__tests__/peer-listener.test.ts | 28 ++++ .../src/__tests__/peer-native.test.ts | 16 +- .../src/__tests__/protocol.test.ts | 30 ++++ .../runtime-host/src/client/connection.ts | 8 +- .../runtime-host/src/client/host-profile.ts | 11 +- .../runtime-host/src/client/peer-client.ts | 4 + packages/runtime-host/src/peer-mesh/node.ts | 19 ++- packages/runtime-host/src/peer-mesh/owner.ts | 4 + .../runtime-host/src/protocol/host-status.ts | 46 ++++++ packages/runtime-host/src/protocol/index.ts | 4 +- .../runtime-host/src/protocol/operations.ts | 1 + .../src/server/execution-service.ts | 3 + .../runtime-host/src/server/host-kernel.ts | 10 ++ .../runtime-host/src/server/listener-set.ts | 21 ++- .../runtime-host/src/server/peer-listener.ts | 4 + 22 files changed, 462 insertions(+), 117 deletions(-) diff --git a/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts b/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts index 9e502095c6..c6f2e90813 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts @@ -44,17 +44,36 @@ test('enabling remote access hands the same root to one managed service before D const handlers = new Map[1]>(); let retired = false; let resumed = false; + const peer = { + peerId: '12D3KooWpeer', + routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRelays: [], + }; + const livePeer = { + ...peer, + coordinationRelays: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + }; const manager = { async retireOwnedLocalHost() { retired = true; return { kind: 'retired' as const, resume: () => { resumed = true; } }; }, + async waitUntilReady(profileId: string) { + assert.equal(profileId, 'local'); + }, + current(profileId: string) { + assert.equal(profileId, 'local'); + return { + candidate: { + client: { + async status() { + return { peerEndpoint: livePeer }; + }, + }, + }, + }; + }, } as unknown as RuntimeHostDesktopManager; - const peer = { - peerId: '12D3KooWpeer', - routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], - coordinationRelays: [], - }; const deploymentId = '11111111-1111-4111-8111-111111111111'; const operator = { async runSetup(input: { @@ -123,7 +142,7 @@ test('enabling remote access hands the same root to one managed service before D assert.deepEqual(decodeRuntimeHostOwnerConnectionCode(result.connectionCode), { name: decodeRuntimeHostOwnerConnectionCode(result.connectionCode).name, rootId: 'a'.repeat(64), - transport: { kind: 'libp2p-direct', ...peer }, + transport: { kind: 'libp2p-direct', ...livePeer }, credential: 'pending-credential', }); const lifecycle = JSON.parse( @@ -134,6 +153,69 @@ test('enabling remote access hands the same root to one managed service before D assert.equal(lifecycle.deploymentId, deploymentId); }); +test('shares the running Local Host endpoint instead of its persisted startup routes', async (t) => { + const base = await mkdtemp(join(tmpdir(), 'maka-local-live-peer-share-')); + t.after(() => rm(base, { recursive: true, force: true })); + const clientDataRoot = join(base, 'client'); + const rootPath = join(clientDataRoot, 'workspaces', 'default'); + const rootId = 'a'.repeat(64); + await mkdir(rootPath, { recursive: true }); + await writeManagedLifecycle(clientDataRoot, rootPath, rootId); + const configuredPeer = { + peerId: '12D3KooWpeer', + routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRelays: [], + }; + const livePeer = { + ...configuredPeer, + coordinationRelays: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + }; + const service = createDesktopLocalRuntimeHostRemoteAccess({ + ipcMain: { handle() {}, removeHandler() {} }, + clientDataRoot, + rootPath, + rootId, + directPeerAvailable: true, + manager: () => + ({ + current() { + return { + candidate: { + client: { + async status() { + return { peerEndpoint: livePeer }; + }, + }, + }, + }; + }, + }) as unknown as RuntimeHostDesktopManager, + resolveSetupPackage: async () => ({ kind: 'npm', specifier: 'maka-agent@0.2.0' }), + operator: { + async runPeer() { + return { + kind: 'result' as const, + action: 'status' as const, + status: { + state: 'enabled' as const, + serviceState: 'running', + rootId, + ...configuredPeer, + }, + }; + }, + async close() {}, + } as unknown as ReturnType, + }); + t.after(() => service.close()); + + const target = await service.createCollaborationConnectionTarget(); + assert.deepEqual(target, { + name: target.name, + transport: { kind: 'libp2p-direct', ...livePeer }, + }); +}); + test('revokes the one Local sharing authority without changing peer connectivity', async (t) => { const base = await mkdtemp(join(tmpdir(), 'maka-local-shared-access-revoke-')); t.after(() => rm(base, { recursive: true, force: true })); diff --git a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts index d99b90a262..12a9886d42 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts @@ -716,12 +716,27 @@ test("keeps a managed Direct route on the SSH profile credential authority", asy await managedServices.save(MANAGED_PROFILE, MANAGED_SERVICE); const startup = await resolveDesktopRuntimeHostStartup(root, { catalog }); const activated: ResolvedRuntimeHostProfile[] = []; + const livePeer = { + peerId: "12D3KooWpeer", + routeHints: ["/ip4/192.0.2.9/udp/44002/quic-v1"], + coordinationRelays: ["/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay"], + }; + let exposeReadyState = false; const service = createDesktopRuntimeHostProfileService({ clientDataRoot: root, startup, catalog, managedServices, - states: () => [connectingLocal()], + states: () => + exposeReadyState + ? [ + connectingLocal(), + readyWithPeerEndpoint( + { profile: MANAGED_PROFILE, credential: "owner-token" }, + livePeer, + ), + ] + : [connectingLocal()], enable: async (target) => { activated.push(target); }, @@ -755,10 +770,15 @@ test("keeps a managed Direct route on the SSH profile credential authority", asy routeHints: ["/ip4/192.0.2.8/udp/44001/quic-v1"], coordinationRelays: [], }); + exposeReadyState = true; assert.deepEqual( await service.resolveCollaborationConnectionTarget(MANAGED_PROFILE), - { name: MANAGED_PROFILE.name, transport: direct.profile.transport }, + { + name: MANAGED_PROFILE.name, + transport: { kind: "libp2p-direct", ...livePeer }, + }, ); + exposeReadyState = false; assert.equal((await catalog.resolve(MANAGED_PROFILE.id)).credential, "owner-token"); const beforeRejectedRemoval = { @@ -1602,6 +1622,27 @@ function ready(target: ResolvedRuntimeHostProfile): RuntimeHostDesktopTargetStat }; } +function readyWithPeerEndpoint( + target: ResolvedRuntimeHostProfile, + peerEndpoint: { + readonly peerId: string; + readonly routeHints: readonly string[]; + readonly coordinationRelays: readonly string[]; + }, +): RuntimeHostDesktopTargetState { + return { + epoch: `epoch-${target.profile.id}`, + target, + readiness: "ready", + candidate: { + client: { + hostId: target.profile.kind === "remote" ? target.profile.rootId : ROOT_ID, + status: async () => ({ peerEndpoint }), + }, + } as never, + }; +} + function unavailable( target: ResolvedRuntimeHostProfile, error: Error, diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index dae1a3cbd6..45ab7d5fb1 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -270,6 +270,9 @@ if (runtimeHostPeerConfiguration) { ...runtimeHostPeerConfiguration, dataRoot: join(userDataDir, 'peer-mesh'), endpointKind: 'client', + onBackgroundReconcileError: (error) => { + console.error('[runtime-host] Peer Mesh background synchronization failed:', error); + }, }); runtimeHostPeerClient = runtimeHostPeerOwner.client; runtimeHostPeerMesh = runtimeHostPeerOwner.mesh; diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index 9bc0318393..58a2f03268 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -84,6 +84,7 @@ import { type MemoryQueryInput, type MemoryQueryResult, type GoalControlAction, + type HostStatusResult, type GoalProjection, type OperationInput, type OperationOutput, @@ -295,6 +296,10 @@ export class DesktopRuntimeHostClient { return this.#connectionClosed || this.#closeTask ? 'unavailable' : 'ready'; } + status(): Promise { + return this.connection.status(); + } + finalizeAccessCredential( timeoutMs?: number, ): Promise> { diff --git a/apps/desktop/src/main/runtime-host-local-remote-access.ts b/apps/desktop/src/main/runtime-host-local-remote-access.ts index 4ade5c5610..3623734975 100644 --- a/apps/desktop/src/main/runtime-host-local-remote-access.ts +++ b/apps/desktop/src/main/runtime-host-local-remote-access.ts @@ -339,11 +339,14 @@ export function createDesktopLocalRuntimeHostRemoteAccess(input: { }; const completed = await finishSetup(setup, 'request'); if (completed.kind === 'active_tasks') return completed; + const manager = requireManager(input.manager); + await manager.waitUntilReady('local'); + const livePeer = await readLivePeer(localClient(input.manager), completed.peer); return enabledResult( encodeRuntimeHostOwnerConnectionCode({ name: hostName(), rootId: completed.managed.rootId, - transport: { kind: 'libp2p-direct', ...completed.peer }, + transport: { kind: 'libp2p-direct', ...livePeer }, credential: completed.credential, }), ); @@ -515,9 +518,10 @@ export function createDesktopLocalRuntimeHostRemoteAccess(input: { ); const peer = await readPeer(input.operator, managed); if (!peer) throw new Error('Remote access is not enabled on this computer'); + const livePeer = await readLivePeer(localClient(input.manager), peer); return { name: hostName(), - transport: { kind: 'libp2p-direct' as const, ...peer }, + transport: { kind: 'libp2p-direct' as const, ...livePeer }, }; }); @@ -968,6 +972,7 @@ async function issueConnectionCode( peer: LocalPeerDescriptor, client: DesktopRuntimeHostClient, ): Promise { + const livePeer = await readLivePeer(client, peer); const prepared = await client.request('access.credential.prepare', { principalKind: 'remote_owner', principalId: LOCAL_REMOTE_ACCESS_PRINCIPAL_ID, @@ -984,11 +989,30 @@ async function issueConnectionCode( return encodeRuntimeHostOwnerConnectionCode({ name: hostName(), rootId, - transport: { kind: 'libp2p-direct', ...peer }, + transport: { kind: 'libp2p-direct', ...livePeer }, credential, }); } +async function readLivePeer( + client: DesktopRuntimeHostClient, + configured: LocalPeerDescriptor, +): Promise { + const endpoint = (await client.status()).peerEndpoint; + if (!endpoint) { + throw new Error('Runtime Host Direct peer is not available'); + } + if (endpoint.peerId !== configured.peerId) { + throw new Error('Runtime Host Direct peer identity changed'); + } + return requireEnabledPeer({ + state: 'enabled', + peerId: endpoint.peerId, + routeHints: endpoint.routeHints, + coordinationRelays: endpoint.coordinationRelays, + }); +} + async function hasSharedAccess( operator: DesktopRuntimeHostLocalOperator, target: LocalServiceTarget, diff --git a/apps/desktop/src/main/runtime-host-profile-service.ts b/apps/desktop/src/main/runtime-host-profile-service.ts index 33bb1380f7..a310141439 100644 --- a/apps/desktop/src/main/runtime-host-profile-service.ts +++ b/apps/desktop/src/main/runtime-host-profile-service.ts @@ -920,23 +920,49 @@ export function createDesktopRuntimeHostProfileService(input: { if (profile.kind !== 'remote') { throw new Error('This Runtime Host does not expose a shareable peer endpoint'); } - if (profile.transport.kind !== 'ssh') { + if (profile.transport.kind !== 'ssh' && profile.transport.kind !== 'libp2p-direct') { return { name: profile.name, transport: profile.transport }; } - const direct = (await catalog.read()).profiles.find( - (candidate) => candidate.id === managedDirectPeerProfileId(profile.id), - ); - if ( - !direct || - direct.kind !== 'remote' || - direct.rootId !== profile.rootId || - direct.transport.kind !== 'libp2p-direct' - ) { + let configuredPeerId: string | undefined; + if (profile.transport.kind === 'libp2p-direct') { + configuredPeerId = profile.transport.peerId; + } else { + const direct = (await catalog.read()).profiles.find( + (candidate) => candidate.id === managedDirectPeerProfileId(profile.id), + ); + if ( + direct?.kind === 'remote' && + direct.rootId === profile.rootId && + direct.transport.kind === 'libp2p-direct' + ) { + configuredPeerId = direct.transport.peerId; + } + } + if (!configuredPeerId) { throw new Error( 'Enable Direct peer access for this Runtime Host before sharing its Sessions', ); } - return { name: profile.name, transport: direct.transport }; + const active = input.states().find((state) => state.target.profile.id === profile.id); + if (!active || active.readiness !== 'ready') { + throw new Error('Connect this Runtime Host before sharing its Sessions'); + } + const endpoint = (await active.candidate.client.status()).peerEndpoint; + if (!endpoint) { + throw new Error('Runtime Host Direct peer is not available'); + } + if (configuredPeerId !== endpoint.peerId) { + throw new Error('Runtime Host Direct peer identity changed'); + } + return { + name: profile.name, + transport: { + kind: 'libp2p-direct' as const, + peerId: endpoint.peerId, + routeHints: endpoint.routeHints, + coordinationRelays: endpoint.coordinationRelays, + }, + }; }); }, resolveManagedAccess(profileId) { diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index 357901a468..dd171c035d 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -644,14 +644,12 @@ async fn run_endpoint_async( &mut swarm, &mut coordination_relays, &HashMap::new(), - false, Instant::now(), ); let startup_deadline = Instant::now() + Duration::from_secs(10); let mut address_quiet_deadline = None; let mut bound_addresses = HashSet::new(); - let mut startup_external_candidate_ready = false; loop { let deadline = address_quiet_deadline .unwrap_or(startup_deadline) @@ -668,13 +666,9 @@ async fn run_endpoint_async( address_quiet_deadline = Some(Instant::now() + LISTENER_ADDRESS_QUIET_PERIOD); } } - Ok(event) => handle_startup_event( - &mut swarm, - event, - &mut coordination_relays, - &mut startup_external_candidate_ready, - &mut transit, - ), + Ok(event) => { + handle_startup_event(&mut swarm, event, &mut coordination_relays, &mut transit) + } Err(_) if pending_listeners.is_empty() => break, Err(_) => { return Err(PeerError::new( @@ -698,7 +692,6 @@ async fn run_endpoint_async( let (stream_completed_tx, mut stream_completed_rx) = mpsc::channel::(MAX_ESTABLISHED_CONNECTIONS as usize); let mut direct = DirectConnectState::default(); - let mut external_candidate_ready = startup_external_candidate_ready; let mut deadline_tick = tokio::time::interval(Duration::from_millis(100)); deadline_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let mut discovered_relays = options @@ -770,7 +763,6 @@ async fn run_endpoint_async( &mut direct, &coordination_relays, &stream_control, - external_candidate_ready, Instant::now(), ); start_pending_webrtc_upgrades( @@ -831,7 +823,6 @@ async fn run_endpoint_async( &mut direct, &coordination_relays, &stream_control, - external_candidate_ready, Instant::now(), ); let requests = direct.pending.keys().copied().collect::>(); @@ -1002,7 +993,6 @@ async fn run_endpoint_async( &mut swarm, &mut coordination_relays, &direct.active, - external_candidate_ready, Instant::now(), ); } @@ -1095,7 +1085,6 @@ async fn run_endpoint_async( &mut direct, &coordination_relays, &stream_control, - external_candidate_ready, Instant::now(), ); maybe_open_peer_stream( @@ -1190,7 +1179,6 @@ async fn run_endpoint_async( &mut direct, &coordination_relays, &stream_control, - external_candidate_ready, Instant::now(), ); maybe_open_peer_stream( @@ -1210,7 +1198,6 @@ async fn run_endpoint_async( event, &mut coordination_relays, &mut direct, - &mut external_candidate_ready, RouteRuntime { active_coordination_relays: &active_coordination_relays, transit: &mut transit, @@ -1230,7 +1217,6 @@ async fn run_endpoint_async( &mut swarm, &mut coordination_relays, &direct.active, - external_candidate_ready, Instant::now(), ); let requests = direct.pending.keys().copied().collect::>(); @@ -1268,7 +1254,6 @@ async fn run_endpoint_async( &mut swarm, &mut coordination_relays, &direct.active, - external_candidate_ready, now, ); retry_connect_routes( @@ -1276,7 +1261,6 @@ async fn run_endpoint_async( &mut direct, &coordination_relays, &stream_control, - external_candidate_ready, now, ); start_pending_webrtc_upgrades( @@ -1508,13 +1492,7 @@ fn start_connect( referenced.insert(relay_peer), )?; } - maintain_coordination_relays( - swarm, - coordination_relays, - active_streams, - false, - Instant::now(), - ); + maintain_coordination_relays(swarm, coordination_relays, active_streams, Instant::now()); Ok(StartedConnect { direct_routes: direct_targets, coordination_relay_peers: relay_peers, @@ -1793,7 +1771,6 @@ fn handle_swarm_event( event: SwarmEvent, coordination_relays: &mut HashMap, direct: &mut DirectConnectState, - external_candidate_ready: &mut bool, route_runtime: RouteRuntime<'_>, ) { match event { @@ -2016,13 +1993,7 @@ fn handle_swarm_event( if let Some(relay) = coordination_relays.get_mut(&peer_id) { relay.identify_received = true; } - request_coordination_reservation( - swarm, - coordination_relays, - peer_id, - *external_candidate_ready, - Instant::now(), - ); + request_coordination_reservation(swarm, coordination_relays, peer_id, Instant::now()); } SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Sent { peer_id, .. @@ -2030,25 +2001,7 @@ fn handle_swarm_event( if let Some(relay) = coordination_relays.get_mut(&peer_id) { relay.identify_sent = true; } - request_coordination_reservation( - swarm, - coordination_relays, - peer_id, - *external_candidate_ready, - Instant::now(), - ); - } - SwarmEvent::NewExternalAddrCandidate { .. } => { - *external_candidate_ready = true; - for peer_id in coordination_relays.keys().copied().collect::>() { - request_coordination_reservation( - swarm, - coordination_relays, - peer_id, - true, - Instant::now(), - ); - } + request_coordination_reservation(swarm, coordination_relays, peer_id, Instant::now()); } SwarmEvent::ListenerClosed { listener_id, @@ -2093,7 +2046,6 @@ fn handle_startup_event( swarm: &mut Swarm, event: SwarmEvent, coordination_relays: &mut HashMap, - external_candidate_ready: &mut bool, transit: &mut TransitRuntime, ) { handle_swarm_event( @@ -2101,7 +2053,6 @@ fn handle_startup_event( event, coordination_relays, &mut DirectConnectState::default(), - external_candidate_ready, RouteRuntime { active_coordination_relays: &Arc::new(RwLock::new(Vec::new())), transit, @@ -2405,7 +2356,7 @@ fn reconcile_transit_reservations( } relay.direct_connection_addresses.clear(); } - maintain_coordination_relays(swarm, relays, active_streams, true, Instant::now()); + maintain_coordination_relays(swarm, relays, active_streams, Instant::now()); } fn retained_transit_candidates( @@ -2817,7 +2768,6 @@ fn maintain_coordination_relays( swarm: &mut Swarm, relays: &mut HashMap, active_streams: &HashMap, - external_candidate_ready: bool, now: Instant, ) { for peer_id in relays.keys().copied().collect::>() { @@ -2870,7 +2820,7 @@ fn maintain_coordination_relays( } continue; } - request_coordination_reservation(swarm, relays, peer_id, external_candidate_ready, now); + request_coordination_reservation(swarm, relays, peer_id, now); } } @@ -2878,7 +2828,6 @@ fn request_coordination_reservation( swarm: &mut Swarm, relays: &mut HashMap, peer_id: PeerId, - external_candidate_ready: bool, now: Instant, ) { let Some(relay) = relays.get_mut(&peer_id) else { @@ -2898,7 +2847,6 @@ fn request_coordination_reservation( || relay.reservation_listener.is_some() || !identified || (transit_provider && !directly_connected) - || !external_candidate_ready || relay.next_reservation_attempt > now { return; @@ -2979,7 +2927,6 @@ fn retry_connect_routes( direct: &mut DirectConnectState, coordination_relays: &HashMap, stream_control: &application_stream::Control, - external_candidate_ready: bool, now: Instant, ) { for connect in direct.pending.values_mut() { @@ -3011,24 +2958,11 @@ fn retry_connect_routes( .any(|origin| *origin == DialOrigin::Coordination) && (connect.retry_coordination || !stream_control.has_relayed_connection(peer_id)) { - let mut targets = Vec::new(); - for relay in &connect.coordination_relays { - let relay_peer = coordination_relay_peer_id(relay) - .expect("coordination relay was validated before connecting"); - if !external_candidate_ready - || coordination_relays - .get(&relay_peer) - .is_none_or(|relay| !relay.identify_received || !relay.identify_sent) - { - continue; - } - targets.push( - relay - .clone() - .with(Protocol::P2pCircuit) - .with(Protocol::P2p(peer_id)), - ); - } + let targets = coordination_dial_targets( + peer_id, + &connect.coordination_relays, + coordination_relays, + ); if let Some(connection_id) = dial_direct_targets(swarm, peer_id, targets) { connect .dials @@ -3069,6 +3003,29 @@ fn retry_connect_routes( } } +fn coordination_dial_targets( + peer_id: PeerId, + addresses: &[Multiaddr], + relays: &HashMap, +) -> Vec { + addresses + .iter() + .filter(|address| { + let relay_peer = coordination_relay_peer_id(address) + .expect("coordination relay was validated before connecting"); + relays + .get(&relay_peer) + .is_some_and(|relay| relay.identify_received && relay.identify_sent) + }) + .map(|address| { + address + .clone() + .with(Protocol::P2pCircuit) + .with(Protocol::P2p(peer_id)) + }) + .collect() +} + fn dial_direct_targets( swarm: &mut Swarm, peer_id: PeerId, @@ -3121,6 +3078,36 @@ mod tests { ); } + #[test] + fn coordination_dial_only_requires_an_identified_relay() { + let relay_peer_id = PeerId::random(); + let target_peer_id = PeerId::random(); + let relay_address: Multiaddr = format!("/ip4/203.0.113.1/tcp/4001/p2p/{relay_peer_id}") + .parse() + .expect("valid relay address"); + let relays = HashMap::from([( + relay_peer_id, + CoordinationRelay { + identify_received: true, + identify_sent: true, + ..CoordinationRelay::default() + }, + )]); + + assert_eq!( + coordination_dial_targets( + target_peer_id, + std::slice::from_ref(&relay_address), + &relays, + ), + vec![ + relay_address + .with(Protocol::P2pCircuit) + .with(Protocol::P2p(target_peer_id)) + ], + ); + } + #[test] fn completion_at_the_immutable_deadline_cannot_commit() { let now = Instant::now(); diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts index 2a5d900895..e6ec103e18 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts @@ -21,6 +21,7 @@ import assert from 'node:assert/strict'; import { setImmediate as waitForImmediate } from 'node:timers/promises'; import { test } from 'node:test'; import { createRuntimeHostPeerListener } from '../server/peer-listener.js'; +import { createRuntimeHostListenerSet } from '../server/listener-set.js'; import type { RuntimeHostPeerClient } from '../client/peer-client.js'; import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js'; @@ -122,6 +123,33 @@ test('bounds active application streams from one authenticated peer', async () = await listener.cleanup(); }); +test('projects newly accepted coordination relays from the running peer endpoint', async () => { + let coordinationRelays: readonly string[] = []; + const peer = { + ...peerWith([]), + identity: () => ({ + peerId: 'peer', + listenAddresses: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRelays, + }), + }; + const listener = createRuntimeHostPeerListener(peer, {} as never, () => {}); + const listeners = createRuntimeHostListenerSet( + { + kind: 'local_ipc', + endpoint: 'local', + closeAdmission: async () => undefined, + cleanup: async () => undefined, + }, + [listener], + ); + + assert.deepEqual(listeners.peerListeners[0]?.coordinationRelays, []); + coordinationRelays = ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay']; + assert.deepEqual(listeners.peerListeners[0]?.coordinationRelays, coordinationRelays); + await listeners.cleanup(); +}); + function peerWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerClient { return { identity: () => ({ peerId: 'peer', listenAddresses: [], coordinationRelays: [] }), diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index fc12167100..ce5627cf97 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/packages/runtime-host/src/__tests__/peer-native.test.ts @@ -72,7 +72,7 @@ module.exports = { connect: ({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }) => { stats.requests.push({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }); if (peerId === 'unreachable') return Promise.reject(Object.assign(new Error('transit_unavailable: no approved route'), { code: 'GenericFailure' })); - if (peerId === 'ready' || peerId === 'fallback') return Promise.resolve(stream); + if (peerId === 'ready' || peerId === 'fallback' || peerId === 'self-contained') return Promise.resolve(stream); return new Promise((resolve, reject) => pending.set(requestId, { resolve, reject })); }, connectMeshControl: ({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }) => { @@ -100,11 +100,13 @@ module.exports = { `, ); let routesPrepared = false; + const preparedPeerIds: string[] = []; const client = createRuntimeHostPeerClient({ nativePath, keyPath: join(directory, 'peer.key'), routeResolver: { prepareRoutes: async (peerId) => { + preparedPeerIds.push(peerId); routesPrepared = true; if (peerId === 'fallback') throw new Error('Mesh refresh failed'); }, @@ -151,6 +153,11 @@ module.exports = { await assert.rejects(client.connect(peerConnectInput('unreachable')), (failure: unknown) => { return failure instanceof RuntimeHostPeerError && failure.code === 'transit_unavailable'; }); + await client.connect({ + ...peerConnectInput('self-contained'), + coordinationRelays: ['/memory/explicit-relay'], + }); + assert.equal(preparedPeerIds.includes('self-contained'), false); assert.deepEqual(native.default.stats, { starts: 1, closes: 0, @@ -197,6 +204,13 @@ module.exports = { coordinationRelays: ['/memory/relay'], transitRelayPeerIds: ['transit-peer'], }, + { + requestId: 7, + peerId: 'self-contained', + routeHints: ['/memory/discovered', '/memory/1'], + coordinationRelays: ['/memory/relay', '/memory/explicit-relay'], + transitRelayPeerIds: ['transit-peer'], + }, ], cancellations: [1, 1], }); diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index db9f2d70f6..3bbe698872 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -1924,6 +1924,36 @@ describe('Runtime Host bootstrap protocol', () => { ); }); + test('publishes a bounded live Direct peer endpoint through Host status', () => { + const status = { + hostEpoch: 'epoch-1', + compositionId: 'maka.interactive', + compositionRevision: '1', + state: 'ready', + connections: 1, + activeOperations: 0, + activeResidencies: 0, + peerEndpoint: { + peerId: '12D3KooWhost', + routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRelays: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + }, + }; + assert.deepEqual(HOST_BOOTSTRAP_OPERATION_SPECS['host.status'].decodeOutput(status), status); + assert.throws(() => + HOST_BOOTSTRAP_OPERATION_SPECS['host.status'].decodeOutput({ + ...status, + peerEndpoint: { + ...status.peerEndpoint, + coordinationRelays: [ + status.peerEndpoint.coordinationRelays[0], + status.peerEndpoint.coordinationRelays[0], + ], + }, + }), + ); + }); + test('keeps Runtime Host logs within the diagnostics operation contract', () => { for (let index = 0; index < 257; index += 1) { runtimeHostLogBuffer.append('info', `entry ${index}`); diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts index 55a1749f3f..ec4c174272 100644 --- a/packages/runtime-host/src/client/connection.ts +++ b/packages/runtime-host/src/client/connection.ts @@ -197,6 +197,7 @@ export type ConnectRemoteRuntimeHostResult = | 'unreachable' | 'connect_failed' | 'handshake_failed' + | 'handshake_timed_out' | 'root_mismatch' | 'composition_mismatch'; }; @@ -1056,11 +1057,13 @@ export async function connectRuntimeHostMessageTransport( ): Promise { let resourceTransferred = false; let timer: ReturnType | undefined; + let handshakeTimedOut = false; try { const normalized = normalizeConnectRuntimeHostInput(input); const compositionId = requireHostCompositionId(input.compositionId); const expectedRootId = requireHostRootId(input.expectedRootId); timer = setTimeout(() => { + handshakeTimedOut = true; input.transport.abort(new Error('Timed out handshaking with Runtime Host')); }, normalized.handshakeTimeoutMs); const result = await exchangeRuntimeHostHandshake({ @@ -1088,7 +1091,10 @@ export async function connectRuntimeHostMessageTransport( if (error instanceof RuntimeHostCompositionMismatchError) { return { kind: 'unavailable', reason: 'composition_mismatch' }; } - return { kind: 'unavailable', reason: 'handshake_failed' }; + return { + kind: 'unavailable', + reason: handshakeTimedOut ? 'handshake_timed_out' : 'handshake_failed', + }; } finally { if (timer) clearTimeout(timer); if (!resourceTransferred) await input.connectionResource?.close().catch(() => undefined); diff --git a/packages/runtime-host/src/client/host-profile.ts b/packages/runtime-host/src/client/host-profile.ts index ea0a72ac9c..feab8c1fd6 100644 --- a/packages/runtime-host/src/client/host-profile.ts +++ b/packages/runtime-host/src/client/host-profile.ts @@ -71,6 +71,7 @@ const PROFILE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/; const PEER_ID_MAX_BYTES = 160; const PEER_ADDRESS_MAX_BYTES = 2 * 1024; const PEER_ROUTE_MAX = 16; +const DEFAULT_PEER_HANDSHAKE_TIMEOUT_MS = 5_000; const PROFILE_CREDENTIAL_RECORD_PREFIX = 'maka-runtime-host-profile-credential-v1:'; const PROFILE_INCARNATION_ID_MAX_BYTES = 128; export const RUNTIME_HOST_ACCESS_CREDENTIAL_MAX_BYTES = 8 * 1024; @@ -588,6 +589,7 @@ export async function connectPeerRuntimeHost(input: { readonly onConnectionPhase?: (phase: RuntimeHostConnectionPhase) => void; }): Promise { input.signal?.throwIfAborted(); + const handshakeTimeoutMs = input.handshakeTimeoutMs ?? DEFAULT_PEER_HANDSHAKE_TIMEOUT_MS; const stream = await input.peerClient.connect( { peerId: input.transport.peerId, @@ -608,7 +610,7 @@ export async function connectPeerRuntimeHost(input: { await writeRuntimeHostPeerAuthentication(stream, input.credential); const authentication = await readRuntimeHostPeerAuthenticationResult( stream, - input.handshakeTimeoutMs, + handshakeTimeoutMs, ); if (!authentication.accepted) { throw new RuntimeHostProfileConnectionError( @@ -628,9 +630,7 @@ export async function connectPeerRuntimeHost(input: { max: RUNTIME_HOST_PROTOCOL_VERSION, }, clientInstanceId: input.clientInstanceId, - ...(input.handshakeTimeoutMs === undefined - ? {} - : { handshakeTimeoutMs: input.handshakeTimeoutMs }), + handshakeTimeoutMs, ...(stream.path ? { peerPath: stream.path } : {}), }); input.signal?.throwIfAborted(); @@ -700,6 +700,9 @@ export function remoteRuntimeHostUnavailableError( case 'unreachable': message = `${subject} could not reach its endpoint`; break; + case 'handshake_timed_out': + message = `${subject} timed out while establishing its protocol session`; + break; default: message = `${subject} is unavailable (${reason})`; } diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts index 543efdb2f0..e8f639141f 100644 --- a/packages/runtime-host/src/client/peer-client.ts +++ b/packages/runtime-host/src/client/peer-client.ts @@ -226,6 +226,10 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { signal: AbortSignal | undefined, ): Promise { if (!this.#routeResolver?.prepareRoutes) return; + // A freshly issued target with both direct hints and coordination relays is + // already self-contained. Cached Mesh state can still enrich the actual + // dial below, but a stale control plane must not delay this explicit path. + if (input.routeHints.length > 0 && (input.coordinationRelays?.length ?? 0) > 0) return; const deadline = AbortSignal.timeout(Math.min(10_000, input.directDeadlineMs)); const operationSignal = signal ? AbortSignal.any([signal, deadline]) : deadline; try { diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index d49068723e..2e16e5469c 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -224,6 +224,7 @@ export async function openPeerMeshNode(input: { readonly peer: PeerMeshTransport; readonly endpointKind?: 'client' | 'host'; readonly now?: () => number; + readonly onBackgroundReconcileError?: (error: unknown) => void; }): Promise { const store = await openPeerMeshStateStore(input.dataRoot, input.peer.identity().peerId); const node = new PeerMeshNodeImpl({ ...input, store }); @@ -241,6 +242,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { readonly #peer: PeerMeshTransport; readonly #endpointKind: 'client' | 'host' | undefined; readonly #now: () => number; + readonly #onBackgroundReconcileError: ((error: unknown) => void) | undefined; readonly #activeControlStreams = new Set(); readonly #lifetime = new AbortController(); #admissionTail = Promise.resolve(); @@ -256,11 +258,13 @@ class PeerMeshNodeImpl implements PeerMeshNode { readonly peer: PeerMeshTransport; readonly endpointKind?: 'client' | 'host'; readonly now?: () => number; + readonly onBackgroundReconcileError?: (error: unknown) => void; }) { this.#store = input.store; this.#peer = input.peer; this.#endpointKind = input.endpointKind; this.#now = input.now ?? Date.now; + this.#onBackgroundReconcileError = input.onBackgroundReconcileError; } async initialize(): Promise { @@ -910,8 +914,21 @@ class PeerMeshNodeImpl implements PeerMeshNode { } async #runReconciliation(signal: AbortSignal): Promise { + let failureReported = false; while (!signal.aborted) { - await this.reconcile(signal).catch(() => undefined); + try { + await this.reconcile(signal); + failureReported = false; + } catch (error) { + if (!signal.aborted && !failureReported) { + failureReported = true; + try { + this.#onBackgroundReconcileError?.(error); + } catch { + // Diagnostics cannot control Peer Mesh reconciliation. + } + } + } await delay(RECONCILE_INTERVAL_MS, undefined, { signal }).catch(() => undefined); } } diff --git a/packages/runtime-host/src/peer-mesh/owner.ts b/packages/runtime-host/src/peer-mesh/owner.ts index 5bde0a8d10..035ab2a46b 100644 --- a/packages/runtime-host/src/peer-mesh/owner.ts +++ b/packages/runtime-host/src/peer-mesh/owner.ts @@ -44,6 +44,7 @@ export async function openRuntimeHostPeerMeshOwner(input: { readonly coordinationRelays?: readonly string[]; readonly automaticRelayDiscovery?: boolean; readonly webRtcStunUrls?: readonly string[]; + readonly onBackgroundReconcileError?: (error: unknown) => void; }): Promise { let mesh: PeerMeshNode | undefined; let resolverMesh: PeerMeshNode | undefined; @@ -77,6 +78,9 @@ export async function openRuntimeHostPeerMeshOwner(input: { dataRoot: join(input.dataRoot, client.identity().peerId), peer: client, endpointKind: input.endpointKind, + ...(input.onBackgroundReconcileError + ? { onBackgroundReconcileError: input.onBackgroundReconcileError } + : {}), }); resolverMesh = mesh; } catch (error) { diff --git a/packages/runtime-host/src/protocol/host-status.ts b/packages/runtime-host/src/protocol/host-status.ts index 4aac3a9603..a28ef97438 100644 --- a/packages/runtime-host/src/protocol/host-status.ts +++ b/packages/runtime-host/src/protocol/host-status.ts @@ -51,6 +51,9 @@ export type HostUpgradePrepareResult = export const HOST_DIAGNOSTICS_RESULT_MAX_BYTES = 72 * 1024; export const HOST_DIAGNOSTIC_LOG_MAX_ENTRIES = 256; export const HOST_DIAGNOSTIC_LOG_MAX_ENTRY_BYTES = 10 * 1024; +const HOST_PEER_ID_MAX_BYTES = 160; +const HOST_PEER_ADDRESS_MAX_BYTES = 2 * 1024; +const HOST_PEER_ROUTE_MAX = 16; export interface HostStatusResult { hostEpoch: string; @@ -60,6 +63,13 @@ export interface HostStatusResult { connections: number; activeOperations: number; activeResidencies: number; + peerEndpoint?: HostPeerEndpoint; +} + +export interface HostPeerEndpoint { + readonly peerId: string; + readonly routeHints: readonly string[]; + readonly coordinationRelays: readonly string[]; } export interface HostDiagnosticsResult extends HostStatusResult { @@ -100,12 +110,26 @@ export const HOST_BOOTSTRAP_OPERATION_SPECS = { }), } as const; +function decodePeerAddresses(value: unknown, label: string): readonly string[] { + if (!Array.isArray(value) || value.length > HOST_PEER_ROUTE_MAX) { + throw invalidProtocolFrame(`Invalid ${label}`); + } + const addresses = value.map((address) => + requireString(address, label, HOST_PEER_ADDRESS_MAX_BYTES), + ); + if (new Set(addresses).size !== addresses.length) { + throw invalidProtocolFrame(`Duplicate ${label}`); + } + return Object.freeze(addresses); +} + function decodeEmptyHostInput(value: unknown, label: string): HostStatusInput { requireExactRecord(value, label, []); return {}; } function decodeHostStatusResult(value: unknown): HostStatusResult { + const valueRecord = requireRecord(value, 'host.status result'); const record = requireExactRecord(value, 'host.status result', [ 'hostEpoch', 'compositionId', @@ -114,6 +138,7 @@ function decodeHostStatusResult(value: unknown): HostStatusResult { 'connections', 'activeOperations', 'activeResidencies', + ...(valueRecord.peerEndpoint === undefined ? [] : ['peerEndpoint']), ]); return decodeHostStatusFields(record); } @@ -124,6 +149,7 @@ function decodeHostDiagnosticsResult(value: unknown): HostDiagnosticsResult { 'host.diagnostics.query result', HOST_DIAGNOSTICS_RESULT_MAX_BYTES, ); + const valueRecord = requireRecord(value, 'host.diagnostics.query result'); const record = requireExactRecord(value, 'host.diagnostics.query result', [ 'hostEpoch', 'compositionId', @@ -132,6 +158,7 @@ function decodeHostDiagnosticsResult(value: unknown): HostDiagnosticsResult { 'connections', 'activeOperations', 'activeResidencies', + ...(valueRecord.peerEndpoint === undefined ? [] : ['peerEndpoint']), 'compositionModules', 'residencies', 'protocolVersion', @@ -262,6 +289,25 @@ function decodeHostStatusFields(record: Record): HostStatusResu connections: requireCount(record.connections, 'connections'), activeOperations: requireCount(record.activeOperations, 'activeOperations'), activeResidencies: requireCount(record.activeResidencies, 'activeResidencies'), + ...(record.peerEndpoint === undefined + ? {} + : { peerEndpoint: decodeHostPeerEndpoint(record.peerEndpoint) }), + }; +} + +function decodeHostPeerEndpoint(value: unknown): HostPeerEndpoint { + const record = requireExactRecord(value, 'Runtime Host peer endpoint', [ + 'peerId', + 'routeHints', + 'coordinationRelays', + ]); + return { + peerId: requireString(record.peerId, 'Runtime Host peer id', HOST_PEER_ID_MAX_BYTES), + routeHints: decodePeerAddresses(record.routeHints, 'Runtime Host peer route hints'), + coordinationRelays: decodePeerAddresses( + record.coordinationRelays, + 'Runtime Host peer coordination relays', + ), }; } diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index a173fdbc4a..438e906ecb 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -100,7 +100,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 90 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 91 as const; +// 91: Host status publishes the live Direct peer endpoint so newly issued +// connection invitations do not preserve stale startup routes. // 90: `session.create.mode` accepts the Bot session mode. A Host that predates // it rejects the value as an invalid Session start mode. // 89: The Host refreshes its models.dev catalog at startup and announces the diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 1f6f900805..f87902936e 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -71,6 +71,7 @@ export type { HostDiagnosticsResult, HostActivitySnapshot, HostLifecycleState, + HostPeerEndpoint, HostStatusInput, HostStatusResult, HostUpgradePrepareInput, diff --git a/packages/runtime-host/src/server/execution-service.ts b/packages/runtime-host/src/server/execution-service.ts index 1ae258c3ff..baf69cdbc4 100644 --- a/packages/runtime-host/src/server/execution-service.ts +++ b/packages/runtime-host/src/server/execution-service.ts @@ -99,6 +99,9 @@ export async function startExecutionRuntimeHostService( ...options.peer, dataRoot: options.peer.meshDataRoot, endpointKind: 'host', + onBackgroundReconcileError: (error) => { + console.error('[runtime-host] Peer Mesh background synchronization failed:', error); + }, }); } catch (error) { console.error( diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index e86705ce28..491e663212 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -825,6 +825,7 @@ export class RuntimeHostKernel { } #statusSnapshot(): HostStatusResult { + const peer = this.peerListeners[0]; return { hostEpoch: this.hostEpoch, compositionId: this.compositionDescriptor.id, @@ -833,6 +834,15 @@ export class RuntimeHostKernel { connections: this.#acceptedTransports.size, activeOperations: this.#activeOperations, activeResidencies: this.#residencies.activeCount, + ...(peer + ? { + peerEndpoint: { + peerId: peer.peerId, + routeHints: peer.listenAddresses, + coordinationRelays: peer.coordinationRelays, + }, + } + : {}), }; } diff --git a/packages/runtime-host/src/server/listener-set.ts b/packages/runtime-host/src/server/listener-set.ts index c26508e581..cdc48047e5 100644 --- a/packages/runtime-host/src/server/listener-set.ts +++ b/packages/runtime-host/src/server/listener-set.ts @@ -46,11 +46,13 @@ export interface RuntimeHostPeerListener extends RuntimeHostListener { readonly kind: 'libp2p_direct'; readonly peerId: string; readonly listenAddresses: readonly string[]; + readonly coordinationRelays: readonly string[]; } export interface RuntimeHostPeerListenerDescriptor { readonly peerId: string; readonly listenAddresses: readonly string[]; + readonly coordinationRelays: readonly string[]; } export type RuntimeHostListenerKind = 'local_ipc' | 'websocket' | 'libp2p_direct'; @@ -135,14 +137,17 @@ export function createRuntimeHostListenerSet( .filter((listener) => listener.kind === 'websocket') .map((listener) => listener.endpoint), ), - peerListeners: Object.freeze( - additional.filter(isRuntimeHostPeerListener).map((listener) => - Object.freeze({ - peerId: listener.peerId, - listenAddresses: Object.freeze([...listener.listenAddresses]), - }), - ), - ), + get peerListeners() { + return Object.freeze( + additional.filter(isRuntimeHostPeerListener).map((listener) => + Object.freeze({ + peerId: listener.peerId, + listenAddresses: Object.freeze([...listener.listenAddresses]), + coordinationRelays: Object.freeze([...listener.coordinationRelays]), + }), + ), + ); + }, closeAdmission: () => settleListeners(listeners, (listener) => listener.closeAdmission()), cleanup: () => settleListeners([...listeners].reverse(), (listener) => listener.cleanup()), }; diff --git a/packages/runtime-host/src/server/peer-listener.ts b/packages/runtime-host/src/server/peer-listener.ts index 750f40bcce..3907669793 100644 --- a/packages/runtime-host/src/server/peer-listener.ts +++ b/packages/runtime-host/src/server/peer-listener.ts @@ -120,6 +120,10 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { .catch(captureFailure); } + get coordinationRelays(): readonly string[] { + return this.#client.identity().coordinationRelays; + } + closeAdmission(): Promise { this.#closeAdmissionTask ??= (async () => { this.#admitting = false; From 3702a51f1468adaee3b4d7f941143433b80052b5 Mon Sep 17 00:00:00 2001 From: Wang Date: Tue, 1 Sep 2026 16:41:06 +0800 Subject: [PATCH 02/14] fix(peer): expose live relay routes to shared sessions --- ...untime-host-collaboration-ipc-main.test.ts | 38 ++++++++++- .../runtime-host-collaboration-ipc-main.ts | 7 +++ apps/desktop/src/preload/bridge-contract.d.ts | 9 ++- .../locales/session-collaboration-copy.ts | 12 ++++ .../renderer/session-collaboration-dialog.tsx | 24 ++++++- .../execution-service-listeners.test.ts | 63 +++++++++++++++++++ .../src/server/execution-service.ts | 10 ++- 7 files changed, 156 insertions(+), 7 deletions(-) create mode 100644 packages/runtime-host/src/__tests__/execution-service-listeners.test.ts diff --git a/apps/desktop/src/main/__tests__/runtime-host-collaboration-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-collaboration-ipc-main.test.ts index 2863fec8e2..3596921c8f 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-collaboration-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-collaboration-ipc-main.test.ts @@ -29,7 +29,7 @@ import { registerRuntimeHostCollaborationIpc } from '../runtime-host-collaborati const ROOT_ID = 'a'.repeat(64); -test('requires Owner confirmation before issuing a plaintext collaboration invitation', async () => { +test('requires plaintext confirmation and reports the issued invitation routes', async () => { const handlers = new Map(); const ipcMain: ReconnectableReadIpcMain = { handle(channel, listener) { @@ -92,9 +92,43 @@ test('requires Owner confirmation before issuing a plaintext collaboration invit assert.equal(prepareCalls, 1); assert.equal((result as { kind?: unknown }).kind, 'prepared'); const invitation = (result as { - invitation: { invitationCode: string }; + invitation: { invitationCode: string; connectivity: unknown }; }).invitation; + assert.deepEqual(invitation.connectivity, { kind: 'configured' }); const bundle = decodeDesktopCollaborationInvitation(invitation.invitationCode); assert.equal(decodeCollaborationInvitationCode(bundle.invitationCode).rootId, ROOT_ID); assert.equal(bundle.target.transport.kind, 'plaintext'); + + const peerHandlers = new Map(); + registerRuntimeHostCollaborationIpc( + client as unknown as Parameters[0], + { + handle(channel, listener) { + peerHandlers.set(channel, listener); + }, + }, + async () => ({ + name: 'Peer Lab', + transport: { + kind: 'libp2p-direct', + peerId: '12D3KooWpeer', + routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRelays: [ + '/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay', + ], + }, + }), + ); + const preparePeer = peerHandlers.get('session-collaboration:prepare'); + assert.ok(preparePeer); + const peerResult = await preparePeer( + {} as Parameters[0], + 'session-1', + 'observe', + false, + ); + assert.deepEqual( + (peerResult as { invitation: { connectivity: unknown } }).invitation.connectivity, + { kind: 'peer', coordinationRelayCount: 1 }, + ); }); diff --git a/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts b/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts index 622cb75c7a..9c1ae80b7b 100644 --- a/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts @@ -70,6 +70,13 @@ export function registerRuntimeHostCollaborationIpc( invitationCode: prepared.invitationCode, target, }), + connectivity: + target.transport.kind === 'libp2p-direct' + ? { + kind: 'peer' as const, + coordinationRelayCount: target.transport.coordinationRelays.length, + } + : { kind: 'configured' as const }, }, }; }, diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 2bc4ce1aad..88b052f0b6 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -345,7 +345,14 @@ export type DesktopSessionCollaborationCancelResult = SessionCollaborationCancel export type DesktopSessionCollaborationPrepareResult = | { readonly kind: 'prepared'; - readonly invitation: CollaborationInvitationPrepareResult; + readonly invitation: CollaborationInvitationPrepareResult & { + readonly connectivity: + | { + readonly kind: 'peer'; + readonly coordinationRelayCount: number; + } + | { readonly kind: 'configured' }; + }; } | { readonly kind: 'insecure_confirmation_required' }; diff --git a/apps/desktop/src/renderer/locales/session-collaboration-copy.ts b/apps/desktop/src/renderer/locales/session-collaboration-copy.ts index 1ba3fe444d..9b0cf33a53 100644 --- a/apps/desktop/src/renderer/locales/session-collaboration-copy.ts +++ b/apps/desktop/src/renderer/locales/session-collaboration-copy.ts @@ -37,6 +37,12 @@ const ZH = { createInvitation: '创建邀请', invitationCode: '一次性邀请码', invitationHelp: '邀请码包含连接地址和访客凭据，不包含所有者凭据。', + coordinationReady: '已包含跨网络协调路径', + coordinationReadyBody: + 'Runtime Host 已取得 Relay reservation；这能提高不同网络中的设备成功建立连接的机会。', + coordinationUnavailable: '尚未取得 Relay reservation', + coordinationUnavailableBody: + '这枚邀请码当前只包含直接路径，不同网络中的设备可能无法连接。可稍后撤销并重新创建。', copy: '复制邀请码', copied: '邀请码已复制', close: '完成', @@ -109,6 +115,12 @@ const EN = { createInvitation: 'Create invitation', invitationCode: 'One-time invitation code', invitationHelp: 'The code contains the connection address and Guest credential, never the Owner credential.', + coordinationReady: 'Cross-network coordination is included', + coordinationReadyBody: + 'The Runtime Host has obtained a Relay reservation, improving the chance that devices on different networks can connect.', + coordinationUnavailable: 'No Relay reservation yet', + coordinationUnavailableBody: + 'This invitation currently contains direct routes only, so devices on different networks may not connect. Revoke and recreate it later.', copy: 'Copy invitation', copied: 'Invitation copied', close: 'Done', diff --git a/apps/desktop/src/renderer/session-collaboration-dialog.tsx b/apps/desktop/src/renderer/session-collaboration-dialog.tsx index a207762c5f..7f17533cf3 100644 --- a/apps/desktop/src/renderer/session-collaboration-dialog.tsx +++ b/apps/desktop/src/renderer/session-collaboration-dialog.tsx @@ -32,10 +32,10 @@ import { } from '@maka/ui'; import type { CollaborationAccessQueryResult, - CollaborationInvitationPrepareResult, SessionCollaborationGrant, SessionTurnAccessRequest, } from '@maka/runtime-host/protocol'; +import type { DesktopSessionCollaborationPrepareResult } from '../preload/bridge-contract.js'; import { getSessionCollaborationCopy } from './locales/session-collaboration-copy.js'; import { turnRequestStateLabel } from './session-turn-request-composer.js'; @@ -54,6 +54,11 @@ type CollaborationAuthorityState = | 'remote_access_off' | 'unavailable'; +type PreparedInvitation = Extract< + DesktopSessionCollaborationPrepareResult, + { readonly kind: 'prepared' } +>['invitation']; + export function SessionCollaborationDialog(props: Props) { return ; } @@ -63,7 +68,7 @@ function ShareSessionDialog(props: Props) { const toast = useToast(); const [preset, setPreset] = useState<'observe' | 'request_turn'>('observe'); const [access, setAccess] = useState(); - const [invitation, setInvitation] = useState(); + const [invitation, setInvitation] = useState(); const [turnRequests, setTurnRequests] = useState(); const [authorityState, setAuthorityState] = useState('loading'); const [working, setWorking] = useState(false); @@ -269,6 +274,21 @@ function ShareSessionDialog(props: Props) { onChange={() => undefined} /> {copy.invitationHelp} + {invitation.connectivity.kind === 'peer' ? ( + invitation.connectivity.coordinationRelayCount > 0 ? ( + + ) : ( + + ) + ) : null}