diff --git a/provider/copilotprovider/copilot.go b/provider/copilotprovider/copilot.go index adb1daa3..9ca20524 100644 --- a/provider/copilotprovider/copilot.go +++ b/provider/copilotprovider/copilot.go @@ -90,14 +90,17 @@ func (p *provider) run(ctx context.Context, messages []*message.Message, options frameworkSession, _ := agent.GetOption(options, agent.WithSession) isStreaming := p.streaming(options) - copilotSession, err := p.openSession(ctx, frameworkSession, isStreaming, options) + // Register eventHandler through SessionConfig.OnEvent so it is attached + // before the session.create/resume RPC. This matches the SDK's + // no-missed-events guarantee: lifecycle events the CLI emits during + // session creation (e.g. session.start) would otherwise be delivered to + // zero handlers in the window before a post-return session.On call. + copilotSession, err := p.openSession(ctx, frameworkSession, isStreaming, eventHandler, options) if err != nil { yield(nil, err) return } defer func() { _ = copilotSession.Disconnect() }() - unsubscribe := copilotSession.On(eventHandler) - defer unsubscribe() if frameworkSession != nil && frameworkSession.ServiceID() == "" { frameworkSession.SetServiceID(copilotSession.SessionID) @@ -221,27 +224,30 @@ func (p *provider) openSession( ctx context.Context, frameworkSession *agent.Session, streaming bool, + eventHandler copilot.SessionEventHandler, options []agent.Option, ) (*copilot.Session, error) { if frameworkSession != nil && frameworkSession.ServiceID() != "" { - cfg := p.resumeSessionConfig(streaming, options) + cfg := p.resumeSessionConfig(streaming, eventHandler, options) return p.client.ResumeSession(ctx, frameworkSession.ServiceID(), &cfg) } - cfg := p.sessionConfig(streaming, options) + cfg := p.sessionConfig(streaming, eventHandler, options) return p.client.CreateSession(ctx, &cfg) } -func (p *provider) sessionConfig(streaming bool, options []agent.Option) copilot.SessionConfig { +func (p *provider) sessionConfig(streaming bool, eventHandler copilot.SessionEventHandler, options []agent.Option) copilot.SessionConfig { cfg := copySessionConfig(p.cfg.SessionConfig) cfg.Streaming = copilot.Bool(streaming) + cfg.OnEvent = chainSessionEventHandlers(cfg.OnEvent, eventHandler) cfg.SystemMessage = systemMessageWithInstructions(cfg.SystemMessage, slices.Collect(agent.AllOptions(options, agent.WithInstructions))) cfg.Tools = append(cfg.Tools, copilotTools(options)...) return cfg } -func (p *provider) resumeSessionConfig(streaming bool, options []agent.Option) copilot.ResumeSessionConfig { +func (p *provider) resumeSessionConfig(streaming bool, eventHandler copilot.SessionEventHandler, options []agent.Option) copilot.ResumeSessionConfig { cfg := copyResumeSessionConfig(p.cfg.SessionConfig) cfg.Streaming = copilot.Bool(streaming) + cfg.OnEvent = chainSessionEventHandlers(cfg.OnEvent, eventHandler) cfg.SystemMessage = systemMessageWithInstructions(cfg.SystemMessage, slices.Collect(agent.AllOptions(options, agent.WithInstructions))) cfg.Tools = append(cfg.Tools, copilotTools(options)...) return cfg @@ -268,6 +274,7 @@ func copyResumeSessionConfig(source *copilot.SessionConfig) copilot.ResumeSessio AvailableTools: source.AvailableTools, ExcludedTools: source.ExcludedTools, Provider: source.Provider, + OnEvent: source.OnEvent, OnPermissionRequest: source.OnPermissionRequest, OnUserInputRequest: source.OnUserInputRequest, Hooks: source.Hooks, @@ -282,6 +289,23 @@ func copyResumeSessionConfig(source *copilot.SessionConfig) copilot.ResumeSessio } } +// chainSessionEventHandlers composes a caller-supplied handler with the +// per-run handler so a user-configured SessionConfig.OnEvent is preserved +// and runs alongside (before) the provider's per-run handler rather than +// being overwritten. +func chainSessionEventHandlers(existing, added copilot.SessionEventHandler) copilot.SessionEventHandler { + if existing == nil { + return added + } + if added == nil { + return existing + } + return func(event copilot.SessionEvent) { + existing(event) + added(event) + } +} + func copyBoolDefaultTrue(source *bool) *bool { if source == nil { return copilot.Bool(true) diff --git a/provider/copilotprovider/copilot_test.go b/provider/copilotprovider/copilot_test.go index ee27ca1e..8b9e4645 100644 --- a/provider/copilotprovider/copilot_test.go +++ b/provider/copilotprovider/copilot_test.go @@ -215,6 +215,61 @@ func TestCopyResumeSessionConfig_WithStreamingNull_DefaultsToTrue(t *testing.T) assertEqual(t, runtime.lastResumeRequest(t)["streaming"], true, "streaming") } +func TestRun_SurfacesLifecycleEventEmittedDuringSessionResume(t *testing.T) { + runtime := newFakeRuntime(t, idleEvent()) + runtime.resumeEvents = []map[string]any{sessionEvent("session.start", map[string]any{})} + agent := copilotprovider.NewAgent(runtime.client(), copilotprovider.AgentConfig{}) + session, err := agent.CreateSession(context.Background(), agentpkg.WithServiceID("existing-session")) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + + response, err := runText(t, agent, "hello", agentpkg.WithSession(session)) + if err != nil { + t.Fatalf("RunText: %v", err) + } + if !hasRawEventOfType(response, "session.start") { + t.Fatal("lifecycle event emitted before the session.resume response was dropped; OnEvent must be registered before the RPC") + } +} + +func TestRun_PreservesUserSuppliedOnEventHandler(t *testing.T) { + runtime := newFakeRuntime(t, idleEvent()) + var mu sync.Mutex + invocations := 0 + userHandler := func(event copilot.SessionEvent) { + mu.Lock() + invocations++ + mu.Unlock() + } + agent := copilotprovider.NewAgent(runtime.client(), copilotprovider.AgentConfig{ + SessionConfig: &copilot.SessionConfig{OnEvent: userHandler}, + }) + + if _, err := runText(t, agent, "hello"); err != nil { + t.Fatalf("RunText: %v", err) + } + + mu.Lock() + defer mu.Unlock() + if invocations == 0 { + t.Fatal("user-supplied SessionConfig.OnEvent was overwritten by the per-run handler and never invoked") + } +} + +func hasRawEventOfType(response *agentpkg.Response, eventType string) bool { + for content := range response.Contents() { + raw, ok := content.(*message.RawContent) + if !ok { + continue + } + if event, ok := raw.RawRepresentation.(copilot.SessionEvent); ok && string(event.Type()) == eventType { + return true + } + } + return false +} + func TestConvertToAgentResponseUpdate_AssistantMessageEventWhenStreaming_DoesNotEmitTextContent(t *testing.T) { runtime := newFakeRuntime(t, sessionEvent("assistant.message", map[string]any{"messageId": "msg-456", "content": "Some streamed content that was already delivered via delta events"}), @@ -675,6 +730,7 @@ type fakeRuntime struct { mu sync.Mutex sessionID string events []map[string]any + resumeEvents []map[string]any createRequests []map[string]any resumeRequests []map[string]any sendRequests []map[string]any @@ -791,7 +847,14 @@ func (r *fakeRuntime) handle(conn net.Conn, req jsonRPCRequest) { r.mu.Lock() r.sessionID = sessionID r.resumeRequests = append(r.resumeRequests, params) + resumeEvents := append([]map[string]any(nil), r.resumeEvents...) r.mu.Unlock() + // Emit any lifecycle events the CLI produces during session.resume + // before the RPC response, so they land in the window before a + // post-return session.On call would have registered a handler. + for _, event := range resumeEvents { + writeNotification(r.t, conn, "session.event", map[string]any{"sessionId": sessionID, "event": event}) + } writeResponse(r.t, conn, req.ID, map[string]any{"sessionId": sessionID, "workspacePath": ""}) case "session.send": params := decodeParams(r.t, req.Params)