diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 0fe740801..6e7d5ea30 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -131,9 +131,8 @@ public SessionCapabilities Capabilities /// Canvas instances currently known to be open for this session. /// /// - /// Populated from the most recent session.create / session.resume - /// response. This snapshot is not refreshed automatically when canvases open or - /// close after the session is established. + /// Populated from the most recent session.resume response and live + /// session.canvas.opened events. /// [Experimental(Diagnostics.Experimental)] public IReadOnlyList OpenCanvases => _openCanvases; @@ -473,6 +472,8 @@ public IDisposable On(Action handler) where T : SessionEvent /// internal void DispatchEvent(SessionEvent sessionEvent) { + UpdateOpenCanvasesFromEvent(sessionEvent); + // Fire broadcast work concurrently (fire-and-forget with error logging). // This is done outside the channel so broadcast handlers don't block the // consumer loop — important when a secondary client's handler intentionally @@ -889,6 +890,47 @@ internal void SetOpenCanvases(IList? canvases) : Array.Empty(); } + private void UpdateOpenCanvasesFromEvent(SessionEvent sessionEvent) + { + if (sessionEvent is not SessionCanvasOpenedEvent canvasEvent) + return; + + var data = canvasEvent.Data; + if (string.IsNullOrEmpty(data.InstanceId) + || string.IsNullOrEmpty(data.CanvasId) + || string.IsNullOrEmpty(data.ExtensionId) + || string.IsNullOrEmpty(data.Availability.Value)) + { + _logger.LogWarning("failed to deserialize session.canvas.opened payload"); + return; + } + + UpsertOpenCanvas(new OpenCanvasInstance + { + Availability = new CanvasInstanceAvailability(data.Availability.Value), + CanvasId = data.CanvasId, + ExtensionId = data.ExtensionId, + ExtensionName = data.ExtensionName, + Input = data.Input, + InstanceId = data.InstanceId, + Reopen = data.Reopen, + Status = data.Status, + Title = data.Title, + Url = data.Url, + }); + } + + private void UpsertOpenCanvas(OpenCanvasInstance canvas) + { + var canvases = _openCanvases.ToList(); + var index = canvases.FindIndex(open => open.InstanceId == canvas.InstanceId); + if (index >= 0) + canvases[index] = canvas; + else + canvases.Add(canvas); + _openCanvases = canvases.AsReadOnly(); + } + internal void SetCanvasHandler(ICanvasHandler? handler) { ClientSessionApis.Canvas = handler is null ? null : new CanvasHandlerAdapter(handler); diff --git a/dotnet/test/Unit/CanvasTests.cs b/dotnet/test/Unit/CanvasTests.cs index 72b995731..18dec1733 100644 --- a/dotnet/test/Unit/CanvasTests.cs +++ b/dotnet/test/Unit/CanvasTests.cs @@ -3,12 +3,14 @@ *--------------------------------------------------------------------------------------------*/ using System; +using System.IO; using System.Reflection; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using GitHub.Copilot; using GitHub.Copilot.Rpc; +using Microsoft.Extensions.Logging; using Xunit; namespace GitHub.Copilot.Test.Unit; @@ -25,6 +27,76 @@ private static JsonSerializerOptions GetSerializerOptions() return options!; } + private static CopilotSession CreateSession() + { + var options = GetSerializerOptions(); + var rpcType = typeof(CopilotClient).Assembly.GetType("GitHub.Copilot.JsonRpc"); + Assert.NotNull(rpcType); + + var inputStream = new MemoryStream(); + var outputStream = new MemoryStream(); + object? rpc; + try + { + rpc = Activator.CreateInstance( + rpcType!, + BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic, + binder: null, + args: [inputStream, outputStream, options, null], + culture: null); + Assert.NotNull(rpc); + } + catch + { + inputStream.Dispose(); + outputStream.Dispose(); + throw; + } + + var logger = new TestLogger(); + var ctor = typeof(CopilotSession).GetConstructor( + BindingFlags.Instance | BindingFlags.NonPublic, + binder: null, + types: [typeof(string), rpcType!, typeof(ILogger), typeof(CopilotClient), typeof(string)], + modifiers: null); + Assert.NotNull(ctor); + try + { + return (CopilotSession)ctor!.Invoke(["session-1", rpc, logger, new CopilotClient(), null]); + } + catch + { + inputStream.Dispose(); + outputStream.Dispose(); + throw; + } + } + + private sealed class TestLogger : ILogger + { + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => false; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + } + } + + private static void DispatchEvent(CopilotSession session, SessionEvent evt) + { + var method = typeof(CopilotSession).GetMethod( + "DispatchEvent", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(method); + method!.Invoke(session, [evt]); + } + [Fact] public void CanvasDeclaration_Serializes_CamelCase_SkippingNulls() { @@ -67,6 +139,96 @@ public void CanvasProviderOpenResult_Roundtrips_WithCamelCaseFields() Assert.Equal("ready", parsed.Status); } + [Fact] + public void SessionCanvasOpenedEvent_UpdatesOpenCanvasSnapshots() + { + var session = CreateSession(); + + DispatchEvent(session, new SessionCanvasOpenedEvent + { + Id = Guid.NewGuid(), + Timestamp = DateTimeOffset.UtcNow, + Data = new SessionCanvasOpenedData + { + Availability = CanvasOpenedAvailability.Ready, + CanvasId = "", + ExtensionId = "project:counter", + InstanceId = "missing-canvas-id", + Reopen = false, + } + }); + DispatchEvent(session, new SessionCanvasOpenedEvent + { + Id = Guid.NewGuid(), + Timestamp = DateTimeOffset.UtcNow, + Data = new SessionCanvasOpenedData + { + Availability = CanvasOpenedAvailability.Ready, + CanvasId = "counter", + ExtensionId = "project:counter", + ExtensionName = "Counter Provider", + InstanceId = "counter-1", + Title = "Counter", + Status = "ready", + Url = "https://example.test/counter", + Input = JsonDocument.Parse("""{"seed":1}""").RootElement.Clone(), + Reopen = false, + } + }); + DispatchEvent(session, new SessionCanvasOpenedEvent + { + Id = Guid.NewGuid(), + Timestamp = DateTimeOffset.UtcNow, + Data = new SessionCanvasOpenedData + { + Availability = CanvasOpenedAvailability.Stale, + CanvasId = "logs", + ExtensionId = "project:logs", + InstanceId = "logs-1", + Title = "Logs", + Reopen = false, + } + }); + + Assert.Collection( + session.OpenCanvases, + canvas => Assert.Equal("counter-1", canvas.InstanceId), + canvas => Assert.Equal("logs-1", canvas.InstanceId)); + + DispatchEvent(session, new SessionCanvasOpenedEvent + { + Id = Guid.NewGuid(), + Timestamp = DateTimeOffset.UtcNow, + Data = new SessionCanvasOpenedData + { + Availability = CanvasOpenedAvailability.Stale, + CanvasId = "counter", + ExtensionId = "project:counter", + ExtensionName = "Counter Provider", + InstanceId = "counter-1", + Title = "Counter Updated", + Status = "reconnected", + Url = "https://example.test/counter-updated", + Input = JsonDocument.Parse("""{"seed":2}""").RootElement.Clone(), + Reopen = true, + } + }); + + Assert.Collection( + session.OpenCanvases, + canvas => + { + Assert.Equal("counter-1", canvas.InstanceId); + Assert.Equal("Counter Updated", canvas.Title); + Assert.Equal("reconnected", canvas.Status); + Assert.Equal("https://example.test/counter-updated", canvas.Url); + Assert.True(canvas.Reopen); + Assert.Equal(CanvasInstanceAvailability.Stale, canvas.Availability); + Assert.Equal(2, canvas.Input!.Value.GetProperty("seed").GetInt32()); + }, + canvas => Assert.Equal("logs-1", canvas.InstanceId)); + } + [Fact] public void ExtensionInfo_Serializes_SourceAndName() { diff --git a/go/session.go b/go/session.go index e242cc1cf..fa03f5cc7 100644 --- a/go/session.go +++ b/go/session.go @@ -98,9 +98,9 @@ func (s *Session) WorkspacePath() string { return s.workspacePath } -// OpenCanvases returns the open-canvas snapshot last reported by the runtime -// (currently populated from the session.resume response). The returned slice -// is a copy and is safe to mutate by the caller. +// OpenCanvases returns the open-canvas snapshot last reported by the runtime. +// The snapshot is populated from session.resume and live session.canvas.opened +// events. The returned slice is a copy and is safe to mutate by the caller. func (s *Session) OpenCanvases() []rpc.OpenCanvasInstance { s.openCanvasesMu.RLock() defer s.openCanvasesMu.RUnlock() @@ -118,6 +118,41 @@ func (s *Session) setOpenCanvases(canvases []rpc.OpenCanvasInstance) { s.openCanvases = canvases } +func (s *Session) upsertOpenCanvas(canvas rpc.OpenCanvasInstance) { + s.openCanvasesMu.Lock() + defer s.openCanvasesMu.Unlock() + for i := range s.openCanvases { + if s.openCanvases[i].InstanceID == canvas.InstanceID { + s.openCanvases[i] = canvas + return + } + } + s.openCanvases = append(s.openCanvases, canvas) +} + +func (s *Session) updateOpenCanvasesFromEvent(event SessionEvent) { + data, ok := event.Data.(*SessionCanvasOpenedData) + if !ok { + return + } + if data.InstanceID == "" || data.CanvasID == "" || data.ExtensionID == "" || data.Availability == "" { + fmt.Printf("failed to deserialize session.canvas.opened payload\n") + return + } + s.upsertOpenCanvas(rpc.OpenCanvasInstance{ + Availability: rpc.CanvasInstanceAvailability(data.Availability), + CanvasID: data.CanvasID, + ExtensionID: data.ExtensionID, + ExtensionName: data.ExtensionName, + Input: data.Input, + InstanceID: data.InstanceID, + Reopen: data.Reopen, + Status: data.Status, + Title: data.Title, + URL: data.URL, + }) +} + func (s *Session) registerCanvasHandler(handler CanvasHandler) { s.canvasMu.Lock() defer s.canvasMu.Unlock() @@ -1110,6 +1145,7 @@ func fromRPCContent(value rpc.UIElicitationFieldValue) any { // are delivered by a single consumer goroutine (processEvents), guaranteeing // serial, FIFO dispatch without blocking the read loop. func (s *Session) dispatchEvent(event SessionEvent) { + s.updateOpenCanvasesFromEvent(event) go s.handleBroadcastEvent(event) // Send to the event channel in a closure with a recover guard. diff --git a/go/session_test.go b/go/session_test.go index b107fb62c..405d7bf7c 100644 --- a/go/session_test.go +++ b/go/session_test.go @@ -26,6 +26,10 @@ func newTestEvent() SessionEvent { return SessionEvent{Data: &SessionIdleData{}} } +func ptr[T any](value T) *T { + return &value +} + func TestSession_On(t *testing.T) { t.Run("multiple handlers all receive events", func(t *testing.T) { session, cleanup := newTestSession() @@ -435,6 +439,89 @@ func TestSession_Capabilities(t *testing.T) { t.Error("Expected UI.Elicitation to be false after second capabilities.changed event") } }) + + t.Run("session.canvas.opened event updates open canvas snapshots", func(t *testing.T) { + session, cleanup := newTestSession() + defer cleanup() + + session.dispatchEvent(SessionEvent{ + Data: &SessionCanvasOpenedData{ + InstanceID: "missing-canvas-id", + ExtensionID: "project:counter", + Availability: CanvasOpenedAvailabilityReady, + }, + }) + session.dispatchEvent(SessionEvent{ + Data: &SessionCanvasOpenedData{ + ExtensionID: "project:counter", + ExtensionName: ptr("Counter Provider"), + CanvasID: "counter", + InstanceID: "counter-1", + Title: ptr("Counter"), + Status: ptr("ready"), + URL: ptr("https://example.test/counter"), + Input: map[string]any{"seed": float64(1)}, + Reopen: false, + Availability: CanvasOpenedAvailabilityReady, + }, + }) + session.dispatchEvent(SessionEvent{ + Data: &SessionCanvasOpenedData{ + ExtensionID: "project:logs", + CanvasID: "logs", + InstanceID: "logs-1", + Title: ptr("Logs"), + Reopen: false, + Availability: CanvasOpenedAvailabilityStale, + }, + }) + + open := session.OpenCanvases() + if len(open) != 2 { + t.Fatalf("expected 2 open canvases, got %d", len(open)) + } + if open[0].InstanceID != "counter-1" || open[1].InstanceID != "logs-1" { + t.Fatalf("unexpected open canvas order: %+v", open) + } + + session.dispatchEvent(SessionEvent{ + Data: &SessionCanvasOpenedData{ + ExtensionID: "project:counter", + ExtensionName: ptr("Counter Provider"), + CanvasID: "counter", + InstanceID: "counter-1", + Title: ptr("Counter Updated"), + Status: ptr("reconnected"), + URL: ptr("https://example.test/counter-updated"), + Input: map[string]any{"seed": float64(2)}, + Reopen: true, + Availability: CanvasOpenedAvailabilityStale, + }, + }) + + open = session.OpenCanvases() + if len(open) != 2 { + t.Fatalf("expected 2 open canvases after upsert, got %d", len(open)) + } + if open[0].InstanceID != "counter-1" || open[1].InstanceID != "logs-1" { + t.Fatalf("upsert should preserve order, got %+v", open) + } + if open[0].Title == nil || *open[0].Title != "Counter Updated" { + t.Fatalf("expected updated title, got %+v", open[0].Title) + } + if open[0].Status == nil || *open[0].Status != "reconnected" { + t.Fatalf("expected updated status, got %+v", open[0].Status) + } + if open[0].URL == nil || *open[0].URL != "https://example.test/counter-updated" { + t.Fatalf("expected updated URL, got %+v", open[0].URL) + } + if !open[0].Reopen { + t.Fatal("expected reopen to be true") + } + if string(open[0].Availability) != string(CanvasOpenedAvailabilityStale) { + t.Fatalf("expected stale availability, got %q", open[0].Availability) + } + }) } // waitForCapability polls Session.Capabilities() until predicate matches or timeout. diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index b3d99271b..b7b9c217a 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -70,6 +70,23 @@ function deserializeHookInput(raw: unknown): unknown { return { ...rest, timestamp: new Date(obj.timestamp), workingDirectory: cwd }; } +function isOpenCanvasInstance(value: unknown): value is OpenCanvasInstance { + if (!value || typeof value !== "object") { + return false; + } + const instance = value as Partial; + return ( + typeof instance.instanceId === "string" && + instance.instanceId.length > 0 && + typeof instance.extensionId === "string" && + instance.extensionId.length > 0 && + typeof instance.canvasId === "string" && + instance.canvasId.length > 0 && + typeof instance.reopen === "boolean" && + (instance.availability === "ready" || instance.availability === "stale") + ); +} + /** Assistant message event - the final response from the assistant. */ export type AssistantMessageEvent = Extract; @@ -487,6 +504,27 @@ export class CopilotSession { } } else if (event.type === "capabilities.changed") { this._capabilities = { ...this._capabilities, ...event.data }; + } else if (event.type === "session.canvas.opened") { + this.upsertOpenCanvasFromEvent(event.data); + } + } + + private upsertOpenCanvasFromEvent(data: unknown): void { + if (!isOpenCanvasInstance(data)) { + console.warn("failed to deserialize session.canvas.opened payload"); + return; + } + this.upsertOpenCanvas(data); + } + + private upsertOpenCanvas(instance: OpenCanvasInstance): void { + const index = this.openCanvasInstances.findIndex( + (open) => open.instanceId === instance.instanceId + ); + if (index >= 0) { + this.openCanvasInstances[index] = instance; + } else { + this.openCanvasInstances.push(instance); } } @@ -810,10 +848,10 @@ export class CopilotSession { } /** - * Snapshot of canvas instances that were already open when the session was - * resumed. Populated from the `session.resume` response; empty for freshly - * created sessions. Returns a defensive copy — mutating the returned array - * has no effect on the session. + * Snapshot of canvas instances currently known to be open for this session. + * Populated from the `session.resume` response and live `session.canvas.opened` + * events. Returns a defensive copy — mutating the returned array has no effect + * on the session. */ get openCanvases(): OpenCanvasInstance[] { return [...this.openCanvasInstances]; diff --git a/nodejs/test/client.test.ts b/nodejs/test/client.test.ts index ac29c6aa1..f8490e47a 100644 --- a/nodejs/test/client.test.ts +++ b/nodejs/test/client.test.ts @@ -214,6 +214,77 @@ describe("CopilotClient", () => { expect(result).toEqual({ actionName: "increment", input: { amount: 1 } }); }); + it("tracks open canvases from live session.canvas.opened events", () => { + const session = new CopilotSession("session-1", {} as any); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + + (session as any)._dispatchEvent({ + type: "session.canvas.opened", + data: { instanceId: "missing-required-fields" }, + }); + (session as any)._dispatchEvent({ + type: "session.canvas.opened", + data: { + extensionId: "project:counter", + extensionName: "Counter Provider", + canvasId: "counter", + instanceId: "counter-1", + title: "Counter", + status: "ready", + url: "https://example.test/counter", + input: { seed: 1 }, + reopen: false, + availability: "ready", + }, + }); + (session as any)._dispatchEvent({ + type: "session.canvas.opened", + data: { + extensionId: "project:logs", + canvasId: "logs", + instanceId: "logs-1", + title: "Logs", + reopen: false, + availability: "stale", + }, + }); + + expect(warn).toHaveBeenCalledWith("failed to deserialize session.canvas.opened payload"); + expect(session.openCanvases.map((canvas) => canvas.instanceId)).toEqual([ + "counter-1", + "logs-1", + ]); + + (session as any)._dispatchEvent({ + type: "session.canvas.opened", + data: { + extensionId: "project:counter", + extensionName: "Counter Provider", + canvasId: "counter", + instanceId: "counter-1", + title: "Counter Updated", + status: "reconnected", + url: "https://example.test/counter-updated", + input: { seed: 2 }, + reopen: true, + availability: "stale", + }, + }); + + expect(session.openCanvases).toHaveLength(2); + expect(session.openCanvases[0]).toMatchObject({ + instanceId: "counter-1", + title: "Counter Updated", + status: "reconnected", + url: "https://example.test/counter-updated", + input: { seed: 2 }, + reopen: true, + availability: "stale", + }); + expect(session.openCanvases[1].instanceId).toBe("logs-1"); + warn.mockRestore(); + }); + it("returns canvas_action_no_handler when no per-action handler is registered", async () => { const canvas = createCanvas({ id: "counter", diff --git a/python/copilot/session.py b/python/copilot/session.py index 6364561c2..5aac18bfa 100644 --- a/python/copilot/session.py +++ b/python/copilot/session.py @@ -64,6 +64,7 @@ ExternalToolRequestedData, PermissionRequest, PermissionRequestedData, + SessionCanvasOpenedData, SessionErrorData, SessionEvent, SessionIdleData, @@ -1546,6 +1547,19 @@ def _handle_broadcast_event(self, event: SessionEvent) -> None: cap["ui"] = ui_cap self._capabilities = {**self._capabilities, **cap} + case SessionCanvasOpenedData() as data: + try: + if ( + not data.instance_id + or not data.canvas_id + or not data.extension_id + or data.availability is None + ): + raise ValueError("missing required open canvas fields") + self._upsert_open_canvas(OpenCanvasInstance.from_dict(data.to_dict())) + except Exception as exc: + logger.warning("failed to deserialize session.canvas.opened payload: %s", exc) + async def _execute_tool_and_respond( self, request_id: str, @@ -1889,12 +1903,19 @@ def _set_open_canvases(self, instances: list[OpenCanvasInstance]) -> None: with self._open_canvases_lock: self._open_canvases = list(instances) + def _upsert_open_canvas(self, instance: OpenCanvasInstance) -> None: + with self._open_canvases_lock: + for index, existing in enumerate(self._open_canvases): + if existing.instance_id == instance.instance_id: + self._open_canvases[index] = instance + return + self._open_canvases.append(instance) + @property def open_canvases(self) -> list[OpenCanvasInstance]: - """Open canvas instances reported by the most recent ``session.resume``. + """Open canvas instances currently known to be open for this session. - Returns an empty list for sessions created via ``session.create`` or - when the server did not include any open canvases on resume. + Populated from ``session.resume`` and live ``session.canvas.opened`` events. """ with self._open_canvases_lock: return list(self._open_canvases) diff --git a/python/test_canvas.py b/python/test_canvas.py index 08cc816ec..9e12a1850 100644 --- a/python/test_canvas.py +++ b/python/test_canvas.py @@ -2,7 +2,9 @@ from __future__ import annotations +from datetime import UTC, datetime from typing import Any, cast +from uuid import uuid4 import pytest @@ -22,6 +24,12 @@ CanvasProviderOpenRequest, CanvasProviderOpenResult, ) +from copilot.generated.session_events import ( + CanvasOpenedAvailability, + SessionCanvasOpenedData, + SessionEvent, + SessionEventType, +) from copilot.session import CopilotSession @@ -205,3 +213,93 @@ def test_set_open_canvases_round_trip(): session = CopilotSession("sess-1", client=None) session._set_open_canvases([inst]) assert session.open_canvases == [inst] + + +def test_session_canvas_opened_updates_open_canvases(caplog: pytest.LogCaptureFixture): + session = CopilotSession("sess-1", client=None) + + session._dispatch_event( + SessionEvent( + data=SessionCanvasOpenedData( + availability=CanvasOpenedAvailability.READY, + canvas_id="", + extension_id="project:counter", + instance_id="missing-canvas-id", + reopen=False, + ), + id=uuid4(), + timestamp=datetime.now(UTC), + type=SessionEventType.SESSION_CANVAS_OPENED, + ) + ) + session._dispatch_event( + SessionEvent( + data=SessionCanvasOpenedData( + availability=CanvasOpenedAvailability.READY, + canvas_id="counter", + extension_id="project:counter", + extension_name="Counter Provider", + instance_id="counter-1", + reopen=False, + input={"seed": 1}, + status="ready", + title="Counter", + url="https://example.test/counter", + ), + id=uuid4(), + timestamp=datetime.now(UTC), + type=SessionEventType.SESSION_CANVAS_OPENED, + ) + ) + session._dispatch_event( + SessionEvent( + data=SessionCanvasOpenedData( + availability=CanvasOpenedAvailability.STALE, + canvas_id="logs", + extension_id="project:logs", + instance_id="logs-1", + reopen=False, + title="Logs", + ), + id=uuid4(), + timestamp=datetime.now(UTC), + type=SessionEventType.SESSION_CANVAS_OPENED, + ) + ) + + assert "failed to deserialize session.canvas.opened payload" in caplog.text + assert [canvas.instance_id for canvas in session.open_canvases] == [ + "counter-1", + "logs-1", + ] + + session._dispatch_event( + SessionEvent( + data=SessionCanvasOpenedData( + availability=CanvasOpenedAvailability.STALE, + canvas_id="counter", + extension_id="project:counter", + extension_name="Counter Provider", + instance_id="counter-1", + reopen=True, + input={"seed": 2}, + status="reconnected", + title="Counter Updated", + url="https://example.test/counter-updated", + ), + id=uuid4(), + timestamp=datetime.now(UTC), + type=SessionEventType.SESSION_CANVAS_OPENED, + ) + ) + + open_canvases = session.open_canvases + assert len(open_canvases) == 2 + assert open_canvases[0].instance_id == "counter-1" + assert open_canvases[0].title == "Counter Updated" + assert open_canvases[0].status == "reconnected" + assert open_canvases[0].url == "https://example.test/counter-updated" + assert open_canvases[0].input == {"seed": 2} + assert open_canvases[0].reopen is True + assert open_canvases[0].availability == CanvasInstanceAvailability.STALE + assert open_canvases[1].instance_id == "logs-1" diff --git a/rust/src/session.rs b/rust/src/session.rs index 1fcf433a5..c1e845d41 100644 --- a/rust/src/session.rs +++ b/rust/src/session.rs @@ -886,6 +886,7 @@ impl Client { let setup_start = Instant::now(); let capabilities = Arc::new(parking_lot::RwLock::new(SessionCapabilities::default())); let idle_waiter = Arc::new(ParkingLotMutex::new(None)); + let open_canvases = Arc::new(parking_lot::RwLock::new(Vec::new())); let shutdown = CancellationToken::new(); let (event_tx, _) = tokio::sync::broadcast::channel(512); @@ -985,6 +986,7 @@ impl Client { channels, idle_waiter.clone(), capabilities.clone(), + open_canvases.clone(), event_tx.clone(), shutdown.clone(), ); @@ -1013,7 +1015,7 @@ impl Client { shutdown, idle_waiter, capabilities, - open_canvases: Arc::new(parking_lot::RwLock::new(Vec::new())), + open_canvases, event_tx, }; apply_mode_post_create_patch( @@ -1114,6 +1116,7 @@ impl Client { let setup_start = Instant::now(); let channels = self.register_session(&session_id); let idle_waiter = Arc::new(ParkingLotMutex::new(None)); + let open_canvases = Arc::new(parking_lot::RwLock::new(Vec::new())); let shutdown = CancellationToken::new(); let (event_tx, _) = tokio::sync::broadcast::channel(512); let event_loop = spawn_event_loop( @@ -1128,6 +1131,7 @@ impl Client { channels, idle_waiter.clone(), capabilities.clone(), + open_canvases.clone(), event_tx.clone(), shutdown.clone(), ); @@ -1200,9 +1204,16 @@ impl Client { } *capabilities.write() = resume_result.capabilities.unwrap_or_default(); - let open_canvases = Arc::new(parking_lot::RwLock::new( - resume_result.open_canvases.unwrap_or_default(), - )); + // Upsert resume snapshots rather than replacing wholesale. Live + // `session.canvas.opened` notifications can arrive on the event loop + // while `session.resume` is in flight; a wholesale replace would + // discard those updates. + { + let mut snapshots = open_canvases.write(); + for snapshot in resume_result.open_canvases.unwrap_or_default() { + upsert_open_canvas_snapshot(&mut snapshots, snapshot); + } + } tracing::debug!( elapsed_ms = total_start.elapsed().as_millis(), @@ -1297,6 +1308,20 @@ fn build_command_handler_map(commands: Option<&[CommandDefinition]>) -> Arc, + snapshot: OpenCanvasInstance, +) { + if let Some(existing) = snapshots + .iter_mut() + .find(|open| open.instance_id == snapshot.instance_id) + { + *existing = snapshot; + } else { + snapshots.push(snapshot); + } +} + #[allow(clippy::too_many_arguments)] fn spawn_event_loop( session_id: SessionId, @@ -1310,6 +1335,7 @@ fn spawn_event_loop( channels: crate::router::SessionChannels, idle_waiter: Arc>>, capabilities: Arc>, + open_canvases: Arc>>, event_tx: tokio::sync::broadcast::Sender, shutdown: CancellationToken, ) -> JoinHandle<()> { @@ -1336,7 +1362,7 @@ fn spawn_event_loop( _ = shutdown.cancelled() => break, Some(notification) = notifications.recv() => { handle_notification( - &session_id, &client, &handlers, &command_handlers, notification, &idle_waiter, &capabilities, &event_tx, + &session_id, &client, &handlers, &command_handlers, notification, &idle_waiter, &capabilities, &open_canvases, &event_tx, ).await; } Some(request) = requests.recv() => { @@ -1407,6 +1433,7 @@ async fn handle_notification( notification: SessionEventNotification, idle_waiter: &Arc>>, capabilities: &Arc>, + open_canvases: &Arc>>, event_tx: &tokio::sync::broadcast::Sender, ) { let dispatch_start = Instant::now(); @@ -1475,20 +1502,28 @@ async fn handle_notification( _ => {} } - // Fan out the event to runtime subscribers (`Session::subscribe`). `send` - // only errors when there are no receivers, which is the normal case - // before any consumer subscribes. - let _ = event_tx.send(event.clone()); - - // Update capabilities when the CLI reports changes. The CLI sends - // the full updated capabilities object — replace wholesale so removals - // and new subfields are handled correctly. + // Update the snapshot caches BEFORE broadcasting so subscribers that + // call `Session::capabilities()` / `Session::open_canvases()` in + // response to the event observe the new state. if event_type == SessionEventType::CapabilitiesChanged { match serde_json::from_value::(notification.event.data.clone()) { Ok(changed) => *capabilities.write() = changed, Err(e) => warn!(error = %e, "failed to deserialize capabilities.changed payload"), } } + if event_type == SessionEventType::SessionCanvasOpened { + match serde_json::from_value::(notification.event.data.clone()) { + Ok(open_canvas) => { + upsert_open_canvas_snapshot(&mut open_canvases.write(), open_canvas); + } + Err(e) => warn!(error = %e, "failed to deserialize session.canvas.opened payload"), + } + } + + // Fan out the event to runtime subscribers (`Session::subscribe`). `send` + // only errors when there are no receivers, which is the normal case + // before any consumer subscribes. + let _ = event_tx.send(event.clone()); tracing::debug!( elapsed_ms = dispatch_start.elapsed().as_millis(), diff --git a/rust/tests/session_test.rs b/rust/tests/session_test.rs index 85d9a6f3f..caa63e84f 100644 --- a/rust/tests/session_test.rs +++ b/rust/tests/session_test.rs @@ -2643,6 +2643,103 @@ async fn resume_session_sends_canvas_fields_and_captures_open_canvases() { assert_eq!(caps.ui.unwrap().canvases, Some(true)); } +#[tokio::test] +async fn session_canvas_opened_updates_open_canvas_snapshots() { + let (session, mut server) = create_session_pair().await; + assert!(session.open_canvases().is_empty()); + + server + .send_event( + "session.canvas.opened", + serde_json::json!({ + "instanceId": "missing-required-fields", + }), + ) + .await; + server + .send_event( + "session.canvas.opened", + serde_json::json!({ + "extensionId": "project:counter", + "extensionName": "Counter Provider", + "canvasId": "counter", + "instanceId": "counter-1", + "title": "Counter", + "status": "ready", + "url": "https://example.test/counter", + "input": { "seed": 1 }, + "reopen": false, + "availability": "ready" + }), + ) + .await; + server + .send_event( + "session.canvas.opened", + serde_json::json!({ + "extensionId": "project:logs", + "canvasId": "logs", + "instanceId": "logs-1", + "title": "Logs", + "reopen": false, + "availability": "stale" + }), + ) + .await; + + let mut open = Vec::new(); + for _ in 0..50 { + open = session.open_canvases(); + if open.len() == 2 { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!(open.len(), 2); + assert_eq!(open[0].instance_id, "counter-1"); + assert_eq!(open[0].title.as_deref(), Some("Counter")); + assert_eq!(open[0].availability, CanvasInstanceAvailability::Ready); + assert_eq!(open[1].instance_id, "logs-1"); + + server + .send_event( + "session.canvas.opened", + serde_json::json!({ + "extensionId": "project:counter", + "extensionName": "Counter Provider", + "canvasId": "counter", + "instanceId": "counter-1", + "title": "Counter Updated", + "status": "reconnected", + "url": "https://example.test/counter-updated", + "input": { "seed": 2 }, + "reopen": true, + "availability": "stale" + }), + ) + .await; + + for _ in 0..50 { + open = session.open_canvases(); + if open.len() == 2 && open[0].title.as_deref() == Some("Counter Updated") { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!(open.len(), 2); + assert_eq!(open[0].instance_id, "counter-1"); + assert_eq!(open[0].title.as_deref(), Some("Counter Updated")); + assert_eq!(open[0].status.as_deref(), Some("reconnected")); + assert_eq!( + open[0].url.as_deref(), + Some("https://example.test/counter-updated") + ); + assert_eq!(open[0].input, Some(serde_json::json!({ "seed": 2 }))); + assert!(open[0].reopen); + assert_eq!(open[0].availability, CanvasInstanceAvailability::Stale); + assert_eq!(open[1].instance_id, "logs-1"); +} + #[tokio::test] async fn elicitation_methods_fail_without_capability() { let (session, _server) = create_session_pair().await;