From 45d6d24a3722a1323187e73fb4e0e5255ebc3158 Mon Sep 17 00:00:00 2001 From: Pranit Date: Tue, 1 Sep 2026 12:12:15 -0400 Subject: [PATCH 1/6] Strengthen ACP embedding contracts - Validate exact session targets and propagate cancellation in both directions. - Stop advertising unsupported slash commands. - Add stable message IDs to live and replayed content. - Accept, persist, and replay native ACP image prompts. - Publish session title and context usage updates. --- src/acp/prompt.zig | 223 +++++++++++-- src/acp/server.zig | 224 +++++++++++-- src/acp/sessions.zig | 201 +++++++----- src/acp/types.zig | 78 ++++- src/core/images/image_attachments.zig | 64 ++++ src/core/session/session_usage.zig | 49 +++ tests/e2e/acp.test.ts | 431 ++++++++++++++++++++++---- 7 files changed, 1068 insertions(+), 202 deletions(-) diff --git a/src/acp/prompt.zig b/src/acp/prompt.zig index 790e4df28..ae0a147f9 100644 --- a/src/acp/prompt.zig +++ b/src/acp/prompt.zig @@ -7,6 +7,7 @@ const model_provider = @import("../core/config/model_provider.zig"); const host = @import("../core/hosts/host.zig"); const host_target = @import("../core/hosts/target.zig"); const io_mod = @import("../core/shared/io.zig"); +const image_attachments = @import("../core/images/image_attachments.zig"); const jsonrpc = @import("jsonrpc.zig"); const acp_types = @import("types.zig"); const server = @import("server.zig"); @@ -84,6 +85,30 @@ pub const TerminalOutcome = union(enum) { rpc_error: jsonrpc.RpcError, }; +fn promptInputFailure(err: anyerror) anyerror!TerminalOutcome { + return switch (err) { + error.OutOfMemory => err, + error.UnsupportedPromptImage => .{ .rpc_error = .{ + .code = ErrorCode.invalid_params, + .message = "Image prompt blocks are not supported in this runtime", + } }, + error.InvalidPromptImage, + error.InvalidImageId, + error.ImageIdOverflow, + error.UnsupportedImageType, + error.ImageSnapshotMediaTypeMismatch, + => .{ .rpc_error = .{ + .code = ErrorCode.invalid_params, + .message = "Invalid image prompt block", + } }, + error.ImageTooLarge => .{ .rpc_error = .{ + .code = ErrorCode.invalid_params, + .message = "Image prompt exceeds size limit", + } }, + else => err, + }; +} + const ProviderTerminalPublication = enum { not_applicable, pending, @@ -98,6 +123,10 @@ const AcpContext = struct { /// copies of provider call ids so the ID stays stable from permission /// review through execution. published_tool_calls: std.StringHashMapUnmanaged(ProviderTerminalPublication) = .empty, + assistant_message_id: acp_types.MessageIdBuffer = undefined, + operational_message_id: acp_types.MessageIdBuffer = undefined, + assistant_message_id_ready: bool = false, + operational_message_id_ready: bool = false, stop_reason: acp_types.StopReason = .end_turn, auto_classifier: permission_auto_classifier.Classifier = permission_auto_classifier.Classifier.disabled(), @@ -120,12 +149,28 @@ const AcpContext = struct { try self.state.writer.writeNotification(self.alloc, "session/update", out.writer.buffered()); } - fn sendAgentText(self: *AcpContext, text: []const u8) !void { + fn assistantMessageId(self: *AcpContext) []const u8 { + if (!self.assistant_message_id_ready) { + _ = acp_types.generateMessageId(&self.assistant_message_id); + self.assistant_message_id_ready = true; + } + return &self.assistant_message_id; + } + + fn operationalMessageId(self: *AcpContext) []const u8 { + if (!self.operational_message_id_ready) { + _ = acp_types.generateMessageId(&self.operational_message_id); + self.operational_message_id_ready = true; + } + return &self.operational_message_id; + } + + fn sendAgentText(self: *AcpContext, message_id: []const u8, text: []const u8) !void { const plain = try stripAnsiAlloc(self.alloc, text); defer if (plain.ptr != text.ptr) self.alloc.free(plain); var out: std.Io.Writer.Allocating = .init(self.alloc); defer out.deinit(); - try acp_types.writeAgentMessageChunk(&out.writer, plain); + try acp_types.writeAgentMessageChunk(&out.writer, message_id, plain); try self.sendUpdate(out.writer.buffered()); } @@ -502,16 +547,26 @@ pub fn handlePrompt( }, }; - var prompt_input = parsePromptInput(alloc, params) catch |err| switch (err) { - error.UnsupportedPromptImage => return .{ - .rpc_error = .{ - .code = ErrorCode.invalid_params, - .message = "Image prompt blocks are not supported", - }, - }, - else => return err, - }; + const prior_image_catalog = try session.session_rt.snapshotImageCatalog(alloc, &.{}); + defer types.freeImageAttachmentSlice(alloc, prior_image_catalog); + const next_image_id = (try image_attachments.calculate_next_image_id(prior_image_catalog)).next_id; + var prompt_input = parsePromptInputWithFirstImageId(alloc, params, next_image_id) catch |err| + return promptInputFailure(err); defer prompt_input.deinit(alloc); + if (prompt_input.pending_images.len > 0) { + if (comptime host_target.is_wasm) return promptInputFailure(error.UnsupportedPromptImage); + var temporary_snapshot_dir: ?[]u8 = null; + defer if (temporary_snapshot_dir) |path| alloc.free(path); + const snapshot_dir = try session_store.imageSnapshotStorageDir( + alloc, + if (session.store) |store| store.sessions_dir else null, + if (session.store != null) session.session_id else null, + &temporary_snapshot_dir, + ); + defer alloc.free(snapshot_dir); + prompt_input.captureImages(alloc, snapshot_dir) catch |err| + return promptInputFailure(err); + } const prompt_text = prompt_input.text; if (prompt_input.continue_recovery and prompt_text.len != 0) { @@ -613,7 +668,7 @@ pub fn handlePrompt( ); defer alloc.free(root_user_intent_context); - const current_images = if (recovery_checkpoint) |checkpoint| checkpoint.user.images else &.{}; + const current_images = if (recovery_checkpoint) |checkpoint| checkpoint.user.images else prompt_input.images; const authorized_image_catalog = try session.session_rt.snapshotImageCatalog(alloc, current_images); defer types.freeImageAttachmentSlice(alloc, authorized_image_catalog); @@ -694,6 +749,9 @@ pub fn handlePrompt( return err; } }; + prompt_input.retainImageSnapshots(); + try sessions.sendActiveSessionInfoUpdate(state, alloc); + try sessions.sendActiveSessionUsageUpdate(state, alloc); if (session.cancel_flag.load(.seq_cst)) { ctx.stop_reason = .cancelled; @@ -850,12 +908,54 @@ fn buildAgentConfig( }; } +const PendingPromptImage = struct { + id: usize, + bytes: []u8, + media_type: []u8, + + fn deinit(self: *PendingPromptImage, alloc: Allocator) void { + alloc.free(self.bytes); + alloc.free(self.media_type); + self.* = undefined; + } +}; + const ParsedPromptInput = struct { text: []u8, continue_recovery: bool = false, targets: []context_contract.ApplicableTarget = &.{}, omissions: []context_contract.ContextOmissionInput = &.{}, omission_summary: ?context_contract.ContextOmissionSummary = null, + pending_images: []PendingPromptImage = &.{}, + images: []types.ImageAttachment = &.{}, + retain_image_snapshots: bool = false, + + fn captureImages(self: *ParsedPromptInput, alloc: Allocator, snapshot_dir: []const u8) !void { + if (self.pending_images.len == 0) return; + const images = try alloc.alloc(types.ImageAttachment, self.pending_images.len); + var captured: usize = 0; + errdefer { + for (images[0..captured]) |attachment| { + image_attachments.discardImageAttachment(alloc, attachment); + } + alloc.free(images); + } + for (self.pending_images, 0..) |pending, index| { + images[index] = try image_attachments.captureInlineImageBytes( + alloc, + pending.id, + pending.media_type, + pending.bytes, + snapshot_dir, + ); + captured += 1; + } + self.images = images; + } + + fn retainImageSnapshots(self: *ParsedPromptInput) void { + self.retain_image_snapshots = true; + } fn deinit(self: *ParsedPromptInput, alloc: Allocator) void { alloc.free(self.text); @@ -863,11 +963,29 @@ const ParsedPromptInput = struct { if (self.targets.len > 0) alloc.free(self.targets); for (self.omissions) |omission| alloc.free(@constCast(omission.source)); if (self.omissions.len > 0) alloc.free(self.omissions); + for (self.pending_images) |*pending| pending.deinit(alloc); + if (self.pending_images.len > 0) alloc.free(self.pending_images); + if (self.images.len > 0) { + if (self.retain_image_snapshots) { + types.freeImageAttachmentSlice(alloc, self.images); + } else { + image_attachments.discardImageAttachmentSlice(alloc, self.images); + } + } self.* = undefined; } }; fn parsePromptInput(alloc: Allocator, params_json: []const u8) !ParsedPromptInput { + return parsePromptInputWithFirstImageId(alloc, params_json, 1); +} + +fn parsePromptInputWithFirstImageId( + alloc: Allocator, + params_json: []const u8, + first_image_id: usize, +) !ParsedPromptInput { + if (first_image_id == 0) return error.InvalidImageId; const parsed = std.json.parseFromSlice(std.json.Value, alloc, params_json, .{}) catch return .{ .text = try alloc.dupe(u8, "") }; defer parsed.deinit(); @@ -895,6 +1013,11 @@ fn parsePromptInput(alloc: Allocator, params_json: []const u8) !ParsedPromptInpu targets.deinit(alloc); } var omissions: std.ArrayList(context_contract.ContextOmissionInput) = .empty; + var pending_images: std.ArrayList(PendingPromptImage) = .empty; + defer { + for (pending_images.items) |*pending| pending.deinit(alloc); + pending_images.deinit(alloc); + } var omission_summary: context_contract.ContextOmissionSummaryBuilder = .{}; defer { for (omissions.items) |omission| alloc.free(@constCast(omission.source)); @@ -914,7 +1037,43 @@ fn parsePromptInput(alloc: Allocator, params_json: []const u8) !ParsedPromptInpu } } } else if (std.mem.eql(u8, block_type.string, "image")) { - return error.UnsupportedPromptImage; + const data_value = block.object.get("data") orelse return error.InvalidPromptImage; + const media_type_value = block.object.get("mimeType") orelse return error.InvalidPromptImage; + if (data_value != .string or media_type_value != .string or media_type_value.string.len == 0) { + return error.InvalidPromptImage; + } + const decoded_len = std.base64.standard.Decoder.calcSizeForSlice(data_value.string) catch + return error.InvalidPromptImage; + if (decoded_len == 0) return error.InvalidPromptImage; + if (decoded_len > image_attachments.max_image_bytes) { + return error.ImageTooLarge; + } + const decoded = try alloc.alloc(u8, decoded_len); + errdefer alloc.free(decoded); + std.base64.standard.Decoder.decode(decoded, data_value.string) catch + return error.InvalidPromptImage; + const canonical_len = std.base64.standard.Encoder.calcSize(decoded.len); + if (canonical_len != data_value.string.len) return error.InvalidPromptImage; + const canonical = try alloc.alloc(u8, canonical_len); + defer alloc.free(canonical); + const encoded = std.base64.standard.Encoder.encode(canonical, decoded); + if (!std.mem.eql(u8, encoded, data_value.string)) return error.InvalidPromptImage; + + const image_id = std.math.add(usize, first_image_id, pending_images.items.len) catch + return error.ImageIdOverflow; + if (text_buf.items.len > 0) try text_buf.append(alloc, '\n'); + var placeholder: std.Io.Writer.Allocating = .init(alloc); + defer placeholder.deinit(); + try image_attachments.writeImagePlaceholder(&placeholder.writer, image_id); + try text_buf.appendSlice(alloc, placeholder.writer.buffered()); + + const media_type = try alloc.dupe(u8, media_type_value.string); + errdefer alloc.free(media_type); + try pending_images.append(alloc, .{ + .id = image_id, + .bytes = decoded, + .media_type = media_type, + }); } else if (std.mem.eql(u8, block_type.string, "resource")) { if (block.object.get("resource")) |resource| { if (resource == .object) { @@ -971,6 +1130,7 @@ fn parsePromptInput(alloc: Allocator, params_json: []const u8) !ParsedPromptInpu errdefer result.deinit(alloc); result.targets = try targets.toOwnedSlice(alloc); result.omissions = try omissions.toOwnedSlice(alloc); + result.pending_images = try pending_images.toOwnedSlice(alloc); result.omission_summary = omission_summary.finish(); return result; } @@ -1931,13 +2091,13 @@ fn pushRouteRecoveryStatus( fn pushText(raw_ctx: *anyopaque, emission: agent_runtime.TextEmission) !void { const ctx: *AcpContext = @ptrCast(@alignCast(raw_ctx)); - const text = switch (emission) { - .assistant_source => |text| text, + const text, const message_id = switch (emission) { + .assistant_source => |text| .{ text, ctx.assistantMessageId() }, .assistant_rendered => return, - .operational => |text| text, + .operational => |text| .{ text, ctx.operationalMessageId() }, }; if (text.len == 0) return; - ctx.sendAgentText(text) catch {}; + ctx.sendAgentText(message_id, text) catch {}; } fn pushToolLifecycle(raw_ctx: *anyopaque, event: types.ToolLifecycleEvent) !void { @@ -1971,7 +2131,7 @@ fn pushToolLifecycle(raw_ctx: *anyopaque, event: types.ToolLifecycleEvent) !void fn pushSystemNotice(raw_ctx: *anyopaque, text: []const u8) !void { const ctx: *AcpContext = @ptrCast(@alignCast(raw_ctx)); - ctx.sendAgentText(text) catch {}; + ctx.sendAgentText(ctx.operationalMessageId(), text) catch {}; } fn pushContextNotice(raw_ctx: *anyopaque, text: []const u8) !void { @@ -2001,7 +2161,7 @@ fn pushHttpError(raw_ctx: *anyopaque, status: std.http.Status, detail: []const u std.fmt.bufPrint(&buf, "HTTP {d}: {s}", .{ @intFromEnum(status), detail }) catch "HTTP error" else std.fmt.bufPrint(&buf, "HTTP {d}", .{@intFromEnum(status)}) catch "HTTP error"; - ctx.sendAgentText(msg) catch {}; + ctx.sendAgentText(ctx.operationalMessageId(), msg) catch {}; } fn formatToolExecutionError(_: *anyopaque, arena: Allocator, tool_name: []const u8, err: anyerror) ![]const u8 { @@ -2866,10 +3026,29 @@ test "parsePromptInput accepts explicit recovery continuation metadata" { try std.testing.expect(result.continue_recovery); } -test "parsePromptInput rejects image blocks" { +test "parsePromptInput accepts image blocks as owned pending images" { const alloc = std.testing.allocator; const params = "{\"sessionId\":\"s1\",\"prompt\":[{\"type\":\"text\",\"text\":\"Only text\"},{\"type\":\"image\",\"data\":\"aGVsbG8=\",\"mimeType\":\"image/png\"}]}"; - try std.testing.expectError(error.UnsupportedPromptImage, parsePromptInput(alloc, params)); + var parsed = try parsePromptInputWithFirstImageId(alloc, params, 7); + defer parsed.deinit(alloc); + + try std.testing.expectEqualStrings("Only text\n[Image #7]", parsed.text); + try std.testing.expectEqual(@as(usize, 1), parsed.pending_images.len); + try std.testing.expectEqual(@as(usize, 7), parsed.pending_images[0].id); + try std.testing.expectEqualStrings("hello", parsed.pending_images[0].bytes); + try std.testing.expectEqualStrings("image/png", parsed.pending_images[0].media_type); +} + +test "parsePromptInput rejects malformed base64 image data" { + const alloc = std.testing.allocator; + const params = "{\"sessionId\":\"s1\",\"prompt\":[{\"type\":\"image\",\"data\":\"not-base64\",\"mimeType\":\"image/png\"}]}"; + try std.testing.expectError(error.InvalidPromptImage, parsePromptInput(alloc, params)); +} + +test "parsePromptInput rejects empty image data" { + const alloc = std.testing.allocator; + const params = "{\"sessionId\":\"s1\",\"prompt\":[{\"type\":\"image\",\"data\":\"\",\"mimeType\":\"image/png\"}]}"; + try std.testing.expectError(error.InvalidPromptImage, parsePromptInput(alloc, params)); } test "parsePromptInput preserves resource text and accepts only local absolute file targets" { @@ -3224,7 +3403,7 @@ test "ACP stream adapter forwards raw Markdown and suppresses rendered duplicate }; var writer_failed = false; - failed_ctx.sendAgentText("writer failure probe") catch { + failed_ctx.sendAgentText(failed_ctx.assistantMessageId(), "writer failure probe") catch { writer_failed = true; }; try std.testing.expect(writer_failed); diff --git a/src/acp/server.zig b/src/acp/server.zig index 8224fc03d..1cd2a6d23 100644 --- a/src/acp/server.zig +++ b/src/acp/server.zig @@ -48,6 +48,7 @@ const writeJsonStr = jsonrpc.writeJsonStr; const legacy_url_completion_timeout_ms: i64 = 10 * 60 * 1000; const AcpMethod = enum { + request_cancel, initialize, session_cancel, session_new, @@ -62,6 +63,7 @@ const AcpMethod = enum { unknown, fn parse(method: []const u8) AcpMethod { + if (std.mem.eql(u8, method, "$/cancel_request")) return .request_cancel; if (std.mem.eql(u8, method, "initialize")) return .initialize; if (std.mem.eql(u8, method, "session/cancel")) return .session_cancel; if (std.mem.eql(u8, method, "session/new")) return .session_new; @@ -79,6 +81,7 @@ const AcpMethod = enum { fn waitsForActivePrompt(self: AcpMethod) bool { return switch (self) { .initialize, + .request_cancel, .session_cancel, .session_set_mode, .session_new, @@ -96,6 +99,20 @@ const AcpMethod = enum { } }; +const SessionTargetDecision = enum { + exact, + missing, + inactive, + mismatch, +}; + +fn decideSessionTarget(active_session_id: ?[]const u8, supplied: ?std.json.Value) SessionTargetDecision { + const active = active_session_id orelse return .inactive; + const value = supplied orelse return .missing; + if (value != .string or value.string.len == 0) return .missing; + return if (std.mem.eql(u8, active, value.string)) .exact else .mismatch; +} + pub const Config = acp_runner.Config; pub const OutboundKind = enum { @@ -711,7 +728,7 @@ pub fn runWithTransport( } // Release any prompt thread parked on a pending approval before // state.deinit() joins it, or shutdown deadlocks. - handleCancel(&state); + handleCancel(&state, false); } fn shouldRespondToMessage(msg: *const jsonrpc.Message) bool { @@ -790,32 +807,58 @@ pub fn awaitOutboundResponse(state: *ServerState, id: u64, kind: OutboundKind) ? return response; } state.outbound_cond.wait(io_mod.getIo(), &state.outbound_mutex) catch { - cancelOutboundRequestLocked(state, id); + _ = cancelOutboundRequestLocked(state, id); }; } } pub fn cancelOutboundRequest(state: *ServerState, id: u64) void { state.outbound_mutex.lockUncancelable(io_mod.getIo()); - defer state.outbound_mutex.unlock(io_mod.getIo()); - cancelOutboundRequestLocked(state, id); + const changed = cancelOutboundRequestLocked(state, id); + state.outbound_mutex.unlock(io_mod.getIo()); + if (changed) publishRequestCancellation(state, id); } -fn cancelOutboundRequestLocked(state: *ServerState, id: u64) void { - const pending = state.pending_outbound.getPtr(id) orelse return; - if (pending.response != null) return; +fn cancelOutboundRequestLocked(state: *ServerState, id: u64) bool { + const pending = state.pending_outbound.getPtr(id) orelse return false; + if (pending.response != null) return false; pending.response = .{ .cancelled = true }; state.outbound_cond.broadcast(io_mod.getIo()); + return true; } -fn cancelPendingOutbound(state: *ServerState) void { +fn cancelPendingOutbound(state: *ServerState, notify_client: bool) void { + var cancelled_ids: [max_pending_outbound]u64 = undefined; + var cancelled_count: usize = 0; state.outbound_mutex.lockUncancelable(io_mod.getIo()); - defer state.outbound_mutex.unlock(io_mod.getIo()); - var pending = state.pending_outbound.valueIterator(); + var pending = state.pending_outbound.iterator(); while (pending.next()) |entry| { - if (entry.response == null) entry.response = .{ .cancelled = true }; + if (entry.value_ptr.response != null) continue; + entry.value_ptr.response = .{ .cancelled = true }; + cancelled_ids[cancelled_count] = entry.key_ptr.*; + cancelled_count += 1; } state.outbound_cond.broadcast(io_mod.getIo()); + state.outbound_mutex.unlock(io_mod.getIo()); + + if (!notify_client) return; + for (cancelled_ids[0..cancelled_count]) |id| publishRequestCancellation(state, id); +} + +fn publishRequestCancellation(state: *ServerState, id: u64) void { + var params: std.Io.Writer.Allocating = .init(state.alloc); + defer params.deinit(); + params.writer.print("{{\"requestId\":{d}}}", .{id}) catch |err| { + debug_trace.logf("acp", "request cancellation serialization failed id={d} err={s}", .{ id, @errorName(err) }); + return; + }; + state.writer.writeNotification( + state.alloc, + "$/cancel_request", + params.writer.buffered(), + ) catch |err| { + debug_trace.logf("acp", "request cancellation publication failed id={d} err={s}", .{ id, @errorName(err) }); + }; } pub fn reserveLegacyUrl( @@ -1062,11 +1105,15 @@ pub fn cancelPermissionRequest(state: *ServerState, id: u64) void { } fn dispatchNotification(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message) !void { - _ = alloc; reapActivePrompt(state, false); if (!state.initialized) return; switch (AcpMethod.parse(msg.method)) { - .session_cancel => handleCancel(state), + .request_cancel => handleRequestCancellation(state, alloc, msg.params_raw), + .session_cancel => { + if (notificationTargetsActiveSession(state, alloc, msg.params_raw)) { + handleCancel(state, true); + } + }, else => {}, } } @@ -1087,7 +1134,8 @@ fn dispatch(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message) !void } if (method == .session_cancel) { - handleCancel(state); + if (!try requireActiveSessionTarget(state, alloc, msg)) return; + handleCancel(state, true); return state.writer.writeResponse(alloc, msg.id, "null"); } @@ -1124,6 +1172,7 @@ fn dispatch(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message) !void .session_set_config_option => handleSetConfigOption(state, alloc, msg), .session_set_mode => handleSetMode(state, alloc, msg), .initialize, + .request_cancel, .session_cancel, .session_remove, .unknown, @@ -1134,8 +1183,49 @@ fn dispatch(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message) !void }; } +fn requestIdsEqual(lhs: jsonrpc.RequestId, rhs: jsonrpc.RequestId) bool { + return switch (lhs) { + .integer => |value| switch (rhs) { + .integer => |other| value == other, + else => false, + }, + .string => |value| switch (rhs) { + .string => |other| std.mem.eql(u8, value, other), + else => false, + }, + .null => rhs == .null, + }; +} + +fn requestIdFromValue(value: std.json.Value) ?jsonrpc.RequestId { + return switch (value) { + .integer => |id| .{ .integer = id }, + .string => |id| .{ .string = id }, + .null => .null, + else => null, + }; +} + +fn handleRequestCancellation( + state: *ServerState, + alloc: Allocator, + params: ?[]const u8, +) void { + const raw = params orelse return; + const parsed = std.json.parseFromSlice(std.json.Value, alloc, raw, .{}) catch return; + defer parsed.deinit(); + if (parsed.value != .object) return; + const value = parsed.value.object.get("requestId") orelse return; + const requested = requestIdFromValue(value) orelse return; + const active = state.active_prompt orelse return; + const active_id = active.msg.id orelse return; + if (!requestIdsEqual(active_id, requested)) return; + handleCancel(state, true); +} + fn startPrompt(state: *ServerState, alloc: Allocator, msg: *const jsonrpc.Message) !void { - const session = if (state.active_session) |*active| active else return state.writer.writeError(alloc, msg.id, prompt_handler.no_active_session_rpc_error); + if (!try requireActiveSessionTarget(state, alloc, msg)) return; + const session = if (state.active_session) |*active| active else unreachable; const active = try alloc.create(ActivePrompt); errdefer alloc.destroy(active); active.* = .{ @@ -1158,6 +1248,81 @@ fn startPrompt(state: *ServerState, alloc: Allocator, msg: *const jsonrpc.Messag } } +fn parsedSessionTargetDecision(state: *const ServerState, root: std.json.Value) SessionTargetDecision { + const active_id = if (state.active_session) |session| session.session_id else null; + const supplied = if (root == .object) root.object.get("sessionId") else null; + return decideSessionTarget(active_id, supplied); +} + +fn writeSessionTargetError( + state: *ServerState, + alloc: Allocator, + id: ?jsonrpc.RequestId, + decision: SessionTargetDecision, +) !void { + const message: []const u8 = switch (decision) { + .exact => return, + .inactive => "No active session", + .missing => "Missing sessionId", + .mismatch => "Session is not active", + }; + try state.writer.writeError(alloc, id, .{ + .code = ErrorCode.invalid_params, + .message = message, + }); +} + +fn requireParsedActiveSessionTarget( + state: *ServerState, + alloc: Allocator, + id: ?jsonrpc.RequestId, + root: std.json.Value, +) !bool { + const decision = parsedSessionTargetDecision(state, root); + if (decision == .exact) return true; + try writeSessionTargetError(state, alloc, id, decision); + return false; +} + +fn requireActiveSessionTarget( + state: *ServerState, + alloc: Allocator, + msg: *const jsonrpc.Message, +) !bool { + const params = msg.params_raw orelse { + try state.writer.writeError(alloc, msg.id, .{ + .code = ErrorCode.invalid_params, + .message = "Missing params", + }); + return false; + }; + const parsed = std.json.parseFromSlice(std.json.Value, alloc, params, .{}) catch { + try state.writer.writeError(alloc, msg.id, .{ + .code = ErrorCode.invalid_params, + .message = "Invalid params", + }); + return false; + }; + defer parsed.deinit(); + return requireParsedActiveSessionTarget(state, alloc, msg.id, parsed.value); +} + +fn notificationTargetsActiveSession( + state: *const ServerState, + alloc: Allocator, + params: ?[]const u8, +) bool { + const raw = params orelse return false; + const parsed = std.json.parseFromSlice(std.json.Value, alloc, raw, .{}) catch return false; + defer parsed.deinit(); + const decision = parsedSessionTargetDecision(state, parsed.value); + if (decision != .exact) { + debug_trace.logf("acp", "ignored session notification target reason={s}", .{@tagName(decision)}); + return false; + } + return true; +} + fn promptWorkerMain(active: *ActivePrompt) void { const outcome: prompt_handler.TerminalOutcome = prompt_handler.handlePrompt( active.state, @@ -1443,21 +1608,21 @@ fn handleInitialize(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); - try acp_types.writeInitializeResponse(&out.writer); + try acp_types.writeInitializeResponse(&out.writer, !host_target.is_wasm); try state.writer.writeResponse(alloc, msg.id, out.writer.buffered()); } -fn handleCancel(state: *ServerState) void { +fn handleCancel(state: *ServerState, notify_client: bool) void { if (state.active_session) |*session| { debug_trace.eventf("interrupt", "cancel_requested", .{}, "source=acp active_tool_known=false", .{}); session.cancel_flag.store(true, .seq_cst); } - cancelPendingOutbound(state); + cancelPendingOutbound(state, notify_client); clearPendingLegacyUrls(state); } pub fn cancelAndReapActivePrompt(state: *ServerState) void { - handleCancel(state); + handleCancel(state, true); reapActivePrompt(state, true); } @@ -1522,6 +1687,7 @@ fn handleSetConfigOption(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Me const root = parsed.value; if (root != .object) return state.writer.writeError(alloc, msg.id, .{ .code = ErrorCode.invalid_params, .message = "Params must be object" }); + if (!try requireParsedActiveSessionTarget(state, alloc, msg.id, root)) return; const config_id = blk: { if (root.object.get("configId")) |v| { @@ -1848,6 +2014,8 @@ fn handleSetMode(state: *ServerState, alloc: Allocator, msg: *jsonrpc.Message) ! return state.writer.writeError(alloc, msg.id, .{ .code = ErrorCode.invalid_params, .message = "Invalid params" }); defer parsed.deinit(); + if (!try requireParsedActiveSessionTarget(state, alloc, msg.id, parsed.value)) return; + if (parsed.value == .object) { if (parsed.value.object.get("modeId")) |v| { if (v == .string) { @@ -2021,17 +2189,30 @@ test "ACP permission responses map canonical option ids" { } test "ACP outbound waiters resolve to deny on cancellation" { + const Capture = struct { + saw_request_cancellation: bool = false, + + fn write(raw: ?*anyopaque, frame: []const u8) !void { + const self: *@This() = @ptrCast(@alignCast(raw.?)); + if (std.mem.find(u8, frame, "\"method\":\"$/cancel_request\"") != null and + std.mem.find(u8, frame, "\"requestId\":3") != null) + { + self.saw_request_cancellation = true; + } + } + }; + var capture = Capture{}; var state = ServerState{ .alloc = std.testing.allocator, .cfg = undefined, - .writer = jsonrpc.Writer.init(), + .writer = jsonrpc.Writer.initCallback(&capture, Capture.write), }; defer state.pending_outbound.deinit(state.alloc); const id = beginPermissionRequest(&state) orelse return error.TestExpectedEqual; const concurrent = beginPermissionRequest(&state) orelse return error.TestExpectedEqual; - cancelPendingOutbound(&state); + cancelPendingOutbound(&state, false); try std.testing.expectEqual(types.ToolPermissionDecision.deny, awaitPermissionDecision(&state, id)); try std.testing.expectEqual(types.ToolPermissionDecision.deny, awaitPermissionDecision(&state, concurrent)); try std.testing.expectEqual(@as(usize, 0), state.pending_outbound.count()); @@ -2040,6 +2221,7 @@ test "ACP outbound waiters resolve to deny on cancellation" { try std.testing.expect(second != id); cancelPermissionRequest(&state, second); try std.testing.expectEqual(types.ToolPermissionDecision.deny, awaitPermissionDecision(&state, second)); + try std.testing.expect(capture.saw_request_cancellation); } test "ACP outbound responses correlate out of order and ignore unknown ids" { diff --git a/src/acp/sessions.zig b/src/acp/sessions.zig index b34dd0ae6..0976252c0 100644 --- a/src/acp/sessions.zig +++ b/src/acp/sessions.zig @@ -8,6 +8,7 @@ const mcp_servers = @import("mcp_servers.zig"); const server = @import("server.zig"); const session_test_controls = @import("session_test_controls.zig"); const session_codec = @import("../core/session/session_codec.zig"); +const session_display_metadata = @import("../core/session/session_display_metadata.zig"); const session_store = @import("../core/session/session_store.zig"); const legacy_background_migration = @import("../core/session/legacy_background_migration.zig"); const js_host_session_store = @import("../core/session/js_host_session_store.zig"); @@ -21,13 +22,13 @@ const model_catalog = @import("../core/gateway/model_catalog.zig"); const provider_set = @import("../core/gateway/provider_set.zig"); const host = @import("../core/hosts/host.zig"); const host_target = @import("../core/hosts/target.zig"); +const image_attachments = @import("../core/images/image_attachments.zig"); const credentials = @import("../core/auth/credentials.zig"); const model_provider = @import("../core/config/model_provider.zig"); const mode_registry = @import("../core/modes/mode_registry.zig"); const subagent_resume_admission = @import("../core/subagent/resume_admission.zig"); const types = @import("../core/shared/types.zig"); const context_contract = @import("../core/workspace/context_contract.zig"); -const command_specs = @import("../core/slash_commands/command_specs.zig"); const test_builtin_gateway = if (builtin.is_test) @import("../builtins/gateway.zig") else @@ -283,9 +284,7 @@ fn writeNewSessionResponse( try state.writer.writeResponse(alloc, msg.id, out.writer.buffered()); - const commands_json = try buildSlashCommandsJson(alloc); - defer alloc.free(commands_json); - try sendAvailableCommands(state, alloc, session_id, commands_json); + try sendAvailableCommands(state, alloc, session_id, "[]"); } pub fn handleLoadWasmSession(state: *server.ServerState, alloc: Allocator, msg: *jsonrpc.Message) !void { @@ -307,6 +306,7 @@ pub fn handleLoadWasmSession(state: *server.ServerState, alloc: Allocator, msg: if (state.active_session) |*active| { if (sameSessionId(active.session_id, session_id)) { for (active.session_rt.history.items) |turn| try sendHistoryTurnAsUpdates(state, alloc, session_id, turn); + try sendActiveSessionInfoUpdate(state, alloc); return writeLoadSessionResponse(state, alloc, msg, active.model); } } @@ -368,6 +368,7 @@ pub fn handleLoadWasmSession(state: *server.ServerState, alloc: Allocator, msg: model_owned = false; session_rt_owned = false; for (state.active_session.?.session_rt.history.items) |turn| try sendHistoryTurnAsUpdates(state, alloc, session_id, turn); + try sendActiveSessionInfoUpdate(state, alloc); try writeLoadSessionResponse(state, alloc, msg, state.active_session.?.model); } @@ -528,6 +529,7 @@ fn handleRestoreSession( session_id, if (active.writable) |*writable| writable.state.recovery_checkpoint else null, ); + try sendActiveSessionInfoUpdate(state, alloc); return writeLoadSessionResponse( state, alloc, @@ -654,6 +656,7 @@ fn handleRestoreSession( session_id, state.active_session.?.writable.?.state.recovery_checkpoint, ); + try sendActiveSessionInfoUpdate(state, alloc); try writeLoadSessionResponse( state, @@ -754,7 +757,7 @@ fn sendPendingRecoveryUpdate( checkpoint: ?session_codec.RecoveryCheckpoint, ) !void { const recovery = checkpoint orelse return; - try sendUserHistoryChunk(state, alloc, session_id, recovery.user.text); + try sendUserHistoryTurn(state, alloc, session_id, recovery.user); try sendExecutionHistory(state, alloc, session_id, recovery.execution); if (recovery.assistant_source.len > 0) { try sendAgentHistoryChunk(state, alloc, session_id, recovery.assistant_source); @@ -1132,13 +1135,11 @@ fn parseListCursor(raw: []const u8) !session_store.ResumableSessionContinuation } fn sendHistoryTurnAsUpdates(state: *server.ServerState, alloc: Allocator, session_id: []const u8, turn: types.HistoryTurn) !void { - const user_text: []const u8 = switch (turn) { - .assistant => |a| a.user.text, - .interrupted => |i| i.user.text, - .compacted_summary => |c| c.summary, - }; - - try sendUserHistoryChunk(state, alloc, session_id, user_text); + switch (turn) { + .assistant => |assistant| try sendUserHistoryTurn(state, alloc, session_id, assistant.user), + .interrupted => |interrupted| try sendUserHistoryTurn(state, alloc, session_id, interrupted.user), + .compacted_summary => |compacted| try sendUserHistoryText(state, alloc, session_id, compacted.summary), + } switch (turn) { .assistant => |assistant| { @@ -1161,13 +1162,62 @@ fn sendHistoryTurnAsUpdates(state: *server.ServerState, alloc: Allocator, sessio } } -fn sendUserHistoryChunk(state: *server.ServerState, alloc: Allocator, session_id: []const u8, text: []const u8) !void { +fn sendUserHistoryTurn( + state: *server.ServerState, + alloc: Allocator, + session_id: []const u8, + user: types.UserTurn, +) !void { + var message_id: acp_types.MessageIdBuffer = undefined; + const stable_message_id = acp_types.generateMessageId(&message_id); + try sendUserHistoryChunk(state, alloc, session_id, stable_message_id, user.text); + for (user.images) |attachment| { + var snapshot = try image_attachments.loadVerifiedSnapshot(alloc, attachment, .{}); + defer snapshot.deinit(alloc); + var out: std.Io.Writer.Allocating = .init(alloc); + defer out.deinit(); + try out.writer.writeAll("{\"sessionId\":"); + try writeJsonStr(session_id, &out.writer); + try out.writer.writeAll(",\"update\":"); + try acp_types.writeUserImageChunk( + &out.writer, + stable_message_id, + snapshot.media_type, + snapshot.bytes, + ); + try out.writer.writeAll("}"); + try state.writer.writeNotification(alloc, "session/update", out.writer.buffered()); + } +} + +fn sendUserHistoryText(state: *server.ServerState, alloc: Allocator, session_id: []const u8, user_text: []const u8) !void { + var message_id: acp_types.MessageIdBuffer = undefined; + try sendUserHistoryChunk( + state, + alloc, + session_id, + acp_types.generateMessageId(&message_id), + user_text, + ); +} + +fn sendUserHistoryChunk( + state: *server.ServerState, + alloc: Allocator, + session_id: []const u8, + message_id: []const u8, + text: []const u8, +) !void { var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); try out.writer.writeAll("{\"sessionId\":"); try writeJsonStr(session_id, &out.writer); try out.writer.writeAll(",\"update\":"); - try acp_types.writeUserMessageChunk(&out.writer, text); + try acp_types.writeUserMessageChunk( + &out.writer, + message_id, + text, + ); try out.writer.writeAll("}"); try state.writer.writeNotification(alloc, "session/update", out.writer.buffered()); } @@ -1184,12 +1234,17 @@ fn sendExecutionHistory( } fn sendAgentHistoryChunk(state: *server.ServerState, alloc: Allocator, session_id: []const u8, text: []const u8) !void { + var message_id: acp_types.MessageIdBuffer = undefined; var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); try out.writer.writeAll("{\"sessionId\":"); try writeJsonStr(session_id, &out.writer); try out.writer.writeAll(",\"update\":"); - try acp_types.writeAgentMessageChunk(&out.writer, text); + try acp_types.writeAgentMessageChunk( + &out.writer, + acp_types.generateMessageId(&message_id), + text, + ); try out.writer.writeAll("}"); try state.writer.writeNotification(alloc, "session/update", out.writer.buffered()); } @@ -1205,47 +1260,55 @@ fn sendAvailableCommands(state: *server.ServerState, alloc: Allocator, session_i try state.writer.writeNotification(alloc, "session/update", out.writer.buffered()); } -fn buildSlashCommandsJson(alloc: Allocator) ![]u8 { +pub fn sendActiveSessionInfoUpdate(state: *server.ServerState, alloc: Allocator) !void { + const active = if (state.active_session) |*session| session else return; + var metadata = try session_display_metadata.deriveFromHistory( + alloc, + active.session_rt.history.items, + ); + defer metadata.deinit(alloc); + const updated_at_ms = if (active.writable) |*writable| + writable.state.updated_at_ms + else if (active.wasm_state) |durable| + durable.updated_at_ms + else + io_mod.milliTimestamp(); + const updated_at = try formatIso8601(alloc, @max(updated_at_ms, 0)); + defer alloc.free(updated_at); + var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); + try out.writer.writeAll("{\"sessionId\":"); + try writeJsonStr(active.session_id, &out.writer); + try out.writer.writeAll(",\"update\":"); + try acp_types.writeSessionInfoUpdate(&out.writer, metadata.title, updated_at); + try out.writer.writeAll("}"); + try state.writer.writeNotification(alloc, "session/update", out.writer.buffered()); +} - const commands = [_]struct { name: []const u8, description: []const u8, hint: ?[]const u8 }{ - .{ .name = "compact", .description = "Compact conversation history", .hint = null }, - .{ .name = "undo", .description = "Undo last file change", .hint = null }, - .{ .name = "changes", .description = "Show file changes in this session", .hint = null }, - .{ .name = "review", .description = "Toggle post-edit review", .hint = null }, - .{ .name = "clear", .description = "Clear the screen", .hint = null }, - .{ .name = "reset", .description = "Reset session", .hint = null }, - .{ .name = "help", .description = "Show available commands", .hint = null }, - .{ .name = "status", .description = "Show current status", .hint = null }, - .{ .name = "model", .description = "Switch model", .hint = "model name" }, - .{ .name = "permissions", .description = "Show permission settings", .hint = null }, - .{ .name = "allowlist", .description = "Manage persistent allow rules", .hint = "add command \"git *\"" }, - .{ .name = "rules", .description = "Show active rules", .hint = null }, - .{ .name = "settings", .description = "Show settings", .hint = null }, - .{ .name = "credits", .description = "Show credit balance", .hint = null }, - .{ .name = "mcp", .description = "Show MCP server status", .hint = null }, - .{ .name = "skills", .description = "Show installed skills", .hint = null }, - .{ .name = "fast", .description = "Toggle fast mode for supported models", .hint = null }, - }; - - try out.writer.writeAll("["); - for (commands, 0..) |cmd, i| { - if (i > 0) try out.writer.writeAll(","); - try out.writer.writeAll("{\"name\":"); - try writeJsonStr(cmd.name, &out.writer); - try out.writer.writeAll(",\"description\":"); - try writeJsonStr(cmd.description, &out.writer); - if (cmd.hint) |hint| { - try out.writer.writeAll(",\"input\":{\"hint\":"); - try writeJsonStr(hint, &out.writer); - try out.writer.writeAll("}"); - } - try out.writer.writeAll("}"); - } - try out.writer.writeAll("]"); +pub fn sendActiveSessionUsageUpdate(state: *server.ServerState, alloc: Allocator) !void { + const active = if (state.active_session) |*session| session else return; + const usage = active.session_rt.usage.liveContextSnapshot() orelse return; + const provider_bundle = state.cfg.provider_set.select(active.provider); + const capabilities = state.capability_resolver.available( + active.model, + provider_bundle.fallbackModelCapabilities(active.model), + ); + const context_window = capabilities.context_window orelse return; - return try alloc.dupe(u8, out.writer.buffered()); + var out: std.Io.Writer.Allocating = .init(alloc); + defer out.deinit(); + try out.writer.writeAll("{\"sessionId\":"); + try writeJsonStr(active.session_id, &out.writer); + try out.writer.writeAll(",\"update\":"); + try acp_types.writeUsageUpdate( + &out.writer, + usage.used, + context_window, + usage.complete_cost, + ); + try out.writer.writeAll("}"); + try state.writer.writeNotification(alloc, "session/update", out.writer.buffered()); } fn formatIso8601(alloc: Allocator, timestamp_ms: i64) ![]u8 { @@ -1356,40 +1419,6 @@ test "formatIso8601 produces valid format" { try std.testing.expect(std.mem.find(u8, result, "T") != null); } -test "buildSlashCommandsJson produces valid array" { - const alloc = std.testing.allocator; - const json = try buildSlashCommandsJson(alloc); - defer alloc.free(json); - try std.testing.expect(json.len > 0); - try std.testing.expect(json[0] == '['); - try std.testing.expect(json[json.len - 1] == ']'); - try std.testing.expect(std.mem.find(u8, json, "compact") != null); -} - -test "buildSlashCommandsJson includes all expected commands" { - const alloc = std.testing.allocator; - const json = try buildSlashCommandsJson(alloc); - defer alloc.free(json); - - const expected_commands = [_][]const u8{ - "compact", "undo", "changes", "review", "clear", - "reset", "help", "status", "model", "permissions", - "allowlist", "rules", "settings", "credits", "mcp", - "skills", "fast", - }; - for (expected_commands) |cmd| { - try std.testing.expect(std.mem.find(u8, json, cmd) != null); - } - try std.testing.expect(std.mem.find(u8, json, "\"name\":\"summary\"") == null); -} - -test "buildSlashCommandsJson includes input hint for model" { - const alloc = std.testing.allocator; - const json = try buildSlashCommandsJson(alloc); - defer alloc.free(json); - try std.testing.expect(std.mem.find(u8, json, "\"input\":{\"hint\":\"model name\"}") != null); -} - test "formatIso8601 produces known timestamp" { const alloc = std.testing.allocator; const result = try formatIso8601(alloc, 0); diff --git a/src/acp/types.zig b/src/acp/types.zig index 4c0d6737b..4144adb2a 100644 --- a/src/acp/types.zig +++ b/src/acp/types.zig @@ -2,10 +2,20 @@ const std = @import("std"); const build_options = @import("build_options"); const jsonrpc = @import("jsonrpc.zig"); const core_types = @import("../core/shared/types.zig"); +const io_mod = @import("../core/shared/io.zig"); const Allocator = std.mem.Allocator; const writeJsonStr = jsonrpc.writeJsonStr; +pub const MessageIdBuffer = [32]u8; + +pub fn generateMessageId(storage: *MessageIdBuffer) []const u8 { + var random_bytes: [16]u8 = undefined; + io_mod.getIo().random(&random_bytes); + storage.* = std.fmt.bytesToHex(random_bytes, .lower); + return storage; +} + pub const protocol_version: u32 = 1; pub fn writeModelRecoveryInfoUpdate( @@ -126,18 +136,37 @@ pub fn writeSessionUpdate(w: *std.Io.Writer, session_id: []const u8, update_json try w.writeAll("}"); } -pub fn writeAgentMessageChunk(w: *std.Io.Writer, text: []const u8) !void { - try w.writeAll("{\"sessionUpdate\":\"agent_message_chunk\",\"content\":{\"type\":\"text\",\"text\":"); +pub fn writeAgentMessageChunk(w: *std.Io.Writer, message_id: []const u8, text: []const u8) !void { + try w.writeAll("{\"sessionUpdate\":\"agent_message_chunk\",\"messageId\":"); + try writeJsonStr(message_id, w); + try w.writeAll(",\"content\":{\"type\":\"text\",\"text\":"); try writeJsonStr(text, w); try w.writeAll("}}"); } -pub fn writeUserMessageChunk(w: *std.Io.Writer, text: []const u8) !void { - try w.writeAll("{\"sessionUpdate\":\"user_message_chunk\",\"content\":{\"type\":\"text\",\"text\":"); +pub fn writeUserMessageChunk(w: *std.Io.Writer, message_id: []const u8, text: []const u8) !void { + try w.writeAll("{\"sessionUpdate\":\"user_message_chunk\",\"messageId\":"); + try writeJsonStr(message_id, w); + try w.writeAll(",\"content\":{\"type\":\"text\",\"text\":"); try writeJsonStr(text, w); try w.writeAll("}}"); } +pub fn writeUserImageChunk( + w: *std.Io.Writer, + message_id: []const u8, + media_type: []const u8, + bytes: []const u8, +) !void { + try w.writeAll("{\"sessionUpdate\":\"user_message_chunk\",\"messageId\":"); + try writeJsonStr(message_id, w); + try w.writeAll(",\"content\":{\"type\":\"image\",\"data\":\""); + try std.base64.standard.Encoder.encodeWriter(w, bytes); + try w.writeAll("\",\"mimeType\":"); + try writeJsonStr(media_type, w); + try w.writeAll("}}"); +} + pub fn writeToolCall( w: *std.Io.Writer, tool_call_id: []const u8, @@ -191,12 +220,12 @@ pub fn writeToolCallUpdateWithCommandResult( try w.writeAll("}"); } -pub fn writeInitializeResponse(w: *std.Io.Writer) !void { +pub fn writeInitializeResponse(w: *std.Io.Writer, image_prompts: bool) !void { try w.writeAll("{\"protocolVersion\":"); try w.print("{d}", .{protocol_version}); try w.writeAll(",\"agentCapabilities\":{"); try w.writeAll("\"loadSession\":true,"); - try w.writeAll("\"promptCapabilities\":{\"image\":false,\"audio\":false,\"embeddedContext\":true},"); + try w.print("\"promptCapabilities\":{{\"image\":{s},\"audio\":false,\"embeddedContext\":true}},", .{if (image_prompts) "true" else "false"}); try w.writeAll("\"mcpCapabilities\":{\"http\":true,\"sse\":true},"); try w.writeAll("\"sessionCapabilities\":{\"list\":{},\"resume\":{},\"close\":{}}"); try w.writeAll("},\"agentInfo\":{\"name\":\"fx\",\"title\":\"fx\",\"version\":"); @@ -217,12 +246,28 @@ pub fn writeAvailableCommandsUpdate(w: *std.Io.Writer, commands_json: []const u8 try w.writeAll("}"); } +pub fn writeSessionInfoUpdate(w: *std.Io.Writer, title: []const u8, updated_at: []const u8) !void { + try w.writeAll("{\"sessionUpdate\":\"session_info_update\",\"title\":"); + try writeJsonStr(title, w); + try w.writeAll(",\"updatedAt\":"); + try writeJsonStr(updated_at, w); + try w.writeAll("}"); +} + +pub fn writeUsageUpdate(w: *std.Io.Writer, used: u64, size: u64, complete_cost: ?f64) !void { + try w.print("{{\"sessionUpdate\":\"usage_update\",\"used\":{d},\"size\":{d}", .{ used, size }); + if (complete_cost) |amount| { + try w.print(",\"cost\":{{\"amount\":{d},\"currency\":\"USD\"}}", .{amount}); + } + try w.writeAll("}"); +} + test "writeAgentMessageChunk produces valid json" { const alloc = std.testing.allocator; var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); - try writeAgentMessageChunk(&out.writer, "Hello world"); - const expected = "{\"sessionUpdate\":\"agent_message_chunk\",\"content\":{\"type\":\"text\",\"text\":\"Hello world\"}}"; + try writeAgentMessageChunk(&out.writer, "message-1", "Hello world"); + const expected = "{\"sessionUpdate\":\"agent_message_chunk\",\"messageId\":\"message-1\",\"content\":{\"type\":\"text\",\"text\":\"Hello world\"}}"; try std.testing.expectEqualStrings(expected, out.writer.buffered()); } @@ -334,7 +379,7 @@ test "writeInitializeResponse contains required fields" { const alloc = std.testing.allocator; var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); - try writeInitializeResponse(&out.writer); + try writeInitializeResponse(&out.writer, true); var parsed = try std.json.parseFromSlice(std.json.Value, alloc, out.writer.buffered(), .{}); defer parsed.deinit(); @@ -347,7 +392,7 @@ test "writeInitializeResponse contains required fields" { build_options.app_version, parsed.value.object.get("agentInfo").?.object.get("version").?.string, ); - try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "\"image\":false") != null); + try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "\"image\":true") != null); try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "\"list\":{}") != null); try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "\"resume\":{}") != null); try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "\"close\":{}") != null); @@ -359,11 +404,22 @@ test "writeUserMessageChunk produces valid json" { const alloc = std.testing.allocator; var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); - try writeUserMessageChunk(&out.writer, "User says hello"); + try writeUserMessageChunk(&out.writer, "message-2", "User says hello"); try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "\"user_message_chunk\"") != null); try std.testing.expect(std.mem.find(u8, out.writer.buffered(), "User says hello") != null); } +test "writeUsageUpdate omits unproven cost" { + const alloc = std.testing.allocator; + var out: std.Io.Writer.Allocating = .init(alloc); + defer out.deinit(); + try writeUsageUpdate(&out.writer, 8, 128_000, null); + try std.testing.expectEqualStrings( + "{\"sessionUpdate\":\"usage_update\",\"used\":8,\"size\":128000}", + out.writer.buffered(), + ); +} + test "writeSessionUpdate wraps update with sessionId" { const alloc = std.testing.allocator; var out: std.Io.Writer.Allocating = .init(alloc); diff --git a/src/core/images/image_attachments.zig b/src/core/images/image_attachments.zig index 2af5df138..ab6c33521 100644 --- a/src/core/images/image_attachments.zig +++ b/src/core/images/image_attachments.zig @@ -470,6 +470,70 @@ pub fn captureBoundImageAttachment( return attachment; } +/// Captures caller-supplied image bytes into the immutable session snapshot +/// contract. The caller owns the returned attachment and must release it with +/// `types.freeImageAttachment` or `discardImageAttachment`. +pub fn captureInlineImageBytes( + alloc: std.mem.Allocator, + image_id: usize, + declared_media_type: []const u8, + bytes: []const u8, + snapshot_dir: []const u8, +) !types.ImageAttachment { + if (image_id == 0) return error.InvalidImageId; + if (bytes.len == 0 or declared_media_type.len == 0) return error.UnsupportedImageType; + if (bytes.len > max_image_bytes) return error.ImageTooLarge; + + var snapshot_dir_handle = try openOrCreateSnapshotDirectoryNoFollow(snapshot_dir); + defer snapshot_dir_handle.close(io_mod.getIo()); + var random_suffix: u64 = undefined; + io_mod.getIo().random(std.mem.asBytes(&random_suffix)); + const source_name = try std.fmt.allocPrint( + alloc, + "image-{d}.acp-source.{x}", + .{ image_id, random_suffix }, + ); + defer alloc.free(source_name); + defer deleteSnapshotFile(snapshot_dir_handle, source_name, "capture_acp_source"); + + { + var source = try snapshot_dir_handle.createFile( + io_mod.getIo(), + source_name, + .{ + .truncate = false, + .exclusive = true, + .permissions = std.Io.File.Permissions.fromMode(0o600), + .resolve_beneath = true, + }, + ); + defer source.close(io_mod.getIo()); + try source.writeStreamingAll(io_mod.getIo(), bytes); + try source.sync(io_mod.getIo()); + } + + const source_path = try std.fs.path.join(alloc, &.{ snapshot_dir, source_name }); + defer alloc.free(source_path); + var attachment = types.ImageAttachment{ + .id = image_id, + .path = try alloc.dupe(u8, source_path), + .media_type = try alloc.dupe(u8, declared_media_type), + }; + errdefer discardImageAttachment(alloc, attachment); + try captureImageSnapshot(alloc, &attachment, snapshot_dir); + if (!std.mem.eql(u8, attachment.media_type, declared_media_type)) { + return error.ImageSnapshotMediaTypeMismatch; + } + + const durable_path = try alloc.dupe( + u8, + attachment.snapshot_path orelse return error.MissingImageSnapshot, + ); + alloc.free(attachment.path); + attachment.path = durable_path; + return attachment; +} + fn captureImageSnapshotFromOpenFileWithBudget( alloc: std.mem.Allocator, attachment: *types.ImageAttachment, diff --git a/src/core/session/session_usage.zig b/src/core/session/session_usage.zig index 1d354e746..dec603399 100644 --- a/src/core/session/session_usage.zig +++ b/src/core/session/session_usage.zig @@ -113,6 +113,7 @@ pub const InvocationObservation = struct { usage_outcome: stream_provider.UsageOutcome, ) !void { const ledger = self.usage orelse return; + ledger.observeContextUsage(self.sequence, completion.usage); switch (usage_outcome) { .unavailable => |availability| { const delivery: DeliveryOutcome = switch (availability) { @@ -373,6 +374,8 @@ pub const Usage = struct { reasoning_tokens: ?u64, request_count: ?u64, billable_web_search_calls: u64 = 0, + latest_context_sequence: u64 = 0, + latest_context_used: ?u64 = null, lines_added: u64 = 0, lines_removed: u64 = 0, models: std.ArrayList(ModelAggregate) = .empty, @@ -1480,6 +1483,37 @@ pub const Usage = struct { self.dirty = true; } + pub const LiveContextSnapshot = struct { + used: u64, + complete_cost: ?f64, + }; + + /// Returns the latest provider-reported context usage. This state is + /// intentionally runtime-only: a restored billing aggregate cannot prove + /// how much context the next model request currently occupies. + pub fn liveContextSnapshot(self: *Usage) ?LiveContextSnapshot { + self.mutex.lockUncancelable(io_mod.getIo()); + defer self.mutex.unlock(io_mod.getIo()); + return .{ + .used = self.latest_context_used orelse return null, + .complete_cost = if (self.billing == .complete and std.math.isFinite(self.total_cost)) + self.total_cost + else + null, + }; + } + + fn observeContextUsage(self: *Usage, sequence: u64, provider_usage: types.Usage) void { + const input = provider_usage.input_tokens orelse return; + const output = provider_usage.output_tokens orelse return; + const used = std.math.add(u64, input, output) catch return; + self.mutex.lockUncancelable(io_mod.getIo()); + defer self.mutex.unlock(io_mod.getIo()); + if (sequence < self.latest_context_sequence) return; + self.latest_context_sequence = sequence; + self.latest_context_used = used; + } + /// Returns an owned point-in-time snapshot. The caller must call `deinit`. pub fn snapshot(self: *Usage, alloc: Allocator) !Snapshot { self.finishReconciliationIfDone(); @@ -1661,6 +1695,8 @@ pub const Usage = struct { self.active_started_at_ms = session_started_at_ms; } self.active_sequence_count = 0; + self.latest_context_sequence = 0; + self.latest_context_used = null; self.total_cost = copied.total_cost; self.input_tokens = copied.input_tokens; self.output_tokens = copied.output_tokens; @@ -2121,6 +2157,8 @@ pub const Usage = struct { self.wall_duration_ms = 0; self.active_started_at_ms = io_mod.milliTimestamp(); self.active_sequence_count = 0; + self.latest_context_sequence = 0; + self.latest_context_used = null; self.total_cost = 0; self.input_tokens = 0; self.output_tokens = 0; @@ -3275,6 +3313,17 @@ fn exactUsageOrigin(provider: model_provider.ProviderId) []const u8 { }; } +test "live context usage keeps the newest completed provider observation" { + var usage = Usage.initFresh(); + defer usage.deinit(std.testing.allocator); + + usage.observeContextUsage(2, .{ .input_tokens = 30, .output_tokens = 7 }); + usage.observeContextUsage(1, .{ .input_tokens = 1, .output_tokens = 1 }); + const snapshot = usage.liveContextSnapshot().?; + try std.testing.expectEqual(@as(u64, 37), snapshot.used); + try std.testing.expectEqual(@as(?f64, 0), snapshot.complete_cost); +} + test "direct exact generation IDs are deterministic and provider scoped" { var first_buffer: [30]u8 = undefined; var replay_buffer: [30]u8 = undefined; diff --git a/tests/e2e/acp.test.ts b/tests/e2e/acp.test.ts index cb17c139b..05d36e24a 100644 --- a/tests/e2e/acp.test.ts +++ b/tests/e2e/acp.test.ts @@ -562,6 +562,9 @@ class AcpClient { private waiters: Array<(line: string) => void> = []; private _closed = false; private _stderrChunks: Buffer[] = []; + private activeSessionId: string | null = null; + private pendingSessionTargets = new Map(); + private pendingNewSessions = new Set(); readonly rawLines: string[] = []; private permissionOptionId: "allow_once" | "allow_always" | "reject_once" = "reject_once"; private elicitationHandler?: ( @@ -642,7 +645,36 @@ class AcpClient { } send(msg: object): void { - this.proc.stdin!.write(JSON.stringify(msg) + "\n"); + let outgoing = msg as any; + if ( + outgoing.method === "session/new" && + (typeof outgoing.id === "number" || typeof outgoing.id === "string") + ) { + this.pendingNewSessions.add(outgoing.id); + } + if ( + (outgoing.method === "session/load" || outgoing.method === "session/resume") && + (typeof outgoing.id === "number" || typeof outgoing.id === "string") && + typeof outgoing.params?.sessionId === "string" + ) { + this.pendingSessionTargets.set(outgoing.id, outgoing.params.sessionId); + } + if ( + this.activeSessionId !== null && + [ + "session/prompt", + "session/cancel", + "session/set_mode", + "session/set_config_option", + ].includes(outgoing.method) && + outgoing.params?.sessionId === undefined + ) { + outgoing = { + ...outgoing, + params: { ...(outgoing.params ?? {}), sessionId: this.activeSessionId }, + }; + } + this.proc.stdin!.write(JSON.stringify(outgoing) + "\n"); } endStdin(): void { @@ -662,6 +694,18 @@ class AcpClient { }); }); const message = JSON.parse(line) as any; + if (typeof message.id === "number" || typeof message.id === "string") { + if (this.pendingNewSessions.delete(message.id)) { + if (message.error === undefined && typeof message.result?.sessionId === "string") { + this.activeSessionId = message.result.sessionId; + } + } + const sessionId = this.pendingSessionTargets.get(message.id); + if (sessionId !== undefined) { + this.pendingSessionTargets.delete(message.id); + if (message.error === undefined) this.activeSessionId = sessionId; + } + } if (message.method === "session/request_permission" && message.id !== undefined) { this.send({ jsonrpc: "2.0", @@ -689,7 +733,22 @@ class AcpClient { async request(method: string, params?: object, id?: number): Promise { const reqId = id ?? Math.floor(Math.random() * 100000); this.send({ jsonrpc: "2.0", id: reqId, method, params: params ?? {} }); - const resp = await this.readLine(); + let resp: any; + do { + resp = await this.readLine() as any; + } while (resp.id !== reqId); + if (resp.error === undefined) { + if (method === "session/new" && typeof resp.result?.sessionId === "string") { + this.activeSessionId = resp.result.sessionId; + } else if ( + (method === "session/load" || method === "session/resume") && + typeof (params as any)?.sessionId === "string" + ) { + this.activeSessionId = (params as any).sessionId; + } else if (method === "session/close") { + this.activeSessionId = null; + } + } return resp as object; } @@ -963,13 +1022,18 @@ async function runMcpToolPrompt( ).toContain(expectedResult); } -async function continueRecovery(client: AcpClient, timeoutMs = LIVE_TIMEOUT) { +async function continueRecovery( + client: AcpClient, + timeoutMs = LIVE_TIMEOUT, + sessionId?: string, +) { const promptId = Math.floor(Math.random() * 100000) + 1000; client.send({ jsonrpc: "2.0", id: promptId, method: "session/prompt", params: { + ...(sessionId ? { sessionId } : {}), prompt: [], _meta: { fx: { continueRecovery: true } }, }, @@ -1130,7 +1194,7 @@ describe("acp: model-independent", () => { ); test( - "session/new omits the removed summary command", + "session/new advertises no unsupported slash commands", async () => { const root = createIsolatedRoot("fx-acp-available-commands-"); const gateway = startFakeGateway([]); @@ -1147,8 +1211,7 @@ describe("acp: model-independent", () => { const commandNames = notification.params.update.availableCommands.map( (command: any) => command.name, ); - expect(commandNames).toContain("compact"); - expect(commandNames).not.toContain("summary"); + expect(commandNames).toEqual([]); expect(client.stderr).toBe(""); } finally { await client?.close(); @@ -1346,15 +1409,104 @@ describe("acp: model-independent", () => { const result = await runPrompt(client, "Return the Markdown fixture.", TIMEOUT); expect(result.promptResult.result.stopReason).toBe("end_turn"); - const responseText = result.messages - .filter((message) => + const agentChunks = result.messages.filter((message) => message.method === "session/update" && message.params?.update?.sessionUpdate === "agent_message_chunk" - ) + ); + const responseText = agentChunks .map((message) => message.params.update.content.text) .join(""); expect(responseText).toBe(markdown.join("")); expect(responseText).not.toContain("\u001b"); + expect(agentChunks.length).toBeGreaterThan(1); + expect(typeof agentChunks[0]?.params.update.messageId).toBe("string"); + expect(agentChunks[0]?.params.update.messageId.length).toBeGreaterThan(0); + expect(new Set(agentChunks.map((message) => message.params.update.messageId)).size).toBe(1); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "ACP publishes session title and authoritative context usage", + async () => { + const root = createIsolatedRoot("fx-acp-session-metadata-"); + const title = "Publish ACP session metadata"; + const gateway = startFakeGateway( + [finalText("Metadata published.")], + { + models: [{ + id: FAKE_GATEWAY_MODEL, + type: "language", + tags: ["tool-use"], + context_window: 128_000, + }], + }, + ); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + const sessionId = await startCodeSession(client); + + const result = await runPrompt(client, title, TIMEOUT); + expect(result.promptResult.result.stopReason).toBe("end_turn"); + const usage = result.messages.find((message) => + message.method === "session/update" && + message.params?.update?.sessionUpdate === "usage_update" + ); + expect(usage?.params).toMatchObject({ + sessionId, + update: { + sessionUpdate: "usage_update", + used: 8, + size: 128_000, + }, + }); + expect(usage?.params.update.cost).toBeUndefined(); + + const info = result.messages.find((message) => + message.method === "session/update" && + message.params?.update?.sessionUpdate === "session_info_update" && + message.params?.update?.title === title + ); + expect(info?.params.sessionId).toBe(sessionId); + expect(Number.isNaN(Date.parse(info?.params.update.updatedAt))).toBe(false); + + await client.close(); + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + await client.request("initialize", { protocolVersion: 1 }, 90); + client.send({ + jsonrpc: "2.0", + id: 91, + method: "session/load", + params: { sessionId, cwd: root.workspace, mcpServers: [] }, + }); + const replay: any[] = []; + while (true) { + const message = await client.readLine() as any; + if (message.id === 91) break; + replay.push(message); + } + expect(replay).toContainEqual(expect.objectContaining({ + method: "session/update", + params: expect.objectContaining({ + sessionId, + update: expect.objectContaining({ + sessionUpdate: "session_info_update", + title, + }), + }), + })); expect(client.stderr).toBe(""); } finally { await client?.close(); @@ -1428,7 +1580,7 @@ describe("acp: model-independent", () => { expect(occurrenceCount(loadUpdates, toolEvidence)).toBe(1); expect(occurrenceCount(loadUpdates, partialText)).toBe(1); - const resumed = await continueRecovery(client, TIMEOUT); + const resumed = await continueRecovery(client, TIMEOUT, sessionId); expect(resumed.promptResult.error).toBeUndefined(); expect(resumed.promptResult.result.stopReason).toBe("end_turn"); expect(gateway.requests).toHaveLength(12); @@ -1874,7 +2026,7 @@ describe("acp: model-independent", () => { ); test( - "initialize reports that image prompt blocks are unsupported", + "initialize advertises native image prompt support", async () => { const root = createIsolatedRoot("fx-acp-initialize-"); try { @@ -1908,7 +2060,7 @@ describe("acp: model-independent", () => { expect(resp.result.agentInfo.name).toBe("fx"); expect(resp.result.agentInfo.version).toBe(version.stdout.trim()); expect(resp.result.agentCapabilities.loadSession).toBe(true); - expect(resp.result.agentCapabilities.promptCapabilities.image).toBe(false); + expect(resp.result.agentCapabilities.promptCapabilities.image).toBe(true); expect(resp.result.agentCapabilities.mcpCapabilities.http).toBe(true); expect(resp.result.agentCapabilities.mcpCapabilities.sse).toBe(true); expect(resp.result.agentCapabilities.sessionCapabilities.resume).toEqual({}); @@ -4123,6 +4275,7 @@ describe("acp: model-independent", () => { }), ]); let directRequestSeen = false; + let elicitationRequestId: number | string | null = null; try { client = await AcpClient.create({ cwd: root.workspace, @@ -4136,7 +4289,7 @@ describe("acp: model-independent", () => { }, 1, ); - await client.request( + const created = await client.request( "session/new", { cwd: root.workspace, @@ -4148,11 +4301,13 @@ describe("acp: model-independent", () => { )], }, 2, - ); + ) as any; + const sessionId = created.result.sessionId as string; await client.readLine(); - await client.request("session/set_mode", { modeId: "code" }, 3); - client.setElicitationHandler(() => { + await client.request("session/set_mode", { sessionId, modeId: "code" }, 3); + client.setElicitationHandler((_params, id) => { directRequestSeen = true; + elicitationRequestId = id; return undefined; }); @@ -4164,21 +4319,32 @@ describe("acp: model-independent", () => { jsonrpc: "2.0", id: cancelId, method: "session/cancel", - params: {}, + params: { sessionId }, }); const responses = new Map(); + let requestCancellation: any = null; const deadline = Date.now() + 3_000; - while (responses.size < 2 && Date.now() < deadline) { - const message = await client.readLine( - Math.max(100, deadline - Date.now()), - ) as any; + while ((responses.size < 2 || requestCancellation === null) && Date.now() < deadline) { + let message: any; + try { + message = await client.readLine(Math.max(100, deadline - Date.now())); + } catch (error) { + if (error instanceof AcpReadTimeoutError) break; + throw error; + } + if (message.method === "$/cancel_request") requestCancellation = message; if (message.id === promptId || message.id === cancelId) { responses.set(message.id, message); } } expect(responses.get(cancelId)?.result).toBeNull(); expect(responses.get(promptId)?.result?.stopReason).toBe("cancelled"); + expect(requestCancellation).toMatchObject({ + jsonrpc: "2.0", + method: "$/cancel_request", + params: { requestId: elicitationRequestId }, + }); expect(client.stderr).toBe(""); } finally { await client?.close(); @@ -4677,19 +4843,94 @@ describe("acp: model-independent", () => { ); test( - "image prompt rejection preserves history and admits the next text prompt", + "session-scoped requests reject a stale active-session target", async () => { - const root = createIsolatedRoot("fx-acp-image-rejection-"); - const boundary = createPromptTerminalBoundary(root.root); - const gateway = startFakeGateway([finalText("valid image follow-up complete")]); + const root = createIsolatedRoot("fx-acp-stale-session-target-"); + const gateway = startFakeGateway([finalText("stale prompt executed")]); try { client = await AcpClient.create({ cwd: root.workspace, - env: { - ...fakeGatewayEnv(root, gateway), - ...boundary.env, + env: fakeGatewayEnv(root, gateway), + }); + await client.request("initialize", { protocolVersion: 1 }, 1); + const first = await client.request( + "session/new", + { cwd: root.workspace, mcpServers: [] }, + 2, + ) as any; + await client.readLine(); + const second = await client.request( + "session/new", + { cwd: root.workspace, mcpServers: [] }, + 3, + ) as any; + await client.readLine(); + const staleSessionId = first.result.sessionId as string; + const activeSessionId = second.result.sessionId as string; + expect(staleSessionId).not.toBe(activeSessionId); + + for (const [id, method, params] of [ + [4, "session/set_mode", { sessionId: staleSessionId, modeId: "code" }], + [5, "session/set_config_option", { + sessionId: staleSessionId, + configId: "mode", + value: "code", + }], + [6, "session/cancel", { sessionId: staleSessionId }], + ] as const) { + const response = await client.request(method, params, id) as any; + expect(response.error).toMatchObject({ + code: -32602, + message: "Session is not active", + }); + } + + const promptId = 7; + client.send({ + jsonrpc: "2.0", + id: promptId, + method: "session/prompt", + params: { + sessionId: staleSessionId, + prompt: [{ type: "text", text: "Do not execute this stale prompt." }], }, }); + const promptResponse = await readResponse(client, promptId); + expect(promptResponse.error).toMatchObject({ + code: -32602, + message: "Session is not active", + }); + expect(gateway.requests).toHaveLength(0); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "image prompt reaches the Gateway and replays from saved history", + async () => { + const root = createIsolatedRoot("fx-acp-image-prompt-"); + const imageData = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9WlXYX0AAAAASUVORK5CYII="; + const gateway = startFakeGateway( + [finalText("image prompt complete")], + { + models: [{ + id: FAKE_GATEWAY_MODEL, + type: "language", + tags: ["vision", "file-input", "tool-use"], + }], + }, + ); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); const sessionId = await startCodeSession(client); client.send({ jsonrpc: "2.0", @@ -4698,37 +4939,105 @@ describe("acp: model-independent", () => { params: { prompt: [ { type: "text", text: "Describe this image." }, - { type: "image", data: "aGVsbG8=", mimeType: "image/png" }, + { type: "image", data: imageData, mimeType: "image/png" }, ], }, }); - const invalid = await readResponse(client, 94); - expect(invalid.error).toEqual({ - code: -32602, - message: "Image prompt blocks are not supported", - }); - expect(gateway.requests).toHaveLength(0); - const detailBefore = await runFx(["session", "--id", sessionId, "--json"], { + const completed = await readResponse(client, 94); + expect(completed.error).toBeUndefined(); + expect(completed.result.stopReason).toBe("end_turn"); + expect(gateway.requests).toHaveLength(1); + expect(gateway.requests[0]!.body).toContain("image/png"); + expect(gateway.requests[0]!.body).toContain(imageData); + + const detail = await runFx(["session", "--id", sessionId, "--json"], { cwd: root.workspace, env: { HOME: root.home }, timeoutMs: TIMEOUT, }); - expect(detailBefore.code).toBe(0); - expect(JSON.parse(detailBefore.stdout).history_len).toBe(0); - await waitForPath(boundary.terminalReady); + expect(detail.code).toBe(0); + expect(JSON.parse(detail.stdout).history_len).toBe(1); + await client.close(); - sendPrompt(client, 95, "Complete the valid prompt."); - await waitForPath(boundary.reapReady); - releasePromptBoundary(boundary); + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + await client.request("initialize", { protocolVersion: 1 }, 95); + client.send({ + jsonrpc: "2.0", + id: 96, + method: "session/load", + params: { sessionId, cwd: root.workspace, mcpServers: [] }, + }); + const replay: any[] = []; + while (true) { + const message = await client.readLine() as any; + if (message.id === 96) break; + replay.push(message); + } + const imageChunk = replay.find((message) => + message.params?.update?.sessionUpdate === "user_message_chunk" && + message.params?.update?.content?.type === "image" + ); + expect(imageChunk?.params.update.content).toMatchObject({ + type: "image", + mimeType: "image/png", + data: imageData, + }); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); - const valid = await readResponse(client, 95); - expect(valid.error).toBeUndefined(); - expect(valid.result.stopReason).toBe("end_turn"); + test( + "image prompt MIME mismatch fails before the Gateway without an orphaned snapshot", + async () => { + const root = createIsolatedRoot("fx-acp-image-mime-mismatch-"); + const imageData = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9WlXYX0AAAAASUVORK5CYII="; + const gateway = startFakeGateway([finalText("ACP image recovery complete")]); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + const sessionId = await startCodeSession(client); + client.send({ + jsonrpc: "2.0", + id: 95, + method: "session/prompt", + params: { + prompt: [ + { type: "text", text: "Reject the mismatched image." }, + { type: "image", data: imageData, mimeType: "image/jpeg" }, + ], + }, + }); + + const rejected = await readResponse(client, 95); + expect(rejected.error).toEqual({ + code: -32602, + message: "Invalid image prompt block", + }); + expect(gateway.requests).toHaveLength(0); + const imageDir = join(root.home, ".fx", "sessions", sessionId, "images"); + if (existsSync(imageDir)) expect(readdirSync(imageDir)).toEqual([]); + + const recovered = await runPrompt( + client, + "Confirm the ACP connection remains usable.", + TIMEOUT, + ); + expect(recovered.promptResult.result.stopReason).toBe("end_turn"); expect(gateway.requests).toHaveLength(1); expect(client.stderr).toBe(""); } finally { - releasePromptBoundary(boundary); await client?.close(); gateway.stop(); rmSync(root.root, { recursive: true, force: true }); @@ -5977,6 +6286,15 @@ describe("acp: model-independent", () => { .map((message) => message.params.update.content.text); expect(userText).toEqual([promptText]); expect(agentText).toEqual([answer]); + const replayMessageIds = loadMessages + .filter((message) => + message.params?.update?.sessionUpdate === "user_message_chunk" || + message.params?.update?.sessionUpdate === "agent_message_chunk" + ) + .map((message) => message.params.update.messageId); + expect(replayMessageIds).toHaveLength(2); + expect(replayMessageIds.every((id) => typeof id === "string" && id.length > 0)).toBe(true); + expect(new Set(replayMessageIds).size).toBe(2); expect(JSON.stringify(loadMessages)).not.toContain("Previous tool execution:"); expect(client.stderr).toBe(""); } finally { @@ -7030,7 +7348,7 @@ describe("acp: model-independent", () => { ); test( - "ACP cancellation aborts held automatic review and keeps server usable", + "protocol request cancellation aborts held automatic review and keeps server usable", async () => { const root = createIsolatedRoot("fx-acp-auto-review-cancel-"); const marker = join(root.workspace, "cancelled-review-must-not-run.txt"); @@ -7061,23 +7379,12 @@ describe("acp: model-independent", () => { ); client.send({ jsonrpc: "2.0", - id: 397, - method: "session/cancel", - params: {}, + method: "$/cancel_request", + params: { requestId: 396 }, }); - const terminalResponses = new Map(); - const deadline = Date.now() + TIMEOUT; - while (terminalResponses.size < 2 && Date.now() < deadline) { - const message = await client.readLine( - Math.min(3_000, Math.max(100, deadline - Date.now())), - ) as any; - if (message.id === 396 || message.id === 397) { - terminalResponses.set(message.id, message); - } - } - expect(terminalResponses.get(397)?.result).toBeNull(); - expect(terminalResponses.get(396)?.result?.stopReason).toBe("cancelled"); + const promptResponse = await readResponse(client, 396); + expect(promptResponse.result?.stopReason).toBe("cancelled"); heldReview.resolve(fakeGatewayPermissionDecision("clear")); await Bun.sleep(100); From 34ec3d32a4bebe23e37c294b6409b8bb0db53c53 Mon Sep 17 00:00:00 2001 From: Pranit Date: Tue, 1 Sep 2026 12:31:19 -0400 Subject: [PATCH 2/6] Correlate ACP recovery responses - Match recovery responses by JSON-RPC request ID. - Include the active session target on recovery model changes. --- tests/e2e/session-recovery.test.ts | 45 ++++++++++++++++++------------ 1 file changed, 27 insertions(+), 18 deletions(-) diff --git a/tests/e2e/session-recovery.test.ts b/tests/e2e/session-recovery.test.ts index bf0fe0f9d..1e35c9ff1 100644 --- a/tests/e2e/session-recovery.test.ts +++ b/tests/e2e/session-recovery.test.ts @@ -68,6 +68,15 @@ class LineClient { ); } + async readResponse(id: number, timeoutMs = TIMEOUT): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const message = await this.read(Math.max(1, deadline - Date.now())); + if (message.id === id) return message; + } + throw new Error(`timed out waiting for ACP response id=${id}`); + } + kill(): void { this.proc.kill("SIGKILL"); } @@ -101,9 +110,9 @@ async function createSession(cwd: string, home: string): Promise { const client = startAcp(cwd, home); try { client.send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: 1 } }); - expect((await client.read()).result).toBeDefined(); + expect((await client.readResponse(1)).result).toBeDefined(); client.send({ jsonrpc: "2.0", id: 2, method: "session/new", params: { mcpServers: [] } }); - const response = await client.read(); + const response = await client.readResponse(2); expect(response.result?.sessionId).toBeDefined(); return response.result.sessionId; } finally { @@ -141,7 +150,7 @@ describe("session recovery", () => { FX_E2E_SESSION_BOUNDARY_READY: ready, }); first.send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: 1 } }); - expect((await first.read()).result).toBeDefined(); + expect((await first.readResponse(1)).result).toBeDefined(); first.send({ jsonrpc: "2.0", id: 2, method: "session/new", params: { mcpServers: [] } }); await waitForPath(ready); first.kill(); @@ -196,7 +205,7 @@ describe("session recovery", () => { FX_E2E_SESSION_BOUNDARY_READY: ready, }); first.send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: 1 } }); - expect((await first.read()).result).toBeDefined(); + expect((await first.readResponse(1)).result).toBeDefined(); first.send({ jsonrpc: "2.0", id: 2, method: "session/new", params: { mcpServers: [] } }); await waitForPath(ready); first.kill(); @@ -212,9 +221,9 @@ describe("session recovery", () => { const resolver = startAcp(workspaceRoot, home); resolver.send({ jsonrpc: "2.0", id: 3, method: "initialize", params: { protocolVersion: 1 } }); - expect((await resolver.read()).result).toBeDefined(); + expect((await resolver.readResponse(3)).result).toBeDefined(); resolver.send({ jsonrpc: "2.0", id: 4, method: "session/load", params: { sessionId, mcpServers: [] } }); - expect((await resolver.read()).result).toBeDefined(); + expect((await resolver.readResponse(4)).result).toBeDefined(); resolver.kill(); const detail = await runFx(["session", "--id", sessionId, "--json"], { @@ -279,21 +288,21 @@ describe("session recovery", () => { const writer = startAcp(workspaceRoot, home); writer.send({ jsonrpc: "2.0", id: 10, method: "initialize", params: { protocolVersion: 1 } }); - expect((await writer.read()).result).toBeDefined(); + expect((await writer.readResponse(10)).result).toBeDefined(); writer.send({ jsonrpc: "2.0", id: 11, method: "session/load", params: { sessionId, mcpServers: [] }, }); - expect((await writer.read()).result).toBeDefined(); + expect((await writer.readResponse(11)).result).toBeDefined(); writer.send({ jsonrpc: "2.0", id: 12, method: "session/set_config_option", - params: { configId: "model", value: "o4-mini" }, + params: { sessionId, configId: "model", value: "o4-mini" }, }); - expect((await writer.read()).result).toBeDefined(); + expect((await writer.readResponse(12)).result).toBeDefined(); writer.kill(); await Bun.sleep(100); @@ -481,7 +490,7 @@ describe("session recovery", () => { FX_E2E_SESSION_BOUNDARY_READY: ready, }); first.send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: 1 } }); - expect((await first.read()).result).toBeDefined(); + expect((await first.readResponse(1)).result).toBeDefined(); first.send({ jsonrpc: "2.0", id: 2, method: "session/new", params: { mcpServers: [] } }); await waitForPath(ready); first.kill(); @@ -521,9 +530,9 @@ describe("session recovery", () => { FX_TRACE_LOG: resolverTrace, }); resolver.send({ jsonrpc: "2.0", id: 3, method: "initialize", params: { protocolVersion: 1 } }); - expect((await resolver.read()).result).toBeDefined(); + expect((await resolver.readResponse(3)).result).toBeDefined(); resolver.send({ jsonrpc: "2.0", id: 4, method: "session/load", params: { sessionId, mcpServers: [] } }); - const loadResponse = await resolver.read(); + const loadResponse = await resolver.readResponse(4); resolver.kill(); expect(readFileSync(resolverTrace, "utf8")).toContain( "session operation=load outcome=failed error=SessionNotFound", @@ -585,14 +594,14 @@ describe("session recovery", () => { FX_E2E_SESSION_BOUNDARY_READY: ready, }); writer.send({ jsonrpc: "2.0", id: 10, method: "initialize", params: { protocolVersion: 1 } }); - expect((await writer.read()).result).toBeDefined(); + expect((await writer.readResponse(10)).result).toBeDefined(); writer.send({ jsonrpc: "2.0", id: 11, method: "session/load", params: { sessionId, mcpServers: [] } }); - expect((await writer.read()).result).toBeDefined(); + expect((await writer.readResponse(11)).result).toBeDefined(); writer.send({ jsonrpc: "2.0", id: 12, method: "session/set_config_option", - params: { configId: "model", value: "o4-mini" }, + params: { sessionId, configId: "model", value: "o4-mini" }, }); await waitForPath(ready); writer.kill(); @@ -615,9 +624,9 @@ describe("session recovery", () => { const resolver = startAcp(workspaceRoot, home); resolver.send({ jsonrpc: "2.0", id: 20, method: "initialize", params: { protocolVersion: 1 } }); - expect((await resolver.read()).result).toBeDefined(); + expect((await resolver.readResponse(20)).result).toBeDefined(); resolver.send({ jsonrpc: "2.0", id: 21, method: "session/load", params: { sessionId, mcpServers: [] } }); - const loaded = await resolver.read(); + const loaded = await resolver.readResponse(21); expect(loaded.result).toBeDefined(); const loadedModel = loaded.result.configOptions.find( (option: { id: string; currentValue: string }) => option.id === "model", From 736d64e0477decf9f8a3e2f873f24958dc88a2b6 Mon Sep 17 00:00:00 2001 From: Pranit Date: Tue, 1 Sep 2026 12:52:53 -0400 Subject: [PATCH 3/6] Update remaining ACP test clients - Preserve the active session target in web and command fixtures. - Correlate fixture responses by JSON-RPC request ID. --- tests/e2e/tui-command-permissions.test.ts | 32 +++++++++++++++++++++-- tests/e2e/web-fetch-fake-network.test.ts | 32 +++++++++++++++++++++-- tests/e2e/web-search-fake-gateway.test.ts | 32 +++++++++++++++++++++-- 3 files changed, 90 insertions(+), 6 deletions(-) diff --git a/tests/e2e/tui-command-permissions.test.ts b/tests/e2e/tui-command-permissions.test.ts index f15a4bd3e..739911d17 100644 --- a/tests/e2e/tui-command-permissions.test.ts +++ b/tests/e2e/tui-command-permissions.test.ts @@ -3741,6 +3741,7 @@ class AcpClient { private waiters: Array<(line: string) => void> = []; private closed = false; private stderrChunks: Buffer[] = []; + private activeSessionId: string | null = null; private constructor(private proc: ChildProcess) { proc.stdout!.on("data", (chunk: Buffer) => { @@ -3773,7 +3774,23 @@ class AcpClient { } send(message: object) { - this.proc.stdin!.write(`${JSON.stringify(message)}\n`); + let outgoing = message as any; + if ( + this.activeSessionId !== null && + [ + "session/prompt", + "session/cancel", + "session/set_mode", + "session/set_config_option", + ].includes(outgoing.method) && + outgoing.params?.sessionId === undefined + ) { + outgoing = { + ...outgoing, + params: { ...(outgoing.params ?? {}), sessionId: this.activeSessionId }, + }; + } + this.proc.stdin!.write(`${JSON.stringify(outgoing)}\n`); } async readLine(timeoutMs = TIMEOUT): Promise { @@ -3802,7 +3819,18 @@ class AcpClient { async request(method: string, params: object, id: number) { this.send({ jsonrpc: "2.0", id, method, params }); - return this.readLine(); + let response: any; + do { + response = await this.readLine(); + } while (response.id !== id); + if ( + response.error === undefined && + method === "session/new" && + typeof response.result?.sessionId === "string" + ) { + this.activeSessionId = response.result.sessionId; + } + return response; } async close() { diff --git a/tests/e2e/web-fetch-fake-network.test.ts b/tests/e2e/web-fetch-fake-network.test.ts index 69a3bc317..90d89eb1a 100644 --- a/tests/e2e/web-fetch-fake-network.test.ts +++ b/tests/e2e/web-fetch-fake-network.test.ts @@ -200,6 +200,7 @@ class AcpClient { private lines: string[] = []; private waiters: Array<(line: string) => void> = []; private closed = false; + private activeSessionId: string | null = null; private constructor(private proc: ChildProcess) { proc.stdout!.on("data", (chunk: Buffer) => { @@ -232,7 +233,23 @@ class AcpClient { } send(message: object) { - this.proc.stdin!.write(`${JSON.stringify(message)}\n`); + let outgoing = message as any; + if ( + this.activeSessionId !== null && + [ + "session/prompt", + "session/cancel", + "session/set_mode", + "session/set_config_option", + ].includes(outgoing.method) && + outgoing.params?.sessionId === undefined + ) { + outgoing = { + ...outgoing, + params: { ...(outgoing.params ?? {}), sessionId: this.activeSessionId }, + }; + } + this.proc.stdin!.write(`${JSON.stringify(outgoing)}\n`); } async readLine(timeoutMs = TIMEOUT): Promise { @@ -253,7 +270,18 @@ class AcpClient { async request(method: string, params: object, id: number) { this.send({ jsonrpc: "2.0", id, method, params }); - return this.readLine(); + let response: any; + do { + response = await this.readLine(); + } while (response.id !== id); + if ( + response.error === undefined && + method === "session/new" && + typeof response.result?.sessionId === "string" + ) { + this.activeSessionId = response.result.sessionId; + } + return response; } async close() { diff --git a/tests/e2e/web-search-fake-gateway.test.ts b/tests/e2e/web-search-fake-gateway.test.ts index 6a882b5c9..fc58bbe11 100644 --- a/tests/e2e/web-search-fake-gateway.test.ts +++ b/tests/e2e/web-search-fake-gateway.test.ts @@ -353,6 +353,7 @@ class AcpClient { private lines: string[] = []; private waiters: Array<(line: string) => void> = []; private closed = false; + private activeSessionId: string | null = null; private constructor(private proc: ChildProcess) { proc.stdout!.on("data", (chunk: Buffer) => { @@ -385,7 +386,23 @@ class AcpClient { } send(message: object) { - this.proc.stdin!.write(`${JSON.stringify(message)}\n`); + let outgoing = message as any; + if ( + this.activeSessionId !== null && + [ + "session/prompt", + "session/cancel", + "session/set_mode", + "session/set_config_option", + ].includes(outgoing.method) && + outgoing.params?.sessionId === undefined + ) { + outgoing = { + ...outgoing, + params: { ...(outgoing.params ?? {}), sessionId: this.activeSessionId }, + }; + } + this.proc.stdin!.write(`${JSON.stringify(outgoing)}\n`); } async readLine(timeoutMs = TIMEOUT): Promise { @@ -406,7 +423,18 @@ class AcpClient { async request(method: string, params: object, id: number) { this.send({ jsonrpc: "2.0", id, method, params }); - return this.readLine(); + let response: any; + do { + response = await this.readLine(); + } while (response.id !== id); + if ( + response.error === undefined && + method === "session/new" && + typeof response.result?.sessionId === "string" + ) { + this.activeSessionId = response.result.sessionId; + } + return response; } async close() { From fcdbabadeb73d26efa7680084ae91e6156bb23e7 Mon Sep 17 00:00:00 2001 From: Pranit Date: Tue, 1 Sep 2026 13:48:26 -0400 Subject: [PATCH 4/6] Harden ACP replay state boundaries - Clear stale context usage and rotate resumed message IDs. - Transfer image ownership when runtime history adopts a turn. - Canonicalize image replay and report unavailable snapshots. --- src/acp/prompt.zig | 102 +++++++++++++++++++++++---- src/acp/sessions.zig | 72 ++++++++++++++++++- src/core/session/session_usage.zig | 22 +++++- tests/e2e/acp.test.ts | 109 ++++++++++++++++++++++++++++- 4 files changed, 282 insertions(+), 23 deletions(-) diff --git a/src/acp/prompt.zig b/src/acp/prompt.zig index ae0a147f9..85b727570 100644 --- a/src/acp/prompt.zig +++ b/src/acp/prompt.zig @@ -115,6 +115,11 @@ const ProviderTerminalPublication = enum { published, }; +const AgentMessageKind = enum { + assistant, + operational, +}; + const AcpContext = struct { alloc: Allocator, state: *server.ServerState, @@ -123,10 +128,8 @@ const AcpContext = struct { /// copies of provider call ids so the ID stays stable from permission /// review through execution. published_tool_calls: std.StringHashMapUnmanaged(ProviderTerminalPublication) = .empty, - assistant_message_id: acp_types.MessageIdBuffer = undefined, - operational_message_id: acp_types.MessageIdBuffer = undefined, - assistant_message_id_ready: bool = false, - operational_message_id_ready: bool = false, + message_id: acp_types.MessageIdBuffer = undefined, + message_kind: ?AgentMessageKind = null, stop_reason: acp_types.StopReason = .end_turn, auto_classifier: permission_auto_classifier.Classifier = permission_auto_classifier.Classifier.disabled(), @@ -135,6 +138,7 @@ const AcpContext = struct { captured_mode: ?[]const u8 = null, captured_permission_mode: ?PermissionMode = null, retain_external_root_user_turn: bool = false, + current_prompt_input: ?*ParsedPromptInput = null, fn deinitPublishedToolCalls(self: *AcpContext) void { var keys = self.published_tool_calls.keyIterator(); @@ -150,19 +154,19 @@ const AcpContext = struct { } fn assistantMessageId(self: *AcpContext) []const u8 { - if (!self.assistant_message_id_ready) { - _ = acp_types.generateMessageId(&self.assistant_message_id); - self.assistant_message_id_ready = true; - } - return &self.assistant_message_id; + return self.messageId(.assistant); } fn operationalMessageId(self: *AcpContext) []const u8 { - if (!self.operational_message_id_ready) { - _ = acp_types.generateMessageId(&self.operational_message_id); - self.operational_message_id_ready = true; + return self.messageId(.operational); + } + + fn messageId(self: *AcpContext, kind: AgentMessageKind) []const u8 { + if (self.message_kind != kind) { + _ = acp_types.generateMessageId(&self.message_id); + self.message_kind = kind; } - return &self.operational_message_id; + return &self.message_id; } fn sendAgentText(self: *AcpContext, message_id: []const u8, text: []const u8) !void { @@ -603,6 +607,7 @@ pub fn handlePrompt( .session_id = session.session_id, .captured_mode = captured_mode, .captured_permission_mode = captured_permission_mode, + .current_prompt_input = &prompt_input, }; defer ctx.deinitPublishedToolCalls(); @@ -1837,6 +1842,8 @@ fn propagateHistoryTurn(raw_ctx: *anyopaque, turn: HistoryTurn) !void { session, turn, ctx.retain_external_root_user_turn, + ctx.current_prompt_input, + .{}, ); } } @@ -1846,10 +1853,13 @@ fn persistAcpHistoryTurn( session: *server.ActiveSessionState, turn: HistoryTurn, prompt_is_root_authority: bool, + current_prompt_input: ?*ParsedPromptInput, + options: session_log.Options, ) !void { session.session_write_mutex.lockUncancelable(io_mod.getIo()); defer session.session_write_mutex.unlock(io_mod.getIo()); try session.session_rt.appendHistoryEntry(alloc, turn); + if (current_prompt_input) |prompt_input| prompt_input.retainImageSnapshots(); if (comptime host_target.is_wasm) { try sessions.commitWasmSessionLocked(alloc, session); return; @@ -1886,7 +1896,7 @@ fn persistAcpHistoryTurn( } }, io_mod.milliTimestamp(), .retry_expected_tail, - .{}, + options, ) catch |err| switch (err) { error.EventFrameTooLarge => { try commitAcpStateReplacement(alloc, session, writable, true); @@ -1971,7 +1981,7 @@ test "ACP degraded history repair commits the finished turn once" { const turn = try session_runtime.makeAssistantTurn(alloc, "hello", "done"); defer types.freeHistoryTurn(alloc, turn); - try persistAcpHistoryTurn(alloc, &session, turn, true); + try persistAcpHistoryTurn(alloc, &session, turn, true, null, .{}); try std.testing.expect(session.writable.?.degradedTail() == null); try std.testing.expectEqual(@as(usize, 1), session.session_rt.history.items.len); @@ -1980,6 +1990,46 @@ test "ACP degraded history repair commits the finished turn once" { "done", session.writable.?.state.history[0].assistant.assistant, ); + + const image_path = try std.fs.path.join(alloc, &.{ workspace, "image-1.bin" }); + defer alloc.free(image_path); + var image_file = try std.Io.Dir.createFileAbsolute(io_mod.getIo(), image_path, .{}); + image_file.close(io_mod.getIo()); + const input_images = try alloc.alloc(types.ImageAttachment, 1); + input_images[0] = .{ + .id = 1, + .path = try alloc.dupe(u8, "/tmp/image.png"), + .media_type = try alloc.dupe(u8, "image/png"), + .snapshot_path = try alloc.dupe(u8, image_path), + .snapshot_sha256 = try alloc.dupe(u8, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + }; + var prompt_input = ParsedPromptInput{ + .text = try alloc.dupe(u8, "[Image #1]"), + .images = input_images, + }; + const failed_turn: HistoryTurn = .{ .assistant = .{ + .user = .{ + .text = try alloc.dupe(u8, prompt_input.text), + .images = try types.dupeImageAttachmentSlice(alloc, prompt_input.images), + }, + .assistant = try alloc.dupe(u8, "not persisted"), + } }; + defer types.freeHistoryTurn(alloc, failed_turn); + try std.testing.expectError( + error.SessionPersistenceDegraded, + persistAcpHistoryTurn( + alloc, + &session, + failed_turn, + true, + &prompt_input, + .{ .test_controls = .{ .boundary_fn = Failure.boundary } }, + ), + ); + try std.testing.expect(prompt_input.retain_image_snapshots); + prompt_input.deinit(alloc); + try std.Io.Dir.accessAbsolute(io_mod.getIo(), image_path, .{}); + try std.testing.expectEqual(@as(usize, 2), session.session_rt.history.items.len); } fn setRecoveryCheckpoint( @@ -3312,6 +3362,28 @@ test "ACP tool notifications preserve UTF-8 for clipped and unsafe output" { } } +test "ACP message IDs rotate when the logical message kind resumes" { + const alloc = std.testing.allocator; + var state = try initTestAcpState(alloc, "/tmp/workspace", .ask); + defer state.deinit(); + var ctx = AcpContext{ + .alloc = alloc, + .state = &state, + .session_id = "session_1", + }; + + var first: acp_types.MessageIdBuffer = undefined; + @memcpy(&first, ctx.operationalMessageId()); + var second: acp_types.MessageIdBuffer = undefined; + @memcpy(&second, ctx.assistantMessageId()); + var third: acp_types.MessageIdBuffer = undefined; + @memcpy(&third, ctx.operationalMessageId()); + + try std.testing.expect(!std.mem.eql(u8, &first, &second)); + try std.testing.expect(!std.mem.eql(u8, &second, &third)); + try std.testing.expect(!std.mem.eql(u8, &first, &third)); +} + test "ACP stream adapter forwards raw Markdown and suppresses rendered duplicates and writer failure" { var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena_state.deinit(); diff --git a/src/acp/sessions.zig b/src/acp/sessions.zig index 0976252c0..b49c9b75b 100644 --- a/src/acp/sessions.zig +++ b/src/acp/sessions.zig @@ -1170,9 +1170,28 @@ fn sendUserHistoryTurn( ) !void { var message_id: acp_types.MessageIdBuffer = undefined; const stable_message_id = acp_types.generateMessageId(&message_id); - try sendUserHistoryChunk(state, alloc, session_id, stable_message_id, user.text); + const replay_text = try canonicalUserHistoryText(alloc, user); + defer alloc.free(replay_text); + if (replay_text.len > 0) { + try sendUserHistoryChunk(state, alloc, session_id, stable_message_id, replay_text); + } for (user.images) |attachment| { - var snapshot = try image_attachments.loadVerifiedSnapshot(alloc, attachment, .{}); + var snapshot = image_attachments.loadVerifiedSnapshot(alloc, attachment, .{}) catch |err| { + if (err == error.OutOfMemory) return err; + debug_trace.logf( + "acp", + "image history replay omitted id={d} err={s}", + .{ attachment.id, @errorName(err) }, + ); + var unavailable: [96]u8 = undefined; + const notice = try std.fmt.bufPrint( + &unavailable, + "Image #{d} unavailable", + .{attachment.id}, + ); + try sendUserHistoryChunk(state, alloc, session_id, stable_message_id, notice); + continue; + }; defer snapshot.deinit(alloc); var out: std.Io.Writer.Allocating = .init(alloc); defer out.deinit(); @@ -1190,6 +1209,32 @@ fn sendUserHistoryTurn( } } +fn canonicalUserHistoryText(alloc: Allocator, user: types.UserTurn) ![]u8 { + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(alloc); + var index: usize = 0; + while (index < user.text.len) { + const placeholder = if (user.text[index] == '[') + image_attachments.matchImagePlaceholder(user.text, index) + else + null; + if (placeholder) |match| { + if (image_attachments.findById(user.images, match.id) != null) { + var end = index + match.length; + const before_newline = out.items.len > 0 and out.items[out.items.len - 1] == '\n'; + const after_newline = end < user.text.len and user.text[end] == '\n'; + if (before_newline and !after_newline) _ = out.pop(); + if (after_newline) end += 1; + index = end; + continue; + } + } + try out.append(alloc, user.text[index]); + index += 1; + } + return out.toOwnedSlice(alloc); +} + fn sendUserHistoryText(state: *server.ServerState, alloc: Allocator, session_id: []const u8, user_text: []const u8) !void { var message_id: acp_types.MessageIdBuffer = undefined; try sendUserHistoryChunk( @@ -1410,6 +1455,29 @@ fn writeModesArray(w: *std.Io.Writer, registry: mode_registry.Registry) !void { try w.writeAll("]"); } +test "canonical ACP image replay hides owned placeholders" { + const alloc = std.testing.allocator; + const images = [_]types.ImageAttachment{.{ + .id = 7, + .path = @constCast("/tmp/image.png"), + .media_type = @constCast("image/png"), + }}; + const cases = [_]struct { text: []const u8, expected: []const u8 }{ + .{ .text = "A\n[Image #7]\nB", .expected = "A\nB" }, + .{ .text = "[Image #7]", .expected = "" }, + .{ .text = "keep [Image #8] literal", .expected = "keep [Image #8] literal" }, + }; + for (cases) |case| { + const user = types.UserTurn{ + .text = @constCast(case.text), + .images = @constCast(&images), + }; + const replay = try canonicalUserHistoryText(alloc, user); + defer alloc.free(replay); + try std.testing.expectEqualStrings(case.expected, replay); + } +} + test "formatIso8601 produces valid format" { const alloc = std.testing.allocator; const result = try formatIso8601(alloc, 1700000000000); diff --git a/src/core/session/session_usage.zig b/src/core/session/session_usage.zig index dec603399..37fb44fb9 100644 --- a/src/core/session/session_usage.zig +++ b/src/core/session/session_usage.zig @@ -1504,9 +1504,13 @@ pub const Usage = struct { } fn observeContextUsage(self: *Usage, sequence: u64, provider_usage: types.Usage) void { - const input = provider_usage.input_tokens orelse return; - const output = provider_usage.output_tokens orelse return; - const used = std.math.add(u64, input, output) catch return; + const used: ?u64 = if (provider_usage.input_tokens) |input| + if (provider_usage.output_tokens) |output| + std.math.add(u64, input, output) catch null + else + null + else + null; self.mutex.lockUncancelable(io_mod.getIo()); defer self.mutex.unlock(io_mod.getIo()); if (sequence < self.latest_context_sequence) return; @@ -3324,6 +3328,18 @@ test "live context usage keeps the newest completed provider observation" { try std.testing.expectEqual(@as(?f64, 0), snapshot.complete_cost); } +test "newer missing context usage clears the prior observation" { + var usage = Usage.initFresh(); + defer usage.deinit(std.testing.allocator); + + usage.observeContextUsage(1, .{ .input_tokens = 30, .output_tokens = 7 }); + usage.observeContextUsage(2, .{ .input_tokens = 40 }); + try std.testing.expect(usage.liveContextSnapshot() == null); + + usage.observeContextUsage(1, .{ .input_tokens = 90, .output_tokens = 10 }); + try std.testing.expect(usage.liveContextSnapshot() == null); +} + test "direct exact generation IDs are deterministic and provider scoped" { var first_buffer: [30]u8 = undefined; var replay_buffer: [30]u8 = undefined; diff --git a/tests/e2e/acp.test.ts b/tests/e2e/acp.test.ts index 05d36e24a..fc4eb351d 100644 --- a/tests/e2e/acp.test.ts +++ b/tests/e2e/acp.test.ts @@ -1438,7 +1438,17 @@ describe("acp: model-independent", () => { const root = createIsolatedRoot("fx-acp-session-metadata-"); const title = "Publish ACP session metadata"; const gateway = startFakeGateway( - [finalText("Metadata published.")], + [ + finalText("Metadata published."), + fakeGatewaySse([{ + type: "text-delta", + id: "answer_2", + delta: "Usage omitted.", + }, { + type: "finish", + finishReason: { unified: "stop", raw: "stop" }, + }]), + ], { models: [{ id: FAKE_GATEWAY_MODEL, @@ -1479,6 +1489,16 @@ describe("acp: model-independent", () => { expect(info?.params.sessionId).toBe(sessionId); expect(Number.isNaN(Date.parse(info?.params.update.updatedAt))).toBe(false); + const unmeasured = await runPrompt( + client, + "Return a response without provider usage metadata.", + TIMEOUT, + ); + expect(unmeasured.promptResult.result.stopReason).toBe("end_turn"); + expect(unmeasured.messages.some((message) => + message.params?.update?.sessionUpdate === "usage_update" + )).toBe(false); + await client.close(); client = await AcpClient.create({ cwd: root.workspace, @@ -4938,8 +4958,9 @@ describe("acp: model-independent", () => { method: "session/prompt", params: { prompt: [ - { type: "text", text: "Describe this image." }, + { type: "text", text: "Before image." }, { type: "image", data: imageData, mimeType: "image/png" }, + { type: "text", text: "After image." }, ], }, }); @@ -4977,10 +4998,20 @@ describe("acp: model-independent", () => { if (message.id === 96) break; replay.push(message); } - const imageChunk = replay.find((message) => + const userChunks = replay.filter((message) => + message.params?.update?.sessionUpdate === "user_message_chunk" + ); + const imageChunk = userChunks.find((message) => message.params?.update?.sessionUpdate === "user_message_chunk" && message.params?.update?.content?.type === "image" ); + expect(userChunks.map((message) => message.params.update.content.type)).toEqual([ + "text", + "image", + ]); + expect(userChunks[0]?.params.update.content.text).toBe("Before image.\nAfter image."); + expect(JSON.stringify(userChunks)).not.toContain("[Image #"); + expect(new Set(userChunks.map((message) => message.params.update.messageId)).size).toBe(1); expect(imageChunk?.params.update.content).toMatchObject({ type: "image", mimeType: "image/png", @@ -5046,6 +5077,78 @@ describe("acp: model-independent", () => { TIMEOUT, ); + test( + "session load reports an unavailable saved image without failing", + async () => { + const root = createIsolatedRoot("fx-acp-image-replay-missing-"); + const imageData = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9WlXYX0AAAAASUVORK5CYII="; + const gateway = startFakeGateway( + [finalText("image saved")], + { + models: [{ + id: FAKE_GATEWAY_MODEL, + type: "language", + tags: ["vision", "file-input", "tool-use"], + }], + }, + ); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + const sessionId = await startCodeSession(client); + const saved = await runPromptBlocks(client, [ + { type: "text", text: "Save this image." }, + { type: "image", data: imageData, mimeType: "image/png" }, + ], TIMEOUT); + expect(saved.promptResult.result.stopReason).toBe("end_turn"); + await client.close(); + + const imageDir = join(root.home, ".fx", "sessions", sessionId, "images"); + const snapshots = readdirSync(imageDir); + expect(snapshots).toHaveLength(1); + rmSync(join(imageDir, snapshots[0]!)); + + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + await client.request("initialize", { protocolVersion: 1 }, 97); + client.send({ + jsonrpc: "2.0", + id: 98, + method: "session/load", + params: { sessionId, cwd: root.workspace, mcpServers: [] }, + }); + const replay: any[] = []; + let loadResponse: any = null; + while (loadResponse === null) { + const message = await client.readLine() as any; + if (message.id === 98) loadResponse = message; + else replay.push(message); + } + + expect(loadResponse.error).toBeUndefined(); + expect(Array.isArray(loadResponse.result?.configOptions)).toBe(true); + const userText = replay + .filter((message) => + message.params?.update?.sessionUpdate === "user_message_chunk" && + message.params?.update?.content?.type === "text" + ) + .map((message) => message.params.update.content.text); + expect(userText).toEqual(["Save this image.", "Image #1 unavailable"]); + expect(JSON.stringify(replay)).not.toContain("[Image #"); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + test( "ACP automatic ask returns to the agent before requesting permission", async () => { From 23b181ad21d84b9266267b9ac2bfe63f819cb964 Mon Sep 17 00:00:00 2001 From: Pranit Date: Tue, 1 Sep 2026 13:53:33 -0400 Subject: [PATCH 5/6] Preserve ACP image replay text - Keep durable user text byte-identical during image replay. - Cover literal image-marker text alongside attached images. --- src/acp/sessions.zig | 55 ++----------------------------------------- tests/e2e/acp.test.ts | 13 ++++++---- 2 files changed, 10 insertions(+), 58 deletions(-) diff --git a/src/acp/sessions.zig b/src/acp/sessions.zig index b49c9b75b..a66ca0b94 100644 --- a/src/acp/sessions.zig +++ b/src/acp/sessions.zig @@ -1170,10 +1170,8 @@ fn sendUserHistoryTurn( ) !void { var message_id: acp_types.MessageIdBuffer = undefined; const stable_message_id = acp_types.generateMessageId(&message_id); - const replay_text = try canonicalUserHistoryText(alloc, user); - defer alloc.free(replay_text); - if (replay_text.len > 0) { - try sendUserHistoryChunk(state, alloc, session_id, stable_message_id, replay_text); + if (user.text.len > 0) { + try sendUserHistoryChunk(state, alloc, session_id, stable_message_id, user.text); } for (user.images) |attachment| { var snapshot = image_attachments.loadVerifiedSnapshot(alloc, attachment, .{}) catch |err| { @@ -1209,32 +1207,6 @@ fn sendUserHistoryTurn( } } -fn canonicalUserHistoryText(alloc: Allocator, user: types.UserTurn) ![]u8 { - var out: std.ArrayList(u8) = .empty; - errdefer out.deinit(alloc); - var index: usize = 0; - while (index < user.text.len) { - const placeholder = if (user.text[index] == '[') - image_attachments.matchImagePlaceholder(user.text, index) - else - null; - if (placeholder) |match| { - if (image_attachments.findById(user.images, match.id) != null) { - var end = index + match.length; - const before_newline = out.items.len > 0 and out.items[out.items.len - 1] == '\n'; - const after_newline = end < user.text.len and user.text[end] == '\n'; - if (before_newline and !after_newline) _ = out.pop(); - if (after_newline) end += 1; - index = end; - continue; - } - } - try out.append(alloc, user.text[index]); - index += 1; - } - return out.toOwnedSlice(alloc); -} - fn sendUserHistoryText(state: *server.ServerState, alloc: Allocator, session_id: []const u8, user_text: []const u8) !void { var message_id: acp_types.MessageIdBuffer = undefined; try sendUserHistoryChunk( @@ -1455,29 +1427,6 @@ fn writeModesArray(w: *std.Io.Writer, registry: mode_registry.Registry) !void { try w.writeAll("]"); } -test "canonical ACP image replay hides owned placeholders" { - const alloc = std.testing.allocator; - const images = [_]types.ImageAttachment{.{ - .id = 7, - .path = @constCast("/tmp/image.png"), - .media_type = @constCast("image/png"), - }}; - const cases = [_]struct { text: []const u8, expected: []const u8 }{ - .{ .text = "A\n[Image #7]\nB", .expected = "A\nB" }, - .{ .text = "[Image #7]", .expected = "" }, - .{ .text = "keep [Image #8] literal", .expected = "keep [Image #8] literal" }, - }; - for (cases) |case| { - const user = types.UserTurn{ - .text = @constCast(case.text), - .images = @constCast(&images), - }; - const replay = try canonicalUserHistoryText(alloc, user); - defer alloc.free(replay); - try std.testing.expectEqualStrings(case.expected, replay); - } -} - test "formatIso8601 produces valid format" { const alloc = std.testing.allocator; const result = try formatIso8601(alloc, 1700000000000); diff --git a/tests/e2e/acp.test.ts b/tests/e2e/acp.test.ts index fc4eb351d..de84bebb0 100644 --- a/tests/e2e/acp.test.ts +++ b/tests/e2e/acp.test.ts @@ -4958,7 +4958,7 @@ describe("acp: model-independent", () => { method: "session/prompt", params: { prompt: [ - { type: "text", text: "Before image." }, + { type: "text", text: "User literal [Image #1]." }, { type: "image", data: imageData, mimeType: "image/png" }, { type: "text", text: "After image." }, ], @@ -5009,8 +5009,9 @@ describe("acp: model-independent", () => { "text", "image", ]); - expect(userChunks[0]?.params.update.content.text).toBe("Before image.\nAfter image."); - expect(JSON.stringify(userChunks)).not.toContain("[Image #"); + expect(userChunks[0]?.params.update.content.text).toBe( + "User literal [Image #1].\n[Image #1]\nAfter image.", + ); expect(new Set(userChunks.map((message) => message.params.update.messageId)).size).toBe(1); expect(imageChunk?.params.update.content).toMatchObject({ type: "image", @@ -5137,8 +5138,10 @@ describe("acp: model-independent", () => { message.params?.update?.content?.type === "text" ) .map((message) => message.params.update.content.text); - expect(userText).toEqual(["Save this image.", "Image #1 unavailable"]); - expect(JSON.stringify(replay)).not.toContain("[Image #"); + expect(userText).toEqual([ + "Save this image.\n[Image #1]", + "Image #1 unavailable", + ]); expect(client.stderr).toBe(""); } finally { await client?.close(); From 20f6a645dce046894fcf53e323ea3faa7a99611b Mon Sep 17 00:00:00 2001 From: Pranit Date: Tue, 1 Sep 2026 23:12:48 -0400 Subject: [PATCH 6/6] Correct ACP image boundary behavior - Derive image-only session titles from canonical placeholder syntax. - Reject oversized inline images before snapshot work. - Return a stable ACP error when the selected model cannot accept images. --- src/acp/prompt.zig | 35 ++- src/core/images/image_attachments.zig | 21 +- src/core/session/session_display_metadata.zig | 55 +++++ tests/e2e/acp.test.ts | 225 ++++++++++++++++++ 4 files changed, 334 insertions(+), 2 deletions(-) diff --git a/src/acp/prompt.zig b/src/acp/prompt.zig index 85b727570..af0a69703 100644 --- a/src/acp/prompt.zig +++ b/src/acp/prompt.zig @@ -109,6 +109,18 @@ fn promptInputFailure(err: anyerror) anyerror!TerminalOutcome { }; } +fn promptExecutionFailure(err: anyerror) anyerror!TerminalOutcome { + return switch (err) { + error.ModelImageCapabilityUnavailable, + error.SubscriptionNativeImageUnavailable, + => .{ .rpc_error = .{ + .code = ErrorCode.invalid_params, + .message = "Image prompts are unavailable for the selected model", + } }, + else => err, + }; +} + const ProviderTerminalPublication = enum { not_applicable, pending, @@ -751,7 +763,7 @@ pub fn handlePrompt( if (err == error.NonInteractivePermissionRequired) { ctx.stop_reason = .refused; } else { - return err; + return promptExecutionFailure(err); } }; prompt_input.retainImageSnapshots(); @@ -2836,6 +2848,27 @@ test "mapToolKind maps common tools" { try std.testing.expectEqual(acp_types.ToolCallKind.other, mapToolKind("unknown_tool")); } +test "ACP maps selected-model image capability failures to one stable error" { + const failures = [_]anyerror{ + error.ModelImageCapabilityUnavailable, + error.SubscriptionNativeImageUnavailable, + }; + for (failures) |failure| { + const outcome = try promptExecutionFailure(failure); + switch (outcome) { + .rpc_error => |rpc_error| { + try std.testing.expectEqual(ErrorCode.invalid_params, rpc_error.code); + try std.testing.expectEqualStrings( + "Image prompts are unavailable for the selected model", + rpc_error.message, + ); + }, + .stop_reason => return error.UnexpectedStopReason, + } + } + try std.testing.expectError(error.OutOfMemory, promptExecutionFailure(error.OutOfMemory)); +} + test "provider terminal status maps only terminal outcomes" { try std.testing.expectEqual(acp_types.ToolCallStatus.completed, providerTerminalStatus(.completed).?); try std.testing.expectEqual(acp_types.ToolCallStatus.failed, providerTerminalStatus(.failed).?); diff --git a/src/core/images/image_attachments.zig b/src/core/images/image_attachments.zig index ab6c33521..8fb94f691 100644 --- a/src/core/images/image_attachments.zig +++ b/src/core/images/image_attachments.zig @@ -482,7 +482,7 @@ pub fn captureInlineImageBytes( ) !types.ImageAttachment { if (image_id == 0) return error.InvalidImageId; if (bytes.len == 0 or declared_media_type.len == 0) return error.UnsupportedImageType; - if (bytes.len > max_image_bytes) return error.ImageTooLarge; + if (bytes.len > max_image_bytes or !fitsEncodedLimit(bytes.len)) return error.ImageTooLarge; var snapshot_dir_handle = try openOrCreateSnapshotDirectoryNoFollow(snapshot_dir); defer snapshot_dir_handle.close(io_mod.getIo()); @@ -2765,6 +2765,25 @@ test "encoded image limit uses exact padded base64 length" { try std.testing.expect(!fitsEncodedLimit(largest_fitting_raw_image + 1)); } +test "inline capture rejects one byte beyond encoded limit before directory effects" { + const alloc = std.testing.allocator; + const largest_fitting_raw_image = (max_encoded_image_bytes / 4) * 3; + const bytes = try alloc.alloc(u8, largest_fitting_raw_image + 1); + defer alloc.free(bytes); + @memset(bytes, 0); + + try std.testing.expectError( + error.ImageTooLarge, + captureInlineImageBytes( + alloc, + 1, + "image/png", + bytes, + "/path/that/does/not/exist", + ), + ); +} + test "capture rejection distinguishes source size from preparation failure" { try std.testing.expectEqualStrings( image_too_large_notice, diff --git a/src/core/session/session_display_metadata.zig b/src/core/session/session_display_metadata.zig index 5375953cc..d52ba7603 100644 --- a/src/core/session/session_display_metadata.zig +++ b/src/core/session/session_display_metadata.zig @@ -1,5 +1,6 @@ const std = @import("std"); const debug_trace = @import("../shared/debug_trace.zig"); +const image_attachments = @import("../images/image_attachments.zig"); const io_mod = @import("../shared/io.zig"); const session = @import("session.zig"); @@ -87,11 +88,28 @@ fn firstPromptCandidate(history: []const session.HistoryTurn) ?PromptCandidate { fn promptCandidateFromUser(user: session.UserTurn) ?PromptCandidate { const trimmed = std.mem.trim(u8, user.text, " \t\r\n"); + if (isCanonicalImageOnlyText(user.text, user.images)) return .image_only; if (trimmed.len > 0 and !isSlashCommandOnly(trimmed)) return .{ .text = user.text }; if (user.images.len > 0) return .image_only; return null; } +fn isCanonicalImageOnlyText(text: []const u8, images: []const session.ImageAttachment) bool { + if (images.len == 0) return false; + + var text_offset: usize = 0; + for (images, 0..) |image, index| { + if (index > 0) { + if (text_offset >= text.len or text[text_offset] != '\n') return false; + text_offset += 1; + } + const placeholder = image_attachments.matchImagePlaceholder(text, text_offset) orelse return false; + if (placeholder.id != image.id) return false; + text_offset += placeholder.length; + } + return text_offset == text.len; +} + fn isSlashCommandOnly(trimmed: []const u8) bool { return trimmed.len > 0 and trimmed[0] == '/' and std.mem.findScalar(u8, trimmed, '\n') == null; } @@ -359,6 +377,43 @@ test "display metadata skips slash-only turns and handles image-only sessions" { try std.testing.expect(image_metadata.preview == null); } +test "display metadata treats canonical image placeholders as image-only syntax" { + const alloc = std.testing.allocator; + const first = session.ImageAttachment{ + .id = 7, + .path = @constCast("/tmp/first.png"), + .media_type = @constCast("image/png"), + }; + const second = session.ImageAttachment{ + .id = 9, + .path = @constCast("/tmp/second.png"), + .media_type = @constCast("image/png"), + }; + var images = [_]session.ImageAttachment{ first, second }; + + const image_only = [_]session.HistoryTurn{ + makeAssistantTurnWithImages("[Image #7]\n[Image #9]", images[0..]), + }; + var image_metadata = try deriveFromHistory(alloc, &image_only); + defer image_metadata.deinit(alloc); + try std.testing.expectEqualStrings(image_title, image_metadata.title); + try std.testing.expect(image_metadata.preview == null); + + const mixed = [_]session.HistoryTurn{ + makeAssistantTurnWithImages("Describe [Image #7]", images[0..1]), + }; + var mixed_metadata = try deriveFromHistory(alloc, &mixed); + defer mixed_metadata.deinit(alloc); + try std.testing.expectEqualStrings("Describe [Image #7]", mixed_metadata.title); + + const mismatched = [_]session.HistoryTurn{ + makeAssistantTurnWithImages("[Image #8]", images[0..1]), + }; + var mismatched_metadata = try deriveFromHistory(alloc, &mismatched); + defer mismatched_metadata.deinit(alloc); + try std.testing.expectEqualStrings("[Image #8]", mismatched_metadata.title); +} + test "display metadata sidecar round trips and invalid sidecar falls back" { const alloc = std.testing.allocator; const source = DisplayMetadata{ diff --git a/tests/e2e/acp.test.ts b/tests/e2e/acp.test.ts index de84bebb0..b28f9a411 100644 --- a/tests/e2e/acp.test.ts +++ b/tests/e2e/acp.test.ts @@ -5028,6 +5028,231 @@ describe("acp: model-independent", () => { TIMEOUT, ); + test( + "image-only prompt publishes and reloads the shared image title", + async () => { + const root = createIsolatedRoot("fx-acp-image-only-title-"); + const imageData = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9WlXYX0AAAAASUVORK5CYII="; + const gateway = startFakeGateway( + [finalText("image-only prompt complete")], + { + models: [{ + id: FAKE_GATEWAY_MODEL, + type: "language", + tags: ["vision", "file-input", "tool-use"], + }], + }, + ); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + const sessionId = await startCodeSession(client); + const prompted = await runPromptBlocks( + client, + [{ type: "image", data: imageData, mimeType: "image/png" }], + TIMEOUT, + ); + expect(prompted.promptResult.result.stopReason).toBe("end_turn"); + expect(prompted.messages.find((message) => + message.params?.update?.sessionUpdate === "session_info_update" + )?.params.update.title).toBe("Image session"); + expect(gateway.requests).toHaveLength(1); + await client.close(); + + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + await client.request("initialize", { protocolVersion: 1 }, 97); + client.send({ + jsonrpc: "2.0", + id: 98, + method: "session/load", + params: { sessionId, cwd: root.workspace, mcpServers: [] }, + }); + const replay: any[] = []; + let loadResponse: any = null; + while (loadResponse === null) { + const message = await client.readLine() as any; + if (message.id === 98) loadResponse = message; + else replay.push(message); + } + + expect(loadResponse.error).toBeUndefined(); + expect(replay.find((message) => + message.params?.update?.sessionUpdate === "session_info_update" + )?.params.update.title).toBe("Image session"); + const userChunks = replay.filter((message) => + message.params?.update?.sessionUpdate === "user_message_chunk" + ); + expect(userChunks.map((message) => message.params.update.content.type)).toEqual([ + "text", + "image", + ]); + expect(userChunks[0]?.params.update.content.text).toBe("[Image #1]"); + expect(new Set(userChunks.map((message) => message.params.update.messageId)).size).toBe(1); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + + test( + "inline image above the portable encoded limit fails before effects", + async () => { + const root = createIsolatedRoot("fx-acp-inline-image-limit-"); + const maxEncodedImageBytes = 5 * 1024 * 1024; + const largestFittingRawImage = Math.floor(maxEncodedImageBytes / 4) * 3; + const oversized = Buffer.alloc(largestFittingRawImage + 1); + Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]).copy(oversized); + const imageData = oversized.toString("base64"); + expect(Buffer.byteLength(imageData)).toBe(maxEncodedImageBytes + 4); + const gateway = startFakeGateway( + [finalText("ACP image size recovery complete")], + { + models: [{ + id: FAKE_GATEWAY_MODEL, + type: "language", + tags: ["vision", "file-input", "tool-use"], + }], + }, + ); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: fakeGatewayEnv(root, gateway), + }); + const sessionId = await startCodeSession(client); + client.send({ + jsonrpc: "2.0", + id: 99, + method: "session/prompt", + params: { + prompt: [{ + type: "image", + data: imageData, + mimeType: "image/png", + }], + }, + }); + + const rejected = await readResponse(client, 99, LIVE_TIMEOUT); + expect(rejected.error).toEqual({ + code: -32602, + message: "Image prompt exceeds size limit", + }); + expect(gateway.requests).toHaveLength(0); + const imageDir = join(root.home, ".fx", "sessions", sessionId, "images"); + if (existsSync(imageDir)) expect(readdirSync(imageDir)).toEqual([]); + + const recovered = await runPrompt( + client, + "Confirm the ACP connection remains usable after image size rejection.", + TIMEOUT, + ); + expect(recovered.promptResult.result.stopReason).toBe("end_turn"); + expect(gateway.requests).toHaveLength(1); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + LIVE_TIMEOUT, + ); + + test( + "selected text-only model rejects images without leaking an internal error", + async () => { + const root = createIsolatedRoot("fx-acp-image-model-capability-"); + const imageData = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9WlXYX0AAAAASUVORK5CYII="; + const gateway = startFakeGateway([]); + const codex = startAcpFakeCodex(); + writeSeededAcpChatGptLogin(root.home, codex.accessToken); + try { + client = await AcpClient.create({ + cwd: root.workspace, + env: { + ...fakeGatewayEnv(root, gateway), + FX_E2E_OPENAI_CODEX_RESPONSES_URL: codex.responsesUrl, + FX_E2E_OPENAI_CODEX_MODELS_URL: codex.modelsUrl, + FX_E2E_CHATGPT_TOKEN_URL: codex.tokenUrl, + }, + }); + const initialized = await client.request( + "initialize", + { protocolVersion: 1 }, + 1, + ) as any; + expect(initialized.result.agentCapabilities.promptCapabilities.image).toBe(true); + const created = await client.request( + "session/new", + { mcpServers: [] }, + 2, + ) as any; + await client.readLine(); + const sessionId = created.result.sessionId as string; + await client.request("session/set_mode", { modeId: "code" }, 3); + await client.request("session/set_config_option", { + configId: "provider", + value: "codex", + }, 4); + await client.request("session/set_config_option", { + configId: "model", + value: "gpt-5.4-mini", + }, 5); + + client.send({ + jsonrpc: "2.0", + id: 100, + method: "session/prompt", + params: { + prompt: [{ type: "image", data: imageData, mimeType: "image/png" }], + }, + }); + const rejected = await readResponse(client, 100); + expect(rejected.error).toEqual({ + code: -32602, + message: "Image prompts are unavailable for the selected model", + }); + expect(codex.requests).toHaveLength(0); + expect(gateway.requests).toHaveLength(0); + const imageDir = join(root.home, ".fx", "sessions", sessionId, "images"); + if (existsSync(imageDir)) expect(readdirSync(imageDir)).toEqual([]); + + const rejectedDetail = await runFx(["session", "--id", sessionId, "--json"], { + cwd: root.workspace, + env: { HOME: root.home }, + timeoutMs: TIMEOUT, + }); + expect(rejectedDetail.code).toBe(0); + expect(JSON.parse(rejectedDetail.stdout).history_len).toBe(0); + + const recovered = await runPrompt( + client, + "Confirm the ACP connection remains usable after image rejection.", + TIMEOUT, + ); + expect(recovered.promptResult.result.stopReason).toBe("end_turn"); + expect(codex.requests).toHaveLength(1); + expect(client.stderr).toBe(""); + } finally { + await client?.close(); + codex.stop(); + gateway.stop(); + rmSync(root.root, { recursive: true, force: true }); + } + }, + TIMEOUT, + ); + test( "image prompt MIME mismatch fails before the Gateway without an orphaned snapshot", async () => {