diff --git a/Directory.Build.props b/Directory.Build.props index cc50e0f..aa94947 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -9,7 +9,7 @@ -0.13.0 +0.13.1 diff --git a/design/client-rendering-capabilities.md b/design/client-rendering-capabilities.md index b795d5f..ce5f96b 100644 --- a/design/client-rendering-capabilities.md +++ b/design/client-rendering-capabilities.md @@ -356,14 +356,20 @@ These are deliberately out of scope for v1: shared-PVC **path references** (`{ mime, path, fileName? }`), reusing the MCP-attachment scheme — bytes never ride the bus. Producing side: an `attach_image` LLM tool (`AttachmentReplyTools`) validates a model-named file under the shared attachments dir and - stages it in a session-keyed `ReplyAttachmentBuffer`; `UserMessageHandler` drains the buffer - onto the final reply (the **user-message path** only in v1). Rendering side: Blazor + stages it in a **per-turn-keyed** `ReplyAttachmentBuffer` (`session → turn`, issue #482); + `UserMessageHandler` drains it via the shared `DrainForFinalReply` seam onto the final reply + (the **user-message path** only in v1). The buffer is hardened for the deferred multi-producer + paths below: per-turn keying means concurrent producers for one primary session each drain only + their own stage rather than scooping every staged attachment; staged turns are TTL-swept + (default 30 min) and cleared on turn cancel (`UserMessageHandler` finally blocks) and on + `ClearContext` (session-level `Clear`). Rendering side: Blazor co-mounts the shared PVC read-only and serves bytes from a minimal `/attachments` endpoint, rendering images as native `` **outside** the markdown sanitizer (which strips `` and `data:` URLs); the CLI prints a placeholder line per attachment. - Still deferred: scheduled-task / subagent / A2A producing paths (the buffer generalizes - trivially), SVG→PNG rasterization, chat-platform proxies (none exist yet), and attachments in - conversation-history replay (replies are ephemeral; history stores text). + Still deferred: wiring attachments into the scheduled-task / subagent / A2A producing paths + (the buffer is now keyed and swept to make that safe), SVG→PNG rasterization, chat-platform + proxies (none exist yet), and attachments in conversation-history replay (replies are + ephemeral; history stores text). - **Platform-native UI emission** — `DiscordEmbed` / `SlackBlockKit` / `TeamsAdaptiveCard` bits are reserved in the enum but no tooling produces them. Adding them is opportunistic upgrade work in each proxy (e.g., a diff --git a/src/RockBot.Agent/AttachmentReplyTools.cs b/src/RockBot.Agent/AttachmentReplyTools.cs index 1dc4fd0..6de9baa 100644 --- a/src/RockBot.Agent/AttachmentReplyTools.cs +++ b/src/RockBot.Agent/AttachmentReplyTools.cs @@ -13,24 +13,27 @@ namespace RockBot.Agent; /// tool — so the agent never handles bytes itself (the "Nothing trusts the LLM" rule). The /// tool validates containment and existence, infers the MIME type from the extension when not /// given, and stages an in the -/// keyed to this session. The reply-publishing path drains the buffer onto the final reply. +/// keyed to this (session, turn). The reply-publishing path drains the buffer onto the final reply. /// public sealed class AttachmentReplyTools { private readonly IAttachmentStorage _storage; private readonly ReplyAttachmentBuffer _buffer; private readonly string _sessionId; + private readonly string _turnId; private readonly ILogger _logger; public AttachmentReplyTools( IAttachmentStorage storage, ReplyAttachmentBuffer buffer, string sessionId, + string turnId, ILogger logger) { _storage = storage; _buffer = buffer; _sessionId = sessionId; + _turnId = turnId; _logger = logger; Tools = [AIFunctionFactory.Create(AttachImage)]; @@ -74,7 +77,7 @@ public string AttachImage( var leaf = Path.GetFileName(path); var displayName = string.IsNullOrWhiteSpace(fileName) ? leaf : fileName.Trim(); - _buffer.Add(_sessionId, new AgentAttachment + _buffer.Add(_sessionId, _turnId, new AgentAttachment { Mime = resolvedMime, Path = leaf, diff --git a/src/RockBot.Agent/ClearContextHandler.cs b/src/RockBot.Agent/ClearContextHandler.cs index 7932498..f6a97d1 100644 --- a/src/RockBot.Agent/ClearContextHandler.cs +++ b/src/RockBot.Agent/ClearContextHandler.cs @@ -13,6 +13,7 @@ internal sealed class ClearContextHandler( IConversationMemory conversationMemory, ISessionTracker sessionTracker, SessionClientCapabilityStore clientCapabilityStore, + ReplyAttachmentBuffer attachmentBuffer, SessionOriginStore originStore, IMessagePublisher publisher, AgentIdentity agent, @@ -33,6 +34,10 @@ public async Task HandleAsync(ClearContextRequest message, MessageHandlerContext // them from scratch — important if the user returns on a different client. clientCapabilityStore.Clear(message.SessionId); + // Drop any attachments staged but not yet drained for this session, so a cleared context + // can't replay an old turn's images onto a future reply. + attachmentBuffer.Clear(message.SessionId); + // Drop the cached origin so the next turn re-anchors fresh background work. originStore.Clear(message.SessionId); diff --git a/src/RockBot.Agent/UserMessageHandler.cs b/src/RockBot.Agent/UserMessageHandler.cs index e94bb03..f76add3 100644 --- a/src/RockBot.Agent/UserMessageHandler.cs +++ b/src/RockBot.Agent/UserMessageHandler.cs @@ -186,9 +186,9 @@ await conversationMemory.AddTurnAsync( var sessionNamespace = $"session/{message.SessionId}"; var sessionWorkingMemoryTools = new WorkingMemoryTools(workingMemory, sessionNamespace, logger); - // Per-session attachment tool — attach_image stages files for this session's final reply. + // Per-turn attachment tool — attach_image stages files for this turn's final reply. var attachmentReplyTools = new AttachmentReplyTools( - attachmentStorage, attachmentBuffer, message.SessionId, logger); + attachmentStorage, attachmentBuffer, message.SessionId, turnId, logger); // Per-session skill tools with usage tracking var sessionSkillTools = new SkillTools(skillStore, llmClient, logger, message.SessionId, skillUsageStore); @@ -234,11 +234,11 @@ await conversationMemory.AddTurnAsync( // This prevents subagent re-spawn on pod restart (issue #122). logger.LogInformation("Native path: launching background LLM loop for session {SessionId}", message.SessionId); await PublishReplyAsync("I'm working on that — I'll follow up shortly.", - replyTo, correlationId, message.SessionId, isFinal: false, ct); + replyTo, correlationId, message.SessionId, turnId, isFinal: false, ct); turnActivityHandedOff = true; context.Items[WipConstants.DeferredKey] = true; _ = NativeLlmLoopAsync(chatMessages, chatOptions, classification, postInjectionTokenEstimate, - message.SessionId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); + message.SessionId, turnId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); } else { @@ -280,13 +280,13 @@ await PublishReplyAsync("I'm working on that — I'll follow up shortly.", "Tool calls detected on iteration 1; sending ack ({AckLen} chars) and continuing in background", effectiveAck.Length); - await PublishReplyAsync(effectiveAck, replyTo, correlationId, message.SessionId, isFinal: false, ct); + await PublishReplyAsync(effectiveAck, replyTo, correlationId, message.SessionId, turnId, isFinal: false, ct); turnActivityHandedOff = true; context.Items[WipConstants.DeferredKey] = true; _ = BackgroundToolLoopAsync( chatMessages, chatOptions, firstResponse, classification, postInjectionTokenEstimate, - message.SessionId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); + message.SessionId, turnId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); } else { @@ -300,13 +300,13 @@ await PublishReplyAsync("I'm working on that — I'll follow up shortly.", await PublishReplyAsync( "I'm working on that — I'll follow up shortly.", - replyTo, correlationId, message.SessionId, isFinal: false, ct); + replyTo, correlationId, message.SessionId, turnId, isFinal: false, ct); turnActivityHandedOff = true; context.Items[WipConstants.DeferredKey] = true; _ = BackgroundToolLoopAsync( chatMessages, chatOptions, firstResponse, classification, postInjectionTokenEstimate, - message.SessionId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); + message.SessionId, turnId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); } else if (modelBehavior.NudgeOnHallucinatedToolCalls && (HallucinatedActionRegex.IsMatch(text) || AgentLoopRunner.CapabilityDenialRegex.IsMatch(text))) @@ -317,13 +317,13 @@ await PublishReplyAsync( await PublishReplyAsync( "I'm working on that — I'll follow up shortly.", - replyTo, correlationId, message.SessionId, isFinal: false, ct); + replyTo, correlationId, message.SessionId, turnId, isFinal: false, ct); turnActivityHandedOff = true; context.Items[WipConstants.DeferredKey] = true; _ = BackgroundToolLoopAsync( chatMessages, chatOptions, firstResponse, classification, postInjectionTokenEstimate, - message.SessionId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); + message.SessionId, turnId, replyTo, correlationId, sessionHandle.Generation, wipMessageId, turnActivity, sessionCt); } else { @@ -351,7 +351,7 @@ await conversationMemory.AddTurnAsync( { AgentName = agent.Name }, ct); - await PublishReplyAsync(text, replyTo, correlationId, message.SessionId, isFinal: true, ct); + await PublishReplyAsync(text, replyTo, correlationId, message.SessionId, turnId, isFinal: true, ct); sessionTracker.EndSession(message.SessionId, sessionHandle.Generation); turnActivity?.SetTag("rockbot.turn.status", "ok"); turnActivity?.SetStatus(ActivityStatusCode.Ok); @@ -393,7 +393,7 @@ await conversationMemory.AddTurnAsync( message.SessionId); } - await PublishReplyAsync(errorText, replyTo, correlationId, message.SessionId, isFinal: true, ct); + await PublishReplyAsync(errorText, replyTo, correlationId, message.SessionId, turnId, isFinal: true, ct); sessionTracker.EndSession(message.SessionId, sessionHandle.Generation); turnActivity?.SetTag("rockbot.turn.status", "error"); turnActivity?.SetStatus(ActivityStatusCode.Error, ex.Message); @@ -408,7 +408,14 @@ await conversationMemory.AddTurnAsync( // Dispose the turn span only when this method owns it (sync paths). // Background paths pass it to the background task which disposes it. if (!turnActivityHandedOff) + { + // Clear any attachments staged this turn that were never drained onto a final + // reply (e.g. the turn threw before publishing). Idempotent: a no-op after a + // successful final drain. Skipped when handed off — the background method owns + // the turn's stage and clears it in its own finally. + attachmentBuffer.Clear(message.SessionId, turnId); turnActivity?.Dispose(); + } } } @@ -418,6 +425,7 @@ private async Task NativeLlmLoopAsync( TierClassification classification, int? postInjectionTokenEstimate, string sessionId, + string turnId, string replyTo, string? correlationId, long sessionGeneration, @@ -434,7 +442,7 @@ private async Task NativeLlmLoopAsync( void OnFallback(string from, string to, string reason) => _ = PublishReplyAsync( $"Switching models ({reason}) — retrying with {to}…", - replyTo, correlationId, sessionId, isFinal: false, ct); + replyTo, correlationId, sessionId, turnId, isFinal: false, ct); if (fallbackClient is not null) fallbackClient.OnFallback += OnFallback; try @@ -460,26 +468,26 @@ void OnFallback(string from, string to, string reason) => diagnostics: nativeDiag, onPreToolCall: async (desc, ct2) => { - await PublishReplyAsync($"Working on it — checking {desc}…", replyTo, correlationId, sessionId, isFinal: false, ct2); + await PublishReplyAsync($"Working on it — checking {desc}…", replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, onProgress: async (msg, ct2) => { if (DateTimeOffset.UtcNow - lastProgressAt < ProgressMessageThreshold) return; - await PublishReplyAsync(msg, replyTo, correlationId, sessionId, isFinal: false, ct2); + await PublishReplyAsync(msg, replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, onToolTimeout: async (desc, ct2) => { await PublishReplyAsync( $"The {desc} service is taking too long to respond — trying a different approach…", - replyTo, correlationId, sessionId, isFinal: false, ct2); + replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, onStageProgress: async (stage, ct2) => { - await PublishReplyAsync(stage, replyTo, correlationId, sessionId, isFinal: false, ct2); + await PublishReplyAsync(stage, replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, cancellationToken: ct); @@ -517,7 +525,7 @@ await conversationMemory.AddTurnAsync( // and resolves the user-proxy SendAsync TCS. If the loop spawned consolidating // subagents, their Phase 2 synthesis will arrive later as a separate // unsolicited final bubble — that's the consolidated answer. - await PublishReplyAsync(text, replyTo, correlationId, sessionId, isFinal: true, ct); + await PublishReplyAsync(text, replyTo, correlationId, sessionId, turnId, isFinal: true, ct); loopSw.Stop(); turnActivity?.SetTag("rockbot.turn.status", "ok"); turnActivity?.SetStatus(ActivityStatusCode.Ok); @@ -535,7 +543,7 @@ await conversationMemory.AddTurnAsync( await PublishReplyAsync( $"Sorry, I ran into an error while working on your request: {ex.Message}", - replyTo, correlationId, sessionId, isFinal: true, ct); + replyTo, correlationId, sessionId, turnId, isFinal: true, ct); loopSw.Stop(); turnActivity?.SetTag("rockbot.turn.status", "error"); turnActivity?.SetStatus(ActivityStatusCode.Error, ex.Message); @@ -547,6 +555,10 @@ await PublishReplyAsync( finally { if (fallbackClient is not null) fallbackClient.OnFallback -= OnFallback; + // Clear any attachments staged this turn that were never drained — on a clean final + // reply this is a no-op; on cancellation (user sent a new message mid-loop) it removes + // the orphaned stage so it can't land on a later turn's reply. + attachmentBuffer.Clear(sessionId, turnId); sessionTracker.EndSession(sessionId, sessionGeneration); if (wipMessageId is not null) await wipTracker.CompleteAsync(wipMessageId, CancellationToken.None); @@ -561,6 +573,7 @@ private async Task BackgroundToolLoopAsync( TierClassification classification, int? postInjectionTokenEstimate, string sessionId, + string turnId, string replyTo, string? correlationId, long sessionGeneration, @@ -579,7 +592,7 @@ private async Task BackgroundToolLoopAsync( void OnFallback(string from, string to, string reason) => _ = PublishReplyAsync( $"Switching models ({reason}) — retrying with {to}…", - replyTo, correlationId, sessionId, isFinal: false, ct); + replyTo, correlationId, sessionId, turnId, isFinal: false, ct); if (fallbackClient is not null) fallbackClient.OnFallback += OnFallback; try @@ -607,26 +620,26 @@ void OnFallback(string from, string to, string reason) => diagnostics: bgDiag, onPreToolCall: async (desc, ct2) => { - await PublishReplyAsync($"Working on it — checking {desc}…", replyTo, correlationId, sessionId, isFinal: false, ct2); + await PublishReplyAsync($"Working on it — checking {desc}…", replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, onProgress: async (msg, ct2) => { if (DateTimeOffset.UtcNow - lastProgressAt < ProgressMessageThreshold) return; - await PublishReplyAsync(msg, replyTo, correlationId, sessionId, isFinal: false, ct2); + await PublishReplyAsync(msg, replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, onToolTimeout: async (desc, ct2) => { await PublishReplyAsync( $"The {desc} service is taking too long to respond — trying a different approach…", - replyTo, correlationId, sessionId, isFinal: false, ct2); + replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, onStageProgress: async (stage, ct2) => { - await PublishReplyAsync(stage, replyTo, correlationId, sessionId, isFinal: false, ct2); + await PublishReplyAsync(stage, replyTo, correlationId, sessionId, turnId, isFinal: false, ct2); lastProgressAt = DateTimeOffset.UtcNow; }, cancellationToken: ct); @@ -657,7 +670,7 @@ await conversationMemory.AddTurnAsync( { AgentName = agent.Name }, ct); - await PublishReplyAsync(finalContent, replyTo, correlationId, sessionId, isFinal: true, ct); + await PublishReplyAsync(finalContent, replyTo, correlationId, sessionId, turnId, isFinal: true, ct); loopSw.Stop(); turnActivity?.SetTag("rockbot.turn.status", "ok"); turnActivity?.SetStatus(ActivityStatusCode.Ok); @@ -675,7 +688,7 @@ await conversationMemory.AddTurnAsync( await PublishReplyAsync( $"Sorry, I ran into an error while working on your request: {ex.Message}", - replyTo, correlationId, sessionId, isFinal: true, ct); + replyTo, correlationId, sessionId, turnId, isFinal: true, ct); loopSw.Stop(); turnActivity?.SetTag("rockbot.turn.status", "error"); turnActivity?.SetStatus(ActivityStatusCode.Error, ex.Message); @@ -687,6 +700,10 @@ await PublishReplyAsync( finally { if (fallbackClient is not null) fallbackClient.OnFallback -= OnFallback; + // Clear any attachments staged this turn that were never drained — on a clean final + // reply this is a no-op; on cancellation (user sent a new message mid-loop) it removes + // the orphaned stage so it can't land on a later turn's reply. + attachmentBuffer.Clear(sessionId, turnId); sessionTracker.EndSession(sessionId, sessionGeneration); if (wipMessageId is not null) await wipTracker.CompleteAsync(wipMessageId, CancellationToken.None); @@ -696,18 +713,12 @@ await PublishReplyAsync( private async Task PublishReplyAsync( string content, string replyTo, string? correlationId, - string sessionId, bool isFinal, CancellationToken ct) + string sessionId, string turnId, bool isFinal, CancellationToken ct) { - // Only final replies carry attachments — drain the per-session buffer so files staged - // by attach_image ride out with the answer and aren't replayed on a later turn. - // Progress/non-final replies never carry attachments. - IReadOnlyList? attachments = null; - if (isFinal) - { - var drained = attachmentBuffer.Drain(sessionId); - if (drained.Count > 0) - attachments = drained; - } + // Only final replies carry attachments — drain this turn's stage so files staged by + // attach_image ride out with the answer and aren't replayed on a later turn. The shared + // helper returns null for non-final replies and when nothing is staged. + var attachments = attachmentBuffer.DrainForFinalReply(sessionId, turnId, isFinal); var reply = new AgentReply { diff --git a/src/RockBot.Host/ReplyAttachmentBuffer.cs b/src/RockBot.Host/ReplyAttachmentBuffer.cs index f43f928..e9c1929 100644 --- a/src/RockBot.Host/ReplyAttachmentBuffer.cs +++ b/src/RockBot.Host/ReplyAttachmentBuffer.cs @@ -3,54 +3,155 @@ namespace RockBot.Host; /// -/// Session-keyed buffer of the agent has staged for the next -/// final reply. The attach_image LLM tool calls during a turn; the -/// reply-publishing path calls when it emits the final reply, moving the -/// staged attachments onto and clearing the buffer so -/// they aren't replayed on a later turn. +/// Per-turn buffer of the agent has staged for an upcoming final +/// reply. The attach_image LLM tool calls during a turn; the +/// reply-publishing path calls (or ) when it +/// emits the final reply, moving the staged attachments onto +/// and clearing the turn's stage so they aren't replayed on a later turn. /// -/// Mirrors in shape and lifetime: a process-wide -/// singleton holding short-lived per-session metadata, guarded by a single lock. Registered -/// as a singleton in the agent host. +/// Keyed two levels deep: session → turn. Per-turn keying is what makes the buffer safe +/// when more than one producer is in flight for the same primary session concurrently (e.g. an +/// A2A result and a subagent completion). A session-only key let whichever producer published +/// first scoop up every staged attachment and land them on the wrong bubble; per-turn +/// keying scopes each drain to its own producer. +/// +/// +/// Staged turns carry a stamp and expire after a TTL (default 30 +/// minutes — comfortably above any loop duration so legitimate long in-flight stages are never +/// swept). Expired turns are swept lazily at the top of every public method. A process-wide +/// singleton holding short-lived per-session metadata, guarded by a single lock; mirrors +/// in shape and lifetime. /// /// public sealed class ReplyAttachmentBuffer { - private readonly Dictionary> _bySession + /// Default time after which an un-drained staged turn is swept. + public static readonly TimeSpan DefaultTtl = TimeSpan.FromMinutes(30); + + private sealed class StagedTurn + { + public List Items { get; } = new(); + public DateTimeOffset LastStagedAt { get; set; } + } + + private readonly Dictionary> _bySession = new(StringComparer.OrdinalIgnoreCase); private readonly Lock _lock = new(); + private readonly TimeProvider _time; + private readonly TimeSpan _ttl; + + public ReplyAttachmentBuffer(TimeProvider? timeProvider = null, TimeSpan? ttl = null) + { + _time = timeProvider ?? TimeProvider.System; + _ttl = ttl ?? DefaultTtl; + } /// - /// Stage for 's next final reply. + /// Stage for the (, + /// ) turn's next final reply. Refreshes the turn's TTL stamp. /// - public void Add(string sessionId, AgentAttachment attachment) + public void Add(string sessionId, string turnId, AgentAttachment attachment) { lock (_lock) { - if (!_bySession.TryGetValue(sessionId, out var list)) - _bySession[sessionId] = list = new List(); - list.Add(attachment); + Sweep(); + if (!_bySession.TryGetValue(sessionId, out var turns)) + _bySession[sessionId] = turns = new Dictionary(StringComparer.OrdinalIgnoreCase); + if (!turns.TryGetValue(turnId, out var staged)) + turns[turnId] = staged = new StagedTurn(); + staged.Items.Add(attachment); + staged.LastStagedAt = _time.GetUtcNow(); } } /// - /// Returns and clears the attachments staged for . Returns an - /// empty list when nothing is staged. Drains so a later reply in the same session starts clean. + /// Returns and clears the attachments staged for the (, + /// ) turn. Returns an empty list when nothing is staged. Drains so a + /// later reply in the same turn starts clean. /// - public IReadOnlyList Drain(string sessionId) + public IReadOnlyList Drain(string sessionId, string turnId) { lock (_lock) { - if (_bySession.Remove(sessionId, out var list)) - return list; + Sweep(); + if (_bySession.TryGetValue(sessionId, out var turns) + && turns.Remove(turnId, out var staged)) + { + if (turns.Count == 0) + _bySession.Remove(sessionId); + return staged.Items; + } return []; } } - /// Drop any staged attachments for without returning them. + /// + /// The shared drain seam every reply-publish site calls. Returns null when + /// is false (non-final replies never carry attachments) or when + /// nothing is staged for the turn; otherwise returns the drained list. + /// + public IReadOnlyList? DrainForFinalReply(string sessionId, string turnId, bool isFinal) + { + if (!isFinal) + return null; + var drained = Drain(sessionId, turnId); + return drained.Count > 0 ? drained : null; + } + + /// Drop all staged attachments for every turn of (ClearContext / session cancel). public void Clear(string sessionId) { lock (_lock) + { + Sweep(); _bySession.Remove(sessionId); + } + } + + /// Drop the staged attachments for one (, ) turn, leaving sibling turns intact (per-turn cancel cleanup). + public void Clear(string sessionId, string turnId) + { + lock (_lock) + { + Sweep(); + if (_bySession.TryGetValue(sessionId, out var turns) + && turns.Remove(turnId) + && turns.Count == 0) + { + _bySession.Remove(sessionId); + } + } + } + + /// + /// Removes staged turns whose last stage is older than the TTL, pruning emptied session + /// dictionaries. Caller must hold . Cardinality is tiny, so the cost is + /// negligible. + /// + private void Sweep() + { + var now = _time.GetUtcNow(); + List? emptySessions = null; + foreach (var (sessionId, turns) in _bySession) + { + List? expired = null; + foreach (var (turnId, staged) in turns) + { + if (now - staged.LastStagedAt > _ttl) + (expired ??= new List()).Add(turnId); + } + if (expired is not null) + { + foreach (var turnId in expired) + turns.Remove(turnId); + } + if (turns.Count == 0) + (emptySessions ??= new List()).Add(sessionId); + } + if (emptySessions is not null) + { + foreach (var sessionId in emptySessions) + _bySession.Remove(sessionId); + } } } diff --git a/src/RockBot.UserProxy.Cli/AttachmentPlaceholder.cs b/src/RockBot.UserProxy.Cli/AttachmentPlaceholder.cs index fffadad..8794078 100644 --- a/src/RockBot.UserProxy.Cli/AttachmentPlaceholder.cs +++ b/src/RockBot.UserProxy.Cli/AttachmentPlaceholder.cs @@ -4,15 +4,21 @@ namespace RockBot.UserProxy.Cli; /// Formats a one-line placeholder for an the CLI can't render /// inline. Shared by the plain and Spectre frontends so the wording stays consistent, e.g. /// [image: chart.png (image/png)] or [attachment: report.pdf (application/pdf)]. +/// Always surfaces the shared-volume — the file you can +/// actually inspect — and prepends the friendly when it +/// differs, e.g. [image: Q3 chart.png (chart.png, image/png)]. /// internal static class AttachmentPlaceholder { public static string Render(AgentAttachment attachment) { - var label = string.IsNullOrWhiteSpace(attachment.FileName) ? attachment.Path : attachment.FileName; var kind = attachment.Mime.StartsWith("image/", StringComparison.OrdinalIgnoreCase) ? "image" : "attachment"; - return $"[{kind}: {label} ({attachment.Mime})]"; + var hasFriendlyName = !string.IsNullOrWhiteSpace(attachment.FileName) + && !string.Equals(attachment.FileName, attachment.Path, StringComparison.Ordinal); + return hasFriendlyName + ? $"[{kind}: {attachment.FileName} ({attachment.Path}, {attachment.Mime})]" + : $"[{kind}: {attachment.Path} ({attachment.Mime})]"; } } diff --git a/src/RockBot.UserProxy.Cli/ChatCommand.cs b/src/RockBot.UserProxy.Cli/ChatCommand.cs index 51f8dcf..9cca685 100644 --- a/src/RockBot.UserProxy.Cli/ChatCommand.cs +++ b/src/RockBot.UserProxy.Cli/ChatCommand.cs @@ -77,6 +77,15 @@ private static async Task RunOneShotAsync(UserProxyService proxy, Settings } Console.Out.WriteLine(HtmlPlainTextRenderer.StripHtml(reply.Content)); + + // The one-shot form can't render binaries inline, so print a placeholder line per + // attachment — to stdout, alongside the reply content, matching PlainConsoleFrontend — + // so the caller knows an attachment was produced (and where). + if (reply.Attachments is { Count: > 0 }) + { + foreach (var att in reply.Attachments) + Console.Out.WriteLine(AttachmentPlaceholder.Render(att)); + } return 0; } diff --git a/tests/RockBot.Agent.Tests/AttachmentReplyToolsTests.cs b/tests/RockBot.Agent.Tests/AttachmentReplyToolsTests.cs index 196592b..2ab8402 100644 --- a/tests/RockBot.Agent.Tests/AttachmentReplyToolsTests.cs +++ b/tests/RockBot.Agent.Tests/AttachmentReplyToolsTests.cs @@ -18,7 +18,7 @@ public void Init() _baseDir = Path.Combine(Path.GetTempPath(), "rb-attach-tests", Guid.NewGuid().ToString("N")); _storage = new AttachmentStorage(_baseDir); // creates the directory _buffer = new ReplyAttachmentBuffer(); - _tools = new AttachmentReplyTools(_storage, _buffer, "session-1", NullLogger.Instance); + _tools = new AttachmentReplyTools(_storage, _buffer, "session-1", "turn-1", NullLogger.Instance); } [TestCleanup] @@ -38,7 +38,7 @@ public void AttachImage_RecordsIntoBuffer_ForExistingFile() var result = _tools.AttachImage("chart.png"); StringAssert.Contains(result, "Attached"); - var drained = _buffer.Drain("session-1"); + var drained = _buffer.Drain("session-1", "turn-1"); Assert.AreEqual(1, drained.Count); Assert.AreEqual("chart.png", drained[0].Path); Assert.AreEqual("image/png", drained[0].Mime); @@ -52,7 +52,7 @@ public void AttachImage_InfersMime_FromExtension() _tools.AttachImage("photo.jpeg"); - Assert.AreEqual("image/jpeg", _buffer.Drain("session-1")[0].Mime); + Assert.AreEqual("image/jpeg", _buffer.Drain("session-1", "turn-1")[0].Mime); } [TestMethod] @@ -62,7 +62,7 @@ public void AttachImage_UsesExplicitMime_WhenProvided() _tools.AttachImage("diagram.bin", mime: "image/svg+xml"); - Assert.AreEqual("image/svg+xml", _buffer.Drain("session-1")[0].Mime); + Assert.AreEqual("image/svg+xml", _buffer.Drain("session-1", "turn-1")[0].Mime); } [TestMethod] @@ -72,7 +72,7 @@ public void AttachImage_UsesFriendlyName_WhenProvided() _tools.AttachImage("chart.png", fileName: "Quarterly results"); - Assert.AreEqual("Quarterly results", _buffer.Drain("session-1")[0].FileName); + Assert.AreEqual("Quarterly results", _buffer.Drain("session-1", "turn-1")[0].FileName); } [TestMethod] @@ -81,7 +81,7 @@ public void AttachImage_Rejects_NonExistentFile() var result = _tools.AttachImage("missing.png"); StringAssert.Contains(result, "no such file"); - Assert.AreEqual(0, _buffer.Drain("session-1").Count); + Assert.AreEqual(0, _buffer.Drain("session-1", "turn-1").Count); } [TestMethod] @@ -90,7 +90,7 @@ public void AttachImage_Rejects_TraversalOutsideBase() var result = _tools.AttachImage("../escape.png"); StringAssert.Contains(result, "outside"); - Assert.AreEqual(0, _buffer.Drain("session-1").Count); + Assert.AreEqual(0, _buffer.Drain("session-1", "turn-1").Count); } [TestMethod] @@ -103,7 +103,7 @@ public void AttachImage_Rejects_AbsolutePathOutsideBase() var result = _tools.AttachImage(outside); StringAssert.Contains(result, "outside"); - Assert.AreEqual(0, _buffer.Drain("session-1").Count); + Assert.AreEqual(0, _buffer.Drain("session-1", "turn-1").Count); } finally { @@ -117,7 +117,7 @@ public void AttachImage_Rejects_EmptyPath() var result = _tools.AttachImage(""); StringAssert.Contains(result, "no path"); - Assert.AreEqual(0, _buffer.Drain("session-1").Count); + Assert.AreEqual(0, _buffer.Drain("session-1", "turn-1").Count); } [TestMethod] @@ -128,14 +128,14 @@ public void AttachImage_StripsRedundantAttachmentsLeaf() var dir = Path.Combine(Path.GetTempPath(), "rb-attach-leaf", Guid.NewGuid().ToString("N"), "attachments"); var storage = new AttachmentStorage(dir); var buffer = new ReplyAttachmentBuffer(); - var tools = new AttachmentReplyTools(storage, buffer, "s", NullLogger.Instance); + var tools = new AttachmentReplyTools(storage, buffer, "s", "t", NullLogger.Instance); File.WriteAllBytes(Path.Combine(dir, "chart.png"), [1, 2, 3]); try { var result = tools.AttachImage("attachments/chart.png"); StringAssert.Contains(result, "Attached"); - var drained = buffer.Drain("s"); + var drained = buffer.Drain("s", "t"); Assert.AreEqual(1, drained.Count); Assert.AreEqual("chart.png", drained[0].Path); } diff --git a/tests/RockBot.Cli.Tests/AttachmentPlaceholderTests.cs b/tests/RockBot.Cli.Tests/AttachmentPlaceholderTests.cs index 442e284..8cd4660 100644 --- a/tests/RockBot.Cli.Tests/AttachmentPlaceholderTests.cs +++ b/tests/RockBot.Cli.Tests/AttachmentPlaceholderTests.cs @@ -7,21 +7,31 @@ namespace RockBot.Cli.Tests; public sealed class AttachmentPlaceholderTests { [TestMethod] - public void Render_Image_UsesImageLabelAndFriendlyName() + public void Render_Image_ShowsFriendlyNameAndPath_WhenTheyDiffer() { var att = new AgentAttachment { Mime = "image/png", Path = "chart.png", FileName = "Q3 chart.png" }; - Assert.AreEqual("[image: Q3 chart.png (image/png)]", AttachmentPlaceholder.Render(att)); + // The path is always surfaced so it can be inspected/tested; the friendly name leads. + Assert.AreEqual("[image: Q3 chart.png (chart.png, image/png)]", AttachmentPlaceholder.Render(att)); } [TestMethod] - public void Render_Image_FallsBackToPath_WhenNoFileName() + public void Render_Image_UsesPathOnly_WhenNoFileName() { var att = new AgentAttachment { Mime = "image/jpeg", Path = "photo.jpg" }; Assert.AreEqual("[image: photo.jpg (image/jpeg)]", AttachmentPlaceholder.Render(att)); } + [TestMethod] + public void Render_Image_UsesPathOnly_WhenFriendlyNameEqualsPath() + { + var att = new AgentAttachment { Mime = "image/png", Path = "chart.png", FileName = "chart.png" }; + + // No redundant duplication when the friendly name is just the path. + Assert.AreEqual("[image: chart.png (image/png)]", AttachmentPlaceholder.Render(att)); + } + [TestMethod] public void Render_NonImage_UsesAttachmentLabel() { diff --git a/tests/RockBot.Host.Tests/ReplyAttachmentBufferTests.cs b/tests/RockBot.Host.Tests/ReplyAttachmentBufferTests.cs index 5d4d4f6..aa30878 100644 --- a/tests/RockBot.Host.Tests/ReplyAttachmentBufferTests.cs +++ b/tests/RockBot.Host.Tests/ReplyAttachmentBufferTests.cs @@ -13,7 +13,7 @@ public void Drain_ReturnsEmpty_WhenNothingStaged() { var buffer = new ReplyAttachmentBuffer(); - var drained = buffer.Drain("unseen"); + var drained = buffer.Drain("unseen", "t1"); Assert.AreEqual(0, drained.Count); } @@ -22,10 +22,10 @@ public void Drain_ReturnsEmpty_WhenNothingStaged() public void Add_Then_Drain_ReturnsStagedAttachments_InOrder() { var buffer = new ReplyAttachmentBuffer(); - buffer.Add("s1", Att("a.png")); - buffer.Add("s1", Att("b.png")); + buffer.Add("s1", "t1", Att("a.png")); + buffer.Add("s1", "t1", Att("b.png")); - var drained = buffer.Drain("s1"); + var drained = buffer.Drain("s1", "t1"); Assert.AreEqual(2, drained.Count); Assert.AreEqual("a.png", drained[0].Path); @@ -36,10 +36,10 @@ public void Add_Then_Drain_ReturnsStagedAttachments_InOrder() public void Drain_Clears_SoSecondDrainIsEmpty() { var buffer = new ReplyAttachmentBuffer(); - buffer.Add("s1", Att("a.png")); + buffer.Add("s1", "t1", Att("a.png")); - var first = buffer.Drain("s1"); - var second = buffer.Drain("s1"); + var first = buffer.Drain("s1", "t1"); + var second = buffer.Drain("s1", "t1"); Assert.AreEqual(1, first.Count); Assert.AreEqual(0, second.Count); @@ -49,34 +49,172 @@ public void Drain_Clears_SoSecondDrainIsEmpty() public void Sessions_AreIsolated() { var buffer = new ReplyAttachmentBuffer(); - buffer.Add("a", Att("a.png")); - buffer.Add("b", Att("b.png")); + buffer.Add("a", "t1", Att("a.png")); + buffer.Add("b", "t1", Att("b.png")); - var drainedA = buffer.Drain("a"); + var drainedA = buffer.Drain("a", "t1"); Assert.AreEqual(1, drainedA.Count); Assert.AreEqual("a.png", drainedA[0].Path); // Draining a did not touch b. - Assert.AreEqual(1, buffer.Drain("b").Count); + Assert.AreEqual(1, buffer.Drain("b", "t1").Count); + } + + [TestMethod] + public void ConcurrentProducers_DrainOnlyTheirOwnTurn() + { + // The item-1 bug: two producers stage under one primary session. Draining turn A's + // final reply must return only A's attachments and leave B's intact for B's own drain. + var buffer = new ReplyAttachmentBuffer(); + buffer.Add("s1", "turnA", Att("a.png")); + buffer.Add("s1", "turnB", Att("b.png")); + + var drainedA = buffer.Drain("s1", "turnA"); + + Assert.AreEqual(1, drainedA.Count); + Assert.AreEqual("a.png", drainedA[0].Path); + + var drainedB = buffer.Drain("s1", "turnB"); + Assert.AreEqual(1, drainedB.Count); + Assert.AreEqual("b.png", drainedB[0].Path); + } + + [TestMethod] + public void Clear_Session_DropsAllTurns() + { + var buffer = new ReplyAttachmentBuffer(); + buffer.Add("s1", "t1", Att("a.png")); + buffer.Add("s1", "t2", Att("b.png")); + + buffer.Clear("s1"); + + Assert.AreEqual(0, buffer.Drain("s1", "t1").Count); + Assert.AreEqual(0, buffer.Drain("s1", "t2").Count); + } + + [TestMethod] + public void Clear_Turn_DropsOneTurn_LeavesSiblings() + { + var buffer = new ReplyAttachmentBuffer(); + buffer.Add("s1", "t1", Att("a.png")); + buffer.Add("s1", "t2", Att("b.png")); + + buffer.Clear("s1", "t1"); + + Assert.AreEqual(0, buffer.Drain("s1", "t1").Count); + // Sibling turn untouched. + Assert.AreEqual(1, buffer.Drain("s1", "t2").Count); + } + + [TestMethod] + public void Ttl_SweepsExpiredStage() + { + var time = new FakeTimeProvider(new DateTimeOffset(2026, 6, 16, 12, 0, 0, TimeSpan.Zero)); + var buffer = new ReplyAttachmentBuffer(time, ttl: TimeSpan.FromMinutes(30)); + buffer.Add("s1", "t1", Att("a.png")); + + time.Advance(TimeSpan.FromMinutes(30) + TimeSpan.FromSeconds(1)); + + Assert.AreEqual(0, buffer.Drain("s1", "t1").Count, "Stage past TTL should be swept."); + } + + [TestMethod] + public void Ttl_KeepsStageWithinWindow() + { + var time = new FakeTimeProvider(new DateTimeOffset(2026, 6, 16, 12, 0, 0, TimeSpan.Zero)); + var buffer = new ReplyAttachmentBuffer(time, ttl: TimeSpan.FromMinutes(30)); + buffer.Add("s1", "t1", Att("a.png")); + + time.Advance(TimeSpan.FromMinutes(29)); + + Assert.AreEqual(1, buffer.Drain("s1", "t1").Count, "Stage within TTL should survive."); + } + + [TestMethod] + public void Ttl_RefreshedOnEachAdd() + { + // A second Add in the same turn refreshes LastStagedAt, so a long-running stage that + // keeps adding within the window is never swept. + var time = new FakeTimeProvider(new DateTimeOffset(2026, 6, 16, 12, 0, 0, TimeSpan.Zero)); + var buffer = new ReplyAttachmentBuffer(time, ttl: TimeSpan.FromMinutes(30)); + buffer.Add("s1", "t1", Att("a.png")); + + time.Advance(TimeSpan.FromMinutes(20)); + buffer.Add("s1", "t1", Att("b.png")); // refreshes the stamp + time.Advance(TimeSpan.FromMinutes(20)); // 40 min since first add, but only 20 since second + + var drained = buffer.Drain("s1", "t1"); + Assert.AreEqual(2, drained.Count, "Refreshed stage should survive past the original add's TTL."); + } + + [TestMethod] + public void DrainForFinalReply_ReturnsNull_WhenNotFinal() + { + var buffer = new ReplyAttachmentBuffer(); + buffer.Add("s1", "t1", Att("a.png")); + + Assert.IsNull(buffer.DrainForFinalReply("s1", "t1", isFinal: false)); + // The stage is untouched — a later final reply still drains it. + Assert.AreEqual(1, buffer.Drain("s1", "t1").Count); + } + + [TestMethod] + public void DrainForFinalReply_ReturnsNull_WhenEmpty() + { + var buffer = new ReplyAttachmentBuffer(); + + Assert.IsNull(buffer.DrainForFinalReply("s1", "t1", isFinal: true)); + } + + [TestMethod] + public void DrainForFinalReply_ReturnsList_WhenFinalAndPresent() + { + var buffer = new ReplyAttachmentBuffer(); + buffer.Add("s1", "t1", Att("a.png")); + + var drained = buffer.DrainForFinalReply("s1", "t1", isFinal: true); + + Assert.IsNotNull(drained); + Assert.AreEqual(1, drained!.Count); + // Drained — a second final reply gets nothing. + Assert.IsNull(buffer.DrainForFinalReply("s1", "t1", isFinal: true)); } [TestMethod] public void SessionIds_AreCaseInsensitive() { var buffer = new ReplyAttachmentBuffer(); - buffer.Add("Session-A", Att("a.png")); + buffer.Add("Session-A", "t1", Att("a.png")); - Assert.AreEqual(1, buffer.Drain("session-a").Count); + Assert.AreEqual(1, buffer.Drain("session-a", "t1").Count); } [TestMethod] - public void Clear_DropsStagedAttachments() + public void TurnIds_AreCaseInsensitive() { var buffer = new ReplyAttachmentBuffer(); - buffer.Add("s1", Att("a.png")); + buffer.Add("s1", "Turn-A", Att("a.png")); + + Assert.AreEqual(1, buffer.Drain("s1", "turn-a").Count); + } + + [TestMethod] + public void Clear_Session_DropsStagedAttachments() + { + var buffer = new ReplyAttachmentBuffer(); + buffer.Add("s1", "t1", Att("a.png")); buffer.Clear("s1"); - Assert.AreEqual(0, buffer.Drain("s1").Count); + Assert.AreEqual(0, buffer.Drain("s1", "t1").Count); + } + + /// Minimal fake for TTL testing without the test-package dependency. + private sealed class FakeTimeProvider : TimeProvider + { + private DateTimeOffset _now; + public FakeTimeProvider(DateTimeOffset start) => _now = start; + public override DateTimeOffset GetUtcNow() => _now; + public void Advance(TimeSpan by) => _now = _now.Add(by); } }