diff --git a/dotnet/README.md b/dotnet/README.md index 9b9ca42c60..87ded18b52 100644 --- a/dotnet/README.md +++ b/dotnet/README.md @@ -250,9 +250,23 @@ Send a message to the session. - `Prompt` - The message/prompt to send - `Attachments` - File attachments - `Mode` - Delivery mode ("enqueue" or "immediate") +- `Source` - Optional message origin: `MessageSource.User` or `MessageSource.System`. Omitted by default, preserving the runtime's default user behavior. Returns the message ID. +Use `MessageSource.System` for application-generated context. This marks the +message's origin; it does not replace the session's system prompt or change +delivery mode. `SendAndWaitAsync` accepts the same option and still waits for +session idle, returning null if no assistant message was received. + +```csharp +await session.SendAsync(new MessageOptions +{ + Prompt = "The background build completed successfully.", + Source = MessageSource.System, +}); +``` + ##### `On(Action handler): IDisposable` Subscribe to session events. Returns a disposable to unsubscribe. diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 8637da948c..404c0054b7 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -327,6 +327,7 @@ public async Task SendAsync(MessageOptions options, CancellationToken ca Attachments = options.Attachments, Mode = options.Mode, AgentMode = options.AgentMode, + Source = options.Source, Traceparent = traceparent, Tracestate = tracestate, RequestHeaders = options.RequestHeaders, @@ -2270,6 +2271,7 @@ internal record SendMessageRequest public string? Mode { get; init; } [JsonPropertyName("agentMode")] public AgentMode? AgentMode { get; init; } + public MessageSource? Source { get; init; } public string? Traceparent { get; init; } public string? Tracestate { get; init; } public IDictionary? RequestHeaders { get; init; } diff --git a/dotnet/src/Types.cs b/dotnet/src/Types.cs index 9129b118e8..3b03e3192c 100644 --- a/dotnet/src/Types.cs +++ b/dotnet/src/Types.cs @@ -2133,6 +2133,20 @@ public enum AgentMode Shell } +/// +/// Identifies the origin of a message sent to a session. +/// +[JsonConverter(typeof(JsonStringEnumConverter))] +public enum MessageSource +{ + /// The message originates from user input. + [JsonStringEnumMemberName("user")] + User, + /// The message provides application-generated context. + [JsonStringEnumMemberName("system")] + System +} + /// /// Specifies the operation to perform on a system message section. /// @@ -4029,6 +4043,7 @@ private MessageOptions(MessageOptions? other) Attachments = other.Attachments is not null ? [.. other.Attachments] : null; Mode = other.Mode; AgentMode = other.AgentMode; + Source = other.Source; Prompt = other.Prompt; DisplayPrompt = other.DisplayPrompt; RequestHeaders = other.RequestHeaders is not null @@ -4055,6 +4070,11 @@ private MessageOptions(MessageOptions? other) /// public AgentMode? AgentMode { get; set; } /// + /// The message's origin. When unset, the field is omitted and the runtime defaults to user input. + /// This tags message provenance; it does not replace the session's system prompt or change delivery mode. + /// + public MessageSource? Source { get; set; } + /// /// Custom per-turn HTTP headers for outbound model requests. /// public IDictionary? RequestHeaders { get; set; } diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index 04ef6e6405..78929cf225 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/dotnet/test/Unit/ClientSessionLifetimeTests.cs @@ -1477,6 +1477,217 @@ public async Task Generated_Session_Rpc_Throws_When_Session_Disposed() await Assert.ThrowsAsync(() => session.Rpc.Model.GetCurrentAsync()); } + [Fact] + public async Task SendAsync_MessageSource_Is_Omitted_By_Default() + { + await using var server = await FakeCopilotServer.StartAsync(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + await using var session = await client.CreateSessionAsync(new SessionConfig()); + var options = new MessageOptions { Prompt = "User input" }; + Assert.Null(options.Source); + Assert.Null(options.Clone().Source); + + await session.SendAsync(options); + await session.SendAsync("More user input"); + + var requests = server.Requests.Where(request => request.Method == "session.send").ToArray(); + Assert.Equal(2, requests.Length); + Assert.All(requests, request => AssertMessageSource(request.Params, null)); + } + + [Theory] + [InlineData(null, null)] + [InlineData(MessageSource.User, "user")] + [InlineData(MessageSource.System, "system")] + public async Task SendAsync_MessageSource_Preserves_Other_Options(MessageSource? source, string? wireSource) + { + await using var server = await FakeCopilotServer.StartAsync(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + await using var session = await client.CreateSessionAsync(new SessionConfig()); + var options = new MessageOptions { Prompt = "Background context", Source = source }; + + Assert.Equal("message-1", await session.SendAsync(options)); + var request = Assert.Single(server.Requests, request => request.Method == "session.send").Params; + AssertMessageSource(request, wireSource); + foreach (var property in new[] { "mode", "agentMode", "attachments", "displayPrompt", "requestHeaders" }) + { + Assert.False(request.TryGetProperty(property, out _)); + } + + using var activity = new Activity("message-source-test").SetIdFormat(ActivityIdFormat.W3C); + activity.TraceStateString = "test=message-source"; + activity.Start(); + + foreach (var mode in new[] { "enqueue", "immediate" }) + { + server.ClearRequests(); + options.Mode = mode; + options.AgentMode = AgentMode.Plan; + options.DisplayPrompt = "Background update"; + options.Attachments = [new AttachmentFile { Path = "/context.txt", DisplayName = "context.txt" }]; + options.RequestHeaders = new Dictionary { ["X-Test"] = "source-parity" }; + + Assert.Equal("message-1", await session.SendAsync(options)); + + request = Assert.Single(server.Requests, request => request.Method == "session.send").Params; + AssertMessageSource(request, wireSource); + Assert.Equal(session.SessionId, request.GetProperty("sessionId").GetString()); + Assert.Equal(options.Prompt, request.GetProperty("prompt").GetString()); + Assert.Equal(mode, request.GetProperty("mode").GetString()); + Assert.Equal("plan", request.GetProperty("agentMode").GetString()); + Assert.Equal(options.DisplayPrompt, request.GetProperty("displayPrompt").GetString()); + Assert.Equal("source-parity", request.GetProperty("requestHeaders").GetProperty("X-Test").GetString()); + var attachment = Assert.Single(request.GetProperty("attachments").EnumerateArray()); + Assert.Equal("file", attachment.GetProperty("type").GetString()); + Assert.Equal("/context.txt", attachment.GetProperty("path").GetString()); + Assert.Equal("context.txt", attachment.GetProperty("displayName").GetString()); + Assert.Equal(activity.Id, request.GetProperty("traceparent").GetString()); + Assert.Equal(activity.TraceStateString, request.GetProperty("tracestate").GetString()); + Assert.Equal(source, options.Source); + } + } + + [Theory] + [InlineData(null)] + [InlineData("user")] + [InlineData("system")] + public async Task Raw_SendAsync_MessageSource_Remains_Available(string? source) + { + await using var server = await FakeCopilotServer.StartAsync(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + await using var session = await client.CreateSessionAsync(new SessionConfig()); + + var result = await session.Rpc.SendAsync("Context", source: source); + + Assert.Equal("message-1", result.MessageId); + AssertMessageSource(Assert.Single(server.Requests, request => request.Method == "session.send").Params, source); + } + + [Theory] + [InlineData(null, false)] + [InlineData(null, true)] + [InlineData(MessageSource.User, false)] + [InlineData(MessageSource.User, true)] + [InlineData(MessageSource.System, false)] + [InlineData(MessageSource.System, true)] + public async Task SendAndWaitAsync_MessageSource_Completes_On_Idle(MessageSource? source, bool hasAssistantMessage) + { + await using var server = await FakeCopilotServer.StartAsync(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + await using var session = await client.CreateSessionAsync(new SessionConfig()); + var assistantReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = session.On(_ => assistantReceived.TrySetResult()); + + var sendTask = session.SendAndWaitAsync(new MessageOptions { Prompt = "Context", Source = source }); + var request = await WaitForRequestAsync(server, "session.send"); + AssertMessageSource(request.Params, source?.ToString().ToLowerInvariant()); + + if (hasAssistantMessage) + { + await server.SendSessionEventAsync(session.SessionId, "assistant.message", new() + { + ["messageId"] = "assistant-1", + ["content"] = "Acknowledged" + }); + await assistantReceived.Task.WaitAsync(TimeSpan.FromSeconds(5)); + } + Assert.False(sendTask.IsCompleted); + + await server.SendSessionEventAsync(session.SessionId, "session.idle", new()); + var result = await sendTask.WaitAsync(TimeSpan.FromSeconds(5)); + + if (hasAssistantMessage) + { + Assert.NotNull(result); + Assert.Equal("Acknowledged", result.Data.Content); + } + else + { + Assert.Null(result); + } + } + + [Theory] + [InlineData(null, false)] + [InlineData(null, true)] + [InlineData(MessageSource.User, false)] + [InlineData(MessageSource.User, true)] + [InlineData(MessageSource.System, false)] + [InlineData(MessageSource.System, true)] + public async Task SendAndWaitAsync_MessageSource_Propagates_Errors(MessageSource? source, bool rpcError) + { + await using var server = await FakeCopilotServer.StartAsync(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + await using var session = await client.CreateSessionAsync(new SessionConfig()); + if (rpcError) + { + server.FailSessionSend(); + } + + var sendTask = session.SendAndWaitAsync(new MessageOptions { Prompt = "Context", Source = source }); + var request = await WaitForRequestAsync(server, "session.send"); + AssertMessageSource(request.Params, source?.ToString().ToLowerInvariant()); + + if (rpcError) + { + var error = await Assert.ThrowsAsync(() => sendTask.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Contains("session send failed", error.Message); + } + else + { + await server.SendSessionEventAsync(session.SessionId, "session.error", new() + { + ["errorType"] = "query", + ["message"] = "model request failed" + }); + var error = await Assert.ThrowsAsync(() => sendTask.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Equal("Session error: model request failed", error.Message); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task SendAndWaitAsync_MessageSource_Preserves_Timeout_And_Cancellation(bool cancel) + { + await using var server = await FakeCopilotServer.StartAsync(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + await using var session = await client.CreateSessionAsync(new SessionConfig()); + using var cancellation = new CancellationTokenSource(); + + var sendTask = session.SendAndWaitAsync( + new MessageOptions { Prompt = "Context", Source = MessageSource.System }, + timeout: cancel ? TimeSpan.FromSeconds(30) : TimeSpan.FromMilliseconds(50), + cancellationToken: cancellation.Token); + await WaitForRequestAsync(server, "session.send"); + + if (cancel) + { + cancellation.Cancel(); + var error = await Assert.ThrowsAnyAsync(() => sendTask.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Equal(cancellation.Token, error.CancellationToken); + } + else + { + var error = await Assert.ThrowsAsync(() => sendTask.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Contains("SendAndWaitAsync timed out", error.Message); + } + } + + private static void AssertMessageSource(JsonElement request, string? source) + { + if (source is null) + { + Assert.False(request.TryGetProperty("source", out _)); + } + else + { + Assert.Equal(source, request.GetProperty("source").GetString()); + } + Assert.False(request.TryGetProperty("billable", out _)); + Assert.False(request.TryGetProperty("wait", out _)); + } + [Fact] public async Task SendAndWaitAsync_Skips_Autopilot_Continuation_Idle() { @@ -1960,6 +2171,7 @@ private sealed class FakeCopilotServer : IAsyncDisposable private bool _delayDestroy; private bool _failRuntimeShutdown; private bool _failSessionCreate; + private bool _failSessionSend; private FakeCopilotServer(TcpListener listener) { @@ -2026,6 +2238,11 @@ public void FailSessionCreate() _failSessionCreate = true; } + public void FailSessionSend() + { + _failSessionSend = true; + } + public void CloseConnection() { _stream?.Dispose(); @@ -2051,6 +2268,28 @@ public async Task SendRequestAsync(string method, Dictionary data) + { + var stream = _stream ?? throw new InvalidOperationException("Client is not connected."); + return WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["method"] = "session.event", + ["params"] = new Dictionary + { + ["sessionId"] = sessionId, + ["event"] = new Dictionary + { + ["id"] = Guid.NewGuid().ToString(), + ["timestamp"] = DateTimeOffset.UtcNow.ToString("O"), + ["parentId"] = null, + ["type"] = type, + ["data"] = data + } + } + }, _cts.Token); + } + public async ValueTask DisposeAsync() { _allowDestroy.TrySetResult(); @@ -2154,6 +2393,21 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel }, cancellationToken); return; } + if (method == "session.send" && _failSessionSend) + { + _failSessionSend = false; + await WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["id"] = id, + ["error"] = new Dictionary + { + ["code"] = -32000, + ["message"] = "session send failed" + } + }, cancellationToken); + return; + } object? result = method switch { "connect" => new Dictionary diff --git a/dotnet/test/Unit/CloneTests.cs b/dotnet/test/Unit/CloneTests.cs index 2f213525f6..93dcd46c4c 100644 --- a/dotnet/test/Unit/CloneTests.cs +++ b/dotnet/test/Unit/CloneTests.cs @@ -275,21 +275,39 @@ public void ResumeSessionConfig_Clone_PreservesMcpServersComparer() Assert.True(clone.McpServers!.ContainsKey("SERVER")); } - [Fact] - public void MessageOptions_Clone_CopiesAllProperties() + [Theory] + [InlineData(null)] + [InlineData(MessageSource.User)] + [InlineData(MessageSource.System)] + public void MessageOptions_Clone_CopiesAllProperties(MessageSource? source) { var original = new MessageOptions { Prompt = "Hello", Attachments = [new AttachmentFile { Path = "/test.txt", DisplayName = "test.txt" }], - Mode = "chat", + Mode = "immediate", + AgentMode = AgentMode.Plan, + Source = source, + DisplayPrompt = "Display text", + RequestHeaders = new Dictionary { ["X-Test"] = "original" }, }; var clone = original.Clone(); Assert.Equal(original.Prompt, clone.Prompt); Assert.Equal(original.Mode, clone.Mode); + Assert.Equal(original.AgentMode, clone.AgentMode); + Assert.Equal(original.Source, clone.Source); + Assert.Equal(original.DisplayPrompt, clone.DisplayPrompt); Assert.Single(clone.Attachments!); + Assert.NotSame(original.Attachments, clone.Attachments); + Assert.Equal(original.RequestHeaders, clone.RequestHeaders); + Assert.NotSame(original.RequestHeaders, clone.RequestHeaders); + + clone.Source = source == MessageSource.System ? MessageSource.User : MessageSource.System; + clone.RequestHeaders!["X-Test"] = "changed"; + Assert.Equal(source, original.Source); + Assert.Equal("original", original.RequestHeaders["X-Test"]); } [Fact] diff --git a/go/README.md b/go/README.md index 1fa8202419..f97272eba9 100644 --- a/go/README.md +++ b/go/README.md @@ -278,6 +278,24 @@ Initial acquisition runs during session creation or resume. Cancellation, provid - `UI() *SessionUI` - Interactive UI API for elicitation dialogs - `Capabilities() SessionCapabilities` - Host capabilities (e.g. elicitation support) +#### Message source + +Set `MessageOptions.Source` to `copilot.MessageSourceUser` or +`copilot.MessageSourceSystem` to identify the message's origin. Leave it empty to +omit `source` from the request and preserve the runtime's default behavior. + +```go +_, err := session.Send(ctx, copilot.MessageOptions{ + Prompt: "Background check completed. The build passed.", + Source: copilot.MessageSourceSystem, + Mode: "enqueue", +}) +``` + +Source is independent of delivery `Mode` and `AgentMode`; it does not replace the +session's `SystemMessage` configuration. `SendAndWait` accepts the same options +and still waits for session idle, returning `nil` if no assistant message arrives. + ### Helper Functions - `Bool(v bool) *bool` - Helper to create bool pointers (e.g. for `Streaming`) diff --git a/go/message_source_test.go b/go/message_source_test.go new file mode 100644 index 0000000000..480e6c13d0 --- /dev/null +++ b/go/message_source_test.go @@ -0,0 +1,235 @@ +package copilot + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "reflect" + "strings" + "testing" + "time" + + "github.com/github/copilot-sdk/go/internal/jsonrpc2" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" +) + +func TestSession_SendMessageSource(t *testing.T) { + previousPropagator := otel.GetTextMapPropagator() + otel.SetTextMapPropagator(propagation.TraceContext{}) + defer otel.SetTextMapPropagator(previousPropagator) + + const traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + const tracestate = "vendor=value" + + for _, source := range []struct { + name string + value MessageSource + wire string + }{ + {name: "omitted"}, + {name: "user", value: MessageSourceUser, wire: "user"}, + {name: "system", value: MessageSourceSystem, wire: "system"}, + } { + t.Run(source.name, func(t *testing.T) { + for _, mode := range []string{"", "enqueue", "immediate"} { + name := mode + if name == "" { + name = "defaults" + } + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + options := MessageOptions{Prompt: "hello", Source: source.value} + want := map[string]any{"sessionId": "session-1", "prompt": "hello"} + if source.wire != "" { + want["source"] = source.wire + } + if mode != "" { + options.Mode = mode + options.AgentMode = AgentModePlan + options.DisplayPrompt = "display text" + options.Attachments = []Attachment{ + &AttachmentFile{Path: "/workspace/main.go", DisplayName: "main.go"}, + } + options.RequestHeaders = map[string]string{"X-Test": "value"} + ctx = contextWithTraceParent(ctx, traceparent, tracestate) + want["mode"] = mode + want["agentMode"] = "plan" + want["displayPrompt"] = "display text" + want["attachments"] = []any{ + map[string]any{"type": "file", "path": "/workspace/main.go", "displayName": "main.go"}, + } + want["requestHeaders"] = map[string]any{"X-Test": "value"} + want["traceparent"] = traceparent + want["tracestate"] = tracestate + } + + params := captureMessageSourceRequest(t, nil, nil, func(session *Session) { + messageID, err := session.Send(ctx, options) + if err != nil { + t.Fatalf("Send failed: %v", err) + } + if messageID != "message-1" { + t.Fatalf("expected message-1, got %q", messageID) + } + }) + if !reflect.DeepEqual(params, want) { + t.Fatalf("unexpected session.send params:\ngot %#v\nwant %#v", params, want) + } + }) + } + }) + } +} + +func TestSession_SendAndWaitMessageSource(t *testing.T) { + for _, tc := range []struct { + name string + events []SessionEvent + rpcError *jsonrpc2.Error + wantContent string + wantError string + }{ + { + name: "idle without assistant", + events: []SessionEvent{{Data: &SessionIdleData{}}}, + }, + { + name: "assistant then idle", + events: []SessionEvent{ + {Data: &AssistantMessageData{MessageID: "assistant-1", Content: "done"}}, + {Data: &SessionIdleData{}}, + }, + wantContent: "done", + }, + { + name: "session error", + events: []SessionEvent{{Data: &SessionErrorData{Message: "model failed"}}}, + wantError: "session error: model failed", + }, + { + name: "RPC error", + rpcError: &jsonrpc2.Error{Code: -32602, Message: "invalid prompt"}, + wantError: "invalid prompt", + }, + } { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + params := captureMessageSourceRequest(t, tc.rpcError, tc.events, func(session *Session) { + result, err := session.SendAndWait(ctx, MessageOptions{ + Prompt: "background update", + Source: MessageSourceSystem, + }) + if tc.wantError != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantError) { + t.Fatalf("expected error containing %q, got %v", tc.wantError, err) + } + if tc.rpcError != nil { + var rpcError *jsonrpc2.Error + if !errors.As(err, &rpcError) || rpcError.Code != tc.rpcError.Code { + t.Fatalf("expected wrapped RPC error, got %v", err) + } + } + } else if err != nil { + t.Fatalf("SendAndWait failed: %v", err) + } + if tc.wantContent == "" { + if result != nil { + t.Fatalf("expected no assistant message, got %#v", result) + } + } else { + if result == nil { + t.Fatal("expected an assistant message") + } + message, ok := result.Data.(*AssistantMessageData) + if !ok || message.Content != tc.wantContent { + t.Fatalf("unexpected assistant message: %#v", result.Data) + } + } + }) + want := map[string]any{ + "sessionId": "session-1", + "prompt": "background update", + "source": "system", + } + if !reflect.DeepEqual(params, want) { + t.Fatalf("unexpected session.send params:\ngot %#v\nwant %#v", params, want) + } + }) + } +} + +func captureMessageSourceRequest(t *testing.T, rpcError *jsonrpc2.Error, events []SessionEvent, invoke func(*Session)) map[string]any { + t.Helper() + + stdinR, stdinW := io.Pipe() + stdoutR, stdoutW := io.Pipe() + defer stdinR.Close() + defer stdinW.Close() + defer stdoutR.Close() + defer stdoutW.Close() + + client := jsonrpc2.NewClient(stdinW, stdoutR) + client.Start() + defer client.Stop() + + session := newSession("session-1", client, "", false) + defer session.stopEventProcessing() + + paramsCh := make(chan map[string]any, 1) + errCh := make(chan error, 1) + go func() { + frame, err := readTestJSONRPCFrame(stdinR) + if err != nil { + errCh <- err + return + } + var request struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params map[string]any `json:"params"` + } + if err := json.Unmarshal(frame, &request); err != nil { + errCh <- err + return + } + if request.Method != "session.send" { + errCh <- fmt.Errorf("expected session.send, got %s", request.Method) + return + } + response := map[string]any{"jsonrpc": "2.0", "id": request.ID} + if rpcError != nil { + response["error"] = rpcError + } else { + response["result"] = map[string]any{"messageId": "message-1"} + } + data, err := json.Marshal(response) + if err != nil { + errCh <- err + return + } + if _, err := fmt.Fprintf(stdoutW, "Content-Length: %d\r\n\r\n%s", len(data), data); err != nil { + errCh <- err + return + } + for _, event := range events { + session.dispatchEvent(event) + } + paramsCh <- request.Params + }() + + invoke(session) + select { + case params := <-paramsCh: + return params + case err := <-errCh: + t.Fatal(err) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for session.send request") + } + return nil +} diff --git a/go/session.go b/go/session.go index f4bb0d38c7..2dd2a8a9a1 100644 --- a/go/session.go +++ b/go/session.go @@ -432,6 +432,7 @@ func (s *Session) Send(ctx context.Context, options MessageOptions) (string, err req := sessionSendRequest{ SessionID: s.SessionID, Prompt: options.Prompt, + Source: options.Source, DisplayPrompt: options.DisplayPrompt, Attachments: options.Attachments, Mode: options.Mode, diff --git a/go/types.go b/go/types.go index 4a45a4019c..b6a5b69d17 100644 --- a/go/types.go +++ b/go/types.go @@ -2431,10 +2431,23 @@ type ToolBinaryResult struct { Description string `json:"description,omitempty"` } +// MessageSource identifies whether a message originates from a user or the system. +type MessageSource string + +const ( + // MessageSourceUser identifies a user-originated message. + MessageSourceUser MessageSource = "user" + // MessageSourceSystem identifies a system-originated message. + MessageSourceSystem MessageSource = "system" +) + // MessageOptions configures a message to send type MessageOptions struct { // Prompt is the message to send Prompt string + // Source identifies the message origin independently of Mode and AgentMode. + // The empty value omits source from the request, preserving runtime defaults. + Source MessageSource // Attachments are file or directory attachments Attachments []Attachment // Mode is the message delivery mode (default: "enqueue") @@ -2913,6 +2926,7 @@ type sessionAbortRequest struct { type sessionSendRequest struct { SessionID string `json:"sessionId"` Prompt string `json:"prompt"` + Source MessageSource `json:"source,omitempty"` DisplayPrompt string `json:"displayPrompt,omitempty"` Attachments []Attachment `json:"attachments,omitempty"` Mode string `json:"mode,omitempty"` diff --git a/java/README.md b/java/README.md index 69bf8b050c..6a186bc434 100644 --- a/java/README.md +++ b/java/README.md @@ -204,6 +204,24 @@ provider errors, and invalid token responses reject that operation instead of falling back to ambient authentication. Idle sessions refresh only before their next credential-consuming operation; there is no background refresh timer. +## Message source + +Use `MessageSource.SYSTEM` to identify programmatic context or automated messages: + +```java +import com.github.copilot.rpc.MessageOptions; +import com.github.copilot.rpc.MessageSource; + +session.send(new MessageOptions() + .setPrompt("The background build completed successfully.") + .setSource(MessageSource.SYSTEM)).get(); +``` + +Leave `source` unset to omit it from the request and retain the runtime's default +user-input behavior, or set `MessageSource.USER` explicitly. Source is independent +of delivery mode (`enqueue` or `immediate`) and does not configure the session's +system prompt. + ## Permission Handling `PermissionHandler.APPROVE_ALL` approves requests when managed settings are disabled. When `enableManagedSettings` is true, it completes exceptionally. Custom handlers can inspect `request.getManagedApprovalRequired()` for human-facing confirmation logic. diff --git a/java/sdk/src/main/java/com/github/copilot/CopilotSession.java b/java/sdk/src/main/java/com/github/copilot/CopilotSession.java index c072e31ec1..e0aab22e1f 100644 --- a/java/sdk/src/main/java/com/github/copilot/CopilotSession.java +++ b/java/sdk/src/main/java/com/github/copilot/CopilotSession.java @@ -561,6 +561,7 @@ public CompletableFuture send(MessageOptions options) { request.setPrompt(options.getPrompt()); request.setAttachments(options.getAttachments()); request.setMode(options.getMode()); + request.setSource(options.getSource()); request.setAgentMode(options.getAgentMode()); request.setRequestHeaders(options.getRequestHeaders()); request.setDisplayPrompt(options.getDisplayPrompt()); diff --git a/java/sdk/src/main/java/com/github/copilot/rpc/MessageOptions.java b/java/sdk/src/main/java/com/github/copilot/rpc/MessageOptions.java index c781011ff8..10458736e1 100644 --- a/java/sdk/src/main/java/com/github/copilot/rpc/MessageOptions.java +++ b/java/sdk/src/main/java/com/github/copilot/rpc/MessageOptions.java @@ -45,6 +45,7 @@ public class MessageOptions { private String prompt; private List attachments; private String mode; + private MessageSource source; private AgentMode agentMode; private Map requestHeaders; private String displayPrompt; @@ -128,6 +129,30 @@ public String getMode() { return mode; } + /** + * Gets the message source. + * + * @return the source, or {@code null} to use the runtime's default + */ + public MessageSource getSource() { + return source; + } + + /** + * Sets the origin of this message. + *

+ * When unset, the source is omitted from the request and the runtime treats the + * message as user input. This is independent of the delivery mode. + * + * @param source + * the source, or {@code null} to omit it + * @return this options instance for method chaining + */ + public MessageOptions setSource(MessageSource source) { + this.source = source; + return this; + } + /** * Sets the per-message agent UI mode. *

@@ -217,6 +242,7 @@ public MessageOptions clone() { copy.prompt = this.prompt; copy.attachments = this.attachments != null ? new ArrayList<>(this.attachments) : null; copy.mode = this.mode; + copy.source = this.source; copy.agentMode = this.agentMode; copy.requestHeaders = this.requestHeaders != null ? new HashMap<>(this.requestHeaders) : null; copy.displayPrompt = this.displayPrompt; diff --git a/java/sdk/src/main/java/com/github/copilot/rpc/MessageSource.java b/java/sdk/src/main/java/com/github/copilot/rpc/MessageSource.java new file mode 100644 index 0000000000..735466c3e7 --- /dev/null +++ b/java/sdk/src/main/java/com/github/copilot/rpc/MessageSource.java @@ -0,0 +1,64 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + *--------------------------------------------------------------------------------------------*/ + +package com.github.copilot.rpc; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +/** + * The origin of a message sent to a Copilot session. + *

+ * Set on {@link MessageOptions#setSource(MessageSource)} to distinguish user + * input from programmatic context. This does not configure the session's system + * prompt or change the message delivery mode. + * + * @see MessageOptions + */ +public enum MessageSource { + + /** Input originating from the user. */ + USER("user"), + + /** Programmatic context or an automated message. */ + SYSTEM("system"); + + private final String value; + + MessageSource(String value) { + this.value = value; + } + + /** + * Returns the JSON value for this message source. + * + * @return the string value used in JSON serialization + */ + @JsonValue + public String getValue() { + return value; + } + + /** + * Deserializes a JSON string into the corresponding message source. + * + * @param value + * the JSON string value + * @return the matching source, or {@code null} if value is {@code null} + * @throws IllegalArgumentException + * if the value does not match a known message source + */ + @JsonCreator + public static MessageSource fromValue(String value) { + if (value == null) { + return null; + } + for (MessageSource source : values()) { + if (source.value.equals(value)) { + return source; + } + } + throw new IllegalArgumentException("Unknown MessageSource value: " + value); + } +} diff --git a/java/sdk/src/main/java/com/github/copilot/rpc/SendMessageRequest.java b/java/sdk/src/main/java/com/github/copilot/rpc/SendMessageRequest.java index c87dda7623..97cfc76f9d 100644 --- a/java/sdk/src/main/java/com/github/copilot/rpc/SendMessageRequest.java +++ b/java/sdk/src/main/java/com/github/copilot/rpc/SendMessageRequest.java @@ -37,6 +37,9 @@ public final class SendMessageRequest { @JsonProperty("mode") private String mode; + @JsonProperty("source") + private MessageSource source; + @JsonProperty("agentMode") private AgentMode agentMode; @@ -86,6 +89,18 @@ public void setMode(String mode) { this.mode = mode; } + /** Gets the message source. @return the source, or {@code null} if unset */ + public MessageSource getSource() { + return source; + } + + /** + * Sets the message source. @param source the source, or {@code null} to omit it + */ + public void setSource(MessageSource source) { + this.source = source; + } + /** Gets the per-message agent UI mode. @return the agent mode */ public AgentMode getAgentMode() { return agentMode; diff --git a/java/sdk/src/test/java/com/github/copilot/MessageSourceTest.java b/java/sdk/src/test/java/com/github/copilot/MessageSourceTest.java new file mode 100644 index 0000000000..29c2d4367e --- /dev/null +++ b/java/sdk/src/test/java/com/github/copilot/MessageSourceTest.java @@ -0,0 +1,332 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + *--------------------------------------------------------------------------------------------*/ + +package com.github.copilot; + +import static org.junit.jupiter.api.Assertions.*; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.NullSource; +import org.junit.jupiter.params.provider.ValueSource; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.copilot.generated.rpc.SessionSendParams; +import com.github.copilot.rpc.AgentMode; +import com.github.copilot.rpc.Attachment; +import com.github.copilot.rpc.CopilotClientOptions; +import com.github.copilot.rpc.MessageOptions; +import com.github.copilot.rpc.MessageSource; +import com.github.copilot.rpc.PermissionHandler; +import com.github.copilot.rpc.SendMessageRequest; +import com.github.copilot.rpc.SessionConfig; + +@AllowCopilotExperimental +class MessageSourceTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @ParameterizedTest + @ValueSource(strings = {"invalid", "2147483648", "-1"}) + void malformedContentLengthFailsWithIOException(String value) { + var input = new ByteArrayInputStream( + ("Content-Length: " + value + "\r\n\r\n").getBytes(StandardCharsets.US_ASCII)); + var error = assertThrows(IOException.class, () -> SendServer.readMessage(input)); + assertTrue(error.getMessage().startsWith("Invalid Content-Length")); + } + + @ParameterizedTest + @CsvSource({"USER,user", "SYSTEM,system"}) + void sourceUsesLowercaseJson(MessageSource source, String value) throws Exception { + assertEquals(value, source.getValue()); + assertEquals("\"" + value + "\"", MAPPER.writeValueAsString(source)); + assertEquals(source, MAPPER.readValue("\"" + value + "\"", MessageSource.class)); + assertEquals(source, MessageSource.fromValue(value)); + } + + @Test + void sourceRejectsUnknownValuesAndAcceptsNull() throws Exception { + assertNull(MessageSource.fromValue(null)); + assertNull(MAPPER.readValue("null", MessageSource.class)); + assertThrows(IllegalArgumentException.class, () -> MessageSource.fromValue("unknown")); + assertThrows(IllegalArgumentException.class, () -> MessageSource.fromValue("USER")); + assertThrows(IOException.class, () -> MAPPER.readValue("\"unknown\"", MessageSource.class)); + } + + @Test + void defaultSourceIsOmittedAndCanBeCleared() throws Exception { + var options = new MessageOptions().setPrompt("hello"); + var request = new SendMessageRequest(); + request.setPrompt("hello"); + + assertNull(options.getSource()); + assertNull(options.clone().getSource()); + assertNull(request.getSource()); + assertEquals(MAPPER.readTree("{\"prompt\":\"hello\"}"), MAPPER.valueToTree(options)); + assertEquals(MAPPER.readTree("{\"prompt\":\"hello\"}"), MAPPER.valueToTree(request)); + + assertSame(options, options.setSource(MessageSource.SYSTEM)); + options.setSource(null); + request.setSource(MessageSource.USER); + request.setSource(null); + assertFalse(MAPPER.valueToTree(options).has("source")); + assertFalse(MAPPER.valueToTree(request).has("source")); + } + + @ParameterizedTest + @EnumSource(MessageSource.class) + void optionsAndRequestRoundTripSource(MessageSource source) throws Exception { + var options = new MessageOptions().setPrompt("hello").setSource(source); + var request = new SendMessageRequest(); + request.setPrompt("hello"); + request.setSource(source); + + for (Object value : List.of(options, request)) { + JsonNode json = MAPPER.valueToTree(value); + assertEquals(source.getValue(), json.get("source").asText()); + assertEquals(source, MAPPER.treeToValue(json, MessageOptions.class).getSource()); + assertEquals(source, MAPPER.treeToValue(json, SendMessageRequest.class).getSource()); + } + } + + @ParameterizedTest + @EnumSource(MessageSource.class) + void clonePreservesSourceAndOtherOptions(MessageSource source) { + var options = fullOptions().setSource(source); + var copy = options.clone(); + + assertNotSame(options, copy); + assertEquals(source, copy.getSource()); + assertEquals(MAPPER.valueToTree(options), MAPPER.valueToTree(copy)); + copy.setSource(null).setPrompt("changed").setAttachments(List.of()).setMode("enqueue") + .setAgentMode(AgentMode.INTERACTIVE).setRequestHeaders(Map.of()).setDisplayPrompt("changed"); + assertEquals(source, options.getSource()); + assertEquals(MAPPER.valueToTree(fullOptions().setSource(source)), + MAPPER.valueToTree(options)); + } + + @Test + void sendWithoutSourcePreservesLegacyPayload() throws Exception { + try (var server = new SendServer(Outcome.IDLE); + var client = server.createClient(); + var session = client.createSession(sessionConfig()).get(5, TimeUnit.SECONDS)) { + assertEquals("message-1", session.send(new MessageOptions().setPrompt("hello")).get(5, TimeUnit.SECONDS)); + assertEquals(MAPPER.readTree("{\"sessionId\":\"source-session\",\"prompt\":\"hello\"}"), + server.takeSendParams()); + assertEquals("message-1", session.send("hello").get(5, TimeUnit.SECONDS)); + assertFalse(server.takeSendParams().has("source")); + } + } + + @ParameterizedTest + @CsvSource({"USER,enqueue", "USER,immediate", "SYSTEM,enqueue", "SYSTEM,immediate"}) + void sendForwardsSourceWithoutChangingOtherOptions(MessageSource source, String mode) throws Exception { + try (var server = new SendServer(Outcome.IDLE); + var client = server.createClient(); + var session = client.createSession(sessionConfig()).get(5, TimeUnit.SECONDS)) { + var options = fullOptions().setMode(mode).setSource(source); + assertEquals("message-1", session.send(options).get(5, TimeUnit.SECONDS)); + + var expected = MAPPER.createObjectNode().put("sessionId", "source-session").put("prompt", "hello") + .put("mode", mode).put("agentMode", "plan").put("displayPrompt", "display") + .put("source", source.getValue()); + expected.set("attachments", MAPPER.valueToTree(options.getAttachments())); + expected.set("requestHeaders", MAPPER.valueToTree(Map.of("X-Trace", "trace-id"))); + assertEquals(expected, server.takeSendParams()); + } + } + + @Test + void systemSourceCompletesOnIdleWithoutAssistantMessage() throws Exception { + try (var server = new SendServer(Outcome.IDLE); + var client = server.createClient(); + var session = client.createSession(sessionConfig()).get(5, TimeUnit.SECONDS)) { + assertNull(session.sendAndWait(new MessageOptions().setPrompt("context").setSource(MessageSource.SYSTEM)) + .get(5, TimeUnit.SECONDS)); + assertEquals("system", server.takeSendParams().get("source").asText()); + } + } + + @ParameterizedTest + @EnumSource(value = Outcome.class, names = {"SESSION_ERROR", "RPC_ERROR"}) + void systemSourceDoesNotSuppressErrors(Outcome outcome) throws Exception { + try (var server = new SendServer(outcome); + var client = server.createClient(); + var session = client.createSession(sessionConfig()).get(5, TimeUnit.SECONDS)) { + var pending = session + .sendAndWait(new MessageOptions().setPrompt("context").setSource(MessageSource.SYSTEM)); + var error = assertThrows(ExecutionException.class, () -> pending.get(5, TimeUnit.SECONDS)); + assertTrue(error.getCause().getMessage().contains("send failed"), error.toString()); + assertEquals("system", server.takeSendParams().get("source").asText()); + } + } + + @ParameterizedTest + @EnumSource(MessageSource.class) + @NullSource + void generatedRawRpcAlreadyForwardsSource(MessageSource source) throws Exception { + try (var server = new SendServer(Outcome.IDLE); + var client = server.createClient(); + var session = client.createSession(sessionConfig()).get(5, TimeUnit.SECONDS)) { + var json = MAPPER.createObjectNode().put("prompt", "hello"); + if (source != null) { + json.put("source", source.getValue()); + } + var params = MAPPER.treeToValue(json, SessionSendParams.class); + assertEquals("message-1", session.getRpc().send(params).get(5, TimeUnit.SECONDS).messageId()); + + json.put("sessionId", "source-session"); + assertEquals(json, server.takeSendParams()); + } + } + + private static MessageOptions fullOptions() { + return new MessageOptions().setPrompt("hello").setMode("immediate").setAgentMode(AgentMode.PLAN) + .setAttachments(List.of(new Attachment("file", "/workspace/example.java", "example"))) + .setRequestHeaders(Map.of("X-Trace", "trace-id")).setDisplayPrompt("display"); + } + + private static SessionConfig sessionConfig() { + return new SessionConfig().setSessionId("source-session").setOnPermissionRequest(PermissionHandler.APPROVE_ALL); + } + + private enum Outcome { + IDLE, SESSION_ERROR, RPC_ERROR + } + + /** Uses the existing loopback JSON-RPC test pattern through public SDK APIs. */ + private static final class SendServer implements AutoCloseable { + + private final ServerSocket listener; + private final Thread thread; + private final BlockingQueue sends = new LinkedBlockingQueue<>(); + private final Outcome outcome; + private volatile Socket socket; + private volatile boolean closed; + private volatile Exception failure; + + SendServer(Outcome outcome) throws IOException { + this.outcome = outcome; + listener = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + thread = new Thread(this::serve, "message-source-server"); + thread.setDaemon(true); + thread.start(); + } + + CopilotClient createClient() { + return new CopilotClient(new CopilotClientOptions().setCliUrl("localhost:" + listener.getLocalPort())); + } + + JsonNode takeSendParams() throws InterruptedException { + JsonNode params = sends.poll(5, TimeUnit.SECONDS); + assertNotNull(params, "Expected session.send"); + return params; + } + + private void serve() { + try (Socket accepted = listener.accept()) { + socket = accepted; + while (!closed) { + JsonNode request = readMessage(accepted.getInputStream()); + if (request == null) { + return; + } + String method = request.path("method").asText(); + JsonNode params = request.path("params"); + Object result = switch (method) { + case "connect" -> Map.of("ok", true, "protocolVersion", 3, "version", "test"); + case "session.create" -> Map.of("sessionId", params.path("sessionId").asText()); + case "session.send" -> Map.of("messageId", "message-1"); + case "session.detach" -> Map.of("success", true); + default -> Map.of(); + }; + boolean send = "session.send".equals(method); + if (send) { + sends.add(params); + } + var response = MAPPER.createObjectNode().put("jsonrpc", "2.0"); + response.set("id", request.get("id")); + if (send && outcome == Outcome.RPC_ERROR) { + response.set("error", MAPPER.valueToTree(Map.of("code", -32603, "message", "send failed"))); + } else { + response.set("result", MAPPER.valueToTree(result)); + } + writeMessage(response); + if (send && outcome != Outcome.RPC_ERROR) { + String type = outcome == Outcome.IDLE ? "session.idle" : "session.error"; + Map data = outcome == Outcome.IDLE + ? Map.of() + : Map.of("errorType", "test", "message", "send failed"); + writeMessage(Map.of("jsonrpc", "2.0", "method", "session.event", "params", + Map.of("sessionId", params.path("sessionId").asText(), "event", + Map.of("type", type, "id", "00000000-0000-0000-0000-000000000001", "timestamp", + "2026-09-06T00:00:00Z", "data", data)))); + } + } + } catch (Exception ex) { + if (!closed) { + failure = ex; + } + } + } + + private static JsonNode readMessage(InputStream input) throws IOException { + var header = new StringBuilder(); + while (!header.toString().endsWith("\r\n\r\n")) { + int b = input.read(); + if (b < 0) { + return null; + } + header.append((char) b); + } + int length; + try { + length = Integer.parseInt(header.substring(header.indexOf(":") + 1).trim()); + } catch (NumberFormatException ex) { + throw new IOException("Invalid Content-Length", ex); + } + if (length < 0) { + throw new IOException("Invalid Content-Length: " + length); + } + return MAPPER.readTree(input.readNBytes(length)); + } + + private void writeMessage(Object message) throws IOException { + byte[] body = MAPPER.writeValueAsBytes(message); + var output = socket.getOutputStream(); + output.write(("Content-Length: " + body.length + "\r\n\r\n").getBytes(StandardCharsets.UTF_8)); + output.write(body); + output.flush(); + } + + @Override + public void close() throws Exception { + closed = true; + listener.close(); + if (socket != null) { + socket.close(); + } + thread.join(5000); + assertFalse(thread.isAlive(), "RPC test server should stop"); + assertNull(failure, "RPC test server failed: " + failure); + } + } +} diff --git a/nodejs/README.md b/nodejs/README.md index e3d76ba6e4..841a4e06b2 100644 --- a/nodejs/README.md +++ b/nodejs/README.md @@ -267,11 +267,20 @@ Send a message to the session. Returns immediately after the message is queued; **Options:** - `prompt: string` - The message/prompt to send +- `source?: MessageSource` - `"user"` or `"system"` provenance; omitted by default - `attachments?: Array<{type, path, displayName}>` - File attachments - `mode?: "enqueue" | "immediate"` - Delivery mode Returns the message ID. +Use `source: "system"` for automated messages from your application: + +```typescript +await session.send({ prompt: "Context updated", source: "system" }); +``` + +Source is independent of delivery mode. Leaving it unset preserves the existing human-message payload; it does not set billing flags or use the notification API. + ##### `sendAndWait(options: MessageOptions, timeout?: number): Promise` Send a message and wait until the session becomes idle. @@ -279,6 +288,7 @@ Send a message and wait until the session becomes idle. **Options:** - `prompt: string` - The message/prompt to send +- `source?: MessageSource` - Same optional provenance as `send` - `attachments?: Array<{type, path, displayName}>` - File attachments - `mode?: "enqueue" | "immediate"` - Delivery mode - `timeout?: number` - Optional timeout in milliseconds diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts index 2007679d61..6251df4fc7 100644 --- a/nodejs/src/index.ts +++ b/nodejs/src/index.ts @@ -116,6 +116,7 @@ export type { DefaultAgentConfig, BearerTokenProvider, MessageOptions, + MessageSource, ManagedSettings, ManagedSettingsPermissions, ModelBilling, diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index b7fc7837a5..4c2be14299 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -719,6 +719,7 @@ export class CopilotSession { ...(await getTraceContext(this.traceContextProvider)), sessionId: this.sessionId, prompt: options.prompt, + source: options.source, displayPrompt: options.displayPrompt, attachments: options.attachments, mode: options.mode, diff --git a/nodejs/src/types.ts b/nodejs/src/types.ts index 0f15749f71..c34c9a70db 100644 --- a/nodejs/src/types.ts +++ b/nodejs/src/types.ts @@ -3317,12 +3317,23 @@ export interface ProviderModelConfig { */ capabilities?: ModelCapabilitiesOverride; } +/** + * Message provenance, independent of delivery mode. + */ +export type MessageSource = "user" | "system"; + export interface MessageOptions { /** * The prompt/message to send */ prompt: string; + /** + * Optional message provenance. Omitted by default to preserve the runtime's + * default for user messages. Use "system" for automated application messages. + */ + source?: MessageSource; + /** * File, directory, selection, or blob attachments */ diff --git a/nodejs/test/message-source.test.ts b/nodejs/test/message-source.test.ts new file mode 100644 index 0000000000..ad47e96933 --- /dev/null +++ b/nodejs/test/message-source.test.ts @@ -0,0 +1,144 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + *--------------------------------------------------------------------------------------------*/ + +import { PassThrough } from "node:stream"; +import { describe, expect, it, onTestFinished } from "vitest"; +import { + createMessageConnection, + ResponseError, + StreamMessageReader, + StreamMessageWriter, +} from "vscode-jsonrpc/node.js"; +import type { MessageOptions, MessageSource, SessionEvent } from "../src/index.js"; +import { CopilotSession } from "../src/session.js"; + +function sessionPair(traceContextProvider?: ConstructorParameters[3]) { + const clientToServer = new PassThrough(); + const serverToClient = new PassThrough(); + const client = createMessageConnection( + new StreamMessageReader(serverToClient), + new StreamMessageWriter(clientToServer) + ); + const server = createMessageConnection( + new StreamMessageReader(clientToServer), + new StreamMessageWriter(serverToClient) + ); + onTestFinished(() => { + client.dispose(); + server.dispose(); + clientToServer.destroy(); + serverToClient.destroy(); + }); + client.listen(); + server.listen(); + return { + session: new CopilotSession("session-1", client, undefined, traceContextProvider), + server, + }; +} + +const sources: (MessageSource | undefined)[] = [undefined, "user", "system"]; +const modes: MessageOptions["mode"][] = [undefined, "enqueue", "immediate"]; + +it("omits source when sending a plain human prompt", async () => { + const { session, server } = sessionPair(); + server.onRequest("session.send", (params: unknown) => { + expect(params).toEqual({ sessionId: "session-1", prompt: "hello" }); + return { messageId: "message-1" }; + }); + await expect(session.send("hello")).resolves.toBe("message-1"); +}); + +describe.each(sources)("message source %s", (source) => { + it.each(modes)("preserves the wire payload with delivery mode %s", async (mode) => { + const { session, server } = sessionPair(); + const expected = { + sessionId: "session-1", + prompt: "hello", + ...(source === undefined ? {} : { source }), + ...(mode === undefined ? {} : { mode }), + }; + server.onRequest("session.send", (params: unknown) => { + expect(params).toEqual(expected); + return { messageId: "message-1" }; + }); + await expect(session.send({ prompt: "hello", source, mode })).resolves.toBe("message-1"); + }); + + it("preserves attachments, display text, headers, agent mode and tracing", async () => { + const trace = { + traceparent: "00-fedcba0987654321fedcba0987654321-abcdef1234567890-01", + tracestate: "vendor=source", + }; + const { session, server } = sessionPair(() => trace); + const options: MessageOptions = { + prompt: "context updated", + source, + mode: "immediate", + agentMode: "plan", + attachments: [{ type: "blob", data: "aGk=", mimeType: "text/plain" }], + displayPrompt: "Context updated", + requestHeaders: { "X-Tag": "context" }, + }; + const expected = { sessionId: "session-1", ...trace, ...options }; + if (source === undefined) { + delete expected.source; + } + server.onRequest("session.send", (params: unknown) => { + expect(params).toEqual(expected); + return { messageId: "message-1" }; + }); + await expect(session.send(options)).resolves.toBe("message-1"); + }); + + it("allows sendAndWait to finish on idle without assistant output", async () => { + const { session, server } = sessionPair(); + server.onRequest("session.send", (params: unknown) => { + expect(params).toEqual({ + sessionId: "session-1", + prompt: "context updated", + ...(source === undefined ? {} : { source }), + }); + session._dispatchEvent({ + type: "session.idle", + id: "idle-1", + timestamp: new Date().toISOString(), + parentId: null, + data: {}, + }); + return { messageId: "message-1" }; + }); + await expect( + session.sendAndWait({ prompt: "context updated", source }, 1000) + ).resolves.toBeUndefined(); + }); +}); + +describe("system message errors", () => { + it("propagates an RPC failure from sendAndWait", async () => { + const { session, server } = sessionPair(); + server.onRequest("session.send", () => new ResponseError(-32603, "send failed")); + await expect( + session.sendAndWait({ prompt: "context updated", source: "system" }, 1000) + ).rejects.toThrow("send failed"); + }); + + it("propagates a session error from sendAndWait", async () => { + const { session, server } = sessionPair(); + server.onRequest("session.send", () => { + const event: SessionEvent = { + type: "session.error", + id: "error-1", + timestamp: new Date().toISOString(), + parentId: null, + data: { errorType: "notification", message: "agent failed" }, + }; + session._dispatchEvent(event); + return { messageId: "message-1" }; + }); + await expect( + session.sendAndWait({ prompt: "context updated", source: "system" }, 1000) + ).rejects.toThrow("agent failed"); + }); +}); diff --git a/python/README.md b/python/README.md index 0c3526f260..0519772793 100644 --- a/python/README.md +++ b/python/README.md @@ -109,6 +109,25 @@ async def main(): asyncio.run(main()) ``` +### Message Source + +Use `source="system"` for automated messages from your application. Both `send` +and `send_and_wait` accept the optional `MessageSource` type: + +```python +from copilot import MessageSource + +source: MessageSource = "system" +await session.send("Context updated", source=source) +await session.send_and_wait("Background work finished", source=source) +``` + +Leave source unset (or `None`) for ordinary human sends so the field stays omitted. +Use `"user"` when you need to set it explicitly. Source is independent of delivery +mode and does not set billing flags or use the notification API. `send_and_wait` +can return `None` when the session goes idle without an assistant message; errors +still propagate. + ### Manual Resource Management If you need more control over the lifecycle, you can call `start()`, `stop()`, and `disconnect()` manually: diff --git a/python/copilot/__init__.py b/python/copilot/__init__.py index 8e14887a4f..e73ff532e2 100644 --- a/python/copilot/__init__.py +++ b/python/copilot/__init__.py @@ -152,6 +152,7 @@ MCPHTTPServerConfig, MCPServerConfig, MCPStdioServerConfig, + MessageSource, ModelCapabilitiesOverride, ModelLimitsOverride, ModelSupportsOverride, @@ -318,6 +319,7 @@ "McpAuthWwwAuthenticateParams", "ManagedSettings", "ManagedSettingsPermissions", + "MessageSource", "ModelBilling", "ModelBillingTokenPrices", "ModelBillingTokenPricesLongContext", diff --git a/python/copilot/session.py b/python/copilot/session.py index b5823923ee..3b672fde4f 100644 --- a/python/copilot/session.py +++ b/python/copilot/session.py @@ -276,6 +276,9 @@ class BlobAttachment(TypedDict): Attachment = FileAttachment | DirectoryAttachment | SelectionAttachment | BlobAttachment +MessageSource = Literal["user", "system"] +"""Message provenance, independent of delivery mode.""" + # ============================================================================ # System Message Configuration # ============================================================================ @@ -1711,6 +1714,7 @@ async def send( prompt: str, *, attachments: list[Attachment] | None = None, + source: MessageSource | None = None, mode: Literal["enqueue", "immediate"] | None = None, agent_mode: Literal["interactive", "plan", "autopilot", "shell"] | None = None, request_headers: dict[str, str] | None = None, @@ -1726,6 +1730,8 @@ async def send( Args: prompt: The message text to send. attachments: Optional file, directory, or selection attachments. + source: Optional message provenance (``"user"`` or ``"system"``). + Omitted when None, preserving the runtime's default for user messages. mode: Message delivery mode (``"enqueue"`` or ``"immediate"``). agent_mode: The UI mode the agent was in when this message was sent (for example ``"plan"`` or ``"autopilot"``). Defaults to the @@ -1752,6 +1758,8 @@ async def send( } if attachments is not None: params["attachments"] = attachments + if source is not None: + params["source"] = source if mode is not None: params["mode"] = mode if agent_mode is not None: @@ -1780,6 +1788,7 @@ async def send_and_wait( prompt: str, *, attachments: list[Attachment] | None = None, + source: MessageSource | None = None, mode: Literal["enqueue", "immediate"] | None = None, agent_mode: Literal["interactive", "plan", "autopilot", "shell"] | None = None, request_headers: dict[str, str] | None = None, @@ -1798,6 +1807,8 @@ async def send_and_wait( Args: prompt: The message text to send. attachments: Optional file, directory, or selection attachments. + source: Optional message provenance (``"user"`` or ``"system"``), + independent of delivery mode. Omitted when None. mode: Message delivery mode (``"enqueue"`` or ``"immediate"``). agent_mode: The UI mode the agent was in when this message was sent (for example ``"plan"`` or ``"autopilot"``). Defaults to the @@ -1861,6 +1872,7 @@ def handler(event: SessionEventTypeAlias) -> None: await self.send( prompt, attachments=attachments, + source=source, mode=mode, agent_mode=agent_mode, request_headers=request_headers, diff --git a/python/test_session.py b/python/test_session.py index d58ce94091..0d1ff9f0b0 100644 --- a/python/test_session.py +++ b/python/test_session.py @@ -7,11 +7,13 @@ import pytest +from copilot import MessageSource from copilot.session import CopilotSession from copilot.session_events import ( AssistantMessageData, ExternalToolCompletedData, ExternalToolRequestedData, + SessionErrorData, SessionEvent, SessionEventType, SessionIdleData, @@ -29,6 +31,121 @@ def _event(data, event_type: SessionEventType) -> SessionEvent: ) +@pytest.mark.asyncio +async def test_send_omits_source_for_plain_human_prompt(monkeypatch): + monkeypatch.setattr("copilot.session.get_trace_context", lambda: {}) + client = Mock() + client.request = AsyncMock(return_value={"messageId": "message-1"}) + session = CopilotSession("session-1", client) + + assert await session.send("hello") == "message-1" + client.request.assert_awaited_once_with( + "session.send", {"sessionId": "session-1", "prompt": "hello"} + ) + + +@pytest.mark.parametrize("source", [None, "user", "system"]) +@pytest.mark.parametrize("mode", [None, "enqueue", "immediate"]) +@pytest.mark.asyncio +async def test_send_source_is_optional(source: MessageSource | None, mode, monkeypatch): + monkeypatch.setattr("copilot.session.get_trace_context", lambda: {}) + client = Mock() + client.request = AsyncMock(return_value={"messageId": "message-1"}) + session = CopilotSession("session-1", client) + + assert await session.send("hello", source=source, mode=mode) == "message-1" + expected = {"sessionId": "session-1", "prompt": "hello"} + if source is not None: + expected["source"] = source + if mode is not None: + expected["mode"] = mode + client.request.assert_awaited_once_with("session.send", expected) + + +@pytest.mark.parametrize("source", [None, "user", "system"]) +@pytest.mark.asyncio +async def test_send_source_preserves_other_options(source: MessageSource | None, monkeypatch): + trace = { + "traceparent": "00-fedcba0987654321fedcba0987654321-abcdef1234567890-01", + "tracestate": "vendor=source", + } + monkeypatch.setattr("copilot.session.get_trace_context", lambda: trace) + client = Mock() + client.request = AsyncMock(return_value={"messageId": "message-1"}) + session = CopilotSession("session-1", client) + attachments = [{"type": "blob", "data": "aGk=", "mimeType": "text/plain"}] + + await session.send( + "context updated", + source=source, + mode="immediate", + agent_mode="plan", + attachments=attachments, + display_prompt="Context updated", + request_headers={"X-Tag": "context"}, + ) + + expected = { + "sessionId": "session-1", + "prompt": "context updated", + "mode": "immediate", + "agentMode": "plan", + "attachments": attachments, + "displayPrompt": "Context updated", + "requestHeaders": {"X-Tag": "context"}, + **trace, + } + if source is not None: + expected["source"] = source + client.request.assert_awaited_once_with("session.send", expected) + + +@pytest.mark.parametrize("source", [None, "user", "system"]) +@pytest.mark.asyncio +async def test_send_and_wait_source_allows_idle_without_assistant( + source: MessageSource | None, monkeypatch +): + monkeypatch.setattr("copilot.session.get_trace_context", lambda: {}) + client = Mock() + session = CopilotSession("session-1", client) + + async def respond(method, params): + assert method == "session.send" + expected = {"sessionId": "session-1", "prompt": "context updated"} + if source is not None: + expected["source"] = source + assert params == expected + session._dispatch_event(_event(SessionIdleData(), SessionEventType.SESSION_IDLE)) + return {"messageId": "message-1"} + + client.request = AsyncMock(side_effect=respond) + assert await session.send_and_wait("context updated", source=source, timeout=1) is None + + +@pytest.mark.parametrize("rpc_error", [True, False]) +@pytest.mark.asyncio +async def test_send_and_wait_system_source_preserves_errors(rpc_error): + client = Mock() + session = CopilotSession("session-1", client) + + async def respond(method, params): + assert method == "session.send" + assert params["source"] == "system" + if rpc_error: + raise RuntimeError("send failed") + session._dispatch_event( + _event( + SessionErrorData(error_type="notification", message="agent failed"), + SessionEventType.SESSION_ERROR, + ) + ) + return {"messageId": "message-1"} + + client.request = AsyncMock(side_effect=respond) + with pytest.raises(RuntimeError if rpc_error else Exception, match="send failed|agent failed"): + await session.send_and_wait("context updated", source="system", timeout=1) + + @pytest.mark.asyncio async def test_send_and_wait_skips_autopilot_continuation_idle(): client = Mock() diff --git a/rust/README.md b/rust/README.md index 2de8a88a2d..bac7812080 100644 --- a/rust/README.md +++ b/rust/README.md @@ -818,6 +818,40 @@ let client = Client::start(opts).await?; The SDK injects the appropriate environment variables (`COPILOT_OTEL_EXPORTER_TYPE`, `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_PROTOCOL`, ...) into the spawned CLI process. The SDK takes no OpenTelemetry dependency; the CLI itself owns the exporter pipeline. Caller-supplied `ClientOptions::env` entries override telemetry-injected values. +### Message Source + +Use `MessageSource::System` for automated messages sent by your application. Ordinary human sends leave `source` unset, so the field is omitted from the request. Use `MessageSource::User` when you need to set it explicitly. + +```rust,no_run +use github_copilot_sdk::{MessageOptions, MessageSource, session::Session}; + +# async fn example(session: &Session) -> Result<(), github_copilot_sdk::Error> { +session + .send(MessageOptions::new("Context updated").with_source(MessageSource::System)) + .await?; +# Ok(()) +# } +``` + +The raw RPC path supports the same builder, including requests with JSON attachments: + +```rust,no_run +use github_copilot_sdk::{MessageSource, rpc::SendRequest, session::Session}; + +# async fn example(session: &Session) -> Result<(), github_copilot_sdk::Error> { +let mut request = SendRequest::default().with_source(MessageSource::System); +request.prompt = "Context updated".into(); +request.attachments = Some(vec![serde_json::json!({ + "type": "github_url", + "url": "https://github.com/github/copilot-sdk" +})]); +session.rpc().send(request).await?; +# Ok(()) +# } +``` + +Both paths use ordinary `session.send`. Source does not select a delivery mode or set billing flags; the runtime applies its existing source behavior. `send_and_wait` still completes on `session.idle` and may return `Ok(None)` when no assistant message was emitted. Genuine errors still propagate. + ### Progress Reporting (`send_and_wait`) For fire-and-forget messaging where you need to block until the agent finishes: diff --git a/rust/src/rpc.rs b/rust/src/rpc.rs index a08a501cb2..93affa7107 100644 --- a/rust/src/rpc.rs +++ b/rust/src/rpc.rs @@ -10,3 +10,19 @@ pub use crate::generated::api_types::*; pub use crate::generated::rpc::*; + +impl SendRequest { + /// Set the message provenance without changing other request options. + /// + /// When this is not called, the source field is omitted by default. + pub fn with_source(mut self, source: crate::MessageSource) -> Self { + self.source = Some( + match source { + crate::MessageSource::User => "user", + crate::MessageSource::System => "system", + } + .to_string(), + ); + self + } +} diff --git a/rust/src/session.rs b/rust/src/session.rs index 0e64d6061c..82e97b6606 100644 --- a/rust/src/session.rs +++ b/rust/src/session.rs @@ -522,6 +522,9 @@ impl Session { "sessionId": self.id, "prompt": opts.prompt, }); + if let Some(source) = opts.source { + params["source"] = serde_json::to_value(source)?; + } if let Some(m) = opts.mode { params["mode"] = serde_json::to_value(m)?; } diff --git a/rust/src/types.rs b/rust/src/types.rs index a99f00a19f..0f7cb6682f 100644 --- a/rust/src/types.rs +++ b/rust/src/types.rs @@ -5315,6 +5315,20 @@ pub fn ensure_attachment_display_names(attachments: &mut [Attachment]) { } } +/// Provenance of a message sent through `session.send`. +/// +/// Source is independent of delivery mode. Leaving [`MessageOptions::source`] +/// unset omits the field and preserves the runtime's default for user messages. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +#[non_exhaustive] +pub enum MessageSource { + /// A message from a human user. + User, + /// An automated message from the integrating application. + System, +} + /// Message delivery mode for [`MessageOptions::mode`]. /// /// Controls how a prompt is delivered relative to in-flight session work. @@ -5380,6 +5394,9 @@ pub enum AgentMode { pub struct MessageOptions { /// The user prompt to send. pub prompt: String, + /// Optional message provenance. When `None`, the field is omitted, + /// preserving the runtime's default for user messages. + pub source: Option, /// Optional message delivery mode for this turn. /// /// Controls whether the prompt is queued behind in-flight work @@ -5419,6 +5436,7 @@ impl MessageOptions { pub fn new(prompt: impl Into) -> Self { Self { prompt: prompt.into(), + source: None, mode: None, agent_mode: None, attachments: None, @@ -5430,6 +5448,12 @@ impl MessageOptions { } } + /// Set the message provenance without changing its delivery mode. + pub fn with_source(mut self, source: MessageSource) -> Self { + self.source = Some(source); + self + } + /// Set the message delivery mode for this turn. /// /// Pass [`DeliveryMode::Immediate`] to interrupt the session and run diff --git a/rust/tests/session_test.rs b/rust/tests/session_test.rs index e7bafc683c..ef5b8dd145 100644 --- a/rust/tests/session_test.rs +++ b/rust/tests/session_test.rs @@ -19,7 +19,7 @@ use github_copilot_sdk::handler::{ }; use github_copilot_sdk::rpc::{ CanvasProviderInvokeActionRequest, CanvasProviderOpenRequest, CanvasProviderOpenResult, - OpenCanvasInstance, + OpenCanvasInstance, SendAgentMode, SendMode, SendRequest, }; use github_copilot_sdk::session_events::{ ManagedSettingsResolvedSource, McpOauthRequiredData, ReasoningSummary, SessionLimitsConfig, @@ -33,7 +33,9 @@ use github_copilot_sdk::types::{ PermissionDecisionOutcome, PermissionDecisionSource, PermissionDecisionSurface, RequestId, SessionConfig, SessionId, SetModelOptions, Tool, ToolInvocation, ToolResult, }; -use github_copilot_sdk::{Client, ContextTier, ErrorKind, ProtocolErrorKind, tool}; +use github_copilot_sdk::{ + AgentMode, Attachment, Client, ContextTier, ErrorKind, MessageSource, ProtocolErrorKind, tool, +}; use serde_json::Value; use tokio::io::{AsyncWrite, AsyncWriteExt, duplex}; use tokio::sync::Notify; @@ -1865,6 +1867,193 @@ async fn send_injects_session_id() { timeout(TIMEOUT, handle).await.unwrap().unwrap().unwrap(); } +#[test] +fn message_options_source_is_opt_in() { + let prompt = "hello".to_string(); + for options in [ + MessageOptions::new(&prompt), + MessageOptions::from(prompt.as_str()), + MessageOptions::from(prompt.clone()), + MessageOptions::from(&prompt), + ] { + assert_eq!(options.source, None); + assert_eq!( + options + .with_source(MessageSource::System) + .with_source(MessageSource::User) + .source, + Some(MessageSource::User) + ); + } + for (source, wire) in [ + (MessageSource::User, "user"), + (MessageSource::System, "system"), + ] { + assert_eq!(serde_json::to_value(source).unwrap(), wire); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + source + ); + } +} + +#[tokio::test] +async fn send_source_is_optional_and_preserves_other_options() { + let (session, mut server) = create_session_pair().await; + let session = Arc::new(session); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("context.txt"); + + for (source, wire_source) in [ + (None, None), + (Some(MessageSource::User), Some("user")), + (Some(MessageSource::System), Some("system")), + ] { + for (mode, wire_mode) in [ + (None, None), + (Some(DeliveryMode::Enqueue), Some("enqueue")), + (Some(DeliveryMode::Immediate), Some("immediate")), + ] { + for include_options in [false, true] { + let mut options = MessageOptions::new("hello"); + let mut expected = serde_json::json!({ + "sessionId": server.session_id, + "prompt": "hello", + }); + if let Some(source) = source { + options = options.with_source(source); + expected["source"] = serde_json::json!(wire_source.unwrap()); + } + if let Some(mode) = mode { + options = options.with_mode(mode); + expected["mode"] = serde_json::json!(wire_mode.unwrap()); + } + if include_options { + options = options + .with_agent_mode(AgentMode::Plan) + .with_attachments(vec![Attachment::File { + path: path.clone(), + display_name: None, + line_range: None, + }]) + .with_display_prompt("Context updated") + .with_request_headers(HashMap::from([("X-Tag".into(), "context".into())])) + .with_traceparent("00-source-trace-01") + .with_tracestate("vendor=source") + .with_wait_timeout(Duration::from_secs(10)); + expected["agentMode"] = serde_json::json!("plan"); + expected["attachments"] = serde_json::json!([{ + "type": "file", "path": path, "displayName": "context.txt", + }]); + expected["displayPrompt"] = serde_json::json!("Context updated"); + expected["requestHeaders"] = serde_json::json!({"X-Tag": "context"}); + expected["traceparent"] = serde_json::json!("00-source-trace-01"); + expected["tracestate"] = serde_json::json!("vendor=source"); + } + + let handle = tokio::spawn({ + let session = session.clone(); + async move { session.send(options).await } + }); + let request = timeout(TIMEOUT, server.read_request()).await.unwrap(); + assert_eq!(request["method"], "session.send"); + assert_eq!(request["params"], expected); + server + .respond(&request, serde_json::json!({"messageId": "source-message"})) + .await; + assert_eq!( + timeout(TIMEOUT, handle).await.unwrap().unwrap().unwrap(), + "source-message" + ); + } + } + } +} + +#[tokio::test] +async fn rpc_send_source_is_optional_and_preserves_other_options() { + let (session, mut server) = create_session_pair().await; + let session = Arc::new(session); + + for (source, wire_source) in [ + (None, None), + (Some(MessageSource::User), Some("user")), + (Some(MessageSource::System), Some("system")), + ] { + for (mode, wire_mode) in [ + (None, None), + (Some(SendMode::Enqueue), Some("enqueue")), + (Some(SendMode::Immediate), Some("immediate")), + ] { + for include_options in [false, true] { + let mut options = SendRequest::default(); + options.prompt = "hello".into(); + options.mode = mode.clone(); + let mut expected = serde_json::json!({ + "sessionId": server.session_id, + "prompt": "hello", + }); + if let Some(source) = source { + options = options + .with_source(MessageSource::System) + .with_source(source); + expected["source"] = serde_json::json!(wire_source.unwrap()); + } + if let Some(wire_mode) = wire_mode { + expected["mode"] = serde_json::json!(wire_mode); + } + if include_options { + let attachment = serde_json::json!({ + "type": "extension-context", + "data": {"extensionId": "project:example", "context": ["updated"]}, + }); + options.agent_mode = Some(SendAgentMode::Plan); + options.attachments = Some(vec![attachment.clone()]); + options.display_prompt = Some("Context updated".into()); + options.request_headers = + Some(HashMap::from([("X-Tag".into(), "context".into())])); + options.traceparent = Some("00-source-trace-01".into()); + options.tracestate = Some("vendor=source".into()); + options.billable = Some(true); + options.prepend = Some(false); + options.required_tool = Some("read_file".into()); + options.wait = Some(false); + expected["agentMode"] = serde_json::json!("plan"); + expected["attachments"] = serde_json::json!([attachment]); + expected["displayPrompt"] = serde_json::json!("Context updated"); + expected["requestHeaders"] = serde_json::json!({"X-Tag": "context"}); + expected["traceparent"] = serde_json::json!("00-source-trace-01"); + expected["tracestate"] = serde_json::json!("vendor=source"); + expected["billable"] = serde_json::json!(true); + expected["prepend"] = serde_json::json!(false); + expected["requiredTool"] = serde_json::json!("read_file"); + expected["wait"] = serde_json::json!(false); + } + + let handle = tokio::spawn({ + let session = session.clone(); + async move { session.rpc().send(options).await } + }); + let request = timeout(TIMEOUT, server.read_request()).await.unwrap(); + assert_eq!(request["method"], "session.send"); + assert_eq!(request["params"], expected); + server + .respond(&request, serde_json::json!({"messageId": "source-message"})) + .await; + assert_eq!( + timeout(TIMEOUT, handle) + .await + .unwrap() + .unwrap() + .unwrap() + .message_id, + "source-message" + ); + } + } + } +} + #[tokio::test] async fn send_serializes_request_headers() { use std::collections::HashMap; @@ -3463,6 +3652,104 @@ async fn send_and_wait_returns_last_assistant_message_on_idle() { assert_eq!(event.data["message"], "Hello back!"); } +#[tokio::test] +async fn send_and_wait_system_source_returns_none_on_idle_without_assistant() { + let (session, mut server) = create_session_pair().await; + let session = Arc::new(session); + + let handle = tokio::spawn({ + let session = session.clone(); + async move { + session + .send_and_wait( + MessageOptions::new("Context updated") + .with_source(MessageSource::System) + .with_wait_timeout(TIMEOUT), + ) + .await + } + }); + let request = timeout(TIMEOUT, server.read_request()).await.unwrap(); + assert_eq!(request["method"], "session.send"); + assert_eq!(request["params"]["source"], "system"); + server + .respond(&request, serde_json::json!({"messageId": "system-message"})) + .await; + server + .send_event("session.idle", serde_json::json!({})) + .await; + assert!( + timeout(TIMEOUT, handle) + .await + .unwrap() + .unwrap() + .unwrap() + .is_none() + ); + + let handle = tokio::spawn(async move { session.send("human follow-up").await }); + let request = timeout(TIMEOUT, server.read_request()).await.unwrap(); + assert!(request["params"].get("source").is_none()); + server + .respond(&request, serde_json::json!({"messageId": "human-message"})) + .await; + assert_eq!( + timeout(TIMEOUT, handle).await.unwrap().unwrap().unwrap(), + "human-message" + ); +} + +#[tokio::test] +async fn send_and_wait_system_source_preserves_errors() { + let (session, mut server) = create_session_pair().await; + let session = Arc::new(session); + + for rpc_error in [true, false] { + let handle = tokio::spawn({ + let session = session.clone(); + async move { + session + .send_and_wait( + MessageOptions::new("Context updated") + .with_source(MessageSource::System) + .with_wait_timeout(TIMEOUT), + ) + .await + } + }); + let request = timeout(TIMEOUT, server.read_request()).await.unwrap(); + assert_eq!(request["params"]["source"], "system"); + if rpc_error { + server.respond_error(&request, -32603, "send failed").await; + } else { + server + .respond(&request, serde_json::json!({"messageId": "system-message"})) + .await; + server + .send_event( + "session.error", + serde_json::json!({"message": "agent failed"}), + ) + .await; + } + let error = timeout(TIMEOUT, handle) + .await + .unwrap() + .unwrap() + .unwrap_err(); + if rpc_error { + assert!(matches!(error.kind(), ErrorKind::Rpc { code: -32603, .. })); + assert!(error.to_string().contains("send failed")); + } else { + assert!(matches!( + error.kind(), + ErrorKind::Session(github_copilot_sdk::SessionErrorKind::AgentError) + )); + assert!(error.to_string().contains("agent failed")); + } + } +} + #[tokio::test] async fn send_and_wait_returns_error_on_session_error() { let (session, mut server) = create_session_pair().await; @@ -6182,20 +6469,26 @@ async fn on_get_trace_context_called_on_session_send() { let baseline = calls.load(Ordering::Relaxed); assert_eq!(baseline, 1, "create_session should call the provider once"); - let send_handle = tokio::spawn({ - let session = session.clone(); - async move { session.send(MessageOptions::new("hi")).await } - }); - let send_req = server.read_request().await; - assert_eq!(send_req["method"], "session.send"); - assert_eq!(send_req["params"]["traceparent"], "00-send-trace-01"); - server.respond(&send_req, serde_json::json!({})).await; - timeout(TIMEOUT, send_handle) - .await - .unwrap() - .unwrap() - .unwrap(); - assert_eq!(calls.load(Ordering::Relaxed), baseline + 1); + for source in [None, Some(MessageSource::System)] { + let send_handle = tokio::spawn({ + let session = session.clone(); + async move { + let mut options = MessageOptions::new("hi"); + options.source = source; + session.send(options).await + } + }); + let send_req = server.read_request().await; + assert_eq!(send_req["method"], "session.send"); + assert_eq!(send_req["params"]["traceparent"], "00-send-trace-01"); + server.respond(&send_req, serde_json::json!({})).await; + timeout(TIMEOUT, send_handle) + .await + .unwrap() + .unwrap() + .unwrap(); + } + assert_eq!(calls.load(Ordering::Relaxed), baseline + 2); } #[tokio::test] @@ -6233,27 +6526,27 @@ async fn message_options_trace_context_overrides_callback() { let baseline = calls.load(Ordering::Relaxed); - let send_handle = tokio::spawn({ - let session = session.clone(); - async move { - session - .send( - MessageOptions::new("hi") - .with_traceparent("00-override-01") - .with_tracestate("vendor=override"), - ) - .await - } - }); - let send_req = server.read_request().await; - assert_eq!(send_req["params"]["traceparent"], "00-override-01"); - assert_eq!(send_req["params"]["tracestate"], "vendor=override"); - server.respond(&send_req, serde_json::json!({})).await; - timeout(TIMEOUT, send_handle) - .await - .unwrap() - .unwrap() - .unwrap(); + for source in [None, Some(MessageSource::System)] { + let send_handle = tokio::spawn({ + let session = session.clone(); + async move { + let mut options = MessageOptions::new("hi") + .with_traceparent("00-override-01") + .with_tracestate("vendor=override"); + options.source = source; + session.send(options).await + } + }); + let send_req = server.read_request().await; + assert_eq!(send_req["params"]["traceparent"], "00-override-01"); + assert_eq!(send_req["params"]["tracestate"], "vendor=override"); + server.respond(&send_req, serde_json::json!({})).await; + timeout(TIMEOUT, send_handle) + .await + .unwrap() + .unwrap() + .unwrap(); + } // Callback must NOT have been invoked when MessageOptions carried an override. assert_eq!(