fix(claude-managed-agents): validate newest first-contact message

integrations/claude-managed-agents/dotnet/README.md

@@ -57,6 +57,21 @@ The Anthropic client reads `ANTHROPIC_API_KEY` (or `ANTHROPIC_AUTH_TOKEN`) from
 | `session.error` (terminal)                       | `RUN_ERROR` with the error type as `code`                                                 |
 | `session.status_idle` (`end_turn`)               | `RUN_FINISHED`                                                                            |
 
+### First-contact attachment validation
+
+A new session sends only the newest user message and now validates that message
+before session creation. Previously, scanning older messages could reject a valid
+correction because of an old malformed attachment, or create a session before the
+newest attachment failed conversion. A malformed newest attachment returns
+`RUN_ERROR` without creating a session; a blank or unsupported-only newest message
+does not fall back to older text. `RUN_STARTED.input` retains the original history
+and attachment bytes. Existing sessions keep their delivery-cursor and pending
+tool-result rules, and text decoding remains strict UTF-8.
+
+The first-contact tests exercise both message orderings with malformed base64 and
+invalid UTF-8, plus a blank newest message, through the adapter with a fake client.
+They do not call the hosted API.
+
 ### Frontend tools (human-in-the-loop)
 
 Tools passed in `RunAgentInput.Tools` are registered on the session as custom tools. When the agent calls one, the run emits the tool call and finishes, leaving the session parked. The client executes the tool and starts the next run with a `role: "tool"` message carrying `toolCallId`. The adapter forwards it into the session as the tool result and resumes streaming.

integrations/claude-managed-agents/dotnet/src/AGUI.ClaudeManagedAgents/ManagedAgentsAgent.cs

@@ -169,34 +169,8 @@ private async IAsyncEnumerable<BaseEvent> StreamRunAsync(
         // The gate is held only inside this try, so it is released however the run ends.
         try
         {
-            // A deserialized input may carry a null message list; treat it as empty. Check for
-            // something sendable before touching the API, so a malformed run does not create an
-            // orphan session.
+            // A deserialized input may carry a null message list; treat it as empty.
             var messages = input.Messages ?? [];
-            bool hasSendableContent = false;
-            string? attachmentError = null;
-            try
-            {
-                hasSendableContent = HasSendableContent(messages);
-            }
-            catch (Exception error) when (error is FormatException or System.Text.DecoderFallbackException)
-            {
-                attachmentError = error.Message;
-            }
-            if (attachmentError is not null)
-            {
-                yield return new RunErrorEvent { Code = "invalid_attachment", Message = attachmentError };
-                yield break;
-            }
-            if (!hasSendableContent && input.Resume is not { Count: > 0 } && _options.ResolveSession is null && await _store.GetAsync(threadKey, cancellationToken).ConfigureAwait(false) is null)
-            {
-                yield return new RunErrorEvent
-                {
-                    Message = "There is nothing to send: this run has no user message or tool result.",
-                    Code = "empty_run",
-                };
-                yield break;
-            }
 
             var run = new RunContext();
             using var timeout = new CancellationTokenSource(_options.TurnTimeout);
@@ -295,6 +269,39 @@ private async IAsyncEnumerable<BaseEvent> RunCoreAsync(
                 await _store.SetAsync(threadKey, record, cancellationToken).ConfigureAwait(false);
             }
         }
+
+        // Resolve the binding before selecting the preflight. First contact sends only
+        // the newest user message, so validate that message before creating a session.
+        // An existing session converts only its undelivered messages, when it sends them.
+        bool hasFirstContactUserContent = false;
+        bool hasSendableContent = false;
+        string? attachmentError = null;
+        try
+        {
+            if (record is null)
+            {
+                hasFirstContactUserContent = HasFirstContactUserContent(messages);
+                hasSendableContent = hasFirstContactUserContent || messages.Any(static message => message is AGUIToolMessage);
+            }
+        }
+        catch (Exception error) when (error is FormatException or System.Text.DecoderFallbackException)
+        {
+            attachmentError = error.Message;
+        }
+        if (attachmentError is not null)
+        {
+            yield return new RunErrorEvent { Code = "invalid_attachment", Message = attachmentError };
+            yield break;
+        }
+        if (record is null && !hasSendableContent && input.Resume is not { Count: > 0 } && _options.ResolveSession is null)
+        {
+            yield return new RunErrorEvent
+            {
+                Message = "There is nothing to send: this run has no user message or tool result.",
+                Code = "empty_run",
+            };
+            yield break;
+        }
         if (record is not null && record.SessionEnded)
         {
             yield return new RunErrorEvent { Code = "session_terminated", Message = "The bound native session has terminated." };
@@ -311,7 +318,7 @@ private async IAsyncEnumerable<BaseEvent> RunCoreAsync(
         {
             // A tool result only answers a pending call, so it cannot start a thread: creating
             // a session for it would orphan the session.
-            if (!HasUserText(messages))
+            if (!hasFirstContactUserContent)
             {
                 yield return new RunErrorEvent
                 {
@@ -774,20 +781,11 @@ private static string FormatSeconds(TimeSpan timeout)
         return $"{timeout.TotalSeconds.ToString("0.###", System.Globalization.CultureInfo.InvariantCulture)}s";
     }
 
-    /// <summary>
-    /// Whether the run carries a user message with text or a tool result.
-    /// </summary>
-    private static bool HasSendableContent(IList<AGUIMessage> messages)
-    {
-        return messages.Any(static message => message is AGUIToolMessage) || HasUserText(messages);
-    }
-
-    /// <summary>
-    /// Whether the run carries a user message with text, the only thing that can start a thread.
-    /// </summary>
-    private static bool HasUserText(IList<AGUIMessage> messages)
+    /// <summary>Validate the same newest user message that first contact will send.</summary>
+    private static bool HasFirstContactUserContent(IList<AGUIMessage> messages)
     {
-        return messages.Any(static message => message is AGUIUserMessage user && ManagedAgentsContent.Blocks(user.Content).Count > 0);
+        var message = messages.OfType<AGUIUserMessage>().LastOrDefault();
+        return message is not null && ManagedAgentsContent.Blocks(message.Content).Count > 0;
     }
 
     private sealed class RunContext

integrations/claude-managed-agents/dotnet/tests/AGUI.ClaudeManagedAgents.Tests/ManagedAgentsFirstContactTest.cs

@@ -0,0 +1,113 @@
+using System.Text.Json;
+using AGUI.Abstractions;
+using Xunit;
+
+namespace AGUI.ClaudeManagedAgents.Tests;
+
+public class ManagedAgentsFirstContactTest
+{
+    private static async Task<(FakeManagedAgentsClient Fake, List<BaseEvent> Events)> RunAsync(IList<AGUIMessage> messages)
+    {
+        var fake = new FakeManagedAgentsClient(["""{"type":"session.status_idle","id":"idle","stop_reason":{"type":"end_turn"}}"""]);
+        var input = new RunAgentInput { ThreadId = "t", RunId = "r", State = JsonSerializer.SerializeToElement(new { }), Messages = messages, Tools = [] };
+        var original = JsonSerializer.SerializeToElement(input);
+        var agent = new ManagedAgentsAgent(new ManagedAgentsAgentOptions { ManagedAgentId = "a", EnvironmentId = "e", Client = fake });
+        var events = new List<BaseEvent>();
+        await foreach (var item in agent.RunAsync(input)) events.Add(item);
+        Assert.Same(input, Assert.IsType<RunStartedEvent>(events[0]).Input);
+        Assert.True(JsonElement.DeepEquals(original, JsonSerializer.SerializeToElement(input)));
+        return (fake, events);
+    }
+
+    [Theory]
+    [InlineData("%%%", false)]
+    [InlineData("/w==", false)]
+    [InlineData("%%%", true)]
+    [InlineData("/w==", true)]
+    public async Task FirstContactValidatesOnlyNewestUserMessage(string data, bool malformedLast)
+    {
+        var malformed = new AGUIUserMessage
+        {
+            Id = "malformed",
+            Content = new AGUIContent(new List<AGUIInputContent> {
+                new AGUIDocumentInputContent { Source = new AGUIInputContentDataSource { Value = data, MimeType = "text/plain" } },
+            }),
+        };
+        var corrected = new AGUIUserMessage { Id = "corrected", Content = "corrected message" };
+        var (fake, events) = await RunAsync(malformedLast ? [corrected, malformed] : [malformed, corrected]);
+        if (malformedLast)
+        {
+            Assert.Equal("invalid_attachment", Assert.IsType<RunErrorEvent>(events[^1]).Code);
+            Assert.Single(events.OfType<RunErrorEvent>());
+            Assert.Empty(fake.CreatedSessions);
+            Assert.Empty(fake.Sent);
+        }
+        else
+        {
+            Assert.IsType<RunFinishedEvent>(events[^1]);
+            Assert.Single(fake.CreatedSessions);
+            var sent = Assert.Single(fake.Sent.SelectMany(batch => batch), item => item.GetProperty("type").GetString() == "user.message");
+            Assert.Equal("corrected message", Assert.Single(sent.GetProperty("content").EnumerateArray()).GetProperty("text").GetString());
+        }
+    }
+
+    [Fact]
+    public async Task FirstContactDoesNotCreateFromOlderTextWhenNewestUserIsBlank()
+    {
+        var (fake, events) = await RunAsync([
+            new AGUIUserMessage { Id = "old", Content = "older message" },
+            new AGUIUserMessage { Id = "new", Content = " \n" },
+        ]);
+        Assert.IsType<RunErrorEvent>(events[^1]);
+        Assert.Empty(fake.CreatedSessions);
+        Assert.Empty(fake.Sent);
+    }
+
+    [Theory]
+    [InlineData("%%%")]
+    [InlineData("/w==")]
+    public async Task ExistingSessionDoesNotRevalidateDeliveredHistory(string data)
+    {
+        var fake = new FakeManagedAgentsClient([Idle("idle")], [Idle("idle_2")]);
+        var agent = new ManagedAgentsAgent(new ManagedAgentsAgentOptions { ManagedAgentId = "a", EnvironmentId = "e", Client = fake });
+        var corrected = new AGUIUserMessage { Id = "corrected", Content = "corrected message" };
+        Assert.IsType<RunFinishedEvent>((await CollectAsync(agent, [Malformed(data), corrected], "r1"))[^1]);
+        var events = await CollectAsync(agent, [Malformed(data), corrected, new AGUIUserMessage { Id = "next", Content = "next message" }], "r2");
+        Assert.IsType<RunFinishedEvent>(events[^1]);
+        Assert.Single(fake.CreatedSessions);
+        var sent = Assert.Single(fake.Sent[^1], item => item.GetProperty("type").GetString() == "user.message");
+        Assert.Equal("next message", Assert.Single(sent.GetProperty("content").EnumerateArray()).GetProperty("text").GetString());
+    }
+
+    [Theory]
+    [InlineData("%%%")]
+    [InlineData("/w==")]
+    public async Task ExistingSessionRejectsMalformedNewestUserMessage(string data)
+    {
+        var fake = new FakeManagedAgentsClient([Idle("idle")], [Idle("idle_2")]);
+        var agent = new ManagedAgentsAgent(new ManagedAgentsAgentOptions { ManagedAgentId = "a", EnvironmentId = "e", Client = fake });
+        var corrected = new AGUIUserMessage { Id = "corrected", Content = "corrected message" };
+        Assert.IsType<RunFinishedEvent>((await CollectAsync(agent, [corrected], "r1"))[^1]);
+        var sent = fake.Sent.Count;
+        var events = await CollectAsync(agent, [corrected, Malformed(data)], "r2");
+        Assert.Equal("run_failed", Assert.IsType<RunErrorEvent>(events[^1]).Code);
+        Assert.Equal(sent, fake.Sent.Count);
+    }
+
+    private static string Idle(string id) => $$$"""{"type":"session.status_idle","id":"{{{id}}}","stop_reason":{"type":"end_turn"}}""";
+
+    private static AGUIUserMessage Malformed(string data) => new()
+    {
+        Id = "malformed",
+        Content = new AGUIContent(new List<AGUIInputContent> {
+            new AGUIDocumentInputContent { Source = new AGUIInputContentDataSource { Value = data, MimeType = "text/plain" } },
+        }),
+    };
+
+    private static async Task<List<BaseEvent>> CollectAsync(ManagedAgentsAgent agent, IList<AGUIMessage> messages, string runId)
+    {
+        var events = new List<BaseEvent>();
+        await foreach (var item in agent.RunAsync(new RunAgentInput { ThreadId = "t", RunId = runId, Messages = messages, Tools = [] })) events.Add(item);
+        return events;
+    }
+}

integrations/claude-managed-agents/python/README.md

@@ -43,6 +43,21 @@ The Anthropic client reads `ANTHROPIC_API_KEY` from the environment. Pass `clien
 | `session.error` (terminal)                       | `RUN_ERROR` with the error type as `code`                                                 |
 | `session.status_idle` (`end_turn`)               | `RUN_FINISHED`                                                                            |
 
+### First-contact attachment validation
+
+A new session sends only the newest user message and now validates that message
+before session creation. Previously, scanning older messages could reject a valid
+correction because of an old malformed attachment, or create a session before the
+newest attachment failed conversion. A malformed newest attachment returns
+`RUN_ERROR` without creating a session; a blank or unsupported-only newest message
+does not fall back to older text. `RUN_STARTED.input` retains the original history
+and attachment bytes. Existing sessions keep their delivery-cursor and pending
+tool-result rules, and text decoding remains strict UTF-8.
+
+The first-contact tests exercise both message orderings with malformed base64 and
+invalid UTF-8, plus a blank newest message, through the adapter with a fake client.
+They do not call the hosted API.
+
 ### Frontend tools (human in the loop)
 
 Tools passed in `RunAgentInput.tools` are registered on the session as custom tools. When the agent calls one, the run emits the tool call and finishes, leaving the session parked. The client executes the tool and starts the next run with a `role: "tool"` message carrying `toolCallId`. The adapter forwards it into the session as the tool result and resumes streaming.

integrations/claude-managed-agents/python/ag_ui_claude_managed_agents/agent.py

@@ -50,8 +50,13 @@
 """
 
 
-def _has_user_content(message: Any) -> bool:
-    return bool(user_content(message))
+def _has_first_contact_user_content(messages: Sequence[Any]) -> bool:
+    """Validate the same newest user message that first contact will send."""
+    message = next(
+        (message for message in reversed(messages) if getattr(message, "role", None) == "user"),
+        None,
+    )
+    return message is not None and bool(user_content(message))
 
 
 def _tool_result_blocks(content: Any, error: str | None) -> list[dict[str, Any]]:
@@ -349,8 +354,8 @@ async def _run_turn_for_input(
         if (
             not self.resolve_session
             and not getattr(input, "resume", None)
-            and not self._has_sendable_content(input.messages)
             and await maybe_await(self.store.get(store_key)) is None
+            and not self._has_sendable_content(input.messages)
         ):
             emit(
                 RunErrorEvent(
@@ -510,14 +515,8 @@ def _client_tools(self, client_tools: Sequence[Any]) -> dict[str, str]:
         return {normalize_tool_name(tool.name): tool.name for tool in client_tools}
 
     def _has_sendable_content(self, messages: Sequence[Any]) -> bool:
-        """Whether the run carries a user message with text or a tool result."""
-        for message in messages:
-            role = getattr(message, "role", None)
-            if role == "tool":
-                return True
-            if role == "user" and _has_user_content(message):
-                return True
-        return False
+        """Whether a new thread carries supported user content or a tool result."""
+        return any(getattr(message, "role", None) == "tool" for message in messages) or _has_first_contact_user_content(messages)
 
     def _session_key(self, thread_id: str) -> str:
         """The key that identifies this thread's state, in the session store and
@@ -681,10 +680,7 @@ async def _get_or_create_session(
 
         # A tool result only answers a pending call on an existing session;
         # never create a session to receive one.
-        if not any(
-            getattr(m, "role", None) == "user" and _has_user_content(m)
-            for m in input.messages
-        ):
+        if not _has_first_contact_user_content(input.messages):
             return None
 
         # The busy-thread gate serializes runs per thread, so creation cannot race.