feat: support optional message provenance across all SDKs

docs/features/steering-and-queueing.md

@@ -33,6 +33,28 @@ sequenceDiagram
     LLM->>S: Turn completes
 ```
 
+## Message provenance
+
+Set the optional `source` field when forwarding a message from another agent. For example, `agent-sender-id` identifies an agent-originated message. Leave it unset for existing source-less sends. Both send and send-and-wait APIs forward the field, including with `"enqueue"` and `"immediate"` delivery.
+
+| SDK | Source option |
+|-----|---------------|
+| Node.js / TypeScript | `MessageOptions.source` |
+| Python | `source=` keyword on `send` and `send_and_wait` |
+| Go | `MessageOptions.Source`, a `*string` |
+| .NET | `MessageOptions.Source` |
+| Java | `MessageOptions.setSource(...)` |
+| Rust | `MessageOptions::with_source(...)` |
+
+The runtime accepts `user`, `system`, `command-<id>`, `schedule-<numeric-id>`, and `agent-<id>`. The SDK forwards the supplied value unchanged and omits it when unset. Rust callers using the typed RPC API can use `rpc::SendRequest::with_source(...)` without accessing generated internal fields.
+
+Source identifies who originated the message, while `"immediate"` mode requests delivery urgency. Neither guarantees a visible reply. Source does not change billing defaults. The runtime uses provenance for scheduling and to distinguish agent input from human authorization evidence, so derive it from trusted sender metadata rather than message text.
+
+A successful high-level `send` acknowledgement returns a message ID and confirms acceptance, not that the recipient has consumed the message. Do not automatically resend an accepted message merely because no reply appears.
+
+> [!WARNING]
+> Remote backends do not necessarily preserve source end to end. Mission Control accepts it and includes it in the local echo, but drops it from remote HTTP delivery. A local source event does not prove that the remote worker received the same provenance.
+
 ## Steering (immediate mode)
 
 Steering sends a message that is injected directly into the agent's current turn. The agent sees the message in real time and adjusts its response accordingly—useful for course-correcting without aborting the turn.

dotnet/src/Session.cs

@@ -326,6 +326,7 @@ public async Task<string> SendAsync(MessageOptions options, CancellationToken ca
             DisplayPrompt = options.DisplayPrompt,
             Attachments = options.Attachments,
             Mode = options.Mode,
+            Source = options.Source,
             AgentMode = options.AgentMode,
             Traceparent = traceparent,
             Tracestate = tracestate,
@@ -2268,6 +2269,7 @@ internal record SendMessageRequest
         public string? DisplayPrompt { get; init; }
         public IList<Attachment>? Attachments { get; init; }
         public string? Mode { get; init; }
+        public string? Source { get; init; }
         [JsonPropertyName("agentMode")]
         public AgentMode? AgentMode { get; init; }
         public string? Traceparent { get; init; }

dotnet/src/Types.cs

@@ -4028,6 +4028,7 @@ private MessageOptions(MessageOptions? other)
 
         Attachments = other.Attachments is not null ? [.. other.Attachments] : null;
         Mode = other.Mode;
+        Source = other.Source;
         AgentMode = other.AgentMode;
         Prompt = other.Prompt;
         DisplayPrompt = other.DisplayPrompt;
@@ -4050,6 +4051,16 @@ private MessageOptions(MessageOptions? other)
     /// </summary>
     public string? Mode { get; set; }
     /// <summary>
+    /// Optional provenance tag copied to the resulting <c>user.message</c> event.
+    /// Supported forms are <c>user</c>, <c>system</c>, <c>command-&lt;id&gt;</c>,
+    /// <c>schedule-&lt;numeric-id&gt;</c>, and <c>agent-&lt;id&gt;</c>.
+    /// The runtime validates the value; the SDK forwards it unchanged and omits it when null.
+    /// This identifies the message's origin, not a response requirement; the agent may
+    /// complete silently. It does not change delivery mode or billing.
+    /// Remote backends may not retain this metadata end-to-end.
+    /// </summary>
+    public string? Source { get; set; }
+    /// <summary>
     /// The UI mode the agent was in when this message was sent (for example "plan", "autopilot").
     /// Defaults to the session's current mode when unset.
     /// </summary>

dotnet/test/Unit/ClientSessionLifetimeTests.cs

@@ -1477,6 +1477,142 @@ public async Task Generated_Session_Rpc_Throws_When_Session_Disposed()
         await Assert.ThrowsAsync<ObjectDisposedException>(() => session.Rpc.Model.GetCurrentAsync());
     }
 
+    [Theory]
+    [InlineData(null, "enqueue", false)]
+    [InlineData(null, "immediate", true)]
+    [InlineData("user", "enqueue", false)]
+    [InlineData("system", "immediate", true)]
+    [InlineData("command-review", "immediate", false)]
+    [InlineData("schedule-123", "enqueue", true)]
+    [InlineData("agent-reviewer", "enqueue", false)]
+    [InlineData("agent-reviewer", "immediate", true)]
+    [InlineData("", "enqueue", false)]
+    [InlineData("future-source", "immediate", true)]
+    public async Task Send_Source_Is_Serialized_Without_Changing_Other_Fields(
+        string? source, string mode, bool waitForReply)
+    {
+        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 activity = new Activity("send-source").SetIdFormat(ActivityIdFormat.W3C).Start();
+        activity.TraceStateString = "test=value";
+        var options = new MessageOptions
+        {
+            Prompt = "real prompt",
+            DisplayPrompt = "display prompt",
+            Source = source,
+            Mode = mode,
+            AgentMode = AgentMode.Plan,
+            Attachments = [new AttachmentFile { Path = "/test.txt", DisplayName = "test.txt" }],
+            RequestHeaders = new Dictionary<string, string> { ["X-Request-ID"] = "request-1" }
+        }.Clone();
+
+        if (waitForReply)
+        {
+            var replyTask = session.SendAndWaitAsync(options, TimeSpan.FromSeconds(5));
+            await WaitForRequestAsync(server, "session.send");
+            await server.SendEventAsync(session.SessionId, new AssistantMessageEvent
+            {
+                Id = Guid.NewGuid(),
+                Data = new AssistantMessageData { Content = "reply", MessageId = "assistant-1" }
+            });
+            await server.SendEventAsync(session.SessionId, new SessionIdleEvent
+            {
+                Id = Guid.NewGuid(),
+                Data = new SessionIdleData { Mode = SessionMode.Interactive }
+            });
+            var reply = await replyTask;
+            Assert.NotNull(reply);
+            Assert.Equal("reply", reply.Data.Content);
+        }
+        else
+        {
+            Assert.Equal("message-1", await session.SendAsync(options));
+        }
+
+        var request = Assert.Single(server.Requests, request => request.Method == "session.send").Params;
+        if (source is null)
+        {
+            Assert.False(request.TryGetProperty("source", out _));
+        }
+        else
+        {
+            Assert.Equal(source, request.GetProperty("source").GetString());
+        }
+        Assert.Equal(session.SessionId, request.GetProperty("sessionId").GetString());
+        Assert.Equal("real prompt", request.GetProperty("prompt").GetString());
+        Assert.Equal("display prompt", request.GetProperty("displayPrompt").GetString());
+        Assert.Equal(mode, request.GetProperty("mode").GetString());
+        Assert.Equal("plan", request.GetProperty("agentMode").GetString());
+        var attachment = Assert.Single(request.GetProperty("attachments").EnumerateArray());
+        Assert.Equal("file", attachment.GetProperty("type").GetString());
+        Assert.Equal("/test.txt", attachment.GetProperty("path").GetString());
+        Assert.Equal("test.txt", attachment.GetProperty("displayName").GetString());
+        Assert.Equal("request-1", request.GetProperty("requestHeaders").GetProperty("X-Request-ID").GetString());
+        Assert.Equal(activity.Id, request.GetProperty("traceparent").GetString());
+        Assert.Equal("test=value", request.GetProperty("tracestate").GetString());
+        Assert.False(request.TryGetProperty("billable", out _));
+        Assert.False(request.TryGetProperty("wait", out _));
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Send_String_Overloads_Omit_Source_And_Keep_Defaults(bool waitForReply)
+    {
+        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 (waitForReply)
+        {
+            var replyTask = session.SendAndWaitAsync("prompt", TimeSpan.FromSeconds(5));
+            await WaitForRequestAsync(server, "session.send");
+            await server.SendEventAsync(session.SessionId, new SessionIdleEvent
+            {
+                Id = Guid.NewGuid(),
+                Data = new SessionIdleData { Mode = SessionMode.Interactive }
+            });
+            Assert.Null(await replyTask);
+        }
+        else
+        {
+            Assert.Equal("message-1", await session.SendAsync("prompt"));
+        }
+
+        var request = Assert.Single(server.Requests, request => request.Method == "session.send").Params;
+        Assert.Equal("prompt", request.GetProperty("prompt").GetString());
+        Assert.False(request.TryGetProperty("source", out _));
+        Assert.False(request.TryGetProperty("mode", out _));
+        Assert.False(request.TryGetProperty("agentMode", out _));
+        Assert.False(request.TryGetProperty("billable", out _));
+    }
+
+    [Theory]
+    [InlineData(null)]
+    [InlineData("agent-reviewer")]
+    public async Task Rpc_Send_Source_Preserves_Explicit_Billing(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("prompt", source: source, mode: SendMode.Immediate, billable: false);
+
+        Assert.Equal("message-1", result.MessageId);
+        var request = Assert.Single(server.Requests, request => request.Method == "session.send").Params;
+        Assert.Equal("immediate", request.GetProperty("mode").GetString());
+        Assert.False(request.GetProperty("billable").GetBoolean());
+        if (source is null)
+        {
+            Assert.False(request.TryGetProperty("source", out _));
+        }
+        else
+        {
+            Assert.Equal(source, request.GetProperty("source").GetString());
+        }
+    }
+
     [Fact]
     public async Task SendAndWaitAsync_Skips_Autopilot_Continuation_Idle()
     {
@@ -2031,6 +2167,22 @@ public void CloseConnection()
             _stream?.Dispose();
         }
 
+        public async Task SendEventAsync(string sessionId, SessionEvent evt)
+        {
+            var stream = _stream ?? throw new InvalidOperationException("Client is not connected.");
+            using var document = JsonDocument.Parse(evt.ToJson());
+            await WriteMessageAsync(stream, new Dictionary<string, object?>
+            {
+                ["jsonrpc"] = "2.0",
+                ["method"] = "session.event",
+                ["params"] = new Dictionary<string, object?>
+                {
+                    ["sessionId"] = sessionId,
+                    ["event"] = document.RootElement
+                }
+            }, _cts.Token);
+        }
+
         public async Task<JsonElement> SendRequestAsync(string method, Dictionary<string, object?> parameters)
         {
             var stream = _stream ?? throw new InvalidOperationException("Client is not connected.");

dotnet/test/Unit/CloneTests.cs

@@ -275,20 +275,31 @@ public void ResumeSessionConfig_Clone_PreservesMcpServersComparer()
         Assert.True(clone.McpServers!.ContainsKey("SERVER"));
     }
 
-    [Fact]
-    public void MessageOptions_Clone_CopiesAllProperties()
+    [Theory]
+    [InlineData(null)]
+    [InlineData("agent-reviewer")]
+    public void MessageOptions_Clone_CopiesAllProperties(string? source)
     {
         var original = new MessageOptions
         {
             Prompt = "Hello",
             Attachments = [new AttachmentFile { Path = "/test.txt", DisplayName = "test.txt" }],
-            Mode = "chat",
+            Mode = "immediate",
+            Source = source,
+            AgentMode = AgentMode.Plan,
+            DisplayPrompt = "Display hello",
+            RequestHeaders = new Dictionary<string, string> { ["X-Request-ID"] = "request-1" },
         };
 
         var clone = original.Clone();
 
         Assert.Equal(original.Prompt, clone.Prompt);
         Assert.Equal(original.Mode, clone.Mode);
+        Assert.Equal(original.Source, clone.Source);
+        Assert.Equal(original.AgentMode, clone.AgentMode);
+        Assert.Equal(original.DisplayPrompt, clone.DisplayPrompt);
+        Assert.Equal(original.RequestHeaders, clone.RequestHeaders);
+        Assert.NotSame(original.RequestHeaders, clone.RequestHeaders);
         Assert.Single(clone.Attachments!);
     }
 

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,

go/session_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"reflect"
 	"strconv"
 	"strings"
 	"sync"
@@ -95,6 +96,51 @@ func ptr[T any](value T) *T {
 	return &value
 }
 
+func TestSession_SendPreservesSource(t *testing.T) {
+	for _, mode := range []string{"", "enqueue", "immediate"} {
+		for _, source := range []*string{nil, ptr("agent-sender-id")} {
+			t.Run(fmt.Sprintf("%s/source=%v", mode, source != nil), func(t *testing.T) {
+				options := MessageOptions{
+					Prompt:         "Agent update",
+					Source:         source,
+					Mode:           mode,
+					AgentMode:      AgentModePlan,
+					DisplayPrompt:  "Update from sender",
+					RequestHeaders: map[string]string{"X-Custom-Tag": "value-1"},
+					Attachments: []Attachment{
+						&AttachmentFile{Path: "report.txt", DisplayName: "Report"},
+					},
+				}
+				params := captureModelRequest(t, "session.send", func(session *Session) error {
+					ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+					defer cancel()
+					_, err := session.Send(ctx, options)
+					return err
+				})
+				expected := map[string]any{
+					"sessionId":      "session-1",
+					"prompt":         "Agent update",
+					"agentMode":      "plan",
+					"displayPrompt":  "Update from sender",
+					"requestHeaders": map[string]any{"X-Custom-Tag": "value-1"},
+					"attachments": []any{
+						map[string]any{"type": "file", "path": "report.txt", "displayName": "Report"},
+					},
+				}
+				if mode != "" {
+					expected["mode"] = mode
+				}
+				if source != nil {
+					expected["source"] = *source
+				}
+				if !reflect.DeepEqual(params, expected) {
+					t.Fatalf("got params %#v, want %#v", params, expected)
+				}
+			})
+		}
+	}
+}
+
 func TestSession_SetModelForwardsContextTier(t *testing.T) {
 	tier := ContextTierLongContext
 	params := captureSetModelRequest(t, &SetModelOptions{ContextTier: &tier})
@@ -513,6 +559,15 @@ func readTestJSONRPCFrame(r io.Reader) ([]byte, error) {
 }
 
 func TestSession_SendAndWaitSkipsAutopilotContinuationIdle(t *testing.T) {
+	for _, source := range []*string{nil, ptr("agent-sender-id")} {
+		t.Run(fmt.Sprintf("source=%v", source != nil), func(t *testing.T) {
+			testSessionSendAndWaitSkipsAutopilotContinuationIdle(t, source)
+		})
+	}
+}
+
+func testSessionSendAndWaitSkipsAutopilotContinuationIdle(t *testing.T, source *string) {
+	t.Helper()
 	stdinR, stdinW := io.Pipe()
 	stdoutR, stdoutW := io.Pipe()
 	defer stdinR.Close()
@@ -536,6 +591,7 @@ func TestSession_SendAndWaitSkipsAutopilotContinuationIdle(t *testing.T) {
 		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
@@ -545,6 +601,14 @@ func TestSession_SendAndWaitSkipsAutopilotContinuationIdle(t *testing.T) {
 			errCh <- fmt.Errorf("expected session.send, got %s", request.Method)
 			return
 		}
+		if source != nil && request.Params["source"] != *source {
+			errCh <- fmt.Errorf("expected agent source, got %v", request.Params["source"])
+			return
+		}
+		if _, present := request.Params["source"]; source == nil && present {
+			errCh <- fmt.Errorf("expected omitted source, got %v", request.Params["source"])
+			return
+		}
 
 		response := map[string]any{
 			"jsonrpc": "2.0",
@@ -576,7 +640,9 @@ func TestSession_SendAndWaitSkipsAutopilotContinuationIdle(t *testing.T) {
 
 	resultCh := make(chan *SessionEvent, 1)
 	go func() {
-		result, err := session.SendAndWait(t.Context(), MessageOptions{Prompt: "keep going"})
+		result, err := session.SendAndWait(t.Context(), MessageOptions{
+			Prompt: "keep going", Source: source,
+		})
 		if err != nil {
 			errCh <- err
 			return