diff --git a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts index 27f630c2f9..f9e3819a7a 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts @@ -388,7 +388,6 @@ test('drives the renderer Session execution facade through real UDS framing', as { client, observer, - observations: observer, attachmentApprovals: createAttachmentApprovalRegistry(), emitSessionsChanged() {}, stat: async () => ({ size: 0 }), 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 new file mode 100644 index 0000000000..98061fddd2 --- /dev/null +++ b/apps/desktop/src/main/__tests__/runtime-host-collaboration-ipc-main.test.ts @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + decodeCollaborationInvitationCode, + encodeCollaborationInvitationCode, +} from '@maka/runtime-host/protocol'; +import type { IpcHandler, ReconnectableReadIpcMain } from '../ipc-reconnect-policy.js'; +import { decodeDesktopCollaborationInvitation } from '../runtime-host-collaboration-invitation.js'; +import { registerRuntimeHostCollaborationIpc } from '../runtime-host-collaboration-ipc-main.js'; + +const ROOT_ID = 'a'.repeat(64); + +test('requires Owner confirmation before issuing a plaintext collaboration invitation', async () => { + const handlers = new Map(); + const ipcMain: ReconnectableReadIpcMain = { + handle(channel, listener) { + handlers.set(channel, listener); + }, + }; + let prepareCalls = 0; + const client = { + async prepareCollaborationInvitation(sessionId: string, grantKinds: readonly string[]) { + prepareCalls += 1; + assert.equal(sessionId, 'session-1'); + assert.deepEqual(grantKinds, ['session_observation']); + return { + invitationCode: encodeCollaborationInvitationCode({ + schemaVersion: 1, + rootId: ROOT_ID, + credential: 'guest-token', + }), + principalId: 'guest-1', + expiresAt: '2026-08-31T00:00:00.000Z', + grants: [], + }; + }, + async queryCollaborationAccess() { + return { principals: [], grants: [] }; + }, + async revokeCollaborationPrincipal() { + return { revoked: false }; + }, + }; + registerRuntimeHostCollaborationIpc( + client as unknown as Parameters[0], + ipcMain, + async () => ({ + name: 'Lab', + transport: { + kind: 'plaintext', + url: 'ws://runtime.example.com', + acknowledgement: 'plaintext-bearer-v1', + }, + }), + ); + const prepare = handlers.get('session-collaboration:prepare'); + assert.ok(prepare); + + assert.deepEqual(await prepare({} as Parameters[0], 'session-1', false), { + kind: 'insecure_confirmation_required', + }); + assert.equal(prepareCalls, 0); + + const result = await prepare({} as Parameters[0], 'session-1', true); + assert.equal(prepareCalls, 1); + assert.equal((result as { kind?: unknown }).kind, 'prepared'); + const invitation = (result as { + invitation: { invitationCode: string }; + }).invitation; + const bundle = decodeDesktopCollaborationInvitation(invitation.invitationCode); + assert.equal(decodeCollaborationInvitationCode(bundle.invitationCode).rootId, ROOT_ID); + assert.equal(bundle.target.transport.kind, 'plaintext'); +}); diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index 031c2b9363..ae2ec73bb5 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -23,6 +23,7 @@ import test from 'node:test'; import type { IpcMain } from 'electron'; import type { BotIncomingMessage, BotRegistry } from '@maka/runtime/bots'; import type { ComputerUseToolSet } from '@maka/runtime/computer-use-tools'; +import type { ShellRunUpdate } from '@maka/core/events'; import type { MakaTool } from '@maka/runtime/tool-runtime'; import type { ClientCapabilityProvider, @@ -183,6 +184,68 @@ test('owns one complete Desktop candidate generation and can restart cleanly', a assert.equal(ipc.size, 0); }); +test('registers only shared observation IPC and consumes scoped catalog changes for a Guest', async () => { + const ipc = ipcHarness(); + const sharedResource = sharedShellRunUpdate('session-guest'); + const host = connectionHarness('guest', { runtimeResourceUpdate: sharedResource }); + const changes: Array<{ reason: string; sessionId?: string }> = []; + const rendererEvents: Array<{ channel: string; payload: unknown }> = []; + const candidate = await createCandidate( + host.connection, + { + ...deps(ipc), + emitSessionsChanged: (_scope, reason, sessionId) => { + changes.push({ reason, ...(sessionId === undefined ? {} : { sessionId }) }); + }, + renderer: { + send(channel, _scope, payload) { + rendererEvents.push({ channel, payload }); + }, + }, + }, + undefined, + 'external', + 'remote', + 'session_guest', + ); + + assert.deepEqual( + ((await ipc.invoke('sessions:list')) as SessionCatalogProjection[]).map(({ id }) => id), + ['session-guest'], + ); + assert.equal(ipc.channels.includes('sessions:observe'), true); + assert.equal(ipc.channels.includes('sessions:transcript:open'), true); + assert.equal(ipc.channels.includes('sessions:send'), false); + assert.equal(ipc.channels.includes('sessions:stop'), false); + assert.equal(ipc.channels.includes('tasks:list'), false); + assert.equal(ipc.channels.includes('attachments:readBytes'), true); + assert.deepEqual(await ipc.invoke('shell-runs:list', 'session-guest'), [sharedResource]); + assert.equal(ipc.channels.includes('shell-runs:attach'), false); + await ipc.invoke('sessions:observe', 'session-guest', 'guest-observer'); + host.pushSubscriptionFrame({ + kind: 'subscription.session_domain_changed', + hostEpoch: 'host-guest', + subscriptionId: 'subscription-guest', + sequence: 1, + sessionId: 'session-guest', + domain: 'runtime_resource', + resources: [ + { sourceSessionId: 'session-guest', ref: sharedResource.result.ref }, + ], + }); + await waitFor(() => + rendererEvents.some( + ({ channel, payload }) => + channel === 'shell-runs:update' && + (payload as ShellRunUpdate).result.ref === sharedResource.result.ref, + ), + ); + host.publishSessionCatalogChange('session-guest'); + assert.deepEqual(changes, [{ reason: 'updated', sessionId: 'session-guest' }]); + + await candidate.close(); +}); + test('rejects a stale Host identity when raw Session IDs collide', async () => { const ipc = ipcHarness(); const browserReleased: string[] = []; @@ -778,6 +841,48 @@ test('retries candidate startup when a restored observation cannot seed', async await observations.close(); }); +test('drops a stale shared Session observation when Guest access is gone', async () => { + const observations = new RuntimeHostSessionObservationRegistry(); + const firstIpc = ipcHarness(); + const firstHost = connectionHarness('shared-before-revoke', { + sessionId: 'session-1', + subscriptionSnapshot: continuitySnapshot(), + }); + const firstCandidate = await createCandidate( + firstHost.connection, + deps(firstIpc), + observations, + 'external', + 'remote', + 'session_guest', + ); + await firstIpc.invoke('sessions:observe', 'session-1', 'observer-1'); + await firstCandidate.close(); + + const changes: Array<{ reason: string; sessionId?: string }> = []; + const revokedHost = connectionHarness('shared-after-revoke', { + sharedSessionAvailable: false, + }); + const candidate = await createCandidate( + revokedHost.connection, + { + ...deps(ipcHarness()), + emitSessionsChanged: (_scope, reason, sessionId) => { + changes.push({ reason, ...(sessionId === undefined ? {} : { sessionId }) }); + }, + }, + observations, + 'external', + 'remote', + 'session_guest', + ); + + assert.deepEqual(observations.trackedSessionIds(), []); + assert.deepEqual(changes, [{ reason: 'deleted', sessionId: 'session-1' }]); + await candidate.close(); + await observations.close(); +}); + type IpcHandler = Parameters['handle']>[1]; function ipcHarness(onSend?: (channel: string, payload: unknown) => void) { @@ -898,6 +1003,8 @@ function connectionHarness( activeAssistantStreams?: readonly SessionAssistantStreamIdentity[]; subscriptionError?: Error; runtimeResourcePty?: ReturnType; + runtimeResourceUpdate?: ShellRunUpdate; + sharedSessionAvailable?: boolean; } = {}, ) { let resolveClosed: (() => void) | undefined; @@ -909,6 +1016,7 @@ function connectionHarness( resolveTurnStarted = resolve; }); const closeSubscriptions = new Set<() => void>(); + const sessionCatalogListeners = new Set<(frame: { sessionId: string }) => void>(); let provider: ClientCapabilityProvider | undefined; let capabilityRegistrations = 0; let capabilityUnregistrations = 0; @@ -934,6 +1042,21 @@ function connectionHarness( nextCursor: null, }; } + if (operation === 'session.shared.query') { + if (options.sharedSessionAvailable === false) return { session: null }; + const id = options.sessionId ?? `session-${label}`; + return { + session: { + kind: 'shared_session', + id, + revision: 1, + createdAt: 1, + activityAt: 1, + name: `Session ${label}`, + status: 'idle', + }, + }; + } if (operation === 'session.create') { return session((input as { sessionId: string }).sessionId); } @@ -967,11 +1090,23 @@ function connectionHarness( }; } if (operation === 'runtime.resource.query') { + const query = input as { kind: string; sessionId: string; ref?: string }; + if (query.kind === 'get') { + return { + kind: 'resource', + sessionId: query.sessionId, + revision: catalogRevision(`${label}-resource`), + resource: + options.runtimeResourceUpdate?.result.ref === query.ref + ? options.runtimeResourceUpdate + : null, + }; + } return { kind: 'page', - sessionId: (input as { sessionId: string }).sessionId, + sessionId: query.sessionId, revision: catalogRevision(`${label}-resource`), - resources: [], + resources: options.runtimeResourceUpdate ? [options.runtimeResourceUpdate] : [], nextCursor: null, }; } @@ -1052,6 +1187,10 @@ function connectionHarness( capabilityUnregistrations += 1; return { registrationId: `registration-${label}`, revision: 2 }; }, + subscribeSessionCatalogChanges: (listener: (frame: { sessionId: string }) => void) => { + sessionCatalogListeners.add(listener); + return () => sessionCatalogListeners.delete(listener); + }, close: async () => { closeCalls += 1; for (const closeSubscription of closeSubscriptions) closeSubscription(); @@ -1074,6 +1213,9 @@ function connectionHarness( assert.ok(activeSubscriptionFrames); activeSubscriptionFrames.push(frame); }, + publishSessionCatalogChange: (sessionId: string) => { + for (const listener of sessionCatalogListeners) listener({ sessionId }); + }, get capabilityRegistrations() { return capabilityRegistrations; }, @@ -1160,6 +1302,34 @@ function ptySnapshot(ref: string, buffer: string) { }; } +function sharedShellRunUpdate(sessionId: string): ShellRunUpdate { + return { + sessionId, + ownership: { kind: 'local' }, + sourceTurnId: 'turn-shared', + sourceToolCallId: 'tool-shared', + result: { + kind: 'shell_run', + ref: 'maka://runtime/background-tasks/shared', + mode: 'pipes', + status: 'running', + cwd: '/workspace', + cmd: 'echo shared', + startedAt: 1, + updatedAt: 1, + revision: 1, + output: { + mode: 'pipes', + stdout: 'shared output', + stderr: '', + stdoutTruncated: false, + stderrTruncated: false, + redacted: false, + }, + }, + }; +} + class AsyncFrameQueue implements AsyncIterable { readonly #frames: SubscriptionFrame[] = []; readonly #waiters: Array< 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 8af1443e71..83040e9f66 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 @@ -35,6 +35,7 @@ import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, + encodeCollaborationInvitationCode, } from "@maka/runtime-host/protocol"; import { RuntimeHostPairingFinalizationInterruptedError, @@ -49,6 +50,7 @@ import { createDesktopRuntimeHostProfileService, resolveDesktopRuntimeHostStartup, } from "../runtime-host-profile-service.js"; +import { encodeDesktopCollaborationInvitation } from '../runtime-host-collaboration-invitation.js'; const ROOT_ID = "a".repeat(64); const PROFILE = { @@ -501,6 +503,102 @@ test("keeps a separate profile when the same Host is paired through another conn assert.equal((await catalog.resolve("replacement")).credential, "new-token"); }); +test('imports shared access without requiring or persisting an Owner credential', async () => { + const root = await clientRoot(); + const catalog = createClientRuntimeHostProfileCatalog(root); + const startup = await resolveDesktopRuntimeHostStartup(root, { catalog }); + const connected: ResolvedRuntimeHostProfile[] = []; + const finalized: string[] = []; + const service = createDesktopRuntimeHostProfileService({ + clientDataRoot: root, + startup, + catalog, + states: () => [connectingLocal()], + enable: async (target) => { + connected.push(target); + }, + disable: async () => undefined, + setDefault: () => undefined, + finalizePairing: async (profileId) => { + finalized.push(profileId); + }, + }); + + const result = await service.importCollaborationInvitation( + encodeDesktopCollaborationInvitation({ + invitationCode: encodeCollaborationInvitationCode({ + schemaVersion: 1, + rootId: ROOT_ID, + credential: 'guest-token', + }), + target: { + name: PROFILE.name, + transport: PROFILE.transport, + }, + }), + false, + ); + + assert.equal(result.kind, 'connected'); + if (result.kind !== 'connected') return; + assert.equal((await catalog.read()).profiles.length, 1); + const sharedProfileId = connected[0]?.profile.id; + assert.ok(sharedProfileId); + const shared = await catalog.resolve(sharedProfileId); + assert.equal(shared.profile.kind, 'remote'); + assert.equal(shared.profile.kind === 'remote' ? shared.profile.access : undefined, 'session_guest'); + assert.equal(shared.credential, 'guest-token'); + assert.deepEqual(finalized, [sharedProfileId]); +}); + +test('requires explicit confirmation before importing plaintext shared access', async () => { + const root = await clientRoot(); + const catalog = createClientRuntimeHostProfileCatalog(root); + const startup = await resolveDesktopRuntimeHostStartup(root, { catalog }); + const connected: ResolvedRuntimeHostProfile[] = []; + const service = createDesktopRuntimeHostProfileService({ + clientDataRoot: root, + startup, + catalog, + states: () => [connectingLocal()], + enable: async (target) => { + connected.push(target); + }, + disable: async () => undefined, + setDefault: () => undefined, + finalizePairing: async () => undefined, + }); + + const code = encodeDesktopCollaborationInvitation({ + invitationCode: encodeCollaborationInvitationCode({ + schemaVersion: 1, + rootId: ROOT_ID, + credential: 'guest-token', + }), + target: { + name: 'Lab', + transport: { + kind: 'plaintext', + url: 'ws://runtime.example.com', + acknowledgement: 'plaintext-bearer-v1', + }, + }, + }); + assert.deepEqual(await service.importCollaborationInvitation(code, false), { + kind: 'error', + reason: 'insecure_confirmation_required', + }); + assert.equal(connected.length, 0); + + const result = await service.importCollaborationInvitation(code, true); + assert.equal(result.kind, 'connected'); + assert.equal(connected[0]?.profile.kind, 'remote'); + assert.equal( + connected[0]?.profile.kind === 'remote' ? connected[0].profile.transport.kind : undefined, + 'plaintext', + ); +}); + test('classifies connection-code failures without exposing transport errors to the renderer', async () => { const root = await clientRoot(); const startup = await resolveDesktopRuntimeHostStartup(root); diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts index d2c3400c29..957a930072 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts @@ -21,7 +21,28 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import type { SessionCatalogProjection } from '@maka/runtime-host/protocol'; import type { IpcHandler } from '../ipc-reconnect-policy.js'; -import { registerRuntimeHostSessionCatalogIpc } from '../runtime-host-session-catalog-ipc-main.js'; +import { + registerRuntimeHostSessionCatalogIpc, + registerRuntimeHostSharedSessionCatalogIpc, +} from '../runtime-host-session-catalog-ipc-main.js'; + +test('registers a read-only Session catalog for shared access', async () => { + const handlers = new Map(); + registerRuntimeHostSharedSessionCatalogIpc( + { getSession: async () => ({ id: 'shared' }) as never }, + { + handle: (channel, listener) => handlers.set(channel, listener), + handleReconnectableRead: (channel, listener) => handlers.set(channel, listener), + }, + ); + + assert.deepEqual([...handlers.keys()], ['sessions:list']); + assert.deepEqual(await handlers.get('sessions:list')!({} as never), [{ id: 'shared' }]); + assert.deepEqual( + await handlers.get('sessions:list')!({} as never, { subagentParentSessionId: 'parent' }), + [], + ); +}); test('projects observed running Turn identities into renderer Session lists', async () => { const handlers = new Map(); diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts index 8bdf462bed..711c22fb2f 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts @@ -38,14 +38,22 @@ import { createAttachmentApprovalRegistry } from "../attachment-approval.js"; import type { DesktopRuntimeHostSession } from "../runtime-host-client.js"; import { registerRuntimeHostSessionExecutionIpc, + registerRuntimeHostSessionObservationIpc, type RuntimeHostSessionExecutionIpcDeps, } from "../runtime-host-session-execution-ipc-main.js"; +import { RuntimeHostSessionObservationRegistry } from '../runtime-host-session-observation-registry.js'; import { RuntimeHostSessionObserver } from "../runtime-host-session-observer.js"; import { runtimeHostSessionFixture } from "./runtime-host-session-test-fixture.js"; test('registers Session observation as one reconnectable operation', () => { const ipc = ipcHarness(); - registerExecutionIpc({ client: executionClient({}) }, ipc); + registerRuntimeHostSessionObservationIpc( + { + observations: new RuntimeHostSessionObservationRegistry(), + resolveSideConversation: async () => false, + }, + ipc, + ); assert.equal(ipc.reconnectableChannels.has('sessions:observe'), true); }); @@ -1681,7 +1689,6 @@ function registerExecutionIpc( resizeImage: async (bytes) => bytes, beforeStop() {}, ...deps, - observations: deps.observations ?? observer, sessionCopyCleanup: deps.sessionCopyCleanup ?? unusedSessionCopyCleanup(), onBackgroundError: deps.onBackgroundError ?? (() => undefined), }, diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index b8c7e90dc5..4af7966a1b 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -1631,9 +1631,10 @@ test("abandons a watched Turn when the initial Host subscription fails", async ( await observer.close(); }); -test("abandons a watched Turn when the Session is removed", async () => { +test("abandons a watched Turn and removes it from the catalog when Guest access ends", async () => { const events = new AsyncFrameQueue(); const finishedTurns: Array<[string, "completed" | "abandoned"]> = []; + const sessionChanges: string[] = []; let closeCount = 0; const observer = new RuntimeHostSessionObserver({ client: { @@ -1648,7 +1649,9 @@ test("abandons a watched Turn when the Session is removed", async () => { }, }), }, - emitSessionsChanged() {}, + emitSessionsChanged(reason) { + sessionChanges.push(reason); + }, onWatchedTurnFinished: (sessionId, outcome) => { finishedTurns.push([sessionId, outcome]); }, @@ -1660,11 +1663,12 @@ test("abandons a watched Turn when the Session is removed", async () => { hostEpoch: "host-1", subscriptionId: "subscription-1", sequence: 1, - reason: "session_removed", + reason: "access_revoked", }); await waitFor(() => closeCount === 1); assert.deepEqual(finishedTurns, [["session-1", "abandoned"]]); + assert.deepEqual(sessionChanges, ["deleted"]); await observer.close(); }); diff --git a/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts b/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts index dc8ed70083..4032293ef4 100644 --- a/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-artifacts-ipc-main.ts @@ -44,6 +44,11 @@ interface RuntimeHostArtifactsIpcDeps { readonly presentationRoot?: string; } +type RuntimeHostAttachmentPreviewIpcDeps = Pick< + RuntimeHostArtifactsIpcDeps, + 'ipcMain' | 'client' +>; + const ATTACHMENT_PREVIEW_LIMIT_EXCEEDED = Symbol("attachment-preview-limit-exceeded"); export function registerRuntimeHostArtifactsIpc( @@ -92,52 +97,7 @@ export function registerRuntimeHostArtifactsIpc( return result; }, ); - handleReconnectableRead( - deps.ipcMain, - "attachments:readBytes", - async (_event, sessionId: string, artifactId: string) => { - const artifact = await deps.client.getArtifact(sessionId, artifactId); - if ( - !artifact || - artifact.status === "deleted" - ) { - return { ok: false as const, reason: "not_found" }; - } - const preview = resolveArtifactImagePreview(artifact); - if (preview.kind === "unsupported") { - return { - ok: false as const, - reason: preview.reason === "oversize" ? "too_large" : "unsupported_mime", - }; - } - const mimeType = normalizeArtifactImagePreviewMime(artifact.mimeType, artifact.name); - if (!mimeType) return { ok: false as const, reason: "unsupported_mime" }; - const chunks: Buffer[] = []; - let received = 0; - try { - await deps.client.streamArtifact(sessionId, artifactId, async (chunk) => { - received += chunk.byteLength; - if (received > ARTIFACT_IMAGE_PREVIEW_MAX_BYTES) { - throw ATTACHMENT_PREVIEW_LIMIT_EXCEEDED; - } - chunks.push(Buffer.from(chunk)); - }); - } catch (error) { - if (error === ATTACHMENT_PREVIEW_LIMIT_EXCEEDED) { - return { ok: false as const, reason: "too_large" }; - } - throw error; - } - if (received !== artifact.sizeBytes) { - return { ok: false as const, reason: "read_failed" }; - } - return { - ok: true as const, - base64: Buffer.concat(chunks, received).toString("base64"), - mimeType, - }; - }, - ); + registerRuntimeHostAttachmentPreviewIpc(deps); deps.ipcMain.handle( "app:openArtifactPath", async (_event, sessionId: string, artifactId: string) => { @@ -192,6 +152,58 @@ export function registerRuntimeHostArtifactsIpc( ); } +/** Guest-safe projection used by transcript attachment thumbnails. */ +export function registerRuntimeHostAttachmentPreviewIpc( + deps: RuntimeHostAttachmentPreviewIpcDeps, +): void { + handleReconnectableRead( + deps.ipcMain, + "attachments:readBytes", + async (_event, sessionId: string, artifactId: string) => { + const artifact = await deps.client.getArtifact(sessionId, artifactId); + if ( + !artifact || + artifact.status === "deleted" + ) { + return { ok: false as const, reason: "not_found" }; + } + const preview = resolveArtifactImagePreview(artifact); + if (preview.kind === "unsupported") { + return { + ok: false as const, + reason: preview.reason === "oversize" ? "too_large" : "unsupported_mime", + }; + } + const mimeType = normalizeArtifactImagePreviewMime(artifact.mimeType, artifact.name); + if (!mimeType) return { ok: false as const, reason: "unsupported_mime" }; + const chunks: Buffer[] = []; + let received = 0; + try { + await deps.client.streamArtifact(sessionId, artifactId, async (chunk) => { + received += chunk.byteLength; + if (received > ARTIFACT_IMAGE_PREVIEW_MAX_BYTES) { + throw ATTACHMENT_PREVIEW_LIMIT_EXCEEDED; + } + chunks.push(Buffer.from(chunk)); + }); + } catch (error) { + if (error === ATTACHMENT_PREVIEW_LIMIT_EXCEEDED) { + return { ok: false as const, reason: "too_large" }; + } + throw error; + } + if (received !== artifact.sizeBytes) { + return { ok: false as const, reason: "read_failed" }; + } + return { + ok: true as const, + base64: Buffer.concat(chunks, received).toString("base64"), + mimeType, + }; + }, + ); +} + async function materializeArtifact( client: DesktopRuntimeHostClient, sessionId: string, diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index e002cadeb8..8212b199c0 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -929,6 +929,8 @@ runtimeHostManager = await startRuntimeHostDesktopManager( registerClientIpc: registerHostClientIpc, openSshTunnel: runtimeHostSshTerminal.openSshTunnel, activateSshOperator: runtimeHostSshTerminal.activateSshOperator, + resolveLocalCollaborationConnectionTarget: () => + localRuntimeHostRemoteAccess.createCollaborationConnectionTarget(), }, { upgradePrompts: createRuntimeHostUpgradePrompts( diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts index 22aaf6b55a..74ef6579d8 100644 --- a/apps/desktop/src/main/runtime-host-client.ts +++ b/apps/desktop/src/main/runtime-host-client.ts @@ -104,6 +104,11 @@ import { type ScheduledTaskChangedFrame, type SessionCatalogItem, type SessionCatalogProjection, + type SharedSessionCatalogProjection, + type CollaborationAccessQueryResult, + type CollaborationInvitationPrepareResult, + type CollaborationPrincipalRevokeResult, + type SessionCollaborationGrantKind, type SessionConfigurationPatch, type SessionAssistantStreamIdentity, type SessionContinuitySnapshot, @@ -271,6 +276,26 @@ export class DesktopRuntimeHostClient { return this.request('access.credential.finalize', {}, timeoutMs); } + prepareCollaborationInvitation( + sessionId: string, + grantKinds: readonly SessionCollaborationGrantKind[], + ): Promise { + return this.request('collaboration.invitation.prepare', { sessionId, grantKinds }); + } + + queryCollaborationAccess(sessionId?: string): Promise { + return this.request( + 'collaboration.access.query', + sessionId === undefined ? {} : { sessionId }, + ); + } + + revokeCollaborationPrincipal( + principalId: string, + ): Promise { + return this.request('collaboration.principal.revoke', { principalId }); + } + subscribeConfigurationChanges(listener: (revision: number) => void): () => void { this.#assertOpen(); return this.connection.subscribeConfigurationChanges(listener); @@ -572,6 +597,11 @@ export class DesktopRuntimeHostClient { } } + async getSharedSession(): Promise { + this.#assertOpen(); + return (await this.request('session.shared.query', {})).session; + } + async listProjects( includeLocations = true, ): Promise<(ProjectCatalogProject | ProjectCatalogProjectDetails)[]> { diff --git a/apps/desktop/src/main/runtime-host-collaboration-invitation.ts b/apps/desktop/src/main/runtime-host-collaboration-invitation.ts new file mode 100644 index 0000000000..7b9890c4fc --- /dev/null +++ b/apps/desktop/src/main/runtime-host-collaboration-invitation.ts @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + decodeRemoteRuntimeHostProfile, + type RuntimeHostRemoteTransport, +} from '@maka/runtime-host/client'; +import { decodeCollaborationInvitationCode } from '@maka/runtime-host/protocol'; + +const SCHEMA_VERSION = 1; +const CODE_MAX_BYTES = 32 * 1024; + +export interface DesktopCollaborationConnectionTarget { + readonly name: string; + readonly transport: RuntimeHostRemoteTransport; +} + +export interface DesktopCollaborationInvitation { + readonly invitationCode: string; + readonly target: DesktopCollaborationConnectionTarget; +} + +export function encodeDesktopCollaborationInvitation( + value: DesktopCollaborationInvitation, +): string { + const invitation = decodeCollaborationInvitationCode(value.invitationCode); + const target = decodeTarget(value.target, invitation.rootId); + return Buffer.from( + JSON.stringify({ + schemaVersion: SCHEMA_VERSION, + invitationCode: value.invitationCode, + target, + }), + 'utf8', + ).toString('base64url'); +} + +export function decodeDesktopCollaborationInvitation( + code: string, +): DesktopCollaborationInvitation { + if (!code || Buffer.byteLength(code, 'utf8') > CODE_MAX_BYTES) { + throw new Error('Invalid Desktop collaboration invitation'); + } + let value: unknown; + try { + value = JSON.parse(Buffer.from(code, 'base64url').toString('utf8')) as unknown; + } catch { + throw new Error('Invalid Desktop collaboration invitation'); + } + if (!isRecord(value) || !hasExactKeys(value, ['schemaVersion', 'invitationCode', 'target'])) { + throw new Error('Invalid Desktop collaboration invitation'); + } + if (value.schemaVersion !== SCHEMA_VERSION || typeof value.invitationCode !== 'string') { + throw new Error('Unsupported Desktop collaboration invitation'); + } + const invitation = decodeCollaborationInvitationCode(value.invitationCode); + return { + invitationCode: value.invitationCode, + target: decodeTarget(value.target, invitation.rootId), + }; +} + +function decodeTarget(value: unknown, rootId: string): DesktopCollaborationConnectionTarget { + if (!isRecord(value) || !hasExactKeys(value, ['name', 'transport'])) { + throw new Error('Invalid Desktop collaboration connection target'); + } + const profile = decodeRemoteRuntimeHostProfile({ + id: 'collaboration-target', + name: value.name, + kind: 'remote', + rootId, + transport: value.transport, + }); + return { + name: profile.name, + transport: profile.transport, + }; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function hasExactKeys(value: Record, expected: readonly string[]): boolean { + const keys = Object.keys(value).sort(); + return keys.length === expected.length && expected.slice().sort().every((key, index) => key === keys[index]); +} diff --git a/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts b/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts new file mode 100644 index 0000000000..0377498090 --- /dev/null +++ b/apps/desktop/src/main/runtime-host-collaboration-ipc-main.ts @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { DesktopRuntimeHostClient } from './runtime-host-client.js'; +import { + encodeDesktopCollaborationInvitation, + type DesktopCollaborationConnectionTarget, +} from './runtime-host-collaboration-invitation.js'; +import { + handleReconnectableRead, + type ReconnectableReadIpcMain, +} from './ipc-reconnect-policy.js'; + +export function registerRuntimeHostCollaborationIpc( + client: Pick< + DesktopRuntimeHostClient, + | 'prepareCollaborationInvitation' + | 'queryCollaborationAccess' + | 'revokeCollaborationPrincipal' + >, + ipcMain: ReconnectableReadIpcMain, + resolveConnectionTarget: () => + | DesktopCollaborationConnectionTarget + | Promise, +): void { + ipcMain.handle( + 'session-collaboration:prepare', + async (_event, sessionId: unknown, allowInsecure: unknown) => { + const target = await resolveConnectionTarget(); + if (target.transport.kind === 'plaintext' && allowInsecure !== true) { + return { kind: 'insecure_confirmation_required' } as const; + } + const prepared = await client.prepareCollaborationInvitation( + requiredId(sessionId, 'Session'), + ['session_observation'], + ); + return { + kind: 'prepared', + invitation: { + ...prepared, + invitationCode: encodeDesktopCollaborationInvitation({ + invitationCode: prepared.invitationCode, + target, + }), + }, + }; + }, + ); + handleReconnectableRead( + ipcMain, + 'session-collaboration:getAccess', + (_event, sessionId: unknown) => + client.queryCollaborationAccess(requiredId(sessionId, 'Session')), + ); + ipcMain.handle( + 'session-collaboration:revokePrincipal', + (_event, principalId: unknown) => + client.revokeCollaborationPrincipal(requiredId(principalId, 'Principal')), + ); +} + +function requiredId(value: unknown, label: string): string { + if ( + typeof value !== 'string' || + value.length === 0 || + value.length > 512 || + /[\u0000-\u001f\u007f]/u.test(value) + ) { + throw new Error(`Invalid ${label} identity`); + } + return value; +} diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index 03ce20e79e..51ff440b9f 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -22,6 +22,7 @@ import type { IpcMain } from "electron"; import type { ActiveInteractionRequestEvent } from '@maka/core/events'; import { redactSecrets } from '@maka/core/redaction'; import type { CreateSessionRequestInput } from '@maka/core/runtime-inputs'; +import { isSideConversationSession } from '@maka/core/side-conversation'; import type { SessionChangedEvent, SessionChangedReason } from '@maka/core/session'; import type { BotRegistry } from '@maka/runtime/bots'; import { @@ -38,6 +39,8 @@ import { type RuntimeHostCandidateLaunchBarrier, type RuntimeHostSpawnedProcess, type PersistedRuntimeHostProfile, + runtimeHostProfileAccess, + type RuntimeHostProfileAccess, type CandidateExitDetails, } from "@maka/runtime-host/client"; import type { RuntimeHostActivationResult } from "@maka/runtime-host/operator"; @@ -66,16 +69,28 @@ import { type DesktopNativeCapabilityProvider, type DesktopNativeCapabilityProviderInput, } from "./runtime-host-native-capabilities.js"; -import { registerRuntimeHostSessionCatalogIpc } from "./runtime-host-session-catalog-ipc-main.js"; +import { + registerRuntimeHostSharedSessionCatalogIpc, + registerRuntimeHostSessionCatalogIpc, + toDesktopHostSharedSessionSummary, +} from "./runtime-host-session-catalog-ipc-main.js"; import { registerRuntimeHostWorkHubIpc } from "./runtime-host-workhub-ipc-main.js"; import { registerRuntimeHostExternalSessionsIpc } from "./runtime-host-external-sessions-ipc-main.js"; +import { registerRuntimeHostCollaborationIpc } from './runtime-host-collaboration-ipc-main.js'; +import type { DesktopCollaborationConnectionTarget } from './runtime-host-collaboration-invitation.js'; +import { registerRuntimeHostAttachmentPreviewIpc } from './runtime-host-artifacts-ipc-main.js'; import { registerRuntimeHostSessionDomainsIpc, type RuntimeHostSessionDomainsIpcDeps, type RuntimeHostSessionDomainsIpcHandle, } from "./runtime-host-session-domains-ipc-main.js"; +import { + registerRuntimeHostShellRunQueriesIpc, + type RuntimeHostShellRunQueriesIpcHandle, +} from './runtime-host-shell-runs-ipc-main.js'; import { registerRuntimeHostSessionExecutionIpc, + registerRuntimeHostSessionObservationIpc, type RuntimeHostSessionExecutionIpcDeps, } from "./runtime-host-session-execution-ipc-main.js"; import { RuntimeHostSessionObservationRegistry } from "./runtime-host-session-observation-registry.js"; @@ -134,6 +149,8 @@ export interface DesktopRuntimeHostCandidateDeps { readonly activateSshOperator?: ( input: RuntimeHostSshOperatorActivationInput, ) => Promise; + readonly resolveLocalCollaborationConnectionTarget?: () => + Promise; readonly createSessionCopyCleanup: (input: { removeSession: (sessionId: string) => Promise; resumeSessionCopy: (input: { @@ -157,6 +174,7 @@ export interface DesktopRuntimeHostCandidateDeps { export interface DesktopRuntimeHostTargetPolicy { readonly kind: RuntimeHostProfileKind; readonly rootId: string; + readonly access: RuntimeHostProfileAccess; } export interface DesktopRuntimeHostCandidateControls { @@ -326,6 +344,7 @@ export async function startDesktopRuntimeHostCandidate( ? 'owned_ephemeral' : 'supervised', "local", + 'owner', connection.registration.pid, connection.spawnedProcess, ), @@ -336,6 +355,16 @@ export async function startDesktopRuntimeHostCandidate( } } +function noGuestBotService(): BotIncomingMainService { + return { + handleBotIncomingMessage: async () => { + throw new Error('Session Guest profiles cannot receive bot messages'); + }, + invalidateSessionBindings: () => undefined, + close: async () => undefined, + }; +} + function observeLocalRuntimeHostProcess(spawnedProcess: RuntimeHostSpawnedProcess | undefined): void { if (!spawnedProcess) return; void spawnedProcess.exited.then( @@ -422,6 +451,15 @@ async function startProfileDesktopRuntimeHostCandidate( observationRegistry, 'external', profileTarget.profile.kind, + runtimeHostProfileAccess(profileTarget.profile), + undefined, + undefined, + profileTarget.profile.kind === 'remote' + ? { + name: profileTarget.profile.name, + transport: profileTarget.profile.transport, + } + : undefined, ), }; } catch (error) { @@ -436,12 +474,15 @@ export async function createDesktopRuntimeHostCandidate( observationRegistry: RuntimeHostSessionObservationRegistry | undefined, hostOwnership: DesktopRuntimeHostOwnership, targetKind: DesktopRuntimeHostTargetPolicy["kind"], + targetAccess: RuntimeHostProfileAccess = 'owner', hostPid?: number, ownedProcess?: RuntimeHostSpawnedProcess, + collaborationConnectionTarget?: DesktopCollaborationConnectionTarget, ): Promise { const target: DesktopRuntimeHostTargetPolicy = { kind: targetKind, rootId: connection.rootId, + access: targetAccess, }; const ipcMain = deps.ipcMain; const scope = { hostId: connection.rootId, targetEpoch: ipcMain.epoch }; @@ -527,6 +568,7 @@ export async function createDesktopRuntimeHostCandidate( }; let observer: RuntimeHostSessionObserver | undefined; let closeSessionDomains: (() => Promise) | undefined; + let sharedShellRuns: RuntimeHostShellRunQueriesIpcHandle | undefined; let disposeClientIpc: (() => void | Promise) | undefined; let observationsAttached = false; let capabilitiesRegistered = false; @@ -545,39 +587,78 @@ export async function createDesktopRuntimeHostCandidate( client, emitSessionsChanged: (reason, sessionId, extra) => emitSessionsChanged(reason, sessionId, extra), - emitSessionDomainChanged: (change) => domains?.sessionDomainChanged(change), + emitSessionDomainChanged: (change) => + target.access === 'session_guest' + ? sharedShellRuns?.sessionDomainChanged(change) + : domains?.sessionDomainChanged(change), emitRuntimeResourcePtyData: (event) => domains?.runtimeResourcePtyData(event), emitAgentGraphChanged: (event) => domains?.agentGraphChanged(event), emitActiveInteractionsChanged, emitSubscriptionRecovered: (sessionId) => - domains?.sessionSubscriptionRecovered(sessionId), + target.access === 'session_guest' + ? sharedShellRuns?.sessionSubscriptionRecovered(sessionId) + : domains?.sessionSubscriptionRecovered(sessionId), emitObservationSeed: (sessionId, phase) => sendToRenderer?.('sessions:observation-seed', { sessionId, phase }), - onWatchedTurnFinished: (sessionId, outcome) => - outcome === "completed" - ? deps.completeComputerUseTurn( - desktopSessionResourceKey({ ...scope, sessionId }), - ) - : deps.nativeCapabilities.releaseComputerUseSession( - desktopSessionResourceKey({ ...scope, sessionId }), - ), + ...(target.access === 'owner' + ? { + onWatchedTurnFinished: (sessionId: string, outcome: 'completed' | 'abandoned') => + outcome === 'completed' + ? deps.completeComputerUseTurn( + desktopSessionResourceKey({ ...scope, sessionId }), + ) + : deps.nativeCapabilities.releaseComputerUseSession( + desktopSessionResourceKey({ ...scope, sessionId }), + ), + } + : {}), recoverConnectionClosed: observationRegistry !== undefined, ...(deps.now ? { now: deps.now } : {}), }); observer = sessionObserver; - domains = registerRuntimeHostSessionDomainsIpc( + if (target.access === 'owner') { + domains = registerRuntimeHostSessionDomainsIpc( + { + client, + sessionObserver, + emitModeChanged, + ...(deps.renderer ? { sendToRenderer } : {}), + ...(deps.onError ? { onError: reportError } : {}), + ...(deps.newId ? { newId: deps.newId } : {}), + ...(deps.now ? { now: deps.now } : {}), + }, + ipc, + ); + closeSessionDomains = domains.close; + } else { + sharedShellRuns = registerRuntimeHostShellRunQueriesIpc( + { client, sendToRenderer, onError: reportError }, + ipc, + ); + } + registerRuntimeHostSessionObservationIpc( { - client, - sessionObserver, - emitModeChanged, - ...(deps.renderer ? { sendToRenderer } : {}), - ...(deps.onError ? { onError: reportError } : {}), - ...(deps.newId ? { newId: deps.newId } : {}), - ...(deps.now ? { now: deps.now } : {}), + observations: sessionObservations, + resolveSideConversation: async (sessionId) => { + if (target.access === 'session_guest') return false; + const session = await client.getSession(sessionId); + if (!session) throw new Error(`Runtime Host Session not found: ${sessionId}`); + return isSideConversationSession(session.labels); + }, }, ipc, ); - closeSessionDomains = domains.close; + if (target.access === 'session_guest') { + const trackedSessionIds = sessionObservations.trackedSessionIds(); + if (trackedSessionIds.length > 0) { + const sharedSessionId = (await client.getSharedSession())?.id; + for (const sessionId of trackedSessionIds) { + if (sessionId === sharedSessionId) continue; + await sessionObservations.forgetSession(sessionId); + emitSessionsChanged('deleted', sessionId); + } + } + } const observedSessionIds = sessionObservations.observedSessionIds(); for (const sessionId of observedSessionIds) { sendToRenderer('sessions:observation-seed', { sessionId, phase: 'pending' }); @@ -610,7 +691,7 @@ export async function createDesktopRuntimeHostCandidate( sendToRenderer('sessions:observation-seed', { sessionId, phase: 'ready' }); emitSessionsChanged("message-appended", sessionId); emitSessionsChanged("goal-change", sessionId); - domains.sessionSubscriptionRecovered(sessionId); + domains?.sessionSubscriptionRecovered(sessionId); emitActiveInteractionsChanged( sessionId, sessionObserver.listActiveInteractions(sessionId) ?? [], @@ -642,13 +723,15 @@ export async function createDesktopRuntimeHostCandidate( providers.add(provider); return provider; }; - const nativeCapabilities = createNativeProvider(); - if ( - nativeCapabilities.offers().length > 0 || - (nativeCapabilities.services?.().length ?? 0) > 0 - ) { - await client.replaceClientCapabilities(nativeCapabilities); - capabilitiesRegistered = true; + if (target.access === 'owner') { + const nativeCapabilities = createNativeProvider(); + if ( + nativeCapabilities.offers().length > 0 || + (nativeCapabilities.services?.().length ?? 0) > 0 + ) { + await client.replaceClientCapabilities(nativeCapabilities); + capabilitiesRegistered = true; + } } let capabilityRefresh = Promise.resolve(); const refreshClientCapabilities = (): Promise => { @@ -666,96 +749,131 @@ export async function createDesktopRuntimeHostCandidate( }); return capabilityRefresh; }; - const sessionCopyCleanup = deps.createSessionCopyCleanup({ - removeSession: async (sessionId) => { - const disposition = await client.removeSessionCopy(sessionId); - if (disposition === "retained") return disposition; - await releaseNativeSession(sessionId).catch(reportError); - emitSessionsChanged("deleted", sessionId); - return disposition; - }, - resumeSessionCopy: async ({ sessionId, kind, sourceSessionId, sourceTurnId, intent }) => { - await client.copySession(kind, { - sourceSessionId, - targetSessionId: sessionId, - sourceTurnId, - ...(intent ? { intent } : {}), - }); - }, - }); - const registeredClientIpc = deps.registerClientIpc?.( - client, - ipc, - { refreshClientCapabilities }, - target, - scope, - isTargetActive, - ); - disposeClientIpc = - typeof registeredClientIpc === "function" + const sessionCopyCleanup = target.access === 'owner' + ? deps.createSessionCopyCleanup({ + removeSession: async (sessionId) => { + const disposition = await client.removeSessionCopy(sessionId); + if (disposition === "retained") return disposition; + await releaseNativeSession(sessionId).catch(reportError); + emitSessionsChanged("deleted", sessionId); + return disposition; + }, + resumeSessionCopy: async ({ sessionId, kind, sourceSessionId, sourceTurnId, intent }) => { + await client.copySession(kind, { + sourceSessionId, + targetSessionId: sessionId, + sourceTurnId, + ...(intent ? { intent } : {}), + }); + }, + }) + : undefined; + const registeredClientIpc = target.access === 'owner' + ? deps.registerClientIpc?.( + client, + ipc, + { refreshClientCapabilities }, + target, + scope, + isTargetActive, + ) + : undefined; + disposeClientIpc = target.access === 'session_guest' + ? client.subscribeSessionCatalogChanges(({ sessionId }) => + emitSessionsChanged('updated', sessionId), + ) + : typeof registeredClientIpc === 'function' ? registeredClientIpc : undefined; - registerRuntimeHostSessionCatalogIpc( - { - client, - runningTurnIds: (sessionId) => sessionObserver.observedRunningTurnIds(sessionId), - resolveCreateProject: (input) => deps.resolveSessionCreateProject(input, target), - emitSessionsChanged, - releaseSessionResources: releaseNativeSession, - sessionCopyCleanup, - ...(deps.newId ? { newId: deps.newId } : {}), - }, - ipc, - ); - registerRuntimeHostWorkHubIpc(client, ipc, { - resolveCreateProject: () => deps.resolveSessionCreateProject({}, target), - emitSessionsChanged, - }); - registerRuntimeHostExternalSessionsIpc( - { - client, - emitSessionsChanged, - }, - ipc, - ); - const stopSession = registerRuntimeHostSessionExecutionIpc( - { - client, - observer: sessionObserver, - observations: sessionObservations, - attachmentApprovals: deps.attachmentApprovals, - emitSessionsChanged, - stat: deps.stat, - resizeImage: deps.resizeImage, - beforeStop: (sessionId) => - deps.nativeCapabilities.releaseComputerUseSession( - desktopSessionResourceKey({ ...scope, sessionId }), - ), - sessionCopyCleanup, - onBackgroundError: reportError, - ...(deps.e2eInteractions - ? { e2eInteractions: deps.e2eInteractions } - : {}), - ...(deps.newId ? { newId: deps.newId } : {}), - }, - ipc, - ); - const botIncoming = createBotIncomingMainService({ - botRegistry: deps.botRegistry, - sessions: createRuntimeHostBotSessionAdapter({ - client, - resolveCreateTarget: () => deps.resolveBotCreateTarget(target), + if (target.access === 'session_guest') { + registerRuntimeHostAttachmentPreviewIpc({ ipcMain: ipc, client }); + registerRuntimeHostSharedSessionCatalogIpc( + { + getSession: async () => { + const session = await client.getSharedSession(); + return session ? toDesktopHostSharedSessionSummary(session) : null; + }, + }, + ipc, + ); + } else { + if (!sessionCopyCleanup) throw new Error('Owner Session copy authority is unavailable'); + registerRuntimeHostSessionCatalogIpc( + { + client, + runningTurnIds: (sessionId) => sessionObserver.observedRunningTurnIds(sessionId), + resolveCreateProject: (input) => deps.resolveSessionCreateProject(input, target), + emitSessionsChanged, + releaseSessionResources: releaseNativeSession, + sessionCopyCleanup, + ...(deps.newId ? { newId: deps.newId } : {}), + }, + ipc, + ); + } + if (target.access === 'owner') { + registerRuntimeHostCollaborationIpc(client, ipc, async () => { + if (collaborationConnectionTarget) return collaborationConnectionTarget; + if (target.kind === 'local' && deps.resolveLocalCollaborationConnectionTarget) { + return deps.resolveLocalCollaborationConnectionTarget(); + } + throw new Error('This Runtime Host does not have a shareable connection target'); + }); + registerRuntimeHostWorkHubIpc(client, ipc, { + resolveCreateProject: () => deps.resolveSessionCreateProject({}, target), emitSessionsChanged, - ...(deps.newId ? { newId: deps.newId } : {}), - }), - }); + }); + registerRuntimeHostExternalSessionsIpc( + { + client, + emitSessionsChanged, + }, + ipc, + ); + } + const stopSession = sessionCopyCleanup + ? registerRuntimeHostSessionExecutionIpc( + { + client, + observer: sessionObserver, + attachmentApprovals: deps.attachmentApprovals, + emitSessionsChanged, + stat: deps.stat, + resizeImage: deps.resizeImage, + beforeStop: (sessionId) => + deps.nativeCapabilities.releaseComputerUseSession( + desktopSessionResourceKey({ ...scope, sessionId }), + ), + sessionCopyCleanup, + onBackgroundError: reportError, + ...(deps.e2eInteractions + ? { e2eInteractions: deps.e2eInteractions } + : {}), + ...(deps.newId ? { newId: deps.newId } : {}), + }, + ipc, + ) + : async () => { + throw new Error('Shared Sessions cannot be stopped'); + }; + const botIncoming = target.access === 'owner' + ? createBotIncomingMainService({ + botRegistry: deps.botRegistry, + sessions: createRuntimeHostBotSessionAdapter({ + client, + resolveCreateTarget: () => deps.resolveBotCreateTarget(target), + emitSessionsChanged, + ...(deps.newId ? { newId: deps.newId } : {}), + }), + }) + : noGuestBotService(); return new DesktopRuntimeHostCandidateImpl({ client, observer: sessionObserver, ipc, botIncoming, closeNativeCapabilities, - closeSessionDomains: domains.close, + closeSessionDomains: domains?.close ?? (() => Promise.resolve()), disposeClientIpc, detachSessionObservations: () => sessionObservations.detach(sessionObserver), 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 49bc2b76a7..01e084c277 100644 --- a/apps/desktop/src/main/runtime-host-local-remote-access.ts +++ b/apps/desktop/src/main/runtime-host-local-remote-access.ts @@ -95,6 +95,15 @@ export interface DesktopRuntimeHostLocalManagementTarget export interface DesktopLocalRuntimeHostRemoteAccess { getSnapshot(): Promise; + createCollaborationConnectionTarget(): Promise<{ + readonly name: string; + readonly transport: { + readonly kind: 'libp2p-direct'; + readonly peerId: string; + readonly routeHints: readonly string[]; + readonly coordinationRelays: readonly string[]; + }; + }>; enable(value: unknown): Promise; disable(): Promise; uninstall(value: unknown): Promise<{ readonly kind: 'active_tasks' | 'uninstalled' }>; @@ -489,6 +498,19 @@ export function createDesktopLocalRuntimeHostRemoteAccess(input: { return issueConnectionCode(input.rootPath, managed.rootId, peer, localClient(input.manager)); }); + const createCollaborationConnectionTarget = () => + serialize(async () => { + const managed = requireManaged( + await readLifecycle(lifecyclePath, input.rootPath, input.rootId), + ); + const peer = await readPeer(input.operator, managed); + if (!peer) throw new Error('Remote access is not enabled on this computer'); + return { + name: hostName(), + transport: { kind: 'libp2p-direct' as const, ...peer }, + }; + }); + const revokeSharedAccess = (): Promise => serialize(async () => { const managed = requireManaged( @@ -653,6 +675,7 @@ export function createDesktopLocalRuntimeHostRemoteAccess(input: { return { getSnapshot, + createCollaborationConnectionTarget, enable, disable, uninstall, diff --git a/apps/desktop/src/main/runtime-host-profile-service.ts b/apps/desktop/src/main/runtime-host-profile-service.ts index 1747d1343f..dd91bf7eb6 100644 --- a/apps/desktop/src/main/runtime-host-profile-service.ts +++ b/apps/desktop/src/main/runtime-host-profile-service.ts @@ -37,6 +37,7 @@ import { type RuntimeHostProfileCatalog, } from "@maka/runtime-host/client"; import { runtimeHostAccessCredentialFingerprint } from "@maka/runtime-host/operator"; +import { decodeCollaborationInvitationCode } from '@maka/runtime-host/protocol'; import type { CredentialStore } from "@maka/storage/credential-store"; import { withFileUpdateLock } from "@maka/storage/file-update-lock"; import type { @@ -45,6 +46,7 @@ import type { DesktopRuntimeHostProfileEntry, DesktopRuntimeHostProfileSnapshot, DesktopRuntimeHostConnectionCodeImportResult, + DesktopSessionCollaborationImportResult, } from "../preload/bridge-contract.js"; import { RuntimeHostPairingFinalizationInterruptedError, @@ -66,6 +68,7 @@ import { type DesktopRuntimeHostManagedServiceBinding, type DesktopRuntimeHostManagedServiceStore, } from "./runtime-host-managed-services.js"; +import { decodeDesktopCollaborationInvitation } from './runtime-host-collaboration-invitation.js'; const PREFERENCES_SCHEMA_VERSION = 2; const PREFERENCES_FILE = "runtime-host-profile-selection.json"; @@ -99,6 +102,10 @@ export interface DesktopRuntimeHostProfileService { }, ): Promise<{ readonly profileId: string }>; importConnectionCode(code: string): Promise; + importCollaborationInvitation( + code: string, + allowInsecure: boolean, + ): Promise; resolveManagedService( profileId: string, ): Promise; @@ -203,9 +210,13 @@ export async function resolveDesktopRuntimeHostStartup( }; } const profileIds = new Set(document.profiles.map((profile) => profile.id)); + const defaultProfile = document.profiles.find( + (profile) => profile.id === preferences.defaultProfileId, + ); const defaultProfileId = preferences.defaultProfileId === LOCAL_RUNTIME_HOST_PROFILE.id || - profileIds.has(preferences.defaultProfileId) + (profileIds.has(preferences.defaultProfileId) && + !(defaultProfile?.kind === 'remote' && defaultProfile.access === 'session_guest')) ? preferences.defaultProfileId : LOCAL_RUNTIME_HOST_PROFILE.id; const enabledRemoteProfileIds = new Set( @@ -764,6 +775,39 @@ export function createDesktopRuntimeHostProfileService(input: { return { kind: 'error', reason: connectionCodeImportFailure(error) }; } }, + async importCollaborationInvitation(code, allowInsecure) { + let bundle; + let invitation; + try { + bundle = decodeDesktopCollaborationInvitation(code); + invitation = decodeCollaborationInvitationCode(bundle.invitationCode); + } catch { + return { kind: 'error', reason: 'invalid_code' }; + } + if (bundle.target.transport.kind === 'plaintext' && !allowInsecure) { + return { kind: 'error', reason: 'insecure_confirmation_required' }; + } + try { + await addAndEnableVerified({ + profile: { + id: `shared-${randomUUID()}`, + name: `${bundle.target.name} · Shared`, + kind: 'remote', + rootId: invitation.rootId, + transport: bundle.target.transport, + access: 'session_guest', + }, + credential: invitation.credential, + }); + return { kind: 'connected' }; + } catch (error) { + return { + kind: 'error', + reason: 'connection_failed', + message: asError(error).message, + }; + } + }, rotateManagedCredential(expected, credential) { return mutateProfiles(async () => { const profileId = expected.profile.id; @@ -1099,6 +1143,12 @@ export function createDesktopRuntimeHostProfileService(input: { ) { throw new Error("Enable a Runtime Host before making it the default"); } + if (profileId !== LOCAL_RUNTIME_HOST_PROFILE.id) { + const target = await catalog.resolve(profileId); + if (target.profile.kind === 'remote' && target.profile.access === 'session_guest') { + throw new Error('A shared Session connection cannot be the default Runtime Host'); + } + } const next = { ...preferences, defaultProfileId: profileId }; await persist(next); input.setDefault(profileId); @@ -1251,6 +1301,7 @@ export function registerDesktopRuntimeHostProfileIpc( "runtime-host-profiles:set-default", "runtime-host-profiles:remove", "runtime-host-profiles:resolve-pairing-recovery", + 'session-collaboration:import', ] as const; ipcMain.handle(channels[0], () => service.getSnapshot()); ipcMain.handle(channels[1], (_event, value: DesktopRuntimeHostProfileAddInput) => @@ -1263,6 +1314,9 @@ export function registerDesktopRuntimeHostProfileIpc( ipcMain.handle(channels[4], (_event, profileId: string) => service.setDefault(profileId)); ipcMain.handle(channels[5], (_event, profileId: string) => service.remove(profileId)); ipcMain.handle(channels[6], () => service.resolvePairingRecovery()); + ipcMain.handle(channels[7], (_event, code: string, allowInsecure: boolean) => + service.importCollaborationInvitation(code, allowInsecure), + ); return () => { for (const channel of channels) ipcMain.removeHandler(channel); }; diff --git a/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts index 1c9325ec0d..4abdaa5ff7 100644 --- a/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-catalog-ipc-main.ts @@ -27,6 +27,7 @@ import { type SessionChangedEvent, type SessionChangedReason, type SessionCatalo import { projectSessionCatalogSummary } from '@maka/runtime-host/client'; import type { SessionCatalogProjection, + SharedSessionCatalogProjection, SessionCreateInput, WorkspaceTarget, SessionModelTarget, @@ -59,6 +60,7 @@ type RuntimeHostSessionCatalogClient = Pick< export interface DesktopHostSessionSummary extends SessionCatalogSummary { labelsTruncated: boolean; + shared?: true; } export interface RuntimeHostSessionCatalogIpcDeps { @@ -78,6 +80,10 @@ export interface RuntimeHostSessionCatalogIpcDeps { newId?: () => string; } +export interface RuntimeHostSharedSessionCatalogIpcDeps { + getSession(): Promise; +} + export function registerRuntimeHostSessionCatalogIpc( deps: RuntimeHostSessionCatalogIpcDeps, ipcMain: ReconnectableReadIpcMain, @@ -221,6 +227,50 @@ export function registerRuntimeHostSessionCatalogIpc( }); } +export function registerRuntimeHostSharedSessionCatalogIpc( + deps: RuntimeHostSharedSessionCatalogIpcDeps, + ipcMain: ReconnectableReadIpcMain, +): void { + handleReconnectableRead(ipcMain, 'sessions:list', async (_event, filter?: unknown) => { + if (normalizeSessionListFilter(filter)?.subagentParentSessionId) return []; + const session = await deps.getSession(); + return session ? [session] : []; + }); +} + +export function toDesktopHostSharedSessionSummary( + session: SharedSessionCatalogProjection, +): DesktopHostSessionSummary { + return { + id: session.id, + name: session.name, + activityAt: session.activityAt, + isFlagged: false, + isArchived: false, + labels: [], + labelsTruncated: false, + hasUnread: false, + ...(session.lastMessageAt === undefined ? {} : { lastMessageAt: session.lastMessageAt }), + ...(session.lastMessagePreview === undefined + ? {} + : { lastMessagePreview: session.lastMessagePreview }), + status: session.status, + ...(session.liveRunState === undefined + ? {} + : { runningTurnIds: [...session.liveRunState.runningTurnIds] }), + ...(session.blockedReason === undefined ? {} : { blockedReason: session.blockedReason }), + ...(session.statusUpdatedAt === undefined + ? {} + : { statusUpdatedAt: session.statusUpdatedAt }), + backend: 'ai-sdk', + llmConnectionSlug: '', + connectionLocked: true, + model: '', + permissionMode: 'ask', + shared: true, + }; +} + /** * Reads the archived premise off the remove options. * diff --git a/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts index c0fd5aaa38..e08afc2397 100644 --- a/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-domains-ipc-main.ts @@ -46,6 +46,7 @@ import { type ReconnectableReadIpcMain, } from './ipc-reconnect-policy.js'; import { + registerRuntimeHostShellRunQueriesIpc, registerRuntimeHostShellRunsIpc, type RuntimeHostShellRunsClient, } from './runtime-host-shell-runs-ipc-main.js'; @@ -104,6 +105,14 @@ export function registerRuntimeHostSessionDomainsIpc( { client: deps.client, newId, sessionObserver: deps.sessionObserver }, ipcMain, ); + const shellRunQueries = registerRuntimeHostShellRunQueriesIpc( + { + client: deps.client, + sendToRenderer: deps.sendToRenderer, + onError: deps.onError, + }, + ipcMain, + ); handleReconnectableRead(ipcMain, 'tasks:list', (_event, sessionId: unknown) => deps.client.listTasks(requiredId(sessionId, 'Session')), @@ -340,7 +349,7 @@ export function registerRuntimeHostSessionDomainsIpc( deps.sendToRenderer?.('usage:changed', { sessionId: change.sessionId }); break; case 'runtime_resource': - void refreshRuntimeResources(deps, change.sessionId, change.resources); + shellRunQueries.sessionDomainChanged(change); break; } }; @@ -359,27 +368,12 @@ export function registerRuntimeHostSessionDomainsIpc( sessionDomainChanged({ sessionId, domain: 'plan' }); sessionDomainChanged({ sessionId, domain: 'usage' }); deps.sendToRenderer?.('graphs:resync', { rootSessionId: sessionId }); - deps.sendToRenderer?.('shell-runs:resync', { sessionId }); + shellRunQueries.sessionSubscriptionRecovered(sessionId); }, close: () => shellRuns.close(), }; } -async function refreshRuntimeResources( - deps: RuntimeHostSessionDomainsIpcDeps, - sessionId: string, - resources: readonly { ref: string }[], -): Promise { - for (const resource of resources) { - try { - const update = await deps.client.getRuntimeResource(sessionId, resource.ref); - if (update) deps.sendToRenderer?.('shell-runs:update', update); - } catch (error) { - deps.onError?.(error); - } - } -} - interface CanonicalGoalArmRequest { readonly sessionId: string; readonly condition: string; diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts index 51e656253e..9e7139044b 100644 --- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts @@ -20,11 +20,11 @@ import { randomUUID } from "node:crypto"; import type { IpcMainInvokeEvent } from "electron"; import { MAX_ATTACHMENT_COUNT } from '@maka/core/attachments'; +import { isSideConversationSession } from '@maka/core/side-conversation'; import { RuntimeHostOperationError, RuntimeHostRequestInterruptedError, } from '@maka/runtime-host/client'; -import { isSideConversationSession } from '@maka/core/side-conversation'; import { type SessionChangedEvent, type SessionChangedReason, @@ -145,13 +145,6 @@ async function submitMessageWithReconnect( export interface RuntimeHostSessionExecutionIpcDeps { client: RuntimeHostSessionExecutionClient; observer: RuntimeHostSessionObserver; - observations: Pick< - RuntimeHostSessionObservationRegistry, - | 'loadTranscriptAround' - | 'loadTranscriptBefore' - | 'observe' - | 'openTranscript' - >; attachmentApprovals: AttachmentApprovalRegistry; emitSessionsChanged: ( reason: SessionChangedReason, @@ -176,6 +169,58 @@ export interface RuntimeHostSessionExecutionIpcDeps { newId?: () => string; } +export interface RuntimeHostSessionObservationIpcDeps { + observations: Pick< + RuntimeHostSessionObservationRegistry, + | 'loadTranscriptAround' + | 'loadTranscriptBefore' + | 'observe' + | 'openTranscript' + >; + resolveSideConversation(sessionId: string): Promise; +} + +/** Register the complete Desktop surface available to an observation-only Session. */ +export function registerRuntimeHostSessionObservationIpc( + deps: RuntimeHostSessionObservationIpcDeps, + ipcMain: ReconnectableReadIpcMain, +): void { + handleReconnectableRead( + ipcMain, + 'sessions:observe', + async (event, sessionId: unknown, observerId: unknown) => { + const normalizedSessionId = requiredId(sessionId, 'Session'); + await deps.observations.observe( + normalizedSessionId, + requiredId(observerId, 'Session observer'), + event.sender as RuntimeHostSessionObserverTarget, + await deps.resolveSideConversation(normalizedSessionId), + ); + }, + ); + ipcMain.handle( + 'sessions:transcript:open', + async (event, sessionId: unknown, consumerId: unknown) => + deps.observations.openTranscript( + requiredId(sessionId, 'Session'), + requiredId(consumerId, 'Transcript consumer'), + event.sender as RuntimeHostTranscriptTarget, + ), + ); + ipcMain.handle('sessions:transcript:load-before', async (event, input: unknown) => { + await deps.observations.loadTranscriptBefore( + normalizeTranscriptRangeRequest(input), + event.sender.id, + ); + }); + ipcMain.handle('sessions:transcript:load-around', async (event, input: unknown) => { + await deps.observations.loadTranscriptAround( + normalizeTranscriptRangeRequest(input), + event.sender.id, + ); + }); +} + /** * Project Host-owned Session execution onto the Desktop renderer IPC contract. * The adapter owns client validation and presentation events, never Runtime @@ -220,47 +265,6 @@ export function registerRuntimeHostSessionExecutionIpc( }, ); - handleReconnectableRead( - ipcMain, - "sessions:observe", - async (event, sessionId: unknown, observerId: unknown) => { - const normalizedSessionId = requiredId(sessionId, "Session"); - const normalizedObserverId = requiredId(observerId, "Session observer"); - const session = await deps.client.getSession(normalizedSessionId); - if (!session) { - throw new Error(`Runtime Host Session not found: ${normalizedSessionId}`); - } - await deps.observations.observe( - normalizedSessionId, - normalizedObserverId, - event.sender as RuntimeHostSessionObserverTarget, - isSideConversationSession(session.labels), - ); - }, - ); - ipcMain.handle( - 'sessions:transcript:open', - async (event, sessionId: unknown, consumerId: unknown) => { - const result = await deps.observations.openTranscript( - requiredId(sessionId, 'Session'), - requiredId(consumerId, 'Transcript consumer'), - event.sender as RuntimeHostTranscriptTarget, - ); - return result; - }, - ); - ipcMain.handle('sessions:transcript:load-before', async (event, input: unknown) => { - await deps.observations.loadTranscriptBefore( - normalizeTranscriptRangeRequest(input), - event.sender.id, - ); - }); - ipcMain.handle('sessions:transcript:load-around', async (event, input: unknown) => { - await deps.observations.loadTranscriptAround( - normalizeTranscriptRangeRequest(input), - event.sender.id, - ); - }); handleReconnectableRead(ipcMain, 'sessions:listTurns', async (_event, sessionId: unknown) => deps.client.listSessionTurns(requiredId(sessionId, 'Session')), ); diff --git a/apps/desktop/src/main/runtime-host-session-observation-registry.ts b/apps/desktop/src/main/runtime-host-session-observation-registry.ts index 0e95057c59..289c154b63 100644 --- a/apps/desktop/src/main/runtime-host-session-observation-registry.ts +++ b/apps/desktop/src/main/runtime-host-session-observation-registry.ts @@ -209,6 +209,37 @@ export class RuntimeHostSessionObservationRegistry { return [...new Set([...this.#registrations.values()].map((registration) => registration.sessionId))]; } + trackedSessionIds(): string[] { + return [ + ...new Set([ + ...[...this.#registrations.values()].map((registration) => registration.sessionId), + ...[...this.#transcripts.values()].map((registration) => registration.sessionId), + ]), + ]; + } + + async forgetSession(sessionId: string): Promise { + const source = this.#source; + const observations = [...this.#registrations].filter( + ([, registration]) => registration.sessionId === sessionId, + ); + const transcripts = [...this.#transcripts].filter( + ([, registration]) => registration.sessionId === sessionId, + ); + for (const [observerId, registration] of observations) { + this.#deleteRegistration(observerId, registration); + } + for (const [consumerId, registration] of transcripts) { + this.#deleteTranscript(consumerId, registration); + } + if (source) { + await Promise.allSettled([ + ...observations.map(([observerId]) => source.unobserve(observerId)), + ...transcripts.map(([consumerId]) => source.closeTranscript?.(consumerId)), + ]); + } + } + detach(source: SessionObservationSource): void { if (this.#source === source) { this.#source = undefined; diff --git a/apps/desktop/src/main/runtime-host-session-subscription-owner.ts b/apps/desktop/src/main/runtime-host-session-subscription-owner.ts index 096683bf58..4f4b7ed1f8 100644 --- a/apps/desktop/src/main/runtime-host-session-subscription-owner.ts +++ b/apps/desktop/src/main/runtime-host-session-subscription-owner.ts @@ -404,18 +404,22 @@ export class RuntimeHostSessionSubscriptionOwner { } function subscriptionClosedError( - reason: "access_revoked" | "slow_consumer" | "session_removed", + reason: "slow_consumer" | "session_removed" | 'access_revoked', ): Error { - return reason !== "slow_consumer" - ? new SessionRemovedSubscriptionError( - reason === "access_revoked" - ? "Runtime Host Session access was revoked" - : "Runtime Host Session was removed while it was observed", - ) - : new RuntimeHostSubscriptionError( - "slow_consumer", - "Runtime Host Session subscription closed for a slow consumer", - ); + if (reason === 'session_removed') { + return new SessionRemovedSubscriptionError( + 'Runtime Host Session was removed while it was observed', + ); + } + if (reason === 'access_revoked') { + return new SessionRemovedSubscriptionError( + 'Access to the shared Runtime Host Session was revoked', + ); + } + return new RuntimeHostSubscriptionError( + 'slow_consumer', + 'Runtime Host Session subscription closed for a slow consumer', + ); } function isRecoverableSubscriptionFailure(error: unknown): boolean { diff --git a/apps/desktop/src/main/runtime-host-shell-runs-ipc-main.ts b/apps/desktop/src/main/runtime-host-shell-runs-ipc-main.ts index 7c92763735..fa7651209e 100644 --- a/apps/desktop/src/main/runtime-host-shell-runs-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-shell-runs-ipc-main.ts @@ -20,6 +20,7 @@ import { randomUUID } from 'node:crypto'; import type { ShellRunUpdate } from '@maka/core/events'; import type { ShellRunPtySnapshot } from '@maka/runtime/shell-run-contract'; +import type { SessionDomainChange } from '@maka/runtime-host/protocol'; import { RuntimeHostOperationError } from '@maka/runtime-host/client'; import { handleReconnectableRead, @@ -39,6 +40,38 @@ export type RuntimeHostShellRunsClient = Pick< | 'stopRuntimeResource' >; +export type RuntimeHostShellRunQueriesClient = Pick< + DesktopRuntimeHostClient, + 'getRuntimeResource' | 'listRuntimeResources' +>; + +export interface RuntimeHostShellRunQueriesIpcHandle { + sessionDomainChanged(change: SessionDomainChange): void; + sessionSubscriptionRecovered(sessionId: string): void; +} + +export function registerRuntimeHostShellRunQueriesIpc( + deps: { + client: RuntimeHostShellRunQueriesClient; + sendToRenderer?(channel: string, payload: unknown): void; + onError?(error: unknown): void; + }, + ipcMain: ReconnectableReadIpcMain, +): RuntimeHostShellRunQueriesIpcHandle { + handleReconnectableRead(ipcMain, 'shell-runs:list', (_event, sessionId: unknown) => + deps.client.listRuntimeResources(requiredId(sessionId, 'Session')), + ); + return { + sessionDomainChanged(change) { + if (change.domain !== 'runtime_resource') return; + void refreshRuntimeResources(deps, change.sessionId, change.resources); + }, + sessionSubscriptionRecovered(sessionId) { + deps.sendToRenderer?.('shell-runs:resync', { sessionId }); + }, + }; +} + export function registerRuntimeHostShellRunsIpc( deps: { client: RuntimeHostShellRunsClient; @@ -60,10 +93,6 @@ export function registerRuntimeHostShellRunsIpc( newId, deps.sessionObserver, ); - - handleReconnectableRead(ipcMain, 'shell-runs:list', (_event, sessionId: unknown) => - deps.client.listRuntimeResources(requiredId(sessionId, 'Session')), - ); ipcMain.handle('shell-runs:start', async (_event, sessionId: unknown) => { const normalizedSessionId = requiredId(sessionId, 'Session'); const started = await deps.client.startRuntimeResource({ @@ -95,6 +124,25 @@ export function registerRuntimeHostShellRunsIpc( return { close: () => controllers.close() }; } +async function refreshRuntimeResources( + deps: { + client: Pick; + sendToRenderer?(channel: string, payload: unknown): void; + onError?(error: unknown): void; + }, + sessionId: string, + resources: readonly { ref: string }[], +): Promise { + for (const resource of resources) { + try { + const update = await deps.client.getRuntimeResource(sessionId, resource.ref); + if (update) deps.sendToRenderer?.('shell-runs:update', update); + } catch (error) { + deps.onError?.(error); + } + } +} + interface RuntimeResourceIdentity { readonly sessionId: string; readonly ref: string; diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 94dcc8db3b..917aec24ba 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -109,6 +109,11 @@ import type { OperationOutcome, OperationOutput, } from '@maka/runtime-host/protocol'; +import type { + CollaborationAccessQueryResult, + CollaborationInvitationPrepareResult, + CollaborationPrincipalRevokeResult, +} from '@maka/runtime-host/protocol'; import type { AgentGraphEpochDirectory } from '@maka/runtime-host/client'; import type { RuntimeHostServiceManagementFrame, @@ -314,6 +319,24 @@ export interface DesktopRuntimeHostProfileSnapshot { readonly pairingRecoveryPending?: true; } +export type DesktopSessionCollaborationImportResult = + | { readonly kind: 'connected' } + | { + readonly kind: 'error'; + readonly reason: + | 'invalid_code' + | 'insecure_confirmation_required' + | 'connection_failed'; + readonly message?: string; + }; + +export type DesktopSessionCollaborationPrepareResult = + | { + readonly kind: 'prepared'; + readonly invitation: CollaborationInvitationPrepareResult; + } + | { readonly kind: 'insecure_confirmation_required' }; + export interface DesktopRuntimeHostRef { readonly profileId: string; readonly hostId: string; @@ -650,6 +673,22 @@ export interface DesktopSessionUsageSummary extends UsageSummaryV2 { } export interface MakaBridge { + sessionCollaboration: { + prepareInvitation( + sessionId: string, + allowInsecure?: boolean, + ): Promise; + getAccess(sessionId: string): Promise; + revokePrincipal( + sessionId: string, + principalId: string, + ): Promise; + importInvitation(input: { + readonly code: string; + readonly allowInsecure?: boolean; + }): Promise; + }; + runtimeHost: { query( operation: K, diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 0ad27fe096..158a8e1008 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -500,7 +500,13 @@ async function loadNewTaskCatalog(): Promise { ) as DesktopRuntimeHostProfileSnapshot; await runtimeHostScopeList(); const hosts = await Promise.all( - profiles.entries.filter((entry) => entry.enabled).map(async (entry): Promise => { + profiles.entries + .filter( + (entry) => + entry.enabled && + (entry.profile.kind !== 'remote' || entry.profile.access !== 'session_guest'), + ) + .map(async (entry): Promise => { if (entry.readiness !== 'ready' || !entry.hostId) { return { profile: entry.profile, @@ -546,7 +552,7 @@ async function loadNewTaskCatalog(): Promise { message: error instanceof Error ? error.message : String(error), }; } - }), + }), ); if (generation !== activeRuntimeHostGeneration) continue; return { defaultProfileId: profiles.defaultProfileId, hosts }; @@ -1173,6 +1179,36 @@ const browserSelection = createBrowserSelectionCoordinator(runtimeHostSessionRef const makaBridge = { runtimeHost, + sessionCollaboration: { + async prepareInvitation(sessionId, allowInsecure = false) { + const session = await runtimeHostSessionRef(sessionId); + return ipcRenderer.invoke( + 'session-collaboration:prepare', + session.scope, + session.sessionId, + allowInsecure, + ); + }, + async getAccess(sessionId) { + const session = await runtimeHostSessionRef(sessionId); + return ipcRenderer.invoke( + 'session-collaboration:getAccess', + session.scope, + session.sessionId, + ); + }, + async revokePrincipal(sessionId, principalId) { + const session = await runtimeHostSessionRef(sessionId); + return ipcRenderer.invoke( + 'session-collaboration:revokePrincipal', + session.scope, + principalId, + ); + }, + importInvitation({ code, allowInsecure = false }) { + return ipcRenderer.invoke('session-collaboration:import', code, allowInsecure); + }, + }, runtimeHostProfiles: { getSnapshot() { return ipcRenderer.invoke('runtime-host-profiles:getSnapshot'); diff --git a/apps/desktop/src/renderer/app-shell-effects.ts b/apps/desktop/src/renderer/app-shell-effects.ts index fa2fd09a1f..da3017850d 100644 --- a/apps/desktop/src/renderer/app-shell-effects.ts +++ b/apps/desktop/src/renderer/app-shell-effects.ts @@ -516,6 +516,7 @@ export function useActiveSessionEvents(options: { export function useShellRunUpdates(options: { activeId: string | undefined; + hydrate?: boolean; setShellRunUpdatesBySession: (updater: (current: ShellRunUpdatesBySession) => ShellRunUpdatesBySession) => void; }) { const applyUpdates = useEffectEvent( @@ -544,6 +545,7 @@ export function useShellRunUpdates(options: { let retryTimer: ReturnType | undefined; let retryDelayMs = 250; const hydration = new ShellRunHydration(); + if (options.hydrate === false) hydration.commit(0); const unsubscribe = window.maka.shellRuns.subscribeUpdates((update) => { if (disposed) return; const live = hydration.accept(update); @@ -554,6 +556,7 @@ export function useShellRunUpdates(options: { } }); const hydrate = (epoch: number) => { + if (options.hydrate === false) return; void window.maka.shellRuns .list(sessionId) .then((updates) => { @@ -577,7 +580,7 @@ export function useShellRunUpdates(options: { }); }; const unsubscribeResync = window.maka.shellRuns.subscribeResync((event) => { - if (disposed || event.sessionId !== sessionId) return; + if (disposed || options.hydrate === false || event.sessionId !== sessionId) return; const epoch = hydration.begin(); retryDelayMs = 250; if (retryTimer !== undefined) { @@ -586,14 +589,14 @@ export function useShellRunUpdates(options: { } hydrate(epoch); }); - hydrate(hydration.begin()); + if (options.hydrate !== false) hydrate(hydration.begin()); return () => { disposed = true; if (retryTimer !== undefined) globalThis.clearTimeout(retryTimer); unsubscribe(); unsubscribeResync(); }; - }, [options.activeId]); + }, [options.activeId, options.hydrate]); } export function useSessionEventHealthPolling(options: { diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx index e390ab2775..edb6e038fa 100644 --- a/apps/desktop/src/renderer/app-shell.tsx +++ b/apps/desktop/src/renderer/app-shell.tsx @@ -107,6 +107,9 @@ import { type TaskEntryError, } from './features/task-entry'; import { useNewTaskChoice } from './use-new-task-choice'; +import { SessionCollaborationDialog } from './session-collaboration-dialog'; +import { getSessionCollaborationCopy } from './locales/session-collaboration-copy'; +import { useSessionCollaborationDialog } from './use-session-collaboration-dialog'; import { NEW_TASK_PENDING_KEY } from './pending-items'; import { parseDesktopSlashCommand } from './desktop-slash-command'; import { @@ -345,6 +348,7 @@ function AppShellContent({ }) { const toastApi = useToast(); const [appUpdateStatus, setAppUpdateStatus] = useState(null); + const sharedSessionDialog = useSessionCollaborationDialog(); const updateInstallInFlightRef = useRef(false); const notifiedInstallErrorRef = useRef(null); const previousInterruptionShownRef = useRef(false); @@ -374,6 +378,10 @@ function AppShellContent({ setMessageLoadPending, sessionUiController, } = useAppShellSessionWorkspace(toastApi); + const activeCatalogSession = sessions.find((session) => session.id === activeId); + const sharedSessionActive = + (activeCatalogSession as DesktopSessionSummary | undefined)?.shared === true; + const ownerActiveId = activeCatalogSession && !sharedSessionActive ? activeId : undefined; const interactionHydrationEpochRef = useRef(new Map()); const markInteractionChanged = useCallback((sessionId: string) => { const epochs = interactionHydrationEpochRef.current; @@ -535,7 +543,8 @@ function AppShellContent({ const { memoryActive, refreshMemoryActive } = useShellMemoryPill({ toastApi, uiLocale, - sessionId: activeId, + sessionId: ownerActiveId, + disabled: sharedSessionActive, }); const newTaskHost = taskEntry.selectors.selectedHost ? { @@ -556,7 +565,7 @@ function AppShellContent({ const sessionHostConnections = useShellConnections({ toastApi, uiLocale, - target: { kind: 'session', sessionId: activeId }, + target: { kind: 'session', sessionId: ownerActiveId }, }); const startupConnectionSnapshot = initialOnboardingSnapshot ?? onboarding.mountedSnapshotHandoff; @@ -586,13 +595,13 @@ function AppShellContent({ return Promise.all([ defaultHostConnections.refreshConnections(), newTaskConnections.refreshConnections(), - ...(activeId ? [sessionHostConnections.refreshConnections()] : []), + ...(ownerActiveId ? [sessionHostConnections.refreshConnections()] : []), ]).then(() => undefined); } function handleConnectionEvent(event: ConnectionEvent): void { defaultHostConnections.handleConnectionEvent(event); newTaskConnections.handleConnectionEvent(event); - if (activeId) sessionHostConnections.handleConnectionEvent(event); + if (ownerActiveId) sessionHostConnections.handleConnectionEvent(event); } const onboardingState = onboarding.snapshot?.state; const onboardingSettled = hasSettledInitialOnboarding(onboarding.snapshot?.milestones ?? []); @@ -800,10 +809,10 @@ function AppShellContent({ resumePendingSessionId, resumeParkDescriptionBySession, resumeInterruptedSession, - } = useShellResume({ activeId, toastApi, shellCopy, uiLocale }); + } = useShellResume({ activeId: ownerActiveId, toastApi, shellCopy, uiLocale }); const rendererMountedRef = useRef(true); const goals = useGoalController({ - activeSessionId: activeId, + activeSessionId: ownerActiveId, reportError: showSessionError, }); // Set of session ids whose backend / connection is no longer usable — @@ -818,11 +827,11 @@ function AppShellContent({ }), [sessions, onboarding.snapshot?.sessionSendOutcomes], ); - const activeInteraction = activeInteractionFor(interactionBySession, activeId); + const activeInteraction = activeInteractionFor(interactionBySession, ownerActiveId); const activeSandboxBoundary = activeInteraction?.type === 'sandbox_boundary_request' ? activeInteraction : undefined; const activeQuestion = activeInteraction?.type === 'user_question_request' ? activeInteraction : undefined; - const activeSession = sessions.find((session) => session.id === activeId); + const activeSession = activeCatalogSession; const activeMessageQueue = activeId ? messageQueueBySession[activeId] : undefined; const activeMessageSubmitting = transientMessages.length > 0; const activeDesktopSession = activeSession; @@ -1220,32 +1229,32 @@ function AppShellContent({ unreadable: activeExecutionBoundaryUnreadable, reading: activeExecutionBoundaryReading, reload: reloadActiveExecutionBoundary, - } = useActiveExecutionBoundary(activeId, activeSessionForView?.permissionMode); + } = useActiveExecutionBoundary(ownerActiveId, activeSessionForView?.permissionMode); // The session view only subscribes to the session it shows, so a request // raised while another session was active never reaches this surface as a // live event — and neither does one raised before the window existed. The // runtime holds every unanswered request, so read them back whenever the // active session changes (#2072). useEffect(() => { - if (!activeId) return; + if (!ownerActiveId) return; let cancelled = false; - const hydrationEpoch = interactionHydrationEpochRef.current.get(activeId) ?? 0; + const hydrationEpoch = interactionHydrationEpochRef.current.get(ownerActiveId) ?? 0; void window.maka.sessions - .listActiveInteractions(activeId) + .listActiveInteractions(ownerActiveId) .then((requests) => { if ( cancelled || - (interactionHydrationEpochRef.current.get(activeId) ?? 0) !== hydrationEpoch + (interactionHydrationEpochRef.current.get(ownerActiveId) ?? 0) !== hydrationEpoch ) { return; } - sessionUiController.setInteractionBySession((current) => reconcileInteractions(current, activeId, requests)); + sessionUiController.setInteractionBySession((current) => reconcileInteractions(current, ownerActiveId, requests)); }) .catch(() => {}); return () => { cancelled = true; }; - }, [activeId, sessionUiController.setInteractionBySession]); + }, [ownerActiveId, sessionUiController.setInteractionBySession]); useEffect( () => window.maka.sessions.subscribeActiveInteractions(({ sessionId, interactions }) => { @@ -1262,7 +1271,7 @@ function AppShellContent({ activeId ? (activeSessionForView?.permissionMode ?? 'ask') : newTaskPermissionMode, ); const activePermissionMode = activeBoundarySurface.permissionMode; - const planMode = usePlanModeState(activeSessionForView); + const planMode = usePlanModeState(sharedSessionActive ? undefined : activeSessionForView); const planConversationItems = (planMode.state?.proposals ?? []).map((proposal) => ({ id: proposal.proposalId, afterTurnId: proposal.turnId, @@ -1446,10 +1455,10 @@ function AppShellContent({ } = useAppShellProjectContext({ uiLocale, rendererMountedRef, - sessionId: activeId, - sessionCwd: activeSession?.cwd, - sessionProjectId: activeSession?.projectId, - sessionProfileKind: activeDesktopSession?.profileKind, + sessionId: ownerActiveId, + sessionCwd: sharedSessionActive ? undefined : activeSession?.cwd, + sessionProjectId: sharedSessionActive ? undefined : activeSession?.projectId, + sessionProfileKind: sharedSessionActive ? undefined : activeDesktopSession?.profileKind, onProjectSelected: (ownerSessionId) => { void refreshProjectSkillsRef.current(); if (ownerSessionId && activeIdRef.current === ownerSessionId) openNewTaskSurface(); @@ -1528,7 +1537,7 @@ function AppShellContent({ const taskReadiness = useTaskSubmissionReadiness( taskReadinessRequest, onboarding.snapshot, - activeId, + ownerActiveId, activeId ? undefined : taskEntry.selectors.target, ); const taskReadinessNotice = deriveTaskReadinessNotice(taskReadiness.snapshot, uiLocale); @@ -1542,10 +1551,12 @@ function AppShellContent({ // The titlebar names the directory the ACTIVE session runs in, so it reads // the same projected project state the picker does — `projectInfo` already // resolves to the session's own cwd once a session owns it. - const titlebarProjectName = deriveTitlebarProjectName({ - projectName: currentProject?.name, - projectPath: projectInfo?.projectPath, - }); + const titlebarProjectName = sharedSessionActive + ? undefined + : deriveTitlebarProjectName({ + projectName: currentProject?.name, + projectPath: projectInfo?.projectPath, + }); const { startModeSession } = useStableActions(createAppShellSessionStartActions, { uiLocale, activeIdRef, @@ -1625,8 +1636,12 @@ function AppShellContent({ // `ComposerMentionsProvider` below, so its reloads do not re-render the shell. const composerMentionsSurface: ComposerMentionsSurface = { skillCatalogRevision: moduleHub.selectors.skillCatalogRevision, - sessionId: activeId, - projectPath: activeId ? projectInfo?.projectPath : taskEntry.selectors.projectPath, + sessionId: ownerActiveId, + projectPath: activeId + ? ownerActiveId + ? projectInfo?.projectPath + : undefined + : taskEntry.selectors.projectPath, newTaskTarget: activeId ? undefined : taskEntry.selectors.target, newSessionModel: newChatModel, newSessionCollaborationMode: newChatPlanModeActive ? 'plan' : 'agent', @@ -1635,7 +1650,7 @@ function AppShellContent({ newSessionPermissionMode: newTaskPermissionMode, }; - const hasModalOpen = helpOpen || paletteOpen || searchModalOpen; + const hasModalOpen = helpOpen || paletteOpen || searchModalOpen || sharedSessionDialog.target !== undefined; const shellObscured = hasModalOpen || settingsOpen; const contextCompactionPresentation = useMemo( () => @@ -2763,6 +2778,14 @@ function AppShellContent({ next one. A remount ties the edit to the session it belongs to. */ key={activeSessionForView.id} sessionName={activeSessionForView.name} + readOnly={sharedSessionActive} + action={sharedSessionActive ? undefined : { + label: getSessionCollaborationCopy(uiLocale).shareAction, + onClick: () => sharedSessionDialog.open({ + sessionId: activeSessionForView.id, + sessionName: activeSessionForView.name, + }), + }} onRenameSession={(name) => { void sessionNavigationCommandsRef.current?.renameSession(activeSessionForView.id, name); }} @@ -2779,7 +2802,7 @@ function AppShellContent({ parentSession={titlebarParentSession} /> )} - {!VIEWS_WITHOUT_WORKSPACE_ACTIONS.has(agentsView) && ( + {!sharedSessionActive && !VIEWS_WITHOUT_WORKSPACE_ACTIONS.has(agentsView) && (