Something went wrong. Try again.
My version of https://github.com/vercel-labs/fx
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360const std = @import("std");const lifecycle_hooks = @import("../../hooks/hooks.zig");const types = @import("../../shared/types.zig");const debug_trace = @import("../../shared/debug_trace.zig");const execution_memory_helpers = @import("../execution_memory.zig");
const runtime_execution_memory = @import("execution_memory.zig");const runtime_finalization = @import("finalization.zig");const runtime_deps = @import("deps.zig");const runtime_telemetry = @import("telemetry.zig");const worker_runtime = @import("../worker_runtime.zig");
const Allocator = std.mem.Allocator;const HistoryTurn = types.HistoryTurn;const ToolCall = types.ToolCall;const TraceContext = debug_trace.TraceContext;const AgentRuntimeDeps = runtime_deps.AgentRuntimeDeps;const TurnFinalizationGuard = runtime_finalization.TurnFinalizationGuard;const QueuedPrompt = worker_runtime.QueuedPrompt;
pub fn persistInterruptedTurnOnce( hooks: *const AgentRuntimeDeps, finalization: *TurnFinalizationGuard, job: QueuedPrompt, partial_assistant: ?[]const u8, active_tool_call: ?ToolCall, completed_tool_names: [][]u8, persisted: *bool, trace_ctx: TraceContext, current_turn_messages: []const types.ChatMessage, retained_candidate: ?[]const u8, terminal_materializing: *bool,) !void { return persistInterruptedTurnWithPresentation( hooks, finalization, job, partial_assistant, active_tool_call, completed_tool_names, persisted, trace_ctx, current_turn_messages, retained_candidate, terminal_materializing, null, .turn, );}
pub fn persistCompactionInterruptedTurnOnce( hooks: *const AgentRuntimeDeps, finalization: *TurnFinalizationGuard, job: QueuedPrompt, completed_tool_names: [][]u8, persisted: *bool, trace_ctx: TraceContext, current_turn_messages: []const types.ChatMessage, retained_candidate: ?[]const u8, terminal_materializing: *bool,) !void { return persistInterruptedTurnWithPresentation( hooks, finalization, job, null, null, completed_tool_names, persisted, trace_ctx, current_turn_messages, retained_candidate, terminal_materializing, null, .compaction, );}
pub fn persistInterruptedCommandTurnOnce( hooks: *const AgentRuntimeDeps, finalization: *TurnFinalizationGuard, job: QueuedPrompt, partial_assistant: ?[]const u8, active_tool_call: ToolCall, completed_tool_names: [][]u8, persisted: *bool, trace_ctx: TraceContext, current_turn_messages: []const types.ChatMessage, retained_candidate: ?[]const u8, terminal_materializing: *bool, cancelled_command: ?types.CancelledCommandPresentation,) !void { return persistInterruptedTurnWithPresentation( hooks, finalization, job, partial_assistant, active_tool_call, completed_tool_names, persisted, trace_ctx, current_turn_messages, retained_candidate, terminal_materializing, cancelled_command, .turn, );}
fn persistInterruptedTurnWithPresentation( hooks: *const AgentRuntimeDeps, finalization: *TurnFinalizationGuard, job: QueuedPrompt, partial_assistant: ?[]const u8, active_tool_call: ?ToolCall, completed_tool_names: [][]u8, persisted: *bool, trace_ctx: TraceContext, current_turn_messages: []const types.ChatMessage, retained_candidate: ?[]const u8, terminal_materializing: *bool, cancelled_command: ?types.CancelledCommandPresentation, cancellation_origin: types.CancellationOrigin,) !void { if (persisted.*) return;
const durable_active_tool_call = if (active_tool_call) |call| try execution_memory_helpers.dupePersistedToolCall( std.heap.c_allocator, call, ) else null; defer if (durable_active_tool_call) |call| { types.freeToolCall(std.heap.c_allocator, call); }; const full_execution = try runtime_execution_memory.buildInterruptedExecutionMemory( std.heap.c_allocator, current_turn_messages, active_tool_call, ); defer types.freeExecutionMemory(std.heap.c_allocator, full_execution); var projection_arena = std.heap.ArenaAllocator.init(std.heap.c_allocator); defer projection_arena.deinit(); const execution = try finalization.compacted_execution.project(projection_arena.allocator(), full_execution); terminal_materializing.* = true;
if (retained_candidate) |candidate| { const assistant = try runtime_finalization.stopTerminalText( std.heap.c_allocator, candidate, partial_assistant, ); defer if (assistant.presentation) |text| std.heap.c_allocator.free(text); const turn: HistoryTurn = .{ .interrupted = .{ .user = .{ .text = job.prompt, .images = job.images }, .assistant = @constCast(assistant.history), .tool_call = durable_active_tool_call, .completed_tool_names = completed_tool_names, .execution = execution, .cancelled_command = cancelled_command, .cancellation_origin = cancellation_origin, } }; const finished = try types.dupeFinishedPrompt( std.heap.c_allocator, .{ .turn = turn, .terminal_projection = .assistant_text, .presentation_text = assistant.presentation, }, );
persisted.* = true; var propagation_error: ?anyerror = null; hooks.propagate_history_turn(hooks.ctx, turn) catch |err| { propagation_error = err; }; try traceInterruptedPersistence( job, assistant.history, active_tool_call, completed_tool_names, trace_ctx, ); try finalization.finish(.interrupted, null, finished); debug_trace.eventf( "interrupt", "finish_event_emitted", trace_ctx, "outcome_kind=interrupted", .{}, ); if (propagation_error) |err| return err; return; }
persisted.* = true; const turn: HistoryTurn = .{ .interrupted = .{ .user = .{ .text = job.prompt, .images = job.images }, .assistant = if (partial_assistant) |text| if (text.len > 0) @constCast(text) else null else null, .tool_call = durable_active_tool_call, .completed_tool_names = completed_tool_names, .execution = execution, .cancelled_command = cancelled_command, .cancellation_origin = cancellation_origin, } };
var propagation_error: ?anyerror = null; hooks.propagate_history_turn(hooks.ctx, turn) catch |err| { propagation_error = err; }; try traceInterruptedPersistence( job, partial_assistant, active_tool_call, completed_tool_names, trace_ctx, ); try finalization.finish(.interrupted, null, .{ .turn = try types.dupeHistoryTurn(std.heap.c_allocator, turn), }); debug_trace.eventf("interrupt", "finish_event_emitted", trace_ctx, "outcome_kind=interrupted", .{}); if (propagation_error) |err| return err;}
pub fn persistFailedPartialTurnOnce( hooks: *const AgentRuntimeDeps, finalization: *TurnFinalizationGuard, job: QueuedPrompt, partial_assistant: []const u8, persisted: *bool, trace_ctx: TraceContext, current_turn_messages: []const types.ChatMessage, terminal_materializing: *bool,) !void { if (persisted.*) return; if (partial_assistant.len == 0) return;
const full_execution = try runtime_execution_memory.buildInterruptedExecutionMemory( std.heap.c_allocator, current_turn_messages, null, ); defer types.freeExecutionMemory(std.heap.c_allocator, full_execution); var projection_arena = std.heap.ArenaAllocator.init(std.heap.c_allocator); defer projection_arena.deinit(); const execution = try finalization.compacted_execution.project(projection_arena.allocator(), full_execution); terminal_materializing.* = true;
const turn: HistoryTurn = .{ .interrupted = .{ .user = .{ .text = job.prompt, .images = job.images }, .assistant = @constCast(partial_assistant), .execution = execution, .terminal_reason = .failed, } }; const finished = try types.dupeFinishedPrompt( std.heap.c_allocator, .{ .turn = turn, .terminal_projection = .assistant_text, }, );
persisted.* = true; var propagation_error: ?anyerror = null; hooks.propagate_history_turn(hooks.ctx, turn) catch |err| { propagation_error = err; }; debug_trace.eventf( "gateway", "stream_failure_history_persisted", trace_ctx, "prompt_bytes={d} partial_assistant_bytes={d}", .{ job.prompt.len, partial_assistant.len }, ); try finalization.finish(.failed, null, finished); debug_trace.eventf( "gateway", "stream_failure_finish_event_emitted", trace_ctx, "outcome_kind=failed", .{}, ); if (propagation_error) |err| return err;}
fn traceInterruptedPersistence( job: QueuedPrompt, partial_assistant: ?[]const u8, active_tool_call: ?ToolCall, completed_tool_names: [][]u8, trace_ctx: TraceContext,) !void { const completed_tool_names_text = try runtime_telemetry.formatCompletedToolNamesCompact(std.heap.c_allocator, completed_tool_names); defer std.heap.c_allocator.free(completed_tool_names_text); debug_trace.logf("agent", "interrupted marker persisted prompt_bytes={d} partial_assistant_bytes={d} active_tool={s} completed_tool_count={d} completed_tool_names={s}", .{ job.prompt.len, if (partial_assistant) |text| text.len else 0, if (active_tool_call != null) "true" else "false", completed_tool_names.len, completed_tool_names_text, }); debug_trace.eventf("interrupt", "interrupted_history_persisted", trace_ctx, "prompt_bytes={d} partial_assistant_bytes={d} active_tool_known={s} completed_tool_count={d} completed_tool_names={s}", .{ job.prompt.len, if (partial_assistant) |text| text.len else 0, if (active_tool_call != null) "true" else "false", completed_tool_names.len, completed_tool_names_text, }); debug_trace.eventf("interrupt", "interrupt_persisted", trace_ctx, "prompt_bytes={d} partial_assistant_bytes={d} active_tool={s} completed_tool_count={d} completed_tool_names={s} active_tool_reason={s}", .{ job.prompt.len, if (partial_assistant) |text| text.len else 0, if (active_tool_call != null) "true" else "false", completed_tool_names.len, completed_tool_names_text, interruptPersistenceReason(partial_assistant, active_tool_call, completed_tool_names), }); if (active_tool_call) |tool_call| { debug_trace.logf("agent", "aborted tool output persisted call_id={s} name={s}", .{ tool_call.id, tool_call.name }); debug_trace.eventf("interrupt", "aborted_tool_persisted", trace_ctx, "call_id={s} name={s}", .{ tool_call.id, tool_call.name }); }}
pub fn recordCompletedToolName(arena: Allocator, completed_tool_names: *std.ArrayList([]u8), name: []const u8) !void { try completed_tool_names.append(arena, try arena.dupe(u8, name));}
pub fn countInterruptedHistory(history: []const HistoryTurn) usize { var count: usize = 0; for (history) |turn| { if (turn == .interrupted) count += 1; } return count;}
pub fn countPartialTextInterruptedClosures(history: []const HistoryTurn) usize { var count: usize = 0; for (history) |turn| { const entry = switch (turn) { .interrupted => |value| value, else => continue, }; if (entry.tool_call != null) continue; if (entry.completed_tool_names.len > 0) continue; if (entry.assistant) |assistant| { if (assistant.len > 0) count += 1; } } return count;}
fn interruptPersistenceReason(partial_assistant: ?[]const u8, active_tool_call: ?ToolCall, completed_tool_names: []const []u8) []const u8 { if (active_tool_call != null) return "active_tool_call_present"; if (completed_tool_names.len > 0) return "completed_tools_present"; if (partial_assistant) |text| { if (text.len > 0) return "partial_assistant_only"; } return "no_assistant_output";}