diff --git a/nodejs/README.md b/nodejs/README.md index 93f9c3fa6..6e5fba7dd 100644 --- a/nodejs/README.md +++ b/nodejs/README.md @@ -160,6 +160,29 @@ Initial acquisition runs during session creation or resume. Cancellation, provid Resume an existing session. Returns the session with `workspacePath` populated if infinite sessions were enabled. +##### `watchSharedSession(sessionId: string): Promise` + +Watch a session another user shared with the authenticated user. The handle is +passive: it exposes `sessionId`, `metadata`, `readOnly`, `on(...)`, and +`close()`, but no send, steer, permission, configuration, or cancellation APIs. +History is delivered first through the ordinary session event stream, followed +by live updates. Terminal connection loss is reported through the client's +existing `session.disconnected` lifecycle event. + +```typescript +const disconnected = client.onLifecycle("session.disconnected", ({ sessionId }) => { + console.log(`Watch ${sessionId} disconnected`); +}); +await using watch = await client.watchSharedSession(sharedSessionId); +watch.on((event) => { + console.log(event.type, event.data); +}); +``` + +Authentication, viewer identity, lane credentials, channel derivation, and +reconnection remain internal to the runtime. Register the lifecycle handler +before opening the watch so an immediate terminal disconnect cannot be missed. + ##### `ping(message?: string): Promise<{ message: string; timestamp: string }>` Ping the server to check connectivity. diff --git a/nodejs/src/client.ts b/nodejs/src/client.ts index 9b853aa59..79e877e89 100644 --- a/nodejs/src/client.ts +++ b/nodejs/src/client.ts @@ -41,7 +41,7 @@ import type { SessionUpdateOptionsParams, } from "./generated/rpc.js"; import { getSdkProtocolVersion } from "./sdkProtocolVersion.js"; -import { CopilotSession } from "./session.js"; +import { CopilotSession, SharedSessionWatch } from "./session.js"; import type { FfiRuntimeHost } from "./ffiRuntimeHost.js"; import { createSessionFsAdapter, type SessionFsProvider } from "./sessionFsProvider.js"; import { createCopilotRequestAdapter } from "./copilotRequestHandler.js"; @@ -486,6 +486,7 @@ export class CopilotClient { private actualHost: string = "localhost"; private state: "disconnected" | "connecting" | "connected" | "error" = "disconnected"; private sessions: Map = new Map(); + private sharedSessionWatches: Map = new Map(); private stderrBuffer: string = ""; // Captures CLI stderr for error messages /** Resolved connection mode chosen in the constructor. */ private connectionConfig: InternalRuntimeConnection; @@ -1029,6 +1030,19 @@ export class CopilotClient { async stop(): Promise { const errors: Error[] = []; + const activeWatches = [...this.sharedSessionWatches.values()]; + for (const watch of activeWatches) { + try { + await watch.close(); + } catch (error) { + errors.push( + new Error( + `Failed to close shared-session watch ${watch.sessionId}: ${error instanceof Error ? error.message : String(error)}` + ) + ); + } + } + // Disconnect all active sessions with retry logic const activeSessions = [...this.sessions.values()]; // TEMPORARY: over the in-process (FFI) transport the runtime shares this @@ -1078,6 +1092,7 @@ export class CopilotClient { session._markDisconnected(); } this.sessions.clear(); + this.sharedSessionWatches.clear(); this.githubTokenProviders.clear(); // Ask SDK-owned runtimes to flush and clean up before we tear down @@ -1261,6 +1276,7 @@ export class CopilotClient { session._markDisconnected(); } this.sessions.clear(); + this.sharedSessionWatches.clear(); this.githubTokenProviders.clear(); // Force close connection. Suppress writer failures first so teardown @@ -1773,6 +1789,54 @@ export class CopilotClient { return session; } + /** + * Watch a session shared with the authenticated user. + * + * The returned handle exposes canonical history and live events but no + * interactive session operations. Authentication and lane routing remain + * entirely inside the runtime. Register a `session.disconnected` lifecycle + * handler before calling this method if terminal connection loss must not + * be missed. + * + * @param sessionId - The owner's shared session ID. + */ + async watchSharedSession(sessionId: string): Promise { + if (!this.connection) { + await this.start(); + } + + const result = await this.rpc.sessions.watch({ sessionId }); + if (result.readOnly !== true) { + await this.rpc.sessions.close({ sessionId: result.sessionId }); + throw new Error("Runtime returned an interactive shared-session watch"); + } + + const routedSession = new CopilotSession( + result.sessionId, + this.connection!, + undefined, + this.onGetTraceContext + ); + const closeWatch = async (): Promise => { + try { + await this.rpc.sessions.close({ sessionId: result.sessionId }); + } finally { + routedSession._markDisconnected(); + this.sessions.delete(result.sessionId); + this.sharedSessionWatches.delete(result.sessionId); + } + }; + const watch = new SharedSessionWatch( + result.sessionId, + result.metadata, + routedSession, + closeWatch + ); + this.sessions.set(result.sessionId, routedSession); + this.sharedSessionWatches.set(result.sessionId, watch); + return watch; + } + /** * Resumes an existing conversation session by its ID. * @@ -3102,11 +3166,18 @@ export class CopilotClient { }; } - const event = { - type: raw.type, - sessionId: raw.sessionId, - metadata, - } as SessionLifecycleEvent; + const event = ( + raw.type === "session.disconnected" + ? { + type: raw.type, + sessionId: raw.sessionId, + } + : { + type: raw.type, + sessionId: raw.sessionId, + metadata, + } + ) as SessionLifecycleEvent; // Dispatch to typed handlers for this specific event type const typedHandlers = this.typedLifecycleHandlers.get(event.type); @@ -3128,6 +3199,15 @@ export class CopilotClient { // Ignore handler errors } } + + if ( + event.type === "session.disconnected" && + this.sharedSessionWatches.has(event.sessionId) + ) { + this.sessions.get(event.sessionId)?._markDisconnected(); + this.sessions.delete(event.sessionId); + this.sharedSessionWatches.delete(event.sessionId); + } } private async handleUserInputRequest(params: { diff --git a/nodejs/src/generated/rpc.ts b/nodejs/src/generated/rpc.ts index 9e7304bde..69c3c8d56 100644 --- a/nodejs/src/generated/rpc.ts +++ b/nodejs/src/generated/rpc.ts @@ -22144,6 +22144,37 @@ export interface VisibilitySetResult { */ shareUrl?: string; } +/** + * Parameters for watching a session another user has shared with the authenticated user. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "WatchSharedSessionParams". + */ +/** @experimental */ +export interface WatchSharedSessionParams { + /** + * Session ID to watch. The session belongs to another user and must already be shared with the authenticated user. The watcher's own identity is deliberately not accepted here: it is resolved from the connection's authenticated credential, so a caller cannot ask to watch as somebody else. + */ + sessionId: string; +} +/** + * Result of attaching to a shared session as a read-only watcher. History replays as ordered `session.event` notifications after this result is delivered, not inside it. + * + * This interface was referenced by `_RpcSchemaRoot`'s JSON-Schema + * via the `definition` "WatchSharedSessionResult". + */ +/** @experimental */ +export interface WatchSharedSessionResult { + /** + * SDK session ID for the watched session. + */ + sessionId: string; + /** + * Always true. A watched session observes and replays only: it cannot send or queue input, steer, answer prompts, approve tools, change session configuration or cancel turns. Server-side denial remains the authority; this flag lets a client refuse the interaction up front rather than surfacing a late failure. + */ + readOnly: true; + metadata: ConnectedRemoteSessionMetadata; +} /** * A single changed file and its unified diff. * @@ -23240,6 +23271,15 @@ export function createServerRpc(connection: MessageConnection) { */ connect: async (params: ConnectRemoteSessionParams): Promise => connection.sendRequest("sessions.connect", params), + /** + * Attaches to a session another user has shared with the authenticated user, as a read-only watcher, and exposes it as an SDK session. The watched session replays and streams over the ordinary `session.event` notification channel, but cannot be driven: sending, steering, answering prompts, approving tools, changing session configuration and cancelling turns are all refused. The watcher's own identity is resolved from the connection's credential, never from the caller. + * + * @param params Parameters for watching a session another user has shared with the authenticated user. + * + * @returns Result of attaching to a shared session as a read-only watcher. History replays as ordered `session.event` notifications after this result is delivered, not inside it. + */ + watch: async (params: WatchSharedSessionParams): Promise => + connection.sendRequest("sessions.watch", params), /** * Lists sessions, optionally filtered by source and working-directory context. Returned entries are discriminated by `isRemote`: local entries carry only the lightweight `LocalSessionMetadataValue` shape; remote entries carry the full `RemoteSessionMetadataValue` shape (repository, PR number, taskType, etc.). * diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts index 9d55ab1d1..21c9a3349 100644 --- a/nodejs/src/index.ts +++ b/nodejs/src/index.ts @@ -11,7 +11,12 @@ export { CopilotClient } from "./client.js"; export { DisableBypassPermissionsModes, RuntimeConnection } from "./types.js"; export { BuiltInTools, ToolSet } from "./toolSet.js"; -export { CopilotSession, type AssistantMessageEvent } from "./session.js"; +export { + CopilotSession, + SharedSessionWatch, + type AssistantMessageEvent, + type SharedSessionMetadata, +} from "./session.js"; export { defineFactory, FactoryResumeError, isFactoryRunTerminal } from "./factory.js"; export { Canvas, @@ -158,6 +163,7 @@ export type { SessionHooks, SessionCreatedEvent, SessionDeletedEvent, + SessionDisconnectedEvent, SessionUpdatedEvent, SessionForegroundEvent, SessionBackgroundEvent, diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 65ff00921..d446fad8c 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -14,6 +14,8 @@ import { createSessionRpc } from "./generated/rpc.js"; import type { ClientSessionApiHandlers, CanvasActionInvokeResult, + ConnectedRemoteSessionMetadata, + ConnectedRemoteSessionMetadataRepository, CurrentToolMetadata, McpOauthPendingRequestResponse, FactoryLogLine, @@ -81,6 +83,94 @@ import { type FactoryStepOptions, } from "./factory.js"; +/** Immutable metadata describing a watched shared session. */ +export type SharedSessionMetadata = Readonly< + Omit & { + repository: Readonly; + } +>; + +/** + * Passive, read-only attachment to a session shared with the authenticated user. + * + * History and live updates are delivered through {@link on}. Interactive session + * operations are intentionally absent from this type. + */ +export class SharedSessionWatch { + /** Always `true`; watched sessions cannot be driven by this client. */ + readonly readOnly = true as const; + /** Immutable metadata returned by the runtime when the watch is attached. */ + readonly metadata: SharedSessionMetadata; + + private readonly handlers = new Set(); + private readonly pendingEvents: SessionEvent[] = []; + private closePromise: Promise | undefined; + + /** @internal */ + constructor( + readonly sessionId: string, + metadata: ConnectedRemoteSessionMetadata, + session: CopilotSession, + private readonly closeWatch: () => Promise + ) { + this.metadata = Object.freeze({ + ...metadata, + repository: Object.freeze({ ...metadata.repository }), + }); + session.on((event) => { + if (this.handlers.size === 0) { + this.pendingEvents.push(event); + return; + } + for (const handler of this.handlers) { + try { + handler(event); + } catch { + // A failing subscriber must not prevent delivery to others. + } + } + }); + } + + /** + * Subscribe to canonical replay and live session events. + * + * Events replayed before the first handler is attached are retained and + * delivered in order when that handler is registered. + */ + on(handler: SessionEventHandler): () => void { + this.handlers.add(handler); + if (this.pendingEvents.length > 0) { + const pending = this.pendingEvents.splice(0); + for (const event of pending) { + try { + handler(event); + } catch { + // Mirrors live delivery: a failing subscriber must not abort + // the remaining replay events or prevent `on` from returning + // its unsubscribe function. + } + } + } + return () => this.handlers.delete(handler); + } + + /** + * Close the watch and release its local event routing. + * + * Repeated calls share the same close operation. + */ + close(): Promise { + this.closePromise ??= this.closeWatch(); + return this.closePromise; + } + + /** Close the watch when used with `await using`. */ + async [Symbol.asyncDispose](): Promise { + await this.close(); + } +} + function isFactoryResumeErrorCode(value: unknown): value is FactoryResumeErrorCode { return ( value === "not_found" || diff --git a/nodejs/src/types.ts b/nodejs/src/types.ts index 616e15a46..17d7de361 100644 --- a/nodejs/src/types.ts +++ b/nodejs/src/types.ts @@ -3500,11 +3500,12 @@ export type SessionLifecycleEventType = | "session.deleted" | "session.updated" | "session.foreground" - | "session.background"; + | "session.background" + | "session.disconnected"; /** * Metadata payload for session lifecycle events. Not present on - * `session.deleted` events. + * `session.deleted` or `session.disconnected` events. */ export interface SessionLifecycleEventMetadata { /** Time the session was created. */ @@ -3519,7 +3520,7 @@ export interface SessionLifecycleEventMetadata { interface SessionLifecycleEventBase { /** ID of the session this event relates to. */ sessionId: string; - /** Session metadata (not included for `session.deleted`). */ + /** Session metadata (not included for deleted or disconnected events). */ metadata?: SessionLifecycleEventMetadata; } @@ -3553,6 +3554,12 @@ export interface SessionBackgroundEvent extends SessionLifecycleEventBase { metadata: SessionLifecycleEventMetadata; } +/** Emitted when a session connection is terminally lost. */ +export interface SessionDisconnectedEvent extends SessionLifecycleEventBase { + type: "session.disconnected"; + metadata?: undefined; +} + /** * Discriminated union of all session lifecycle events emitted in TUI+server mode. * Switch on `type` to access the variant-specific metadata. @@ -3562,7 +3569,8 @@ export type SessionLifecycleEvent = | SessionDeletedEvent | SessionUpdatedEvent | SessionForegroundEvent - | SessionBackgroundEvent; + | SessionBackgroundEvent + | SessionDisconnectedEvent; /** * Handler for session lifecycle events. diff --git a/nodejs/test/client.test.ts b/nodejs/test/client.test.ts index 3ffda2fa7..b3939266f 100644 --- a/nodejs/test/client.test.ts +++ b/nodejs/test/client.test.ts @@ -15,8 +15,10 @@ import { type GitHubTelemetryNotification, type ManagedSettings, type ModelInfo, + type SessionLifecycleEventType, } from "../src/index.js"; import { CopilotSession } from "../src/session.js"; +import type { WatchSharedSessionParams, WatchSharedSessionResult } from "../src/generated/rpc.js"; import { defaultJoinSessionPermissionHandler } from "../src/types.js"; // This file is for unit tests. Where relevant, prefer to add e2e tests in e2e/*.test.ts instead @@ -1186,6 +1188,154 @@ describe("CopilotClient", () => { expect(received).toEqual([notification]); }); + it("watches a shared session without exposing interactive session methods", async () => { + const { createMessageConnection, StreamMessageReader, StreamMessageWriter } = + await import("vscode-jsonrpc/node.js"); + + const clientToServer = new PassThrough(); + const serverToClient = new PassThrough(); + const clientConn = createMessageConnection( + new StreamMessageReader(serverToClient), + new StreamMessageWriter(clientToServer) + ); + const serverConn = createMessageConnection( + new StreamMessageReader(clientToServer), + new StreamMessageWriter(serverToClient) + ); + const client = new CopilotClient({ + connection: RuntimeConnection.forUri("localhost:1234"), + }); + (client as any).connection = clientConn; + (client as any).attachConnectionHandlers(); + + onTestFinished(() => { + clientConn.dispose(); + serverConn.dispose(); + }); + + const closeRequests: unknown[] = []; + serverConn.onRequest("sessions.watch", async (params) => { + expect(params).toEqual({ sessionId: "shared-session" }); + return { + sessionId: "watch-session", + readOnly: true, + metadata: { + sessionId: "watch-session", + startTime: "2025-01-01T00:00:00Z", + modifiedTime: "2025-01-01T00:01:00Z", + repository: { owner: "github", name: "copilot-sdk", branch: "main" }, + kind: "remote-session", + }, + }; + }); + serverConn.onRequest("sessions.close", async (params) => { + closeRequests.push(params); + return {}; + }); + clientConn.listen(); + serverConn.listen(); + + const disconnected = new Promise((resolve) => { + client.onLifecycle("session.disconnected", (event) => { + expect(event).toEqual({ + type: "session.disconnected", + sessionId: "watch-session", + }); + expect(Object.keys(event).sort()).toEqual(["sessionId", "type"]); + resolve(); + }); + }); + const watch = await client.watchSharedSession("shared-session"); + expect((client as any).sessions.has("watch-session")).toBe(true); + const replay = { + type: "assistant.message", + id: "replay-event", + parentId: null, + timestamp: "2025-01-01T00:00:30Z", + data: { content: "history" }, + } as const; + + await serverConn.sendNotification("session.event", { + sessionId: "watch-session", + event: replay, + }); + await serverConn.sendNotification("session.lifecycle", { + type: "session.disconnected", + sessionId: "watch-session", + }); + + const received: string[] = []; + await new Promise((resolve) => { + watch.on((event) => { + received.push(event.type); + resolve(); + }); + }); + await disconnected; + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(watch.sessionId).toBe("watch-session"); + expect(watch.readOnly).toBe(true); + expect(watch.metadata.sessionId).toBe("watch-session"); + expect(Object.isFrozen(watch.metadata)).toBe(true); + expect(Object.isFrozen(watch.metadata.repository)).toBe(true); + expect(received).toEqual(["assistant.message"]); + expect((client as any).sessions.has("watch-session")).toBe(false); + expect((client as any).sharedSessionWatches.has("watch-session")).toBe(false); + expect("send" in watch).toBe(false); + expect("abort" in watch).toBe(false); + // A shared-session watch is intentionally passive: `send`/`abort` must not + // exist on its type, so this resolves to `true` only while they are absent. + type PassiveWatch = "send" | "abort" extends keyof T ? never : true; + const watchIsPassive: PassiveWatch = true; + expect(watchIsPassive).toBe(true); + + await watch.close(); + await watch.close(); + expect(closeRequests).toEqual([{ sessionId: "watch-session" }]); + }); + + it("generates the exact credential-free shared-session watch payloads", () => { + const params: WatchSharedSessionParams = { sessionId: "shared-session" }; + const result: WatchSharedSessionResult = { + sessionId: "watch-session", + readOnly: true, + metadata: { + sessionId: "watch-session", + startTime: "2025-01-01T00:00:00Z", + modifiedTime: "2025-01-01T00:01:00Z", + repository: { owner: "github", name: "copilot-sdk", branch: "main" }, + kind: "remote-session", + }, + }; + + expect(params).toEqual({ sessionId: "shared-session" }); + expect(Object.keys(result).sort()).toEqual(["metadata", "readOnly", "sessionId"]); + expect(JSON.stringify(result)).not.toMatch( + /viewerId|baseUrl|wps|lane|channel|credential|token/i + ); + }); + + it("pins the complete runtime lifecycle event type set", () => { + const eventTypes = { + "session.created": true, + "session.deleted": true, + "session.updated": true, + "session.foreground": true, + "session.background": true, + "session.disconnected": true, + } satisfies Record; + + expect(Object.keys(eventTypes)).toEqual([ + "session.created", + "session.deleted", + "session.updated", + "session.foreground", + "session.background", + "session.disconnected", + ]); + }); + it("registers no gitHubTelemetry handler when onGitHubTelemetry is omitted", () => { const client = new CopilotClient(); onTestFinished(() => stopClient(client)); diff --git a/rust/README.md b/rust/README.md index 323d525d3..b6af87962 100644 --- a/rust/README.md +++ b/rust/README.md @@ -172,6 +172,32 @@ session session.disconnect().await?; ``` +### Shared Session Watch + +`Client::watch_shared_session` attaches to a session another user shared with +the authenticated user. The returned `SharedSessionWatch` is passive and +exposes metadata, ordered canonical history/live events, and `close()` without +interactive session methods. Terminal connection loss is reported through the +client's existing `SessionLifecycleEventType::Disconnected` subscription. + +```rust,ignore +let mut lifecycle = client.subscribe_lifecycle(); +let mut watch = client.watch_shared_session(shared_session_id).await?; +tokio::select! { + Some(event) = watch.events().recv() => { + println!("{}: {}", event.event_type, event.data); + } + Ok(event) = lifecycle.recv() => { + println!("{:?}: {}", event.event_type, event.session_id); + } +} +watch.close().await?; +``` + +Authentication, viewer identity, lane credentials, channel derivation, and +reconnection remain internal to the runtime. Subscribe to lifecycle events +before opening the watch so an immediate terminal disconnect cannot be missed. + #### Typed RPC namespace High-level helpers are convenience wrappers over a fully-typed diff --git a/rust/src/generated/api_types.rs b/rust/src/generated/api_types.rs index bdf291e59..b6cf69888 100644 --- a/rust/src/generated/api_types.rs +++ b/rust/src/generated/api_types.rs @@ -137,6 +137,8 @@ pub mod rpc_methods { pub const SESSIONS_FORK: &str = "sessions.fork"; /// `sessions.connect` pub const SESSIONS_CONNECT: &str = "sessions.connect"; + /// `sessions.watch` + pub const SESSIONS_WATCH: &str = "sessions.watch"; /// `sessions.list` pub const SESSIONS_LIST: &str = "sessions.list"; /// `sessions.getMetadata` @@ -20755,6 +20757,40 @@ pub struct VisibilitySetResult { pub synced: bool, } +/// Parameters for watching a session another user has shared with the authenticated user. +/// +///
+/// +/// **Experimental.** This type is part of an experimental wire-protocol surface +/// and may change or be removed in future SDK or CLI releases. +/// +///
+#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WatchSharedSessionParams { + /// Session ID to watch. The session belongs to another user and must already be shared with the authenticated user. The watcher's own identity is deliberately not accepted here: it is resolved from the connection's authenticated credential, so a caller cannot ask to watch as somebody else. + pub session_id: SessionId, +} + +/// Result of attaching to a shared session as a read-only watcher. History replays as ordered `session.event` notifications after this result is delivered, not inside it. +/// +///
+/// +/// **Experimental.** This type is part of an experimental wire-protocol surface +/// and may change or be removed in future SDK or CLI releases. +/// +///
+#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WatchSharedSessionResult { + /// Metadata for the watched session. + pub metadata: ConnectedRemoteSessionMetadata, + /// Always true. A watched session observes and replays only: it cannot send or queue input, steer, answer prompts, approve tools, change session configuration or cancel turns. Server-side denial remains the authority; this flag lets a client refuse the interaction up front rather than surfacing a late failure. + pub read_only: bool, + /// SDK session ID for the watched session. + pub session_id: SessionId, +} + /// A single changed file and its unified diff. /// ///
@@ -21659,6 +21695,25 @@ pub struct SessionsConnectResult { pub session_id: SessionId, } +/// Result of attaching to a shared session as a read-only watcher. History replays as ordered `session.event` notifications after this result is delivered, not inside it. +/// +///
+/// +/// **Experimental.** This type is part of an experimental wire-protocol surface +/// and may change or be removed in future SDK or CLI releases. +/// +///
+#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionsWatchResult { + /// Metadata for the watched session. + pub metadata: ConnectedRemoteSessionMetadata, + /// Always true. A watched session observes and replays only: it cannot send or queue input, steer, answer prompts, approve tools, change session configuration or cancel turns. Server-side denial remains the authority; this flag lets a client refuse the interaction up front rather than surfacing a late failure. + pub read_only: bool, + /// SDK session ID for the watched session. + pub session_id: SessionId, +} + /// Sessions matching the filter, ordered most-recently-modified first. /// ///
diff --git a/rust/src/generated/rpc.rs b/rust/src/generated/rpc.rs index 60bf9d804..590f5b192 100644 --- a/rust/src/generated/rpc.rs +++ b/rust/src/generated/rpc.rs @@ -1800,6 +1800,37 @@ impl<'a> ClientRpcSessions<'a> { Ok(serde_json::from_value(_value)?) } + /// Attaches to a session another user has shared with the authenticated user, as a read-only watcher, and exposes it as an SDK session. The watched session replays and streams over the ordinary `session.event` notification channel, but cannot be driven: sending, steering, answering prompts, approving tools, changing session configuration and cancelling turns are all refused. The watcher's own identity is resolved from the connection's credential, never from the caller. + /// + /// Wire method: `sessions.watch`. + /// + /// # Parameters + /// + /// * `params` - Parameters for watching a session another user has shared with the authenticated user. + /// + /// # Returns + /// + /// Result of attaching to a shared session as a read-only watcher. History replays as ordered `session.event` notifications after this result is delivered, not inside it. + /// + ///
+ /// + /// **Experimental.** This API is part of an experimental wire-protocol surface + /// and may change or be removed in future SDK or CLI releases. Pin both the + /// SDK and CLI versions if your code depends on it. + /// + ///
+ pub async fn watch( + &self, + params: WatchSharedSessionParams, + ) -> Result { + let wire_params = serde_json::to_value(params)?; + let _value = self + .client + .call(rpc_methods::SESSIONS_WATCH, Some(wire_params)) + .await?; + Ok(serde_json::from_value(_value)?) + } + /// Lists sessions, optionally filtered by source and working-directory context. Returned entries are discriminated by `isRemote`: local entries carry only the lightweight `LocalSessionMetadataValue` shape; remote entries carry the full `RemoteSessionMetadataValue` shape (repository, PR number, taskType, etc.). /// /// Wire method: `sessions.list`. diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 4a9f73ca4..618e677d3 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -54,6 +54,7 @@ pub mod trace_context; pub mod transforms; /// Protocol types shared between the SDK and the GitHub Copilot CLI. pub mod types; +mod watch; mod wire; /// Session event payload types — auto-generated from the protocol schema. @@ -95,6 +96,7 @@ pub(crate) use jsonrpc::{ }; pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet}; pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs}; +pub use watch::{SharedSessionWatch, SharedSessionWatchEvents}; /// Re-exported JSON-RPC internals for integration tests (requires `test-support` feature). #[cfg(feature = "test-support")] @@ -2067,6 +2069,24 @@ impl Client { self.inner.router.unregister(session_id); } + pub(crate) fn register_watch_session( + &self, + session_id: &SessionId, + ) -> crate::router::SessionChannels { + self.inner.router.ensure_started( + &self.inner.notification_tx, + &self.inner.request_rx, + self.inner.llm_inference.get().cloned(), + self.inner.on_github_telemetry.clone(), + self.inner.github_token_registry.clone(), + ); + self.inner.router.register_watch(session_id) + } + + pub(crate) fn unregister_watch_session(&self, session_id: &SessionId) { + self.inner.router.unregister(session_id); + } + pub(crate) fn register_github_token_provider( &self, provider: Arc, @@ -2306,12 +2326,14 @@ impl Client { pub async fn cleanup_sessions_for_test(&self) -> Result<()> { let mut first_error = None; - for session_id in self.inner.router.session_ids() { + for (session_id, is_watch) in self.inner.router.session_entries() { + let method = if is_watch { + generated::api_types::rpc_methods::SESSIONS_CLOSE + } else { + "session.destroy" + }; if let Err(error) = self - .call( - "session.destroy", - Some(serde_json::json!({ "sessionId": session_id })), - ) + .call(method, Some(serde_json::json!({ "sessionId": session_id }))) .await && first_error.is_none() { @@ -2469,12 +2491,14 @@ impl Client { // Snapshot the registered session IDs without holding the router // lock across the destroy RPCs. - for session_id in self.inner.router.session_ids() { + for (session_id, is_watch) in self.inner.router.session_entries() { + let method = if is_watch { + generated::api_types::rpc_methods::SESSIONS_CLOSE + } else { + "session.destroy" + }; match self - .call( - "session.destroy", - Some(serde_json::json!({ "sessionId": session_id })), - ) + .call(method, Some(serde_json::json!({ "sessionId": session_id }))) .await { Ok(_) => {} @@ -2482,7 +2506,8 @@ impl Client { warn!( session_id = %session_id, error = %e, - "session.destroy failed during Client::stop", + method, + "session cleanup failed during Client::stop", ); errors.push(e); } @@ -3304,7 +3329,7 @@ mod tests { handle.abort(); let _ = handle.await; - assert!(client.inner.router.session_ids().is_empty()); + assert!(client.inner.router.session_entries().is_empty()); client.force_stop(); } @@ -3444,7 +3469,7 @@ mod tests { async fn wait_for_pending_session_registration(client: &Client) { let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1); - while client.inner.router.session_ids().is_empty() { + while client.inner.router.session_entries().is_empty() { assert!( tokio::time::Instant::now() < deadline, "session was not registered" diff --git a/rust/src/router.rs b/rust/src/router.rs index 1dec9d16f..874588afc 100644 --- a/rust/src/router.rs +++ b/rust/src/router.rs @@ -6,7 +6,9 @@ use tokio::sync::{broadcast, mpsc}; use tracing::warn; use crate::jsonrpc::{JsonRpcNotification, JsonRpcRequest}; -use crate::types::{SessionEventNotification, SessionId}; +use crate::types::{ + SessionEventNotification, SessionId, SessionLifecycleEvent, SessionLifecycleEventType, +}; /// Per-session channels created by the router during session registration. pub(crate) struct SessionChannels { @@ -19,26 +21,37 @@ pub(crate) struct SessionChannels { struct SessionSenders { notifications: mpsc::UnboundedSender, requests: mpsc::UnboundedSender, + is_watch: bool, } /// Routes notifications and requests by sessionId to per-session channels. /// /// Internal to the SDK — consumers interact via `Client::register_session()`. +#[derive(Clone)] pub(crate) struct SessionRouter { sessions: Arc>>, - started: Mutex, + started: Arc>, } impl SessionRouter { pub(crate) fn new() -> Self { Self { sessions: Arc::new(Mutex::new(HashMap::new())), - started: Mutex::new(false), + started: Arc::new(Mutex::new(false)), } } /// Register a session to receive filtered events and requests. pub(crate) fn register(&self, session_id: &SessionId) -> SessionChannels { + self.register_with_kind(session_id, false) + } + + /// Register a passive shared-session watch. + pub(crate) fn register_watch(&self, session_id: &SessionId) -> SessionChannels { + self.register_with_kind(session_id, true) + } + + fn register_with_kind(&self, session_id: &SessionId, is_watch: bool) -> SessionChannels { let (notif_tx, notif_rx) = mpsc::unbounded_channel(); let (req_tx, req_rx) = mpsc::unbounded_channel(); self.sessions.lock().insert( @@ -46,6 +59,7 @@ impl SessionRouter { SessionSenders { notifications: notif_tx, requests: req_tx, + is_watch, }, ); SessionChannels { @@ -59,13 +73,13 @@ impl SessionRouter { self.sessions.lock().remove(session_id.as_str()); } - /// Snapshot every currently-registered session ID. - /// - /// Used by [`Client::stop`](crate::Client::stop) to iterate active - /// sessions for cooperative shutdown without holding the router lock - /// across `.await`. - pub(crate) fn session_ids(&self) -> Vec { - self.sessions.lock().keys().cloned().collect() + /// Snapshot registered session IDs with their cleanup classification. + pub(crate) fn session_entries(&self) -> Vec<(SessionId, bool)> { + self.sessions + .lock() + .iter() + .map(|(session_id, senders)| (session_id.clone(), senders.is_watch)) + .collect() } /// Drop all registered session channels. @@ -136,6 +150,33 @@ impl SessionRouter { } continue; } + if notification.method == "session.lifecycle" { + let Some(ref params) = notification.params else { + continue; + }; + match serde_json::from_value::(params.clone()) { + Ok(event) + if event.event_type + == SessionLifecycleEventType::Disconnected => + { + let mut guard = sessions.lock(); + if guard + .get(&event.session_id) + .is_some_and(|senders| senders.is_watch) + { + guard.remove(&event.session_id); + } + } + Ok(_) => {} + Err(e) => { + warn!( + error = %e, + "failed to deserialize session.lifecycle notification" + ); + } + } + continue; + } if notification.method != "session.event" { continue; } diff --git a/rust/src/types.rs b/rust/src/types.rs index 6e451eb45..36a1b034b 100644 --- a/rust/src/types.rs +++ b/rust/src/types.rs @@ -78,6 +78,9 @@ pub enum SessionLifecycleEventType { /// A session moved into the background. #[serde(rename = "session.background")] Background, + /// A session connection was terminally lost. + #[serde(rename = "session.disconnected")] + Disconnected, } /// Optional metadata attached to a [`SessionLifecycleEvent`]. @@ -105,6 +108,7 @@ pub struct SessionLifecycleEvent { #[serde(rename = "sessionId")] pub session_id: SessionId, /// Optional metadata describing the session at the time of the event. + /// Absent for deleted and disconnected events. #[serde(skip_serializing_if = "Option::is_none")] pub metadata: Option, } @@ -5972,8 +5976,8 @@ mod tests { InfiniteSessionConfig, LargeToolOutputConfig, McpServerConfig, McpStdioServerConfig, MemoryConfiguration, NamedProviderConfig, PermissionResponseCapability, ProviderConfig, ProviderModelConfig, ReasoningSummary, ResumeSessionConfig, SessionConfig, SessionEvent, - SessionId, SystemMessageConfig, Tool, ToolBinaryResult, ToolResult, ToolResultExpanded, - ToolResultResponse, ensure_attachment_display_names, + SessionId, SessionLifecycleEventType, SystemMessageConfig, Tool, ToolBinaryResult, + ToolResult, ToolResultExpanded, ToolResultResponse, ensure_attachment_display_names, }; use crate::generated::session_events::TypedSessionEvent; @@ -7371,6 +7375,38 @@ mod tests { let _ = ConnectionState::Error; } + #[test] + fn session_lifecycle_event_types_match_runtime_contract() { + let event_types = [ + SessionLifecycleEventType::Created, + SessionLifecycleEventType::Deleted, + SessionLifecycleEventType::Updated, + SessionLifecycleEventType::Foreground, + SessionLifecycleEventType::Background, + SessionLifecycleEventType::Disconnected, + ]; + let wire_names = event_types.map(|event_type| match event_type { + SessionLifecycleEventType::Created => "session.created", + SessionLifecycleEventType::Deleted => "session.deleted", + SessionLifecycleEventType::Updated => "session.updated", + SessionLifecycleEventType::Foreground => "session.foreground", + SessionLifecycleEventType::Background => "session.background", + SessionLifecycleEventType::Disconnected => "session.disconnected", + }); + + assert_eq!( + wire_names, + [ + "session.created", + "session.deleted", + "session.updated", + "session.foreground", + "session.background", + "session.disconnected", + ] + ); + } + #[test] fn deserializes_runtime_attachment_variants() { let attachments: Vec = serde_json::from_value(json!([ diff --git a/rust/src/watch.rs b/rust/src/watch.rs new file mode 100644 index 000000000..6ab01190a --- /dev/null +++ b/rust/src/watch.rs @@ -0,0 +1,193 @@ +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use parking_lot::Mutex as ParkingLotMutex; +use tokio::sync::mpsc; +use tokio_stream::Stream; + +use crate::generated::api_types::{ + ConnectedRemoteSessionMetadata, SessionsCloseRequest, WatchSharedSessionParams, + WatchSharedSessionResult, rpc_methods, +}; +use crate::router::SessionChannels; +use crate::types::{SessionEvent, SessionEventNotification, SessionId}; +use crate::{Client, Error, ErrorKind}; + +/// Passive, read-only attachment to a session shared with the authenticated user. +/// +/// History and live updates are available through [`events`](Self::events). +/// Interactive session operations are intentionally absent from this type. +/// Call [`close`](Self::close) before dropping the handle; client shutdown also +/// closes any watch that remains registered. +pub struct SharedSessionWatch { + session_id: SessionId, + metadata: ConnectedRemoteSessionMetadata, + client: Client, + events: SharedSessionWatchEvents, + closed: tokio::sync::Mutex, +} + +impl SharedSessionWatch { + pub(crate) fn new( + client: Client, + session_id: SessionId, + metadata: ConnectedRemoteSessionMetadata, + channels: SessionChannels, + ) -> Self { + let SessionChannels { + notifications, + requests: _, + } = channels; + Self { + session_id, + metadata, + client, + events: SharedSessionWatchEvents { notifications }, + closed: tokio::sync::Mutex::new(false), + } + } + + /// SDK session ID assigned to this watch. + pub fn session_id(&self) -> &SessionId { + &self.session_id + } + + /// Metadata for the watched shared session. + pub fn metadata(&self) -> &ConnectedRemoteSessionMetadata { + &self.metadata + } + + /// Whether this attachment is read-only. + pub const fn is_read_only(&self) -> bool { + true + } + + /// Ordered canonical history and live events for the watched session. + /// + /// The receiver is registered before the watch response is delivered, so + /// replay events remain buffered until the caller begins consuming them. + pub fn events(&mut self) -> &mut SharedSessionWatchEvents { + &mut self.events + } + + /// Close the watch and release local event routing. + /// + /// Repeated calls are idempotent. + pub async fn close(&self) -> Result<(), Error> { + let mut closed = self.closed.lock().await; + if *closed { + return Ok(()); + } + + let result = self + .client + .rpc() + .sessions() + .close(SessionsCloseRequest { + session_id: self.session_id.clone(), + }) + .await + .map(|_| ()); + self.client.unregister_watch_session(&self.session_id); + *closed = true; + result + } +} + +/// Event stream retained by a [`SharedSessionWatch`]. +pub struct SharedSessionWatchEvents { + notifications: mpsc::UnboundedReceiver, +} + +impl SharedSessionWatchEvents { + /// Receive the next canonical session event. + /// + /// Returns `None` after the watch is closed or the client disconnects. + pub async fn recv(&mut self) -> Option { + self.notifications + .recv() + .await + .map(|notification| notification.event) + } +} + +impl Stream for SharedSessionWatchEvents { + type Item = SessionEvent; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.notifications) + .poll_recv(cx) + .map(|notification| notification.map(|notification| notification.event)) + } +} + +impl Client { + /// Watch a session shared with the authenticated user. + /// + /// The runtime derives viewer identity, authorization and lane routing from + /// the existing authenticated connection. The returned handle exposes no + /// send, steer, prompt, permission, configuration or cancellation methods. + /// Subscribe with [`Client::subscribe_lifecycle`] before calling this method + /// if terminal connection loss must not be missed. + pub async fn watch_shared_session( + &self, + session_id: impl Into, + ) -> Result { + let params = WatchSharedSessionParams { + session_id: session_id.into(), + }; + let wire_params = serde_json::to_value(params)?; + let registration = Arc::new(ParkingLotMutex::new(None)); + let registration_for_callback = registration.clone(); + let client = self.clone(); + + let value = self + .call_with_inline_callback( + rpc_methods::SESSIONS_WATCH, + Some(wire_params), + Some(Box::new(move |response| { + let value = response.result.as_ref().ok_or_else(|| { + Error::with_message( + ErrorKind::Rpc { code: -32603 }, + "sessions.watch response did not include a result", + ) + })?; + let result: WatchSharedSessionResult = serde_json::from_value(value.clone())?; + let channels = client.register_watch_session(&result.session_id); + *registration_for_callback.lock() = Some(channels); + Ok(()) + })), + ) + .await?; + let result: WatchSharedSessionResult = serde_json::from_value(value)?; + + if !result.read_only { + let _ = self + .rpc() + .sessions() + .close(SessionsCloseRequest { + session_id: result.session_id.clone(), + }) + .await; + self.unregister_watch_session(&result.session_id); + return Err(Error::with_message( + ErrorKind::Rpc { code: -32603 }, + "runtime returned an interactive shared-session watch", + )); + } + + let channels = registration.lock().take().ok_or_else(|| { + Error::with_message( + ErrorKind::Rpc { code: -32603 }, + "sessions.watch response was not registered for event routing", + ) + })?; + Ok(SharedSessionWatch::new( + self.clone(), + result.session_id, + result.metadata, + channels, + )) + } +} diff --git a/rust/tests/api_types_test.rs b/rust/tests/api_types_test.rs index 9b86b1367..d204f8ca8 100644 --- a/rust/tests/api_types_test.rs +++ b/rust/tests/api_types_test.rs @@ -6,6 +6,7 @@ use github_copilot_sdk::rpc::{ Extension, ExtensionList, ExtensionSource, ExtensionStatus, ExtensionsDisableRequest, ExtensionsEnableRequest, FleetStartRequest, FleetStartResult, TasksStartAgentRequest, + WatchSharedSessionParams, WatchSharedSessionResult, }; use github_copilot_sdk::session_events::{PermissionRequest, PermissionRequestedData}; @@ -104,6 +105,50 @@ fn permission_event_exposes_managed_approval_required() { assert_eq!(request.managed_approval_required, Some(true)); } +#[test] +fn shared_session_watch_payloads_are_generated_without_credentials() { + let params = WatchSharedSessionParams { + session_id: "shared-session".into(), + }; + assert_eq!( + serde_json::to_value(params).unwrap(), + serde_json::json!({ "sessionId": "shared-session" }) + ); + + let result: WatchSharedSessionResult = serde_json::from_value(serde_json::json!({ + "sessionId": "watch-session", + "readOnly": true, + "metadata": { + "sessionId": "watch-session", + "startTime": "2025-01-01T00:00:00Z", + "modifiedTime": "2025-01-01T00:01:00Z", + "repository": { + "owner": "github", + "name": "copilot-sdk", + "branch": "main" + }, + "kind": "remote-session" + } + })) + .unwrap(); + let serialized = serde_json::to_value(result).unwrap(); + + assert_eq!(serialized["readOnly"], true); + assert_eq!(serialized["sessionId"], "watch-session"); + let debug = serialized.to_string().to_ascii_lowercase(); + for forbidden in [ + "viewerid", + "baseurl", + "wps", + "lane", + "channel", + "credential", + "token", + ] { + assert!(!debug.contains(forbidden), "unexpected field: {forbidden}"); + } +} + fn running_extension(id: &str, name: &str) -> Extension { Extension { id: id.to_string(), diff --git a/rust/tests/session_test.rs b/rust/tests/session_test.rs index a51d61910..043d2d81d 100644 --- a/rust/tests/session_test.rs +++ b/rust/tests/session_test.rs @@ -258,6 +258,152 @@ where (session, server) } +#[tokio::test] +async fn shared_session_watch_retains_replay_and_closes_once() { + let (client, server_read, server_write) = make_client(); + let mut lifecycle = client.subscribe_lifecycle(); + let mut server = FakeServer { + read: server_read, + write: server_write, + session_id: "watch-session".to_string(), + }; + + let watch_handle = tokio::spawn({ + let client = client.clone(); + async move { client.watch_shared_session("shared-session").await.unwrap() } + }); + let watch_request = server.read_request().await; + assert_eq!(watch_request["method"], "sessions.watch"); + assert_eq!( + watch_request["params"], + serde_json::json!({ "sessionId": "shared-session" }) + ); + server + .respond( + &watch_request, + serde_json::json!({ + "sessionId": "watch-session", + "readOnly": true, + "metadata": { + "sessionId": "watch-session", + "startTime": "2025-01-01T00:00:00Z", + "modifiedTime": "2025-01-01T00:01:00Z", + "repository": { + "owner": "github", + "name": "copilot-sdk", + "branch": "main" + }, + "kind": "remote-session" + } + }), + ) + .await; + server + .send_event( + "assistant.message", + serde_json::json!({ "content": "history" }), + ) + .await; + server + .send_notification( + "session.lifecycle", + serde_json::json!({ + "type": "session.disconnected", + "sessionId": "watch-session" + }), + ) + .await; + + let mut watch = timeout(TIMEOUT, watch_handle).await.unwrap().unwrap(); + assert_eq!(watch.session_id(), "watch-session"); + assert!(watch.is_read_only()); + assert_eq!(watch.metadata().session_id, "watch-session"); + + let replay = timeout(TIMEOUT, watch.events().recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(replay.event_type, "assistant.message"); + let terminal = timeout(TIMEOUT, lifecycle.recv()).await.unwrap().unwrap(); + assert_eq!( + terminal.event_type, + github_copilot_sdk::SessionLifecycleEventType::Disconnected + ); + assert_eq!(terminal.session_id, "watch-session"); + assert!(terminal.metadata.is_none()); + assert!( + timeout(TIMEOUT, watch.events().recv()) + .await + .unwrap() + .is_none() + ); + + let close_handle = tokio::spawn({ + async move { + watch.close().await.unwrap(); + watch.close().await.unwrap(); + } + }); + let close_request = server.read_request().await; + assert_eq!(close_request["method"], "sessions.close"); + assert_eq!( + close_request["params"], + serde_json::json!({ "sessionId": "watch-session" }) + ); + server.respond(&close_request, serde_json::json!({})).await; + timeout(TIMEOUT, close_handle).await.unwrap().unwrap(); +} + +#[tokio::test] +async fn client_stop_closes_shared_session_watches() { + let (client, server_read, server_write) = make_client(); + let mut server = FakeServer { + read: server_read, + write: server_write, + session_id: "watch-session".to_string(), + }; + + let watch_handle = tokio::spawn({ + let client = client.clone(); + async move { client.watch_shared_session("shared-session").await.unwrap() } + }); + let watch_request = server.read_request().await; + server + .respond( + &watch_request, + serde_json::json!({ + "sessionId": "watch-session", + "readOnly": true, + "metadata": { + "sessionId": "watch-session", + "startTime": "2025-01-01T00:00:00Z", + "modifiedTime": "2025-01-01T00:01:00Z", + "repository": { + "owner": "github", + "name": "copilot-sdk", + "branch": "main" + }, + "kind": "remote-session" + } + }), + ) + .await; + let _watch = timeout(TIMEOUT, watch_handle).await.unwrap().unwrap(); + + let stop_handle = tokio::spawn({ + let client = client.clone(); + async move { client.stop().await.unwrap() } + }); + let close_request = server.read_request().await; + assert_eq!(close_request["method"], "sessions.close"); + assert_eq!( + close_request["params"], + serde_json::json!({ "sessionId": "watch-session" }) + ); + server.respond(&close_request, serde_json::json!({})).await; + timeout(TIMEOUT, stop_handle).await.unwrap().unwrap(); +} + #[tokio::test] async fn github_token_provider_uses_global_registration_and_maps_results() { let (client, server_read, server_write) = make_client();