diff --git a/.changeset/reconnect-keeps-supports.md b/.changeset/reconnect-keeps-supports.md new file mode 100644 index 00000000..83552388 --- /dev/null +++ b/.changeset/reconnect-keeps-supports.md @@ -0,0 +1,48 @@ +--- +"@smooai/smooth-operator": patch +--- + +fix(all five): a reconnect no longer silently turns Rich Interactions off + +`supports` — the client render-capability list that gates the **entire** Rich +Interactions framework — was kept somewhere that a reconnect destroys, in every +implementation. + +A **reconnect is a resume**: the client re-opens the socket and re-issues +`create_conversation_session` with the same `conversationId`, which mints a +**new session id** on a **new dispatcher**. So unless the client re-declared +`supports` on every single reconnect, the server forgot the client could render +cards at all and every interaction kind quietly fell back to conversational +collection — no error, no event, nothing on the wire to notice. The parked-card +flow (raise tool → `interaction_required` → `submit_interaction` → resume) +simply stopped happening. Reconnects are routine (network blips, mobile +backgrounding, deploys), so a shipped feature was degrading in the field with no +signal. + +- **Rust** kept it in `Session.metadata.supports` — the per-pod session registry. +- **Go** (`FrameDispatcher.supports`), **Python** (`_session_supports`) and + **.NET** (`_sessionSupports`) kept a per-connection map, also never pruned. +- **TypeScript** already stored it through the `SessionStore`, but on the + **session** record, so a resumed session started empty just the same. + +The session was already the wrong home, and the repo had said so once: th-c12df5 +moved the workflow step pointer off it for exactly this reason ("this per-pod +session map resets on reconnect/pod hop"). `supports` now lives on the +**conversation** in all five, mirroring whatever conversation-scoped mechanism +each store already had — Rust/Go/Python/TypeScript write `clientSupports` into +`conversations.metadata_json`; .NET follows its own store's documented hold for +`currentStepId`/`otpVerified` (session-row metadata) under the same key name. + +A list the frame **does** declare always wins, including `[]` — which is now how +a text-only channel resuming a rich conversation opts out, and the opt-out is +durable so the next reconnect that omits the key cannot resurrect the old +capabilities. Because `[]` and an absent key now mean different things, each port +had to stop collapsing them (`*[]string` in Go, `IReadOnlyList?` in .NET, +`undefined` vs `[]` across the TS store interface). + +The rule lives in the `supports` description in +`spec/actions/create-conversation-session.schema.json` — the source of truth — +and the TS/Go/Python/.NET wire types are regenerated from it rather than +restating it by hand. Each implementation adds a reconnect test that drives a +**second, fresh dispatcher over the same store**, and each was verified to fail +against its own pre-fix code. diff --git a/docs/Architecture/Rich Interactions.md b/docs/Architecture/Rich Interactions.md index 0c66d6ae..25c46e09 100644 --- a/docs/Architecture/Rich Interactions.md +++ b/docs/Architecture/Rich Interactions.md @@ -103,10 +103,41 @@ Intake = **collect** (who are you?), OTP = **verify** (prove it). Shared machine | Channel | `supports` | identity_intake behavior | | --- | --- | --- | | Chat widget (web) | `["identity_form"]` (registry-derived) | parked inline form card; server-validated; decline button | -| SMS | — | conversational fallback: field-by-field ask, `submit_interaction` tool validates each value | -| Voice | — | same fallback (spoken turn-by-turn) | +| SMS | `[]` | conversational fallback: field-by-field ask, `submit_interaction` tool validates each value | +| Voice | `[]` | same fallback (spoken turn-by-turn) | | Future rich client | declares the kinds its cards cover | rich per declared kind, fallback for the rest | +### Capabilities survive a reconnect + +A reconnect is a **resume**: the client re-opens the socket and re-issues +`create_conversation_session` with the same `conversationId`, which mints a NEW +session id. So the declared list cannot live on the session — it is persisted on +the **conversation** (`metadata.clientSupports`, written by +`handle_create_session`), the same durability the workflow step pointer moved to +for the same reason (th-c12df5: the session registry is per-pod and resets on +reconnect/pod hop). A resume that **omits** `supports` inherits the conversation's +last declared set, so Rich Interactions keep working across a network blip, a +backgrounded mobile app, or a deploy. + +A frame that **does** declare wins, in both directions — which is why a text-only +channel sends an explicit `[]` rather than omitting the key: omitting it on a +resume would inherit the rich set. Both directions are pinned by +`reconnect_resuming_a_conversation_keeps_the_declared_capabilities` +(`rust/smooth-operator-server/tests/interactions.rs`). + +All five implementations hold to this. Before the fix each kept the set somewhere +a reconnect destroys: Rust in `Session.metadata.supports` (the per-pod session +registry), Go / Python / .NET in a per-connection map on the dispatcher (also +never pruned), TypeScript on the store-backed **session** record — which a resume +mints fresh. Every one of them silently degraded each kind to its conversational +fallback after a reconnect: no error, no event, nothing to notice (th-13df6d). + +Rust, Go, Python and TypeScript store it as `clientSupports` on +`conversations.metadata_json`; the .NET store follows its own documented hold for +`currentStepId` / `otpVerified` (session-row metadata) under the same key name, so +a database driven by both Rust and .NET would not share this one value — the same +pre-existing divergence those two keys already have. + ## What changes where | Layer | Change | diff --git a/dotnet/server/integration-tests/SubmitInteractionTests.cs b/dotnet/server/integration-tests/SubmitInteractionTests.cs index 70da95d0..dfee18e4 100644 --- a/dotnet/server/integration-tests/SubmitInteractionTests.cs +++ b/dotnet/server/integration-tests/SubmitInteractionTests.cs @@ -39,16 +39,23 @@ public class SubmitInteractionTests } """; - private static WebApplication BuildApp() + /// scripted send_message turns, each "call request_choices, then + /// answer": a rich turn consumes the tool call when it parks and the text when the submit resumes + /// it; a fallback turn consumes both back-to-back. Either way one turn = one pair, so a test that + /// settles each turn before starting the next stays in step with the queue. + private static WebApplication BuildApp(int turns = 1) { var chat = new MockChatClient(); var args = JsonSerializer.Deserialize(ChoicesArgsJson); - chat.PushToolCall("call-1", "request_choices", new Dictionary + for (var i = 0; i < turns; i++) { - ["questions"] = args.GetProperty("questions"), - ["reason"] = args.GetProperty("reason"), - }); - chat.PushText("Great — routing you to the right team now."); + chat.PushToolCall($"call-{i + 1}", "request_choices", new Dictionary + { + ["questions"] = args.GetProperty("questions"), + ["reason"] = args.GetProperty("reason"), + }); + chat.PushText("Great — routing you to the right team now."); + } var builder = WebApplication.CreateBuilder(); builder.WebHost.UseTestServer(); @@ -92,7 +99,14 @@ private static async Task NextEventAsync(WebSocket socket) } } - private static async Task CreateSessionAsync(WebSocket socket, string[]? supports = null) + private static async Task CreateSessionAsync(WebSocket socket, string[]? supports = null) => + (await CreateSessionDataAsync(socket, supports: supports))["sessionId"]!.GetValue(); + + /// The create_conversation_session response's data (sessionId + conversationId). + /// A null OMITS the key entirely; an empty array declares it explicitly + /// — the two are different frames and, on a resume, mean different things (inherit vs. opt out). + /// A non-null makes this a resume, i.e. a reconnect. + private static async Task CreateSessionDataAsync(WebSocket socket, string? conversationId = null, string[]? supports = null) { var frame = new JsonObject { @@ -102,6 +116,10 @@ private static async Task CreateSessionAsync(WebSocket socket, string[]? ["userName"] = "Alice", ["userEmail"] = "alice@example.com", }; + if (conversationId is not null) + { + frame["conversationId"] = conversationId; + } if (supports is not null) { frame["supports"] = new JsonArray(supports.Select(s => (JsonNode)s).ToArray()); @@ -112,7 +130,7 @@ private static async Task CreateSessionAsync(WebSocket socket, string[]? var ev = await NextEventAsync(socket); if (ev["type"]!.GetValue() == "immediate_response") { - return ev["data"]!["sessionId"]!.GetValue(); + return ev["data"]!.AsObject(); } } } @@ -444,4 +462,114 @@ public async Task SubmitWithoutPending_IsACleanError() await socket.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", CancellationToken.None); await app.StopAsync(); } + + /// Send one message and settle the raise: the interaction_required event when the + /// kind took the RICH (card) path, or null when the turn ran straight to + /// eventual_response on the conversational fallback. Returns the tool results seen either + /// way, so a fallback can be asserted on its directive rather than merely on the absence of a card. + private static async Task<(JsonObject? Card, List ToolResults)> RunRaiseTurnAsync(WebSocket socket, string sessionId, string requestId) + { + await SendAsync(socket, new JsonObject + { + ["action"] = "send_message", + ["requestId"] = requestId, + ["sessionId"] = sessionId, + ["message"] = "help me choose", + }); + + var toolResults = new List(); + while (true) + { + var ev = await NextEventAsync(socket); + var type = ev["type"]!.GetValue(); + if (type == "stream_chunk" && ev["data"]?["state"]?["rawResponse"]?["toolResult"]?.AsObject() is { } tr) + { + toolResults.Add(tr); + } + else if (type == "interaction_required") + { + return (ev, toolResults); + } + else if (type == "eventual_response") + { + return (null, toolResults); + } + } + } + + /// + /// A RECONNECT is a resume: the client re-opens the socket and re-issues + /// create_conversation_session with the same conversationId, which mints a NEW session + /// on a NEW . While the declared capabilities lived in that + /// dispatcher's per-connection map, every routine reconnect (network blip, mobile backgrounding, + /// deploy) silently turned the whole Rich Interactions framework off unless the client re-declared + /// supports every single time — no error, no event, nothing on the wire to notice. Each + /// connection below is a genuinely fresh dispatcher over the SAME singleton store. th-13df6d. + /// + [Fact] + public async Task Reconnect_ResumingAConversation_KeepsTheDeclaredCapabilities() + { + await using var app = BuildApp(turns: 3); + await app.StartAsync(); + var server = app.GetTestServer(); + + // 1. First connection declares the capability, then drops (the blip). + string conversationId; + using (var first = await ConnectAsync(server)) + { + var created = await CreateSessionDataAsync(first, supports: new[] { "choice_chips" }); + conversationId = created["conversationId"]!.GetValue(); + await first.CloseAsync(WebSocketCloseStatus.NormalClosure, "blip", CancellationToken.None); + } + + // 2. The reconnect: same conversation, `supports` OMITTED — exactly what a widget resuming from + // its stored conversationId sends. This is where the feature used to go dark. + using (var reconnect = await ConnectAsync(server)) + { + var data = await CreateSessionDataAsync(reconnect, conversationId: conversationId); + Assert.Equal(conversationId, data["conversationId"]!.GetValue()); + var sessionId = data["sessionId"]!.GetValue(); + + var (card, _) = await RunRaiseTurnAsync(reconnect, sessionId, "r-reconnect"); + Assert.True(card is not null, "a reconnect that omits 'supports' must inherit the conversation's declared capabilities"); + Assert.Equal("choices", card!["data"]!["data"]!["kind"]!.GetValue()); + + // Unpark so this turn is fully settled before the next one starts. + await SendAsync(reconnect, new JsonObject + { + ["action"] = "submit_interaction", + ["requestId"] = "r-reconnect", + ["sessionId"] = sessionId, + ["interactionId"] = card["data"]!["data"]!["interactionId"]!.GetValue(), + ["values"] = JsonNode.Parse("""{ "answers": [ { "header": "Plan", "options": ["Pro"] }, { "header": "Topics", "options": ["Sales"] } ] }"""), + }); + await ReadUntilAsync(reconnect, "eventual_response"); + await reconnect.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", CancellationToken.None); + } + + // 3. A resume that DECLARES wins, including an explicit `[]` — how a text-only channel (SMS, + // voice) resuming a rich conversation opts out instead of being handed cards it can't render. + using (var textOnly = await ConnectAsync(server)) + { + var data = await CreateSessionDataAsync(textOnly, conversationId: conversationId, supports: Array.Empty()); + var (card, toolResults) = await RunRaiseTurnAsync(textOnly, data["sessionId"]!.GetValue(), "r-text-only"); + Assert.True(card is null, "an explicit empty 'supports' declares text-only and must never inherit"); + Assert.Contains(toolResults, tr => + tr["name"]!.GetValue() == "request_choices" + && tr["result"]!.GetValue().Contains("cannot display choice chips", StringComparison.Ordinal)); + await textOnly.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", CancellationToken.None); + } + + // 4. ...and that opt-out is itself durable: the next reconnect that omits the key must inherit + // the OPT-OUT, not resurrect the capability from a stale record. + using (var after = await ConnectAsync(server)) + { + var data = await CreateSessionDataAsync(after, conversationId: conversationId); + var (card, _) = await RunRaiseTurnAsync(after, data["sessionId"]!.GetValue(), "r-after-opt-out"); + Assert.True(card is null, "the text-only declaration replaced the stored capabilities"); + await after.CloseAsync(WebSocketCloseStatus.NormalClosure, "done", CancellationToken.None); + } + + await app.StopAsync(); + } } diff --git a/dotnet/server/postgres/src/PostgresSessionStore.cs b/dotnet/server/postgres/src/PostgresSessionStore.cs index 17f2999e..69917f76 100644 --- a/dotnet/server/postgres/src/PostgresSessionStore.cs +++ b/dotnet/server/postgres/src/PostgresSessionStore.cs @@ -16,9 +16,20 @@ namespace SmooAI.SmoothOperator.Server.Postgres; /// The interface stays CONVERSATION-keyed (GetWorkflowStepAsync(conversationId) and friends) /// while Rust/Go key the same metadata by session. That is a deliberate hold: this host's persisted /// workflow step and OTP bit survive a resume today, and flipping to session-keyed would silently -/// require re-verification after every resume — a product decision, not a schema one. The DATA lives -/// in the shared place under the shared keys either way; a conversation-keyed write touches every -/// session row of the conversation and a read takes the most recent. +/// require re-verification after every resume — a product decision, not a schema one. A +/// conversation-keyed write touches every session row of the conversation and a read takes the most +/// recent. +/// +/// KNOWN DIVERGENCE — the data does NOT live in the same place as the other ports, only under the +/// same key names. This store writes conversation_sessions.metadata; the Rust/Go/Python/ +/// TypeScript stores write conversations.metadata_json (Rust: +/// rust/adapters/postgres/src/lib.rs:618). So one database driven by BOTH a .NET server and +/// one of the others would not share currentStepId, otpVerified, or +/// clientSupports — each would read its own table and see the other's writes as absent. +/// Nothing enforces this at build time and no test pins it: you deploy ONE server implementation, +/// so the mixed-driver case is theoretical. Do not "fix" it by moving one key — that would split +/// this store's three keys across two tables, which is strictly worse. Unifying all three at once +/// is tracked as its own piece of work (th-13df6d follow-up). /// public sealed class PostgresSessionStore : ISessionStore, IAsyncDisposable { @@ -335,6 +346,28 @@ ORDER BY sub.seq ASC public Task SetWorkflowStepAsync(string conversationId, string stepId, CancellationToken cancellationToken = default) => MergeSessionMetadataAsync(conversationId, SessionMetadata.CurrentStepIdKey, stepId, cancellationToken); + public async Task> GetClientSupportsAsync(string conversationId, CancellationToken cancellationToken = default) + { + // `->>` renders the stored JSON array as its text (`["choice_chips"]`), so parse it back. + // Missing / unparseable ⇒ empty, i.e. the text-only behavior every kind already falls back to. + var value = await ReadSessionMetadataAsync(conversationId, SessionMetadata.ClientSupportsKey, cancellationToken).ConfigureAwait(false); + if (string.IsNullOrEmpty(value)) + { + return Array.Empty(); + } + try + { + return JsonSerializer.Deserialize(value) ?? Array.Empty(); + } + catch (JsonException) + { + return Array.Empty(); + } + } + + public Task SetClientSupportsAsync(string conversationId, IReadOnlyList supports, CancellationToken cancellationToken = default) => + MergeSessionMetadataAsync(conversationId, SessionMetadata.ClientSupportsKey, supports.ToArray(), cancellationToken); + public async Task GetSessionAuthenticatedAsync(string conversationId, CancellationToken cancellationToken = default) { var value = await ReadSessionMetadataAsync(conversationId, SessionMetadata.OtpVerifiedKey, cancellationToken).ConfigureAwait(false); @@ -402,6 +435,11 @@ internal sealed class SessionMetadata internal const string OtpVerifiedKey = "otpVerified"; internal const string CurrentStepIdKey = "currentStepId"; + /// The conversation's last-declared render capabilities (a JSON array of strings). Named + /// to match the Rust reference's conversation-metadata key so a database driven by either server + /// reads the same record. th-13df6d. + internal const string ClientSupportsKey = "clientSupports"; + [System.Text.Json.Serialization.JsonPropertyName(ContactEmailKey)] [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull)] public string? ContactEmail { get; set; } diff --git a/dotnet/server/postgres/tests/SessionStoreContractTests.cs b/dotnet/server/postgres/tests/SessionStoreContractTests.cs index b3ea39b1..4a838af5 100644 --- a/dotnet/server/postgres/tests/SessionStoreContractTests.cs +++ b/dotnet/server/postgres/tests/SessionStoreContractTests.cs @@ -99,6 +99,29 @@ public async Task WorkflowStep_DefaultsNull_ThenUpsertsAndScopesByConversation() Assert.Null(await store.GetWorkflowStepAsync(b.ConversationId)); } + [SkippableFact] + public async Task ClientSupports_DefaultsEmpty_ThenUpsertsAndScopesByConversation() + { + var store = await CreateStoreAsync(); + var a = await store.CreateSessionAsync("", null, null); + var b = await store.CreateSessionAsync("", null, null); + + // Fresh conversation → nothing declared, i.e. text-only. + Assert.Empty(await store.GetClientSupportsAsync(a.ConversationId)); + + await store.SetClientSupportsAsync(a.ConversationId, new[] { "choice_chips", "identity_form" }); + Assert.Equal(new[] { "choice_chips", "identity_form" }, await store.GetClientSupportsAsync(a.ConversationId)); + + // Upsert REPLACES, including back to empty — the text-only opt-out has to be durable or the + // next reconnect that omits `supports` resurrects capabilities the client just gave up. + await store.SetClientSupportsAsync(a.ConversationId, Array.Empty()); + Assert.Empty(await store.GetClientSupportsAsync(a.ConversationId)); + + // Scoped per conversation. + await store.SetClientSupportsAsync(a.ConversationId, new[] { "choice_chips" }); + Assert.Empty(await store.GetClientSupportsAsync(b.ConversationId)); + } + [SkippableFact] public async Task SessionAuthenticated_DefaultsFalse_ThenUpsertsAndScopesByConversation() { diff --git a/dotnet/server/src/FrameDispatcher.cs b/dotnet/server/src/FrameDispatcher.cs index 24a27e1d..b16b73af 100644 --- a/dotnet/server/src/FrameDispatcher.cs +++ b/dotnet/server/src/FrameDispatcher.cs @@ -1,4 +1,3 @@ -using System.Collections.Concurrent; using System.Text.Json.Nodes; using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging; @@ -45,13 +44,14 @@ public sealed class FrameDispatcher private readonly InteractionCatalog? _interactions; private readonly InteractionParkRegistry _interactionPark = new(); // The in-memory, session-keyed contact overlay stamped by an interaction kind's host effect - // (identity_intake) and read by the OTP contact seam. Per-connection, like the park + supports maps. + // (identity_intake) and read by the OTP contact seam. Per-connection, like the park registry. private readonly SessionIdentityRegistry _sessionIdentity = new(); - // Per-connection render capabilities declared at create_conversation_session (the `supports` array), - // keyed by sessionId. The rich-vs-fallback interaction branch reads this. Captured per connection - // (like the confirmation registry) rather than persisted on the session — a client re-declares its - // capabilities on every connect, and create always precedes send on a connection. - private readonly ConcurrentDictionary> _sessionSupports = new(); + // NOTE: the render capabilities declared at create_conversation_session (`supports`) used to live + // here in a per-connection, session-keyed map. They do NOT any more — a reconnect is a resume onto + // a NEW dispatcher whose map was empty, so every routine network blip silently turned Rich + // Interactions off until the client re-declared. They now ride the conversation-scoped session + // store (ISessionStore.Get/SetClientSupportsAsync), which also retired the map's unbounded growth + // (nothing ever pruned it). th-13df6d. private readonly IAgentConfigResolver? _agentConfigResolver; private readonly IWorkflowJudge? _judge; private readonly ISessionAuthenticator _authenticator; @@ -389,11 +389,22 @@ private async Task HandleCreateSessionAsync(JsonObject frame, string? requestId, string.IsNullOrEmpty(conversationId) ? null : conversationId, cancellationToken).ConfigureAwait(false); - // Capture the client's render capabilities (`supports`) for this session so a mid-turn Rich - // Interaction takes the rich (card) path only on a channel that can render it; an omitted / - // empty `supports` (text-only channels: SMS, voice) leaves the set empty → every kind degrades - // to its conversational fallback. Unknown values are kept but simply never match a kind. - _sessionSupports[session.SessionId] = ParseSupports(frame["supports"]); + // Record the client's render capabilities (`supports`) on the CONVERSATION so a mid-turn Rich + // Interaction takes the rich (card) path only on a channel that can render it; no capability + // (text-only channels: SMS, voice) → every kind degrades to its conversational fallback. + // Unknown values are kept but simply never match a kind. + // + // A frame that DECLARES the key always wins and replaces the stored set — including an explicit + // `[]`, which is how a text-only channel resuming a rich conversation opts out (and the opt-out + // has to be durable, or the next reconnect would resurrect the old set). A frame that OMITS the + // key is not a declaration: on a resume it INHERITS what the conversation last declared, which + // is what keeps Rich Interactions alive across a reconnect (see ISessionStore + // .GetClientSupportsAsync); on a fresh conversation there is nothing to inherit → empty, + // unchanged. th-13df6d. + if (ParseSupports(frame["supports"]) is { } declared) + { + await _store.SetClientSupportsAsync(session.ConversationId, declared, cancellationToken).ConfigureAwait(false); + } // A freshly created session never passes through ScopedSessionAsync, so associate here too. AssociateSession(session); @@ -808,8 +819,11 @@ private async Task HandleSendMessageAsync(JsonObject frame, string? requestId, A // above, and reused to back the built-in knowledge_search tool) — so a user only ever sees // documents their groups grant (ACL enforced on the chat path). // Rich Interactions: hand the turn the hosted-kind catalog, the connection's park registry, and - // THIS session's declared render capabilities (empty for a text-only session → fallback path). - var capabilities = _sessionSupports.TryGetValue(session.SessionId, out var supports) ? supports : null; + // THIS conversation's declared render capabilities (empty for a text-only client → fallback + // path). Read from the store, not from this connection: the create frame that declared them may + // have arrived on a PREVIOUS connection (a reconnect resumes the conversation on a fresh + // dispatcher), and a per-connection map would read empty there. th-13df6d. + var capabilities = await _store.GetClientSupportsAsync(session.ConversationId, cancellationToken).ConfigureAwait(false); var runner = new TurnRunner(_chatClient, _store, scopedKnowledge, _systemPrompt, _reranker, gatedTools, confirmTools, _confirmations, agentConfig, _judge, _limits, _logger, toolHooks: _toolHooks, interactions: _interactions, interactionPark: _interactionPark, capabilities: capabilities, interactionEffects: _sessionIdentity) { ConfirmationTimeout = ConfirmationTimeout, @@ -1155,22 +1169,31 @@ private async Task HandleSubmitInteractionAsync(JsonObject frame, string? reques })); } - /// Parse the create-session supports field (a string array of render capabilities) - /// into a set. Absent/non-array ⇒ empty (text-only ⇒ every interaction kind uses its fallback). - private static HashSet ParseSupports(JsonNode? node) + /// Parse the create-session supports field (a string array of render capabilities). + /// + /// Returns null when the key is ABSENT (or not an array) and an empty list for an explicit + /// [] — that distinction is load-bearing, not pedantry: an explicit [] is a text-only + /// client DECLARING it renders nothing and must replace whatever the conversation had, while an + /// omitted key is no declaration at all and inherits on a resume. Collapsing the two (as this did + /// while the set lived on the connection) is what makes a reconnect either lose the capabilities or + /// resurrect ones the client just opted out of. th-13df6d. + /// + private static IReadOnlyList? ParseSupports(JsonNode? node) { - var set = new HashSet(StringComparer.Ordinal); - if (node is JsonArray array) + if (node is not JsonArray array) { - foreach (var entry in array) + return null; + } + var capabilities = new List(); + foreach (var entry in array) + { + if (entry is JsonValue value && value.TryGetValue(out var capability) && !string.IsNullOrEmpty(capability) + && !capabilities.Contains(capability, StringComparer.Ordinal)) { - if (entry is JsonValue value && value.TryGetValue(out var capability) && !string.IsNullOrEmpty(capability)) - { - set.Add(capability); - } + capabilities.Add(capability); } } - return set; + return capabilities; } /// diff --git a/dotnet/server/src/Interactions.cs b/dotnet/server/src/Interactions.cs index 9924030d..ea1b80ce 100644 --- a/dotnet/server/src/Interactions.cs +++ b/dotnet/server/src/Interactions.cs @@ -154,7 +154,10 @@ public SessionIdentity Merge(string? name, string? email, string? phone) => /// The in-memory, session-keyed contact overlay stamped by identity_intake's host effect and read /// by the OTP contact seam — the C# analog of the Rust reference's AppState session-metadata map /// (an in-process overlay, NOT the durable ). Per-connection lifetime, like -/// InteractionParkRegistry and the declared-capabilities map: a client re-declares on reconnect. +/// InteractionParkRegistry — an intake is captured and consumed within one connection's turn. +/// (The declared render capabilities used to share this lifetime and no longer do: they outlive the +/// connection on the store, because a reconnect resumes the conversation on a fresh dispatcher and the +/// client does NOT re-declare. th-13df6d.) /// public sealed class SessionIdentityRegistry { diff --git a/dotnet/server/src/SessionStore.cs b/dotnet/server/src/SessionStore.cs index 07a45944..a36afe0b 100644 --- a/dotnet/server/src/SessionStore.cs +++ b/dotnet/server/src/SessionStore.cs @@ -149,6 +149,27 @@ public interface ISessionStore /// advances the pointer at the end of a turn. Task SetWorkflowStepAsync(string conversationId, string stepId, CancellationToken cancellationToken = default); + /// The client render capabilities (supports) this conversation last declared — + /// the gate on the entire Rich Interactions framework (a kind whose capability is listed parks the + /// turn and emits a card; one that is missing degrades to the conversational fallback). Empty for a + /// conversation that never declared any, which is exactly the text-only behavior. + /// + /// CONVERSATION-scoped, not session-scoped, on purpose: a reconnect IS a resume — the client + /// re-opens the socket and re-issues create_conversation_session with the same + /// conversationId, which mints a NEW session on a NEW dispatcher. Capabilities that lived on + /// the connection were therefore lost on every network blip / backgrounding / deploy unless the + /// client re-declared them, and Rich Interactions silently stopped being offered — no error, no + /// event, nothing on the wire to notice. th-13df6d. + /// + Task> GetClientSupportsAsync(string conversationId, CancellationToken cancellationToken = default); + + /// Record the capabilities a create_conversation_session frame DECLARED for this + /// conversation (upsert). A declaration always REPLACES what was stored, including an explicit + /// empty list — that is how a text-only channel resuming a rich conversation opts out, and the + /// opt-out has to be durable too or the next reconnect resurrects the old set. A frame that OMITS + /// the key is not a declaration and must not call this. + Task SetClientSupportsAsync(string conversationId, IReadOnlyList supports, CancellationToken cancellationToken = default); + /// Whether this conversation's caller is identity-verified (the persisted /// otpVerified bit — the C# analog of the Rust session's metadata.otpVerified). /// false for a fresh or unknown conversation. Threaded into the end_user auth gate @@ -185,6 +206,10 @@ public sealed class InMemorySessionStore : ISessionStore private readonly Dictionary _sessions = new(); private readonly Dictionary> _messages = new(); private readonly Dictionary _workflowSteps = new(); + + // Each conversation's last-declared render capabilities (`supports`). Conversation-keyed like the + // workflow step pointer, so a reconnect that omits the key still gets its cards. th-13df6d. + private readonly Dictionary> _clientSupports = new(); private readonly HashSet _authenticated = new(); // Each conversation's last activity (creation, then every append) — the sort key + updatedAt @@ -328,6 +353,24 @@ public Task SetWorkflowStepAsync(string conversationId, string stepId, Cancellat return Task.CompletedTask; } + public Task> GetClientSupportsAsync(string conversationId, CancellationToken cancellationToken = default) + { + lock (_gate) + { + return Task.FromResult(_clientSupports.TryGetValue(conversationId, out var supports) ? supports : Array.Empty()); + } + } + + public Task SetClientSupportsAsync(string conversationId, IReadOnlyList supports, CancellationToken cancellationToken = default) + { + lock (_gate) + { + // Copy: the caller's list must not keep mutating the stored record. + _clientSupports[conversationId] = supports.ToArray(); + } + return Task.CompletedTask; + } + public Task GetSessionAuthenticatedAsync(string conversationId, CancellationToken cancellationToken = default) { lock (_gate) diff --git a/dotnet/server/tests/GetConversationMessagesTests.cs b/dotnet/server/tests/GetConversationMessagesTests.cs index 6cce7544..aac8e40e 100644 --- a/dotnet/server/tests/GetConversationMessagesTests.cs +++ b/dotnet/server/tests/GetConversationMessagesTests.cs @@ -257,6 +257,12 @@ public async Task> ListMessagesAsync(string convers public Task SetWorkflowStepAsync(string conversationId, string stepId, CancellationToken cancellationToken = default) => inner.SetWorkflowStepAsync(conversationId, stepId, cancellationToken); + public Task> GetClientSupportsAsync(string conversationId, CancellationToken cancellationToken = default) => + inner.GetClientSupportsAsync(conversationId, cancellationToken); + + public Task SetClientSupportsAsync(string conversationId, IReadOnlyList supports, CancellationToken cancellationToken = default) => + inner.SetClientSupportsAsync(conversationId, supports, cancellationToken); + public Task GetSessionAuthenticatedAsync(string conversationId, CancellationToken cancellationToken = default) => inner.GetSessionAuthenticatedAsync(conversationId, cancellationToken); diff --git a/dotnet/src/Generated/Types.cs b/dotnet/src/Generated/Types.cs index 168fb995..07c6d336 100644 --- a/dotnet/src/Generated/Types.cs +++ b/dotnet/src/Generated/Types.cs @@ -180,7 +180,7 @@ public partial class CreateConversationSessionRequest public string? BrowserFingerprint { get; set; } = default!; /// - /// Client render capabilities for this session — a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` → capability `identity_form`); future kinds add their own values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) omit this and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). + /// Client render capabilities for this session — a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` → capability `identity_form`, kind `choices` → capability `choice_chips`); future kinds add their own values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) declare `[]` and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). Durability: the declared list is persisted on the CONVERSATION, so a reconnect that resumes an existing `conversationId` and OMITS this key inherits the set the conversation last declared — a reconnect is not a downgrade to text-only. Any list the frame does declare (including `[]`) replaces the inherited one, so a text-only client resuming a rich conversation opts out explicitly. /// [System.Text.Json.Serialization.JsonPropertyName("supports")] public System.Collections.Generic.ICollection? Supports { get; set; } = default!; @@ -222,7 +222,7 @@ public partial class CreateConversationSessionResponse /// ID of the agent handling this session. /// [System.Text.Json.Serialization.JsonPropertyName("agentId")] - public System.Guid AgentId { get; set; } = default!; + public System.Guid? AgentId { get; set; } = default!; /// /// Display name of the agent. @@ -1154,10 +1154,10 @@ public partial class Session public System.Guid OrganizationId { get; set; } = default!; /// - /// The agent handling this session. + /// The agent handling this session. OPTIONAL in storage: create_conversation_session REJECTS an absent or blank agentId, so a session created through the protocol always has one. It stays optional here for rows that predate that validation — it used to be filled with a fresh UUID, pointing every agentless session at an agent that had never existed (th-68897a). Absence is represented by omitting the field, never by a fabricated id. /// [System.Text.Json.Serialization.JsonPropertyName("agentId")] - public System.Guid AgentId { get; set; } = default!; + public System.Guid? AgentId { get; set; } = default!; /// /// Human-readable display name of the agent. diff --git a/go/protocol/types_gen.go b/go/protocol/types_gen.go index b28a0de7..b9900ac7 100644 --- a/go/protocol/types_gen.go +++ b/go/protocol/types_gen.go @@ -118,6 +118,35 @@ type Checkpoint struct { // name, bead ID). type CheckpointMetadata map[string]string +// The canonical validated payload the parked turn resumes with (identical on the +// chip and conversational paths). +type ChoicesPayload struct { + // Guidance for the agent when `status` is `declined` / `no_response`. + Message *string `json:"message,omitempty,omitzero"` + + // How the interaction resolved. + Status PayloadStatus `json:"status"` + + // Present when `status` is `submitted`: the validated, normalized answers. + Values *PayloadValues `json:"values,omitempty,omitzero"` +} + +// The `spec` carried on `interaction_required` for kind `choices`: the questions +// to ask. +type ChoicesSpec struct { + // The questions to ask, in display order (1–4). + Questions []SpecQuestionsElem `json:"questions"` +} + +// The `values` a client submits via `submit_interaction` for kind `choices`. +// Validated server-side: every question answered, each selected label is one of +// that question's options, single-select takes exactly one pick. The free-text +// `other` is always accepted (the AskUserQuestion 'Other' escape hatch). +type ChoicesValues struct { + // One entry per question, keyed by the question's `header`. + Answers []ValuesAnswersElem `json:"answers"` +} + // A source the agent used to ground its answer. Each citation points back at one // retrieved knowledge-base document — the chunk the model read, plus enough // metadata to render an attribution link. Citations are collected by the runtime @@ -330,10 +359,16 @@ type CreateConversationSessionRequest struct { // Client render capabilities for this session — a per-kind list gating the Rich // Interactions the server may emit mid-turn (`interaction_required`). Each // interaction kind declares the capability that gates it (e.g. kind - // `identity_intake` → capability `identity_form`); future kinds add their own - // values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) omit - // this and the server degrades each kind to its conversational fallback. Unknown - // values are ignored (forward-compatible). + // `identity_intake` → capability `identity_form`, kind `choices` → capability + // `choice_chips`); future kinds add their own values (`date_picker`, + // `file_upload`, …). Text-only channels (SMS, voice) declare `[]` and the server + // degrades each kind to its conversational fallback. Unknown values are ignored + // (forward-compatible). Durability: the declared list is persisted on the + // CONVERSATION, so a reconnect that resumes an existing `conversationId` and + // OMITS this key inherits the set the conversation last declared — a reconnect is + // not a downgrade to text-only. Any list the frame does declare (including `[]`) + // replaces the inherited one, so a text-only client resuming a rich conversation + // opts out explicitly. Supports []string `json:"supports,omitempty,omitzero"` // Optional email address for the user participant. @@ -347,7 +382,7 @@ type CreateConversationSessionRequest struct { // creation. type CreateConversationSessionResponse struct { // ID of the agent handling this session. - AgentID string `json:"agentId"` + AgentID *string `json:"agentId,omitempty,omitzero"` // Display name of the agent. AgentName string `json:"agentName"` @@ -743,7 +778,7 @@ type IdentityIntakePayload struct { Status PayloadStatus `json:"status"` // Present when `status` is `submitted`: the validated, normalized values. - Values *PayloadValues `json:"values,omitempty,omitzero"` + Values *PayloadValues_1 `json:"values,omitempty,omitzero"` } // The `spec` carried on `interaction_required` for kind `identity_intake`: which @@ -1332,8 +1367,25 @@ const PayloadStatusDeclined PayloadStatus = "declined" const PayloadStatusNoResponse PayloadStatus = "no_response" const PayloadStatusSubmitted PayloadStatus = "submitted" -// Present when `status` is `submitted`: the validated, normalized values. +// Present when `status` is `submitted`: the validated, normalized answers. type PayloadValues struct { + // Answers corresponds to the JSON schema field "answers". + Answers []PayloadValuesAnswersElem `json:"answers"` +} + +type PayloadValuesAnswersElem struct { + // Header corresponds to the JSON schema field "header". + Header string `json:"header"` + + // Options corresponds to the JSON schema field "options". + Options []string `json:"options,omitempty,omitzero"` + + // Other corresponds to the JSON schema field "other". + Other *string `json:"other,omitempty,omitzero"` +} + +// Present when `status` is `submitted`: the validated, normalized values. +type PayloadValues_1 struct { // Email corresponds to the JSON schema field "email". Email *string `json:"email,omitempty,omitzero"` @@ -1548,8 +1600,13 @@ type SendMessageResponse struct { // smooth-operator thread identifier (stored as `langgraph_thread_id` in the DB for // historical reasons; renamed to `threadId` in the protocol). type Session struct { - // The agent handling this session. - AgentID string `json:"agentId"` + // The agent handling this session. OPTIONAL in storage: + // create_conversation_session REJECTS an absent or blank agentId, so a session + // created through the protocol always has one. It stays optional here for rows + // that predate that validation — it used to be filled with a fresh UUID, pointing + // every agentless session at an agent that had never existed (th-68897a). Absence + // is represented by omitting the field, never by a fabricated id. + AgentID *string `json:"agentId,omitempty,omitzero"` // Human-readable display name of the agent. AgentName string `json:"agentName"` @@ -1632,6 +1689,30 @@ const SpecFieldsElemKeyEmail SpecFieldsElemKey = "email" const SpecFieldsElemKeyName SpecFieldsElemKey = "name" const SpecFieldsElemKeyPhone SpecFieldsElemKey = "phone" +type SpecQuestionsElem struct { + // A short label (≤12 chars), unique within the raise. The answer key and the + // chip/tab caption. + Header string `json:"header"` + + // Whether the visitor may select more than one option (default false). + MultiSelect bool `json:"multiSelect,omitempty,omitzero"` + + // The enumerated options. A free-text `other` answer is always available in + // addition to these. + Options []SpecQuestionsElemOptionsElem `json:"options"` + + // The question prompt shown to the visitor. + Question string `json:"question"` +} + +type SpecQuestionsElemOptionsElem struct { + // A short human-readable gloss for the option. + Description *string `json:"description,omitempty,omitzero"` + + // The option label — the value the visitor submits. + Label string `json:"label"` +} + // Event: `stream_chunk`. Emitted each time a node in the smooth-operator workflow // completes. Carries the node name and a filtered state snapshot. Clients use this // to show per-node progress (e.g. `knowledge_search completed`, tool activity) in @@ -1841,6 +1922,19 @@ type SubmitInteractionRequest struct { // stays parked). This schema is provided for documentation completeness only. type SubmitInteractionResponse map[string]interface{} +type ValuesAnswersElem struct { + // Which question this answers — matches the spec question's `header`. + Header string `json:"header"` + + // The selected option label(s). One for single-select; empty when the visitor + // only used `other`. + Options []string `json:"options,omitempty,omitzero"` + + // A free-text answer outside the enumerated options (the 'Other' escape hatch). + // Blank ⇒ omitted. + Other *string `json:"other,omitempty,omitzero"` +} + // Fields sent by the client to submit an OTP code. type VerifyOTPRequest struct { // Action discriminator. diff --git a/go/server/conversations_test.go b/go/server/conversations_test.go index 12f6092a..c562996d 100644 --- a/go/server/conversations_test.go +++ b/go/server/conversations_test.go @@ -249,3 +249,73 @@ func TestCreateSessionRejectsBlankAgentID(t *testing.T) { }) } } + +// A reconnect IS a resume: the client re-opens the socket and re-issues +// create_conversation_session with the same conversationId, which mints a NEW session id +// on a NEW dispatcher. `supports` — the render-capability list gating the whole Rich +// Interactions framework — used to live in a per-connection map, so unless the client +// re-declared it on every reconnect the server forgot it could render cards and every +// kind silently degraded to its conversational fallback: no error, no event, nothing on +// the wire to notice. It now rides the CONVERSATION, so an omitting reconnect inherits +// it. th-13df6d. +func TestReconnectKeepsDeclaredCapabilities(t *testing.T) { + ctx := context.Background() + store := NewInMemorySessionStore() + + // create_conversation_session on a FRESH dispatcher each time — that is what a + // reconnect is — returning what the next turn on that session would gate on. + reconnect := func(t *testing.T, frame map[string]any) map[string]bool { + t.Helper() + sink, events := capture() + dispatchJSON(t, bareDispatcher(store), frame, sink) + if len(*events) != 1 || (*events)[0]["type"] != "immediate_response" { + t.Fatalf("create_conversation_session: want immediate_response, got %+v", *events) + } + data := (*events)[0]["data"].(map[string]any) + if frame["conversationId"] != nil && data["conversationId"] != frame["conversationId"] { + t.Fatalf("reconnect resumed conversation %v, want %v", data["conversationId"], frame["conversationId"]) + } + session, err := store.GetSession(ctx, data["sessionId"].(string)) + if err != nil || session == nil { + t.Fatalf("GetSession(%v): %v", data["sessionId"], err) + } + // The exact expression handleSendMessage hands the turn runner. + return capabilitySet(session.Supports) + } + + // First connection: declares the capability. + first := map[string]any{ + "action": "create_conversation_session", "requestId": "r-conn-1", + "agentId": "agent", "supports": []string{"identity_form"}, + } + sink, events := capture() + dispatchJSON(t, bareDispatcher(store), first, sink) + convID := (*events)[0]["data"].(map[string]any)["conversationId"].(string) + + // Reconnect with `supports` OMITTED — exactly what a widget resuming from its + // stored conversationId sends. This is where the feature used to go dark. + if caps := reconnect(t, map[string]any{ + "action": "create_conversation_session", "requestId": "r-conn-2", + "agentId": "agent", "conversationId": convID, + }); !caps["identity_form"] { + t.Fatalf("a reconnect omitting 'supports' lost the conversation's capabilities: %v", caps) + } + + // A resume that DECLARES wins — including an explicit `[]`, which is how a + // text-only channel resuming a rich conversation opts out of cards it can't render. + if caps := reconnect(t, map[string]any{ + "action": "create_conversation_session", "requestId": "r-conn-3", + "agentId": "agent", "conversationId": convID, "supports": []string{}, + }); len(caps) != 0 { + t.Fatalf("an explicit empty 'supports' must declare text-only, got %v", caps) + } + + // …and that opt-out is itself durable: the next omitting reconnect must not + // resurrect the capability from a stale record. + if caps := reconnect(t, map[string]any{ + "action": "create_conversation_session", "requestId": "r-conn-4", + "agentId": "agent", "conversationId": convID, + }); len(caps) != 0 { + t.Fatalf("the text-only declaration did not replace the durable record, got %v", caps) + } +} diff --git a/go/server/dispatcher.go b/go/server/dispatcher.go index e9695f15..2097c8ae 100644 --- a/go/server/dispatcher.go +++ b/go/server/dispatcher.go @@ -72,13 +72,11 @@ type FrameDispatcher struct { // submit_interaction frame resolves (the Rich Interactions analog of confirmations). // Shared with the turn runner's raise tools. Created on demand in the constructor. interactions *InteractionRegistry - // supports maps sessionId → the render capabilities its client declared at - // create_conversation_session (`supports`). A kind's rich card path is offered only - // when its capability is present; otherwise the turn degrades to the conversational - // fallback. Per-connection (like confirmations): the create + send + submit frames - // for a session all ride the same connection. th-choices. - supports map[string]map[string]bool - supportsMu sync.Mutex + // The client's declared render capabilities (`supports`) are NOT held here. They used + // to be, in a per-connection sessionId→caps map, which lost them on every reconnect + // (a reconnect resumes the conversation on a NEW session id) and never pruned. They + // now live on the conversation in the store and arrive on StoredSession.Supports — + // see capabilitySet. th-13df6d. // turns tracks in-flight spawned send_message turns so the connection loop can wait // for them to finish (and flush their eventual_response) on teardown — the @@ -127,7 +125,6 @@ func NewFrameDispatcher(store SessionStore, client core.ChatClient, access Acces confirmations: confirmations, interactionKinds: DefaultInteractionKinds(), interactions: NewInteractionRegistry(), - supports: map[string]map[string]bool{}, agentConfigs: agentConfigs, judgeModel: judgeModel, authRequiringTools: authRequiringTools, @@ -193,11 +190,17 @@ type inboundFrame struct { AgentID string `json:"agentId"` UserName string `json:"userName"` UserEmail string `json:"userEmail"` - // create_conversation_session — the client's render capabilities for this session + // create_conversation_session — the client's render capabilities for this conversation // (per spec/actions/create-conversation-session.schema.json). A per-kind list gating - // the Rich Interactions the server may emit mid-turn; absent ⇒ text-only, every kind - // degrades to its conversational fallback. Unknown values are ignored (forward-compat). - Supports []string `json:"supports"` + // the Rich Interactions the server may emit mid-turn; a kind whose capability is + // missing degrades to its conversational fallback. Unknown values are kept + // (forward-compat: a future kind may gate on them). + // + // A POINTER because omitted and `[]` mean different things and json.Unmarshal + // collapses both to a nil []string: nil here ⇒ the key was absent, which INHERITS + // the resumed conversation's stored set (a reconnect re-declares nothing); a + // non-nil empty slice ⇒ an explicit "I render nothing", which replaces it. th-13df6d. + Supports *[]string `json:"supports"` // create_conversation_session — optional: resume an existing conversation (bind the new // session to it) when known; absent/unknown → a fresh conversation (unchanged). th-d5b446. ConversationID string `json:"conversationId"` @@ -342,9 +345,23 @@ func (d *FrameDispatcher) handleCreateSession(ctx context.Context, frame inbound } // A freshly created session never passes through scopedSession, so associate here too. d.associateSession(&session) - // Record the client's declared render capabilities for this session so a later turn - // offers a kind's rich card only when its capability is present (else the fallback). - d.setSupports(session.SessionID, frame.Supports) + // Record the client's declared render capabilities on the CONVERSATION, so a later + // turn offers a kind's rich card only when its capability is present (else the + // fallback) — and so does a turn on the session the NEXT reconnect mints. A frame + // that omits the key declares nothing and writes nothing: it inherits whatever the + // conversation last declared (already on session.Supports from the resume). A frame + // that DOES declare wins, including an explicit `[]` — that is the text-only opt-out, + // and it must replace the stored set rather than leave it for the reconnect after it. + // + // Best-effort, like the workflow step pointer: a storage error is logged, not fatal. + // This session already has its capabilities from the frame; the blast radius is a + // LATER reconnect that omits `supports` degrading to text-only. + if frame.Supports != nil { + if err := d.store.SetConversationSupports(ctx, session.ConversationID, *frame.Supports); err != nil { + slog.Warn("failed to persist declared client capabilities; a later reconnect that omits 'supports' will degrade to text-only", + "conversationId", session.ConversationID, "error", err) + } + } data := map[string]any{ "sessionId": session.SessionID, "conversationId": session.ConversationID, @@ -735,7 +752,7 @@ func (d *FrameDispatcher) handleSendMessage(ctx context.Context, frame inboundFr // kind (rich park when the capability is declared, else conversational fallback). runner.interactionKinds = d.interactionKinds runner.interactions = d.interactions - runner.capabilities = d.capabilities(frame.SessionID) + runner.capabilities = capabilitySet(session.Supports) // Span attribution: the owning org (grouped by smooai.org_id on the turn span). runner.orgID = d.access.Principal.Org result, err := runner.Run(turnCtx, frame.SessionID, session.ConversationID, requestID, frame.Message, sink) @@ -816,30 +833,17 @@ func (d *FrameDispatcher) handleConfirmToolAction(frame inboundFrame, sink Event })) } -// setSupports records a session's declared render capabilities (from -// create_conversation_session). An absent/empty list means text-only. Idempotent. -func (d *FrameDispatcher) setSupports(sessionID string, supports []string) { +// capabilitySet turns a conversation's stored `supports` list into the set the turn +// runner gates rich cards on. Empty/nil ⇒ text-only: every kind degrades to its +// conversational fallback. +func capabilitySet(supports []string) map[string]bool { caps := make(map[string]bool, len(supports)) for _, c := range supports { if c != "" { caps[c] = true } } - d.supportsMu.Lock() - d.supports[sessionID] = caps - d.supportsMu.Unlock() -} - -// capabilities returns the render capabilities a session declared at create time (an -// empty set when it declared none or is unknown on this connection → every kind -// degrades to its conversational fallback). -func (d *FrameDispatcher) capabilities(sessionID string) map[string]bool { - d.supportsMu.Lock() - defer d.supportsMu.Unlock() - if caps, ok := d.supports[sessionID]; ok { - return caps - } - return map[string]bool{} + return caps } // handleSubmitInteraction resumes a turn parked on a Rich Interaction. diff --git a/go/server/postgres_store.go b/go/server/postgres_store.go index 6875bbd6..a48cab78 100644 --- a/go/server/postgres_store.go +++ b/go/server/postgres_store.go @@ -271,8 +271,9 @@ func (s *PostgresStore) ResumeSession(ctx context.Context, agentID, userName, us resumed := false owner := scope.Email + var supports []string if conversationID != "" { - existingOwner, found, err := s.conversationOwner(ctx, conversationID, scope.OrgID) + existingOwner, existingSupports, found, err := s.conversationOwner(ctx, conversationID, scope.OrgID) if err != nil { return StoredSession{}, false, err } @@ -280,7 +281,9 @@ func (s *PostgresStore) ResumeSession(ctx context.Context, agentID, userName, us // conversationOwner already filtered by scope.OrgID, so the org half is // satisfied by construction — pass it through rather than re-deriving it. if found && scope.Allows(existingOwner, scope.OrgID) { - resumed, owner = true, existingOwner + // A reconnect that re-declares nothing inherits what the conversation last + // declared; a fresh conversation has none. th-13df6d. + resumed, owner, supports = true, existingOwner, existingSupports } } @@ -304,6 +307,7 @@ func (s *PostgresStore) ResumeSession(ctx context.Context, agentID, userName, us // only matches rows whose organization_id equals it — so both branches stamp // the same org. OwnerOrg: scope.OrgID, + Supports: supports, } metadata, err := json.Marshal(sessionMetadata{ContactEmail: userEmail}) @@ -367,27 +371,56 @@ func (s *PostgresStore) ResumeSession(ctx context.Context, agentID, userName, us } // conversationOwner returns the owner email of an org's conversation ("" when -// ownerless) and whether the conversation exists in that org at all. A conversation -// in ANOTHER org reports found=false — indistinguishable from never having existed. -func (s *PostgresStore) conversationOwner(ctx context.Context, conversationID, orgID string) (string, bool, error) { - var owner string +// ownerless), the render capabilities it last declared (nil when none), and whether the +// conversation exists in that org at all. A conversation in ANOTHER org reports +// found=false — indistinguishable from never having existed. +// +// The capabilities ride this query rather than a second roundtrip: it already reads the +// conversation row the resume path needs them from. +func (s *PostgresStore) conversationOwner(ctx context.Context, conversationID, orgID string) (string, []string, bool, error) { + var ( + owner string + supports []byte + ) err := s.pool.QueryRow(ctx, `SELECT coalesce((SELECT p.email FROM conversation_participants p WHERE p.conversation_id = c.id AND p.type = 'user' - ORDER BY p.created_at, p.id LIMIT 1), '') + ORDER BY p.created_at, p.id LIMIT 1), ''), + coalesce(c.metadata_json -> '`+clientSupportsMetaKey+`', 'null'::jsonb) FROM conversations c WHERE c.id = $1 AND c.organization_id = $2`, - conversationID, orgID).Scan(&owner) + conversationID, orgID).Scan(&owner, &supports) switch { case err == nil: - return owner, true, nil + return owner, decodeClientSupports(supports), true, nil case errors.Is(err, pgx.ErrNoRows): - return "", false, nil + return "", nil, false, nil default: - return "", false, fmt.Errorf("postgres: resolve conversation owner: %w", err) + return "", nil, false, fmt.Errorf("postgres: resolve conversation owner: %w", err) } } +// clientSupportsMetaKey is the conversations.metadata_json key holding the render +// capabilities (`supports`) a client last declared for the conversation — the same key +// and the same home as the Rust reference adapter's, so the two servers can share a +// database and still agree on what a resuming client can render. th-13df6d. +const clientSupportsMetaKey = "clientSupports" + +// decodeClientSupports reads the stored capability list. Anything unreadable (absent, +// JSON null, a non-array someone else wrote) is nil — i.e. exactly the text-only +// behavior, which is the safe direction: a card a client cannot render would park the +// turn until it times out. +func decodeClientSupports(raw []byte) []string { + if len(raw) == 0 { + return nil + } + var supports []string + if err := json.Unmarshal(raw, &supports); err != nil { + return nil + } + return supports +} + // GetSession returns the session for sessionID, or (nil, nil) if unknown. The raw // lookup primitive: ownership is REPORTED (OwnerEmail + OwnerOrg) but not enforced // here — the dispatcher's scopedSession is the gate, and it applies org before owner. @@ -402,19 +435,21 @@ func (s *PostgresStore) conversationOwner(ctx context.Context, conversationID, o // ownerless conversation, on the one backend that actually holds several orgs' data. func (s *PostgresStore) GetSession(ctx context.Context, sessionID string) (*StoredSession, error) { var session StoredSession - var metadata []byte + var metadata, supports []byte err := s.pool.QueryRow(ctx, `SELECT s.conversation_id, coalesce(s.agent_id, ''), s.agent_name, s.user_participant_id, s.agent_participant_id, s.organization_id, coalesce(s.metadata, '{}'::jsonb), coalesce((SELECT p.email FROM conversation_participants p WHERE p.conversation_id = s.conversation_id AND p.type = 'user' - ORDER BY p.created_at, p.id LIMIT 1), '') + ORDER BY p.created_at, p.id LIMIT 1), ''), + coalesce((SELECT c.metadata_json -> '`+clientSupportsMetaKey+`' FROM conversations c + WHERE c.id = s.conversation_id), 'null'::jsonb) FROM conversation_sessions s WHERE s.session_id = $1`, sessionID).Scan(&session.ConversationID, &session.AgentID, &session.AgentName, &session.UserParticipantID, &session.AgentParticipantID, &session.OwnerOrg, - &metadata, &session.OwnerEmail) + &metadata, &session.OwnerEmail, &supports) if errors.Is(err, pgx.ErrNoRows) { return nil, nil } @@ -431,6 +466,9 @@ func (s *PostgresStore) GetSession(ctx context.Context, sessionID string) (*Stor session.ContactPhone = meta.ContactPhone session.OtpVerified = meta.OtpVerified session.CurrentStepID = meta.CurrentStepID + // Conversation-scoped, so every session in the conversation reports the same set — + // including the one a reconnect just minted. th-13df6d. + session.Supports = decodeClientSupports(supports) return &session, nil } @@ -596,6 +634,36 @@ func (s *PostgresStore) AttachSessionContact(ctx context.Context, sessionID, use return s.mergeSessionMetadata(ctx, sessionID, patch) } +// SetConversationSupports replaces a conversation's declared render capabilities. +// `||` is a shallow merge on this one key, so sibling metadata (the caller's own +// `metadata`, anything the Rust server wrote) survives; an empty list DELETES the key +// rather than storing `[]`, so a stale set can never be resurrected by a later +// reconnect that omits `supports`. A no-op for an unknown conversation (no row matches). +func (s *PostgresStore) SetConversationSupports(ctx context.Context, conversationID string, supports []string) error { + if len(supports) == 0 { + if _, err := s.pool.Exec(ctx, + `UPDATE conversations + SET metadata_json = coalesce(metadata_json, '{}'::jsonb) - $2 + WHERE id = $1`, + conversationID, clientSupportsMetaKey); err != nil { + return fmt.Errorf("postgres: clear conversation supports: %w", err) + } + return nil + } + encoded, err := json.Marshal(map[string]any{clientSupportsMetaKey: supports}) + if err != nil { + return fmt.Errorf("postgres: encode conversation supports: %w", err) + } + if _, err := s.pool.Exec(ctx, + `UPDATE conversations + SET metadata_json = coalesce(metadata_json, '{}'::jsonb) || $2::jsonb + WHERE id = $1`, + conversationID, string(encoded)); err != nil { + return fmt.Errorf("postgres: update conversation supports: %w", err) + } + return nil +} + // mergeSessionMetadata merges patch into a session's metadata JSON. `||` on jsonb is a // shallow merge, which is all this flat object needs — and it leaves the other keys // (contactEmail, the sibling flag) alone instead of clobbering them. diff --git a/go/server/postgres_store_test.go b/go/server/postgres_store_test.go index 7b336edd..9a9cf216 100644 --- a/go/server/postgres_store_test.go +++ b/go/server/postgres_store_test.go @@ -882,3 +882,55 @@ func TestSessionWithNoAgentHasNoAgent(t *testing.T) { t.Errorf("round-tripped agentId = %q, want empty", fetched.AgentID) } } + +// The declared render capabilities ride the CONVERSATION, so the session a reconnect +// mints inherits them — the durable half of th-13df6d. Also asserts the edge the +// in-memory store gets for free but SQL does not: an empty list CLEARS the stored key +// rather than leaving a stale set for the next omitting reconnect to resurrect. +func TestPostgresStorePersistsConversationSupports(t *testing.T) { + store := newPostgresStore(t) + ctx := t.Context() + scope := pgScope(t, "alice@example.test") + + session, err := store.CreateSession(ctx, "", "Alice", "alice@example.test", scope) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + if err := store.SetConversationSupports(ctx, session.ConversationID, []string{"identity_form"}); err != nil { + t.Fatalf("SetConversationSupports: %v", err) + } + + // A reconnect: a NEW session on the same conversation, through a fresh store. + reconnected, resumed, err := newPostgresStore(t).ResumeSession(ctx, "", "Alice", "alice@example.test", scope, session.ConversationID) + if err != nil || !resumed { + t.Fatalf("ResumeSession: resumed=%v err=%v", resumed, err) + } + if !capabilitySet(reconnected.Supports)["identity_form"] { + t.Fatalf("resumed session lost the conversation's capabilities: %v", reconnected.Supports) + } + // …and the turn's read path (GetSession) reports the same set. + fetched, err := store.GetSession(ctx, reconnected.SessionID) + if err != nil || fetched == nil { + t.Fatalf("GetSession: %v (session %v)", err, fetched) + } + if !capabilitySet(fetched.Supports)["identity_form"] { + t.Fatalf("GetSession reported capabilities %v, want identity_form", fetched.Supports) + } + + // The text-only opt-out clears the key — a later resume must not resurrect it. + if err := store.SetConversationSupports(ctx, session.ConversationID, nil); err != nil { + t.Fatalf("SetConversationSupports(empty): %v", err) + } + afterOptOut, _, err := store.ResumeSession(ctx, "", "Alice", "alice@example.test", scope, session.ConversationID) + if err != nil { + t.Fatalf("ResumeSession after opt-out: %v", err) + } + if len(afterOptOut.Supports) != 0 { + t.Fatalf("the opt-out left a stale set: %v", afterOptOut.Supports) + } + + // Unknown conversation: a no-op, never an error. + if err := store.SetConversationSupports(ctx, "unknown-conversation", []string{"identity_form"}); err != nil { + t.Fatalf("SetConversationSupports(unknown) must be a no-op, got %v", err) + } +} diff --git a/go/server/session_store.go b/go/server/session_store.go index 94da3f3f..8e86b4d8 100644 --- a/go/server/session_store.go +++ b/go/server/session_store.go @@ -61,6 +61,17 @@ type StoredSession struct { // dispatcher checks before ownership, so a session id alone cannot reach another // org's conversation. OwnerOrg string + // Supports is the client render-capability list (`supports`) last declared for this + // session's CONVERSATION — the gate on which Rich Interactions are offered as parked + // cards rather than degrading to their conversational fallback. + // + // Conversation-scoped, not session-scoped, and populated FROM the conversation on + // every load/resume: a reconnect IS a resume (the client re-opens the socket and + // re-issues create_conversation_session with the same conversationId, minting a NEW + // session id), so a per-session — or, as this was before, per-connection — copy is + // lost on every network blip and the whole framework silently goes text-only with + // nothing on the wire to notice. th-13df6d. + Supports []string } // ConversationScope is the visibility filter for conversation reads: WHO is asking, derived @@ -202,6 +213,17 @@ type SessionStore interface { // become OTP delivery channels on subsequent turns. A no-op for an unknown session. The Go // analog of the Rust attach_session_identity. AttachSessionContact(ctx context.Context, sessionID, userName, email, phone string) error + // SetConversationSupports persists the render capabilities a client DECLARED on + // create_conversation_session, keyed by CONVERSATION so the next reconnect — which + // resumes the conversation on a fresh session and typically re-declares nothing — + // still gets its Rich Interactions (th-13df6d). Every load of a session in this + // conversation reports it back as StoredSession.Supports. + // + // A REPLACE, not a merge: an explicit empty list is how a text-only channel resuming + // a rich conversation opts out, so it must clear the stored set rather than leave a + // stale one for a later omitting reconnect to resurrect. A no-op for an unknown + // conversation. + SetConversationSupports(ctx context.Context, conversationID string, supports []string) error } // InMemorySessionStore is an in-process SessionStore. The Go analog of the Rust @@ -221,6 +243,10 @@ type InMemorySessionStore struct { // owner and never rewritten, for the same reason: a resume must not re-home a // conversation into the resumer's org. org map[string]string + // supports maps conversation id → the render capabilities its client last declared. + // Conversation-scoped so a reconnect (a resume on a NEW session id) inherits them — + // see StoredSession.Supports. th-13df6d. + supports map[string][]string } // NewInMemorySessionStore returns an empty in-memory store. @@ -231,6 +257,7 @@ func NewInMemorySessionStore() *InMemorySessionStore { updatedAt: map[string]time.Time{}, org: map[string]string{}, owner: map[string]string{}, + supports: map[string][]string{}, } } @@ -287,6 +314,9 @@ func (s *InMemorySessionStore) ResumeSession(_ context.Context, agentID, _ /*use if resumed { session.OwnerEmail = s.owner[convID] session.OwnerOrg = s.org[convID] + // A reconnect that re-declares nothing inherits what the conversation last + // declared; a fresh conversation has none. th-13df6d. + session.Supports = s.supports[convID] } s.sessions[session.SessionID] = session if !resumed { @@ -303,6 +333,10 @@ func (s *InMemorySessionStore) GetSession(_ context.Context, sessionID string) ( s.mu.Lock() defer s.mu.Unlock() if session, ok := s.sessions[sessionID]; ok { + // Capabilities are conversation-scoped and read through on every load, so a + // declaration made on a LATER session of the same conversation is what this + // session's turns see (and the session map never holds a stale copy). + session.Supports = s.supports[session.ConversationID] return &session, nil } return nil, nil @@ -395,6 +429,25 @@ func (s *InMemorySessionStore) SetSessionAuthenticated(_ context.Context, sessio return nil } +// SetConversationSupports replaces a conversation's declared render capabilities. An empty +// list clears them (the text-only opt-out), so a later reconnect that omits `supports` +// cannot resurrect them. A no-op for an unknown conversation — which also keeps the map +// from growing on ids nobody minted. +func (s *InMemorySessionStore) SetConversationSupports(_ context.Context, conversationID string, supports []string) error { + s.mu.Lock() + defer s.mu.Unlock() + if _, known := s.messages[conversationID]; !known { + return nil + } + if len(supports) == 0 { + delete(s.supports, conversationID) + return nil + } + // Copy: the caller's slice is frame-owned and must not alias the store's. + s.supports[conversationID] = append([]string(nil), supports...) + return nil +} + // AttachSessionContact stamps a captured identity onto a session (merge — blank fields left // untouched). A no-op for an unknown session. func (s *InMemorySessionStore) AttachSessionContact(_ context.Context, sessionID, userName, email, phone string) error { diff --git a/python/server/src/smooth_operator_server/dispatcher.py b/python/server/src/smooth_operator_server/dispatcher.py index d2db4673..4b5747dd 100644 --- a/python/server/src/smooth_operator_server/dispatcher.py +++ b/python/server/src/smooth_operator_server/dispatcher.py @@ -84,11 +84,6 @@ def __init__( #: this same connection, so (like confirmations) the registry is connection-local. self._interactions = interactions if interactions is not None else InteractionRegistry.default() self._interaction_pending = interaction_pending if interaction_pending is not None else PendingInteractions() - #: ``supports`` (client render capabilities) declared per session at - #: ``create_conversation_session``, connection-local: park/resume happen on this - #: connection, so this need not be persisted. Gates each kind's rich-vs-fallback - #: path per turn. - self._session_supports: dict[str, list[str]] = {} #: Per-agent config resolver (SMOODEV-590). Resolved per turn from the session's #: agent; the default (empty static resolver) returns None → the server-wide #: default prompt drives every turn. @@ -331,13 +326,29 @@ async def _handle_create_session(self, frame: dict, request_id: str | None, sink org_id=self._access.principal.org, ) await self._associate_session(session) - # Capture the session's declared render capabilities (``supports``), connection- - # local, to gate Rich Interactions per turn. Unknown values are kept as-is and - # simply never match a kind's capability (forward-compatible). A non-list is - # ignored (no capabilities → every kind degrades to its conversational fallback). + # Capture the declared render capabilities (``supports``) that gate Rich + # Interactions per turn. Unknown values are kept as-is and simply never match a + # kind's capability (forward-compatible). + # + # They are persisted on the CONVERSATION, not this connection: a RECONNECT is a + # resume — the client re-opens the socket and re-issues this action with the same + # ``conversationId``, minting a NEW session on a NEW dispatcher. Capabilities held + # per connection were therefore dropped on every network blip / backgrounding / + # deploy, and every interaction kind quietly fell back to conversational + # collection with nothing on the wire to notice (th-13df6d). + # + # A frame that DECLARES always wins and overwrites, including an explicit ``[]``: + # that is how a text-only channel resuming a rich conversation opts out, and the + # opt-out has to be durable too — hence a list, empty or not, is always written. + # A frame that OMITS the key writes nothing and so inherits whatever the + # conversation last declared (a non-list is treated as omitted, mirroring the Rust + # reference). A fresh conversation has nothing stored, so omitting still means + # text-only — unchanged behavior. raw_supports = frame.get("supports") if isinstance(raw_supports, list): - self._session_supports[session.session_id] = [s for s in raw_supports if isinstance(s, str)] + await self._store.set_client_supports( + session.conversation_id, [s for s in raw_supports if isinstance(s, str)] + ) data = { "sessionId": session.session_id, "conversationId": session.conversation_id, @@ -561,7 +572,9 @@ async def _handle_send_message(self, frame: dict, request_id: str | None, sink: org_id=self._access.principal.org, interactions=self._interactions, interaction_pending=self._interaction_pending, - capabilities=self._session_supports.get(session_id), + # Read from the conversation, the one source of truth: this turn may be + # running on a reconnect whose frame never re-declared `supports` (th-13df6d). + capabilities=await self._store.get_client_supports(session.conversation_id), ) # Run the turn as a background task, NOT awaited inline. A turn that calls a diff --git a/python/server/src/smooth_operator_server/postgres_store.py b/python/server/src/smooth_operator_server/postgres_store.py index 1b6affac..59186b0c 100644 --- a/python/server/src/smooth_operator_server/postgres_store.py +++ b/python/server/src/smooth_operator_server/postgres_store.py @@ -497,6 +497,38 @@ async def set_current_step_id(self, conversation_id: str, step_id: str | None) - json.dumps({"currentStepId": step_id}), ) + async def get_client_supports(self, conversation_id: str) -> list[str]: + """The client render capabilities (``supports``) the conversation last declared — + durable, so a reconnect that omits the key still gets its Rich Interactions + (th-13df6d). Missing / malformed → empty, i.e. text-only.""" + raw = await self._pool.fetchval( + "SELECT metadata_json->'clientSupports' FROM conversations WHERE id = $1", + conversation_id, + ) + value = json.loads(raw) if isinstance(raw, str) else raw + return [s for s in value if isinstance(s, str)] if isinstance(value, list) else [] + + async def set_client_supports(self, conversation_id: str, supports: list[str]) -> None: + """Persist (or clear) the declared capabilities, next to the workflow-step + pointer and with the same shallow-merge / single-key-delete shape so sibling + metadata survives. An empty list is the text-only opt-out: it must REMOVE the + key, or a later reconnect that omits ``supports`` would resurrect the old set.""" + if not supports: + await self._pool.execute( + """UPDATE conversations + SET metadata_json = COALESCE(metadata_json, '{}'::jsonb) - 'clientSupports' + WHERE id = $1""", + conversation_id, + ) + return + await self._pool.execute( + """UPDATE conversations + SET metadata_json = COALESCE(metadata_json, '{}'::jsonb) || $2::jsonb + WHERE id = $1""", + conversation_id, + json.dumps({"clientSupports": list(supports)}), + ) + async def is_session_authenticated(self, session_id: str) -> bool: """Whether the caller completed OTP verification for this session. ``False`` for an unknown or unverified session.""" diff --git a/python/server/src/smooth_operator_server/session_store.py b/python/server/src/smooth_operator_server/session_store.py index 7b972a7f..e33c9c20 100644 --- a/python/server/src/smooth_operator_server/session_store.py +++ b/python/server/src/smooth_operator_server/session_store.py @@ -195,6 +195,35 @@ async def set_current_step_id(self, conversation_id: str, step_id: str | None) - ``state.currentStepId`` carried across turns).""" ... + async def get_client_supports(self, conversation_id: str) -> list[str]: + """The client render capabilities (``supports``) this conversation last declared. + + Conversation-scoped, exactly like the workflow-step pointer above and for the + same reason (th-c12df5): a RECONNECT is a resume — the client re-opens the + socket and re-issues ``create_conversation_session`` with the same + ``conversationId``, which mints a NEW session on a NEW connection. Capabilities + kept per session/connection are therefore lost on every network blip, and Rich + Interactions silently stop being offered with nothing on the wire to notice + (th-13df6d). + + Empty for an unknown conversation, or one that declared nothing — which is + exactly the text-only behavior (every interaction kind degrades to its + conversational fallback). + + Like :meth:`attach_session_identity` this **base default is a no-op** (returns + empty) rather than an ``@abstractmethod``, so a downstream :class:`SessionStore` + keeps compiling; it simply keeps today's behavior — a reconnect that omits + ``supports`` degrades to text-only. Both bundled stores override it.""" + return [] + + async def set_client_supports(self, conversation_id: str, supports: list[str]) -> None: + """Persist the capability list a ``create_conversation_session`` frame DECLARED, + replacing whatever the conversation held. An empty list is a declaration, not an + absence: it is how a text-only channel resuming a rich conversation opts out, so + it must clear the stored set rather than leave a stale one a later reconnect + would resurrect. No-op default, as above.""" + return None + @abstractmethod async def is_session_authenticated(self, session_id: str) -> bool: """Whether the caller has completed OTP identity verification for this session @@ -247,6 +276,10 @@ def __init__(self) -> None: self._orgs: dict[str, str | None] = {} #: Per-conversation workflow-step pointer (absent = fresh start / no workflow). self._current_step: dict[str, str] = {} + #: Per-conversation client render capabilities last declared in ``supports`` + #: (absent = none → text-only). Lives beside the step pointer because it has the + #: same lifetime: a reconnect resumes the conversation, not the session. + self._client_supports: dict[str, list[str]] = {} #: Per-session OTP-verified bit (absent/False = unverified). Set by a #: successful ``verify_otp``; read by the ``end_user`` auth gate. self._authenticated: dict[str, bool] = {} @@ -362,6 +395,20 @@ async def set_current_step_id(self, conversation_id: str, step_id: str | None) - else: self._current_step[conversation_id] = step_id + async def get_client_supports(self, conversation_id: str) -> list[str]: + with self._gate: + return list(self._client_supports.get(conversation_id, ())) + + async def set_client_supports(self, conversation_id: str, supports: list[str]) -> None: + with self._gate: + # An explicit `[]` REPLACES the stored set (the text-only opt-out), so drop + # the entry rather than keeping an empty one — same shape as clearing the + # step pointer above, and the read is `[]` either way. + if supports: + self._client_supports[conversation_id] = list(supports) + else: + self._client_supports.pop(conversation_id, None) + async def is_session_authenticated(self, session_id: str) -> bool: with self._gate: return self._authenticated.get(session_id, False) diff --git a/python/server/tests/test_postgres_store.py b/python/server/tests/test_postgres_store.py index ad28b430..43250e0d 100644 --- a/python/server/tests/test_postgres_store.py +++ b/python/server/tests/test_postgres_store.py @@ -274,8 +274,12 @@ async def test_isolates_organizations(store) -> None: assert cross_org.conversation_id != in_a.conversation_id -async def test_persists_workflow_step_and_otp_bit(store, postgres_dsn: str) -> None: - """Both survive a reconnect, and both are no-ops for an unknown id.""" +async def test_persists_workflow_step_otp_bit_and_client_supports(store, postgres_dsn: str) -> None: + """All three survive a reconnect, and all three are no-ops for an unknown id. + + ``clientSupports`` is the client render-capability list (``supports``): a reconnect + resumes the conversation on a NEW session, so the capabilities have to outlive the + session exactly like the workflow-step pointer beside them (th-13df6d).""" from smooth_operator_server.postgres_store import PostgresStore session = await store.create_session( @@ -283,11 +287,13 @@ async def test_persists_workflow_step_and_otp_bit(store, postgres_dsn: str) -> N ) await store.set_current_step_id(session.conversation_id, "collect-email") await store.set_session_authenticated(session.session_id, True) + await store.set_client_supports(session.conversation_id, ["identity_form", "choice_chips"]) reopened = await PostgresStore.create(postgres_dsn) try: assert await reopened.get_current_step_id(session.conversation_id) == "collect-email" assert await reopened.is_session_authenticated(session.session_id) is True + assert await reopened.get_client_supports(session.conversation_id) == ["identity_form", "choice_chips"] # The OTP write must not have clobbered the contact email beside it. fetched = await reopened.get_session(session.session_id) assert fetched is not None and fetched.contact_email == "alice@example.test" @@ -295,14 +301,24 @@ async def test_persists_workflow_step_and_otp_bit(store, postgres_dsn: str) -> N # Clearing the step removes only that key. await reopened.set_current_step_id(session.conversation_id, None) assert await reopened.get_current_step_id(session.conversation_id) is None + assert await reopened.get_client_supports(session.conversation_id) == ["identity_form", "choice_chips"] await reopened.set_session_authenticated(session.session_id, False) assert await reopened.is_session_authenticated(session.session_id) is False + # An explicit `[]` is the text-only opt-out: it REPLACES the stored set (a later + # reconnect that omits `supports` must not resurrect it), and disturbs nothing else. + await reopened.set_current_step_id(session.conversation_id, "collect-email") + await reopened.set_client_supports(session.conversation_id, []) + assert await reopened.get_client_supports(session.conversation_id) == [] + assert await reopened.get_current_step_id(session.conversation_id) == "collect-email" + # No-ops for unknown ids, never errors. await reopened.set_current_step_id("unknown-conversation", "whatever") await reopened.set_session_authenticated("unknown-session", True) + await reopened.set_client_supports("unknown-conversation", ["identity_form"]) assert await reopened.is_session_authenticated("unknown-session") is False + assert await reopened.get_client_supports("unknown-conversation") == [] finally: await reopened.close() diff --git a/python/server/tests/test_supports_reconnect.py b/python/server/tests/test_supports_reconnect.py new file mode 100644 index 00000000..8b0cde54 --- /dev/null +++ b/python/server/tests/test_supports_reconnect.py @@ -0,0 +1,129 @@ +"""``supports`` survives a reconnect (th-13df6d). + +``supports`` is the client render-capability list declared on +``create_conversation_session``; it gates the entire Rich Interactions framework (a kind +whose capability is declared parks the turn and emits ``interaction_required``, one +without it degrades to a conversational fallback). + +A RECONNECT is a resume: the client re-opens the socket and re-issues +``create_conversation_session`` with the same ``conversationId``, which mints a NEW +session on a NEW :class:`FrameDispatcher`. While the declared list lived in a +per-connection dict on the dispatcher, every network blip / backgrounding / deploy +silently dropped it and the shipped feature degraded with nothing on the wire to notice. +It now rides the conversation, the same durability the workflow-step pointer has. + +The assertions go through the thing the declaration is FOR: the ``capabilities`` the +dispatcher hands the turn. A spy :class:`TurnRunner` captures them, so a regression that +persists the list but forgets to read it back still fails. +""" + +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from smooth_operator_server import dispatcher as dispatcher_module +from smooth_operator_server.dispatcher import FrameDispatcher +from smooth_operator_server.session_store import InMemorySessionStore +from smooth_operator_server.turn_runner import TurnResult + +_AGENT = "11111111-1111-1111-1111-111111111111" + + +class _SpyRunner: + """Stands in for :class:`TurnRunner`, recording the capabilities it was built with + and completing the turn immediately (no LLM, no streaming).""" + + last_capabilities: list[str] | None = None + + def __init__(self, *_args: Any, capabilities: list[str] | None = None, **_kwargs: Any) -> None: + _SpyRunner.last_capabilities = capabilities + + async def run(self, *_args: Any, **_kwargs: Any) -> TurnResult: + return TurnResult(reply="ok", message_id="m-1") + + +async def _dispatch(dispatcher: FrameDispatcher, frame: dict) -> list[dict]: + """Dispatch one frame, collecting every event emitted to the sink.""" + events: list[dict] = [] + await dispatcher.dispatch(json.dumps(frame), events.append) + return events + + +async def _create(dispatcher: FrameDispatcher, request_id: str, **extra: Any) -> tuple[str, str]: + """``create_conversation_session`` → ``(sessionId, conversationId)``.""" + events = await _dispatch( + dispatcher, {"action": "create_conversation_session", "requestId": request_id, "agentId": _AGENT, **extra} + ) + data = events[0]["data"] + return data["sessionId"], data["conversationId"] + + +async def _turn_capabilities(dispatcher: FrameDispatcher, session_id: str) -> list[str] | None: + """The capabilities a turn on this session is actually handed.""" + _SpyRunner.last_capabilities = None + await _dispatch( + dispatcher, {"action": "send_message", "requestId": "r-msg", "sessionId": session_id, "message": "hi"} + ) + # The turn runs as a background task; capabilities are captured when the runner is + # constructed (before the spawn), but await the task so it never outlives the test. + for task in list(dispatcher._turn_tasks): # noqa: SLF001 — no public handle on the in-flight turn + await task + return _SpyRunner.last_capabilities + + +@pytest.fixture(autouse=True) +def _spy_runner(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(dispatcher_module, "TurnRunner", _SpyRunner) + + +@pytest.mark.asyncio +async def test_reconnect_resuming_a_conversation_keeps_the_declared_capabilities() -> None: + store = InMemorySessionStore() + + # First connection: declares the capability. + first = FrameDispatcher(store, object()) + session_id, conversation_id = await _create(first, "r-conn-1", supports=["identity_form"]) + assert await _turn_capabilities(first, session_id) == ["identity_form"] + + # Reconnect: a SECOND, fresh dispatcher over the same store (that is what a new + # socket is), same conversation, `supports` omitted — exactly what a widget resuming + # from its stored conversationId sends. This is where the feature went dark. + second = FrameDispatcher(store, object()) + resumed_id, resumed_conversation = await _create(second, "r-conn-2", conversationId=conversation_id) + assert resumed_conversation == conversation_id, "the reconnect resumed the same conversation" + assert resumed_id != session_id, "a reconnect mints a NEW session" + assert await _turn_capabilities(second, resumed_id) == ["identity_form"], ( + "a reconnect that omits 'supports' inherits the conversation's declared capabilities" + ) + + # A resume that DECLARES wins over the inherited set, in both directions — so a + # text-only client resuming a rich conversation opts out with `[]` rather than being + # handed cards it cannot render. + third = FrameDispatcher(store, object()) + text_only_id, _ = await _create(third, "r-conn-3", conversationId=conversation_id, supports=[]) + assert not await _turn_capabilities(third, text_only_id), ( + "an explicit empty 'supports' declares text-only and never inherits" + ) + + # ...and that opt-out is itself durable: the NEXT reconnect omitting the key must not + # resurrect the capability from a stale record. + fourth = FrameDispatcher(store, object()) + after_id, _ = await _create(fourth, "r-conn-4", conversationId=conversation_id) + assert not await _turn_capabilities(fourth, after_id), "the text-only declaration replaced the durable record" + + +@pytest.mark.asyncio +async def test_fresh_conversation_without_supports_is_text_only() -> None: + """The inherit rule keys on the CONVERSATION, so a brand-new one still starts empty — + nothing to inherit means unchanged behavior, not someone else's capabilities.""" + store = InMemorySessionStore() + dispatcher = FrameDispatcher(store, object()) + + rich_id, _ = await _create(dispatcher, "r-1", supports=["identity_form"]) + assert await _turn_capabilities(dispatcher, rich_id) == ["identity_form"] + + fresh_id, _ = await _create(dispatcher, "r-2") + assert not await _turn_capabilities(dispatcher, fresh_id) diff --git a/python/src/smooth_operator/_generated.py b/python/src/smooth_operator/_generated.py index 27814e24..7ffc2e8e 100644 --- a/python/src/smooth_operator/_generated.py +++ b/python/src/smooth_operator/_generated.py @@ -228,7 +228,7 @@ class CreateConversationSessionRequest(BaseModel): """ supports: list[str] | None = None """ - Client render capabilities for this session — a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` → capability `identity_form`); future kinds add their own values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) omit this and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). + Client render capabilities for this session — a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` → capability `identity_form`, kind `choices` → capability `choice_chips`); future kinds add their own values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) declare `[]` and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). Durability: the declared list is persisted on the CONVERSATION, so a reconnect that resumes an existing `conversationId` and OMITS this key inherits the set the conversation last declared — a reconnect is not a downgrade to text-only. Any list the frame does declare (including `[]`) replaces the inherited one, so a text-only client resuming a rich conversation opts out explicitly. """ metadata: dict[str, Any] | None = None """ @@ -253,7 +253,7 @@ class CreateConversationSessionResponse(BaseModel): """ ID of the conversation created for this session. """ - agent_id: Annotated[UUID, Field(alias='agentId')] + agent_id: Annotated[UUID | None, Field(alias='agentId')] = None """ ID of the agent handling this session. """ @@ -1894,9 +1894,9 @@ class Session(BaseModel): """ The organization that owns this session. Mirrors `organizationId` on the conversation, participants, and messages so org-scoping is uniform across every domain type and storage backends can write the session's org directly. """ - agent_id: Annotated[UUID, Field(alias='agentId')] + agent_id: Annotated[UUID | None, Field(alias='agentId')] = None """ - The agent handling this session. + The agent handling this session. OPTIONAL in storage: create_conversation_session REJECTS an absent or blank agentId, so a session created through the protocol always has one. It stays optional here for rows that predate that validation — it used to be filled with a fresh UUID, pointing every agentless session at an agent that had never existed (th-68897a). Absence is represented by omitting the field, never by a fabricated id. """ agent_name: Annotated[str, Field(alias='agentName')] """ diff --git a/rust/smooth-operator-server/src/handler.rs b/rust/smooth-operator-server/src/handler.rs index 7bba2f51..7c18837f 100644 --- a/rust/smooth-operator-server/src/handler.rs +++ b/rust/smooth-operator-server/src/handler.rs @@ -458,6 +458,45 @@ async fn handle_create_session( ) .await; + // Client render capabilities (`supports`, per + // create-conversation-session.schema.json) — the per-kind list gating which + // Rich Interactions this session gets as parked cards (e.g. `identity_form` + // for kind `identity_intake`); kinds without their capability degrade to + // the conversational fallback. Unknown values are kept (forward-compatible: + // a future kind may gate on them). `None` = the frame omitted the key + // entirely, which is what the inherit-on-resume rule below keys on; an + // explicit `[]` is a declaration of "I render nothing" and never inherits. + let declared: Option> = + parsed + .get("supports") + .and_then(Value::as_array) + .map(|caps| { + caps.iter() + .filter_map(|c| c.as_str().map(str::to_string)) + .collect() + }); + + // A RECONNECT is a resume: the client re-opens the socket and re-issues + // `create_conversation_session` with the same `conversationId`, which mints a + // NEW session id. Capabilities that live only on the session therefore have + // to be re-declared on every reconnect or Rich Interactions silently stop + // being offered — the shipped feature degrades with no error anyone sees + // (th-13df6d). So the declared set is persisted on the CONVERSATION, the same + // durability the workflow step pointer moved to for the same reason + // (th-c12df5: the session registry is per-pod and resets on reconnect/pod + // hop), and a resume that omits `supports` inherits it. + // + // A resume that declares (including `[]`) always wins, so a text-only client + // resuming a rich conversation opts out by sending `"supports": []`. The + // fallback direction is bounded either way: an offered card a client can't + // render times out (`INTERACTION_TIMEOUT`) into the same conversational + // fallback the capability gate would have chosen. + let supports: Vec = match (&declared, &resume) { + (Some(caps), _) => caps.clone(), + (None, Some(c)) => conversation_supports(c.metadata_json.as_ref()), + (None, None) => Vec::new(), + }; + // Only mint a conversation on a fresh session — a resume reuses the existing // one (and its persisted history), so `create_conversation` is skipped. let conversation = resume.is_none().then(|| Conversation { @@ -466,7 +505,7 @@ async fn handle_create_session( name: format!("Session {session_id}"), organization_id: org_id.clone(), idempotency_key: session_id.clone(), - metadata_json: parsed.get("metadata").cloned(), + metadata_json: with_client_supports(parsed.get("metadata").cloned(), &supports), analytics_json: None, created_at: now, updated_at: now, @@ -508,22 +547,6 @@ async fn handle_create_session( updated_at: now, }; - // Client render capabilities (`supports`, per - // create-conversation-session.schema.json) — the per-kind list gating which - // Rich Interactions this session gets as parked cards (e.g. `identity_form` - // for kind `identity_intake`); kinds without their capability degrade to - // the conversational fallback. Unknown values are kept (forward-compatible: - // a future kind may gate on them). - let supports: Vec = parsed - .get("supports") - .and_then(Value::as_array) - .map(|caps| { - caps.iter() - .filter_map(|c| c.as_str().map(str::to_string)) - .collect() - }) - .unwrap_or_default(); - // Stash the caller's OTP contact on the session so the end_user auth-gate // flow can offer verification without a storage roundtrip (mirrors how the // workflow step pointer lives in session metadata). The reference create path @@ -568,6 +591,8 @@ async fn handle_create_session( let request_id_owned = request_id.map(str::to_string); let session_for_registry = session.clone(); let state_clone = state.clone(); + let conversation_id_owned = conversation_id.clone(); + let redeclared = declared.is_some(); let data = json!({ "sessionId": session_id, @@ -589,6 +614,12 @@ async fn handle_create_session( )); return; } + } else if redeclared { + // A resume that re-declared `supports`: refresh the durable set so the + // NEXT reconnect — which may omit the key — inherits what this client + // actually renders. (A fresh conversation carries it in the metadata + // it was created with, so this write is the resume path only.) + persist_client_supports(storage.as_ref(), &conversation_id_owned, &supports).await; } if let Err(e) = storage.add_participant(user_participant).await { let _ = sink_clone.send(protocol::error( @@ -1134,6 +1165,91 @@ fn sanitize_title(raw: &str) -> String { const WF_STEP_META_KEY: &str = "workflowCurrentStepId"; const WF_ATTEMPTS_META_KEY: &str = "workflowStepAttempts"; +/// Conversation-metadata key holding the client render capabilities (`supports`) +/// last declared for this conversation. Durable, unlike the session registry, so +/// a reconnect that resumes the conversation without re-declaring still gets its +/// Rich Interactions (th-13df6d). +const CLIENT_SUPPORTS_META_KEY: &str = "clientSupports"; + +/// Read the durable capability list off a conversation's metadata. Missing / +/// unreadable → empty, i.e. exactly the text-only behavior (every interaction +/// kind degrades to its conversational fallback). +fn conversation_supports(metadata: Option<&Value>) -> Vec { + metadata + .and_then(|m| m.get(CLIENT_SUPPORTS_META_KEY)) + .and_then(Value::as_array) + .map(|caps| { + caps.iter() + .filter_map(|c| c.as_str().map(str::to_string)) + .collect() + }) + .unwrap_or_default() +} + +/// Fold the declared capabilities into the metadata a fresh conversation is +/// created with, so the very first session already leaves the durable record a +/// later reconnect reads. An empty list writes nothing (keeps the metadata of a +/// text-only conversation byte-for-byte what it was). +fn with_client_supports(metadata: Option, supports: &[String]) -> Option { + if supports.is_empty() { + return metadata; + } + let mut obj = match metadata { + Some(Value::Object(m)) => m, + _ => serde_json::Map::new(), + }; + obj.insert( + CLIENT_SUPPORTS_META_KEY.to_string(), + Value::from(supports.to_vec()), + ); + Some(Value::Object(obj)) +} + +/// Persist the declared capability list onto conversation metadata, +/// read-modify-write so sibling metadata keys (the workflow step pointer, the +/// caller's own `metadata`) survive. Best-effort, like +/// [`persist_workflow_step`]: a storage error is logged, not fatal — this +/// session already has its capabilities from the frame, and the worst case is +/// that a LATER reconnect which omits `supports` falls back to text-only. +async fn persist_client_supports( + storage: &dyn StorageAdapter, + conversation_id: &str, + supports: &[String], +) { + let existing = match storage.get_conversation(conversation_id).await { + Ok(Some(c)) => c.metadata_json, + _ => None, + }; + let mut obj = match existing { + Some(Value::Object(m)) => m, + _ => serde_json::Map::new(), + }; + if supports.is_empty() { + obj.remove(CLIENT_SUPPORTS_META_KEY); + } else { + obj.insert( + CLIENT_SUPPORTS_META_KEY.to_string(), + Value::from(supports.to_vec()), + ); + } + if let Err(e) = storage + .update_conversation( + conversation_id, + ConversationUpdate { + metadata_json: Some(Value::Object(obj)), + ..Default::default() + }, + ) + .await + { + tracing::warn!( + error = %e, + conversation_id, + "failed to persist declared client capabilities; a later reconnect that omits 'supports' will degrade to text-only" + ); + } +} + /// Read the durable `(current_step_id, attempts)` off the conversation's /// metadata. Missing / unreadable → `(None, 0)`, so the runner resolves to the /// workflow's first step exactly as a fresh conversation should. diff --git a/rust/smooth-operator-server/tests/interactions.rs b/rust/smooth-operator-server/tests/interactions.rs index 616af3da..5e0f4804 100644 --- a/rust/smooth-operator-server/tests/interactions.rs +++ b/rust/smooth-operator-server/tests/interactions.rs @@ -757,3 +757,114 @@ fn session_metadata(state: &AppState) -> std::collections::HashMap(); + + let create = |body: Value| { + let state = state.clone(); + let tx = tx.clone(); + async move { + handler::handle_frame( + &state, + &AccessContext::anonymous(), + "conn", + None, + None, + &handler::UserScope::Unscoped, + &body.to_string(), + &tx, + ) + .await; + } + }; + + // First connection: declares the capability. + create(json!({ + "action": "create_conversation_session", + "requestId": "req-conn-1", + "agentId": "11111111-1111-1111-1111-111111111111", + "supports": ["identity_form"] + })) + .await; + let (first, _) = await_event(&mut rx, "immediate_response").await; + let conversation_id = first["data"]["conversationId"] + .as_str() + .expect("conversationId") + .to_string(); + assert!(state + .session_capabilities(first["data"]["sessionId"].as_str().expect("sessionId")) + .contains("identity_form")); + + // Reconnect: same conversation, `supports` omitted (the client re-declares + // nothing — exactly what a widget resuming from its stored conversationId + // does). Without the durable record this is where the feature went dark. + create(json!({ + "action": "create_conversation_session", + "requestId": "req-conn-2", + "agentId": "11111111-1111-1111-1111-111111111111", + "conversationId": conversation_id + })) + .await; + let (resumed, _) = await_event(&mut rx, "immediate_response").await; + assert_eq!( + resumed["data"]["conversationId"].as_str(), + Some(conversation_id.as_str()), + "the reconnect resumed the same conversation" + ); + assert!( + state + .session_capabilities(resumed["data"]["sessionId"].as_str().expect("sessionId")) + .contains("identity_form"), + "a reconnect that omits 'supports' inherits the conversation's declared capabilities" + ); + + // A resume that DECLARES wins over the inherited set, in both directions — + // so a text-only client resuming a rich conversation opts out with `[]` + // rather than being handed cards it cannot render. + create(json!({ + "action": "create_conversation_session", + "requestId": "req-conn-3", + "agentId": "11111111-1111-1111-1111-111111111111", + "conversationId": conversation_id, + "supports": [] + })) + .await; + let (text_only, _) = await_event(&mut rx, "immediate_response").await; + let text_only_session = text_only["data"]["sessionId"].as_str().expect("sessionId"); + assert!( + state.session_capabilities(text_only_session).is_empty(), + "an explicit empty 'supports' declares text-only and never inherits" + ); + + // ...and that explicit opt-out is itself durable: the NEXT reconnect that + // omits `supports` must not resurrect the capability from a stale record. + create(json!({ + "action": "create_conversation_session", + "requestId": "req-conn-4", + "agentId": "11111111-1111-1111-1111-111111111111", + "conversationId": conversation_id + })) + .await; + let (after_opt_out, _) = await_event(&mut rx, "immediate_response").await; + assert!( + state + .session_capabilities( + after_opt_out["data"]["sessionId"] + .as_str() + .expect("sessionId") + ) + .is_empty(), + "the text-only declaration replaced the durable record" + ); +} diff --git a/spec/actions/create-conversation-session.schema.json b/spec/actions/create-conversation-session.schema.json index 46dccc87..d0b6b5b5 100644 --- a/spec/actions/create-conversation-session.schema.json +++ b/spec/actions/create-conversation-session.schema.json @@ -42,7 +42,7 @@ "supports": { "type": "array", "items": { "type": "string" }, - "description": "Client render capabilities for this session \u2014 a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` \u2192 capability `identity_form`, kind `choices` \u2192 capability `choice_chips`); future kinds add their own values (`date_picker`, `file_upload`, \u2026). Text-only channels (SMS, voice) omit this and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible)." + "description": "Client render capabilities for this session \u2014 a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` \u2192 capability `identity_form`, kind `choices` \u2192 capability `choice_chips`); future kinds add their own values (`date_picker`, `file_upload`, \u2026). Text-only channels (SMS, voice) declare `[]` and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). Durability: the declared list is persisted on the CONVERSATION, so a reconnect that resumes an existing `conversationId` and OMITS this key inherits the set the conversation last declared \u2014 a reconnect is not a downgrade to text-only. Any list the frame does declare (including `[]`) replaces the inherited one, so a text-only client resuming a rich conversation opts out explicitly." }, "metadata": { "type": "object", diff --git a/typescript/server/src/frameDispatcher.ts b/typescript/server/src/frameDispatcher.ts index ec428bef..71404a79 100644 --- a/typescript/server/src/frameDispatcher.ts +++ b/typescript/server/src/frameDispatcher.ts @@ -435,8 +435,15 @@ export class FrameDispatcher { const ownerEmail = this.access.authEnabled ? this.access.principal.email : typeof frame.userEmail === 'string' ? frame.userEmail : undefined; // The client's declared render capabilities (`supports`) gate this session's - // Rich Interactions. Non-string entries are dropped (forward-compatible); an - // absent/empty list ⇒ a text-only channel (every kind falls back). + // Rich Interactions. Non-string entries are dropped (forward-compatible). + // + // `undefined` vs `[]` is LOAD-BEARING and must survive to the store: `undefined` + // means the frame omitted the key, and a resume that omits it inherits what the + // conversation last declared — without that a reconnect (which mints a new + // session id) silently downgraded every kind to its text fallback (th-13df6d). + // An explicit `[]` is a declaration of "I render nothing" and replaces the + // stored set. A non-array value is malformed, so it reads as omitted — the same + // forgiving parse the Rust reference does. const supports = Array.isArray(frame.supports) ? frame.supports.filter((s): s is string => typeof s === 'string') : undefined; const session = await this.store.createSession( diff --git a/typescript/server/src/postgresStore.ts b/typescript/server/src/postgresStore.ts index 3a998ac3..b2b98ce5 100644 --- a/typescript/server/src/postgresStore.ts +++ b/typescript/server/src/postgresStore.ts @@ -165,6 +165,17 @@ CREATE INDEX IF NOT EXISTS idx_indexing_runs_org_started /** The agent's display name, mirroring the in-memory store and the Rust `AGENT_NAME`. */ const AGENT_NAME = 'smooth-agent'; +/** + * `conversations.metadata_json` key holding the render capabilities (`supports`) the + * conversation last declared. Conversation-scoped, not session-scoped: a reconnect + * mints a NEW session id, so a set kept on the session row was lost on every network + * blip and every Rich Interaction quietly degraded to its text fallback (th-13df6d). + * + * The literal string is shared with the Rust adapter's `CLIENT_SUPPORTS_META_KEY` — + * these servers write the SAME `conversations` table, so the key must not drift. + */ +const CLIENT_SUPPORTS_META_KEY = 'clientSupports'; + /** * The JSON held in `conversation_sessions.metadata`: the per-session bits * {@link StoredSession} carries that have no dedicated column in the shared schema. @@ -177,7 +188,11 @@ interface SessionMetadata { userName?: string; otpVerified?: boolean; currentStepId?: string; - /** The session's declared render capabilities (`supports`) — the Rich Interactions gate. */ + /** + * The render capabilities (`supports`) in effect for this session — the Rich + * Interactions gate. A SNAPSHOT of the conversation's durable set (see + * {@link CLIENT_SUPPORTS_META_KEY}), copied here at create time. + */ supports?: string[]; } @@ -230,7 +245,12 @@ export class PostgresStore implements SessionStore, AdminStore { } } + // `undefined` means the frame OMITTED `supports`, and only then does a resume + // inherit the set the conversation last declared. A declared list — `[]` + // included — wins and replaces it, which is how a text-only channel resuming a + // rich conversation opts out for good (th-13df6d). const convId = resumeId ?? randomUUID(); + const effectiveSupports = supports ?? (resumeId ? await this.conversationSupports(resumeId) : []); const session: StoredSession = { sessionId: randomUUID(), conversationId: convId, @@ -247,7 +267,7 @@ export class PostgresStore implements SessionStore, AdminStore { // The caller's email doubles as the OTP delivery contact. ...(owner ? { contactEmail: owner } : {}), // The declared render capabilities gate this session's Rich Interactions. - ...(supports && supports.length > 0 ? { supports } : {}), + ...(effectiveSupports.length > 0 ? { supports: effectiveSupports } : {}), }; const now = new Date().toISOString(); @@ -277,9 +297,27 @@ export class PostgresStore implements SessionStore, AdminStore { [session.agentParticipantId, convId, orgId, AGENT_NAME, now], ); } + // Only a frame that DECLARED rewrites the durable record; an omitting resume + // read it above and must leave it exactly as it was. `jsonb_set` rather than a + // read-modify-write so sibling metadata keys (the Rust server's workflow step + // pointer, the caller's own `metadata`) survive without a lost-update race. + // + // Deliberately does NOT touch `updated_at`: that column is the sidebar's + // recency sort and is bumped when a MESSAGE lands (see appendMessage). A bare + // reconnect appends nothing, so bumping it here would float every backgrounded + // tab to the top of the list. The Go, Python and Rust stores leave it alone for + // the same reason — this is a parity-relevant choice, not an omission. + if (supports !== undefined) { + await client.query( + `UPDATE conversations + SET metadata_json = jsonb_set(COALESCE(metadata_json, '{}'::jsonb), $2, $3::jsonb, true) + WHERE id = $1`, + [convId, `{${CLIENT_SUPPORTS_META_KEY}}`, JSON.stringify(effectiveSupports)], + ); + } const metadata: SessionMetadata = { ...(owner ? { contactEmail: owner } : {}), - ...(supports && supports.length > 0 ? { supports } : {}), + ...(effectiveSupports.length > 0 ? { supports: effectiveSupports } : {}), }; await client.query( `INSERT INTO conversation_sessions @@ -298,6 +336,17 @@ export class PostgresStore implements SessionStore, AdminStore { return session; } + /** + * The render capabilities a conversation last declared, off its durable metadata. + * Missing / malformed → empty, i.e. exactly the text-only behavior (every + * interaction kind degrades to its conversational fallback). + */ + private async conversationSupports(conversationId: string): Promise { + const { rows } = await this.pool.query(`SELECT COALESCE(metadata_json, '{}'::jsonb) AS metadata_json FROM conversations WHERE id = $1`, [conversationId]); + const declared = (rows[0]?.metadata_json as Record | undefined)?.[CLIENT_SUPPORTS_META_KEY]; + return Array.isArray(declared) ? declared.filter((c): c is string => typeof c === 'string') : []; + } + /** * The session for `sessionId`, or null. The raw lookup primitive: ownership is * REPORTED (`userEmail`) but not enforced here, matching the in-memory store — the diff --git a/typescript/server/src/sessionStore.ts b/typescript/server/src/sessionStore.ts index 69fa845d..c98e448e 100644 --- a/typescript/server/src/sessionStore.ts +++ b/typescript/server/src/sessionStore.ts @@ -84,12 +84,17 @@ export interface StoredSession { */ otpVerified?: boolean; /** - * The client render capabilities this session declared at create-session - * (`supports`) — the per-kind gate for Rich Interactions. A kind whose - * `capability` is present here parks the turn on a rich card - * (`interaction_required`); anything else degrades to the kind's conversational - * fallback. `undefined`/empty → a text-only channel (every kind falls back). The - * TS analog of the Rust reference server's `session_capabilities`. + * The client render capabilities in effect for this session (`supports`) — the + * per-kind gate for Rich Interactions. A kind whose `capability` is present here + * parks the turn on a rich card (`interaction_required`); anything else degrades + * to the kind's conversational fallback. `undefined`/empty → a text-only channel + * (every kind falls back). The TS analog of the Rust reference server's + * `session_capabilities`. + * + * A SNAPSHOT, not the source of truth: the durable set lives on the CONVERSATION + * (see {@link SessionStore.createSession}) and is copied here at create time. A + * reconnect mints a new session id, so a value that lived only here was lost + * every time the socket dropped (th-13df6d). */ supports?: string[]; } @@ -134,6 +139,24 @@ export interface SessionStore { * so subsequent turns append and history replays). An absent or unknown id mints * a fresh conversation (unchanged behavior). */ + /** + * `supports` — the client's declared render capabilities (see + * {@link StoredSession.supports}). Implementations MUST persist it per + * CONVERSATION, not per session: + * + * - `undefined` (the frame OMITTED the key) on a resume → INHERIT the set the + * conversation last declared. A reconnect is a resume, and re-declaring on + * every reconnect is not something clients do, so anything session-scoped + * silently degraded every Rich Interaction to its text fallback (th-13df6d). + * - a declared list — INCLUDING an explicit `[]` — always wins and REPLACES the + * stored set, so a text-only channel resuming a rich conversation opts out for + * good rather than having the old capabilities resurrected by the next + * reconnect that omits the key. + * - `undefined` on a FRESH conversation → empty (unchanged behavior). + * + * The distinction between "key omitted" and "explicit `[]`" is load-bearing; + * `undefined` vs `[]` is how it crosses this interface. Do not collapse them. + */ createSession(agentId: string, userName?: string, userEmail?: string, conversationId?: string, orgId?: string, supports?: string[]): Promise; getSession(sessionId: string): Promise; /** @@ -214,11 +237,25 @@ export class InMemorySessionStore implements SessionStore { */ private readonly convOrg = new Map(); + /** + * conversation id → the render capabilities (`supports`) the conversation last + * declared. Conversation-scoped for the same reason the owner and org above are: + * a reconnect mints a NEW session, so a set kept only on the session was lost on + * every network blip / backgrounding / deploy and Rich Interactions went dark with + * nothing on the wire to notice (th-13df6d). Unlike those two this IS rewritten on + * a resume that declares — that is the text-only opt-out. + */ + private readonly convSupports = new Map(); + async createSession(agentId: string, _userName?: string, userEmail?: string, conversationId?: string, orgId?: string, supports?: string[]): Promise { // Resume: bind to an existing conversation (reuse its id + persisted log) when // the caller passes a known conversationId. Unknown/absent → mint a fresh one. const resume = conversationId && this.messages.has(conversationId); const convId = resume ? conversationId : randomUUID(); + // `undefined` means the frame OMITTED `supports`, and only then does a resume + // inherit what the conversation last declared. A declared list — `[]` included — + // wins, which is how a text-only client opts a rich conversation out. + const effectiveSupports = supports ?? (resume ? (this.convSupports.get(convId) ?? []) : []); const session: StoredSession = { sessionId: randomUUID(), conversationId: convId, @@ -236,10 +273,14 @@ export class InMemorySessionStore implements SessionStore { // auth-gate flow (mirrors the Rust reference capturing contactEmail). ...(userEmail ? { contactEmail: userEmail } : {}), // The declared render capabilities gate this session's Rich Interactions. - // Empty/absent ⇒ a text-only channel (every kind falls back). - ...(supports && supports.length > 0 ? { supports } : {}), + // Empty ⇒ a text-only channel (every kind falls back). + ...(effectiveSupports.length > 0 ? { supports: effectiveSupports } : {}), }; this.sessions.set(session.sessionId, session); + // Durable per conversation, so the NEXT reconnect — which will omit the key — + // inherits it. A declared list (including `[]`) overwrites, so the opt-out is + // itself durable; a fresh conversation records whatever it was created with. + this.convSupports.set(convId, effectiveSupports); // Only initialize the message log on a fresh conversation — a resume keeps its history. if (!resume) { this.messages.set(convId, []); diff --git a/typescript/server/test/postgres-store.test.ts b/typescript/server/test/postgres-store.test.ts index 91298b92..d7003c26 100644 --- a/typescript/server/test/postgres-store.test.ts +++ b/typescript/server/test/postgres-store.test.ts @@ -519,6 +519,53 @@ describe('PostgresStore (needs Docker)', () => { } }); + // ── render capabilities survive a reconnect (th-13df6d) ───────────────── + // + // A reconnect is a resume, and it mints a NEW session id — so `supports` kept on + // the session row was lost on every network blip and Rich Interactions silently + // degraded to their text fallback. The durable set rides conversation metadata, + // which for THIS store means a `jsonb_set` on `conversations.metadata_json` — a + // statement that fails by quietly writing nothing, so it needs a real database. + // The in-memory sibling of this contract is `test/supports-reconnect.test.ts`. + pgIt('a resume inherits the conversation’s declared capabilities, and an explicit [] replaces them', async () => { + const store = await newStore(); + const orgId = org(); + try { + const first = await store.createSession('agent', 'Alice', 'alice@example.test', undefined, orgId, ['identity_form']); + expect(first.supports).toEqual(['identity_form']); + + // Reconnect: same conversation, `supports` OMITTED (undefined ≠ []). + const resumed = await store.createSession('agent', 'Alice', 'alice@example.test', first.conversationId, orgId); + expect(resumed.conversationId).toBe(first.conversationId); + expect(resumed.sessionId).not.toBe(first.sessionId); + expect(resumed.supports, 'an omitting reconnect inherits the conversation’s set').toEqual(['identity_form']); + // …and through a round trip, not just in the returned object. + expect((await store.getSession(resumed.sessionId))?.supports).toEqual(['identity_form']); + + // Sibling metadata keys must survive the write — the Rust server shares this + // table and keeps its workflow step pointer in the same JSONB column. + const pool = new Pool({ connectionString }); + try { + await pool.query(`UPDATE conversations SET metadata_json = metadata_json || '{"workflowCurrentStepId":"step-2"}'::jsonb WHERE id = $1`, [ + first.conversationId, + ]); + await store.createSession('agent', 'Alice', 'alice@example.test', first.conversationId, orgId, ['choice_chips']); + const row = await pool.query('SELECT metadata_json FROM conversations WHERE id = $1', [first.conversationId]); + expect(row.rows[0].metadata_json).toEqual({ workflowCurrentStepId: 'step-2', clientSupports: ['choice_chips'] }); + } finally { + await pool.end(); + } + + // An explicit [] is the text-only opt-out, and it is itself durable: the next + // reconnect that omits the key must not resurrect the old capabilities. + const textOnly = await store.createSession('agent', 'Alice', 'alice@example.test', first.conversationId, orgId, []); + expect(textOnly.supports).toBeUndefined(); + const afterOptOut = await store.createSession('agent', 'Alice', 'alice@example.test', first.conversationId, orgId); + expect(afterOptOut.supports, 'the text-only declaration replaced the durable record').toBeUndefined(); + } finally { + await store.close(); + } + }); }); describe('memory stays the default', () => { diff --git a/typescript/server/test/supports-reconnect.test.ts b/typescript/server/test/supports-reconnect.test.ts new file mode 100644 index 00000000..8fdbe791 --- /dev/null +++ b/typescript/server/test/supports-reconnect.test.ts @@ -0,0 +1,88 @@ +/** + * `supports` — the client render-capability list that gates the ENTIRE Rich + * Interactions framework — must survive a reconnect. + * + * A reconnect IS a resume: the client re-opens the socket and re-issues + * `create_conversation_session` with the same `conversationId`, which mints a NEW + * session id on a NEW dispatcher. So while the declared list lived only on the + * session record, the server forgot it could render cards unless the client + * re-declared on every single reconnect — and every interaction kind quietly fell + * back to conversational collection with no error, no event, nothing on the wire to + * notice. Reconnects are routine (network blips, mobile backgrounding, deploys), so + * a shipped feature was degrading in the field with no signal (th-13df6d). + * + * The list now rides the CONVERSATION, and a resume that OMITS the key inherits it. + * A frame that DOES declare — `[]` included — replaces the stored set, which is the + * text-only opt-out and is itself durable. + * + * Two dispatchers over one store is the point: a single dispatcher would pass even + * with the state kept per connection. + */ +import { describe, expect, it } from 'vitest'; + +import { MockLlmProvider } from '@smooai/smooth-operator-core'; + +import { FrameDispatcher } from '../src/frameDispatcher.js'; +import type { Frame } from '../src/protocol.js'; +import { InMemorySessionStore } from '../src/sessionStore.js'; + +const AGENT = '11111111-1111-1111-1111-111111111111'; + +describe('declared render capabilities survive a reconnect', () => { + /** A FRESH dispatcher over the shared store — i.e. a new WebSocket connection. */ + function reconnect(store: InMemorySessionStore) { + const dispatcher = new FrameDispatcher({ store, chatClient: new MockLlmProvider() }); + return async (frame: Record): Promise<{ sessionId: string; conversationId: string }> => { + const sink: Frame[] = []; + await dispatcher.dispatch(JSON.stringify({ type: 'action', action: 'create_conversation_session', agentId: AGENT, ...frame }), (f) => sink.push(f)); + const data = sink[0]?.data as { sessionId?: string; conversationId?: string } | undefined; + expect(data?.sessionId, `create_conversation_session failed: ${JSON.stringify(sink[0])}`).toBeTruthy(); + return { sessionId: data!.sessionId!, conversationId: data!.conversationId! }; + }; + } + + /** What the turn reads to decide rich-card vs conversational fallback. */ + async function capabilities(store: InMemorySessionStore, sessionId: string): Promise { + return (await store.getSession(sessionId))?.supports ?? []; + } + + it('inherits an omitted list on reconnect, and an explicit [] opts out for good', async () => { + const store = new InMemorySessionStore(); + + // Connection 1 declares the capability. + const first = await reconnect(store)({ requestId: 'req-conn-1', supports: ['identity_form'] }); + expect(await capabilities(store, first.sessionId)).toContain('identity_form'); + + // Connection 2 — a reconnect: same conversation, `supports` OMITTED, which is + // exactly what a widget resuming from its stored conversationId sends. This is + // where the feature went dark. + const resumed = await reconnect(store)({ requestId: 'req-conn-2', conversationId: first.conversationId }); + expect(resumed.conversationId, 'the reconnect resumed the same conversation').toBe(first.conversationId); + expect(resumed.sessionId, 'a reconnect mints a NEW session id — that is the whole problem').not.toBe(first.sessionId); + expect(await capabilities(store, resumed.sessionId), "a reconnect that omits 'supports' inherits the conversation's declared capabilities").toContain( + 'identity_form', + ); + + // Connection 3 DECLARES `[]` — a text-only channel (SMS, voice) resuming a rich + // conversation opts out rather than being handed cards it cannot render. + const textOnly = await reconnect(store)({ requestId: 'req-conn-3', conversationId: first.conversationId, supports: [] }); + expect(await capabilities(store, textOnly.sessionId), "an explicit empty 'supports' declares text-only and never inherits").toEqual([]); + + // Connection 4 omits again: the opt-out must be durable, not a one-session + // exception that the next reconnect resurrects from a stale record. + const afterOptOut = await reconnect(store)({ requestId: 'req-conn-4', conversationId: first.conversationId }); + expect(await capabilities(store, afterOptOut.sessionId), 'the text-only declaration replaced the durable record').toEqual([]); + }); + + it('leaves a fresh conversation that declares nothing text-only', async () => { + // The inherit rule keys on a RESUME; a brand-new conversation with no `supports` + // is unchanged behavior and must not pick up a neighbour's capabilities. + const store = new InMemorySessionStore(); + const rich = await reconnect(store)({ requestId: 'r1', supports: ['identity_form'] }); + expect(await capabilities(store, rich.sessionId)).toContain('identity_form'); + + const fresh = await reconnect(store)({ requestId: 'r2' }); + expect(fresh.conversationId).not.toBe(rich.conversationId); + expect(await capabilities(store, fresh.sessionId)).toEqual([]); + }); +}); diff --git a/typescript/src/generated/types.ts b/typescript/src/generated/types.ts index 1f840e74..ffce599d 100644 --- a/typescript/src/generated/types.ts +++ b/typescript/src/generated/types.ts @@ -86,7 +86,7 @@ export interface CreateConversationSessionRequest { */ browserFingerprint?: string; /** - * Client render capabilities for this session — a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` → capability `identity_form`, kind `choices` → capability `choice_chips`); future kinds add their own values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) omit this and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). + * Client render capabilities for this session — a per-kind list gating the Rich Interactions the server may emit mid-turn (`interaction_required`). Each interaction kind declares the capability that gates it (e.g. kind `identity_intake` → capability `identity_form`, kind `choices` → capability `choice_chips`); future kinds add their own values (`date_picker`, `file_upload`, …). Text-only channels (SMS, voice) declare `[]` and the server degrades each kind to its conversational fallback. Unknown values are ignored (forward-compatible). Durability: the declared list is persisted on the CONVERSATION, so a reconnect that resumes an existing `conversationId` and OMITS this key inherits the set the conversation last declared — a reconnect is not a downgrade to text-only. Any list the frame does declare (including `[]`) replaces the inherited one, so a text-only client resuming a rich conversation opts out explicitly. */ supports?: string[]; /**