Something went wrong. Try again.
My version of https://github.com/vercel-labs/fx
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746const std = @import("std");const types = @import("../../shared/types.zig");const debug_trace = @import("../../shared/debug_trace.zig");const diff = @import("../../output/diff.zig");const tool_dispatch = @import("../../tooling/tool_dispatch.zig");const tool_result_errors = @import("../../tooling/tool_result_errors.zig");
const runtime_config = @import("config.zig");const runtime_execution_memory = @import("execution_memory.zig");const runtime_deps = @import("deps.zig");const runtime_interruption = @import("interruption.zig");const runtime_parallel_execution = @import("parallel_execution.zig");const runtime_telemetry = @import("telemetry.zig");const runtime_tool_admission = @import("tool_admission.zig");const runtime_tool_contracts = @import("tool_contracts.zig");const runtime_tool_presentation = @import("tool_presentation.zig");
const Allocator = std.mem.Allocator;const ChatMessage = types.ChatMessage;const ToolCall = types.ToolCall;const TraceContext = debug_trace.TraceContext;const AgentRuntimeDeps = runtime_deps.AgentRuntimeDeps;const ToolExecutionResult = runtime_tool_contracts.ToolExecutionResult;
pub const StepBatchState = struct { step_error_count: usize = 0, step_total_count: usize = 0, step_had_writes: bool = false, pending_user_suffix: std.ArrayList(ChatMessage) = .empty, /// Turn-scoped identical-failure tracker owned by the orchestrator step /// loop. Null disables escalation (tests and non-turn callers). identical_failure_escalation: ?*runtime_tool_admission.IdenticalFailureEscalationState = null,
pub fn allToolResultsFailed(self: StepBatchState) bool { return self.step_total_count > 0 and self.step_error_count == self.step_total_count; }};
pub const ToolResultAccounting = struct { increment_total: bool = true, increment_error: bool = false, record_completion: bool = false, mark_write: bool = false, status: ?types.PersistedToolStatus = null,};
pub fn appendAssistantToolCallStep( arena: Allocator, within_turn_suffix: *std.ArrayList(ChatMessage), content: ?[]const u8, tool_calls: []const ToolCall, provider_replay: ?types.ProviderReplay,) !void { try within_turn_suffix.append(arena, .{ .role = .assistant, .content = content, .tool_calls = tool_calls, .provider_replay = provider_replay, });}
pub fn appendToolResultContent( arena: Allocator, within_turn_suffix: *std.ArrayList(ChatMessage), completed_tool_names: *std.ArrayList([]u8), batch: *StepBatchState, tool_call: ToolCall, model_output: []const u8, memory: ?types.ToolResultMemory, accounting: ToolResultAccounting,) !void { if (accounting.increment_total) batch.step_total_count += 1; if (accounting.increment_error) batch.step_error_count += 1; if (accounting.mark_write) batch.step_had_writes = true; var content = model_output; // Only well-formed calls escalate: malformed-argument rejections never // execute and already have their own hard-stop path. if (tool_call.argument_integrity == .valid) { if (batch.identical_failure_escalation) |escalation| { const failure_count = try escalation.observe(arena, tool_call, accounting.increment_error); if (failure_count >= 2) { content = try runtime_tool_admission.appendIdenticalFailureEscalation( arena, model_output, failure_count, ); } } } try within_turn_suffix.append(arena, .{ .role = .tool, .content = content, .tool_call_id = tool_call.id, .tool_name = tool_call.name, .tool_result_status = accounting.status orelse if (accounting.increment_error) .failure else .success, .tool_result_memory = memory, }); if (accounting.record_completion) { try runtime_interruption.recordCompletedToolName(arena, completed_tool_names, tool_call.name); }}
pub fn appendPermissionFeedback( arena: Allocator, batch: *StepBatchState, source_tool_call_id: []const u8, feedback: []const []const u8,) !void { for (feedback) |text| { if (text.len == 0) continue; const owned = try arena.dupe(u8, text); try batch.pending_user_suffix.append(arena, .{ .role = .user, .content = owned, .tool_call_id = source_tool_call_id, .permission_feedback = true, }); }}
pub fn drainPendingUserSuffix( arena: Allocator, batch: *StepBatchState, within_turn_suffix: *std.ArrayList(ChatMessage),) !void { try within_turn_suffix.appendSlice(arena, batch.pending_user_suffix.items); batch.pending_user_suffix.clearRetainingCapacity();}
pub fn reserveCommittedFileBookkeeping( arena: Allocator, within_turn_suffix: *std.ArrayList(ChatMessage), completed_tool_names: *std.ArrayList([]u8), tool_name: []const u8,) ![]u8 { try within_turn_suffix.ensureUnusedCapacity(arena, 1); try completed_tool_names.ensureUnusedCapacity(arena, 1); return try arena.dupe(u8, tool_name);}
pub fn assembleParallelToolResults( arena: Allocator, hooks: *const AgentRuntimeDeps, config: runtime_config.Config, within_turn_suffix: *std.ArrayList(ChatMessage), completed_tool_names: *std.ArrayList([]u8), batch: *StepBatchState, provisional_statuses: *runtime_tool_presentation.ProvisionalToolStatuses, provisional_alloc: Allocator, parallel_calls: []const ToolCall, precomputed_results: []const ?ToolExecutionResult, parallel_attempts: []const runtime_parallel_execution.ParallelToolAttempt, parallel_status_started: []const bool, parallel_status_terminalized: []const bool, turn_id: u64, advertised_dynamic_tool_names: []const []const u8, step_ctx: TraceContext,) !?ToolCall { var parallel_result_index: usize = 0; var cancelled_call: ?ToolCall = null; for (parallel_calls, precomputed_results, 0..) |original_call, precomputed, original_index| { const execution = precomputed orelse blk: { const attempt = parallel_attempts[parallel_result_index]; parallel_result_index += 1; switch (attempt) { .completed => |result| break :blk result.execution, .cancelled => { if (cancelled_call == null) cancelled_call = original_call; _ = try provisional_statuses.finishDeniedCall( hooks, provisional_alloc, arena, turn_id, original_call, parallel_status_started[original_index], null, "Cancelled", advertised_dynamic_tool_names, ); continue; }, } }; var prepared = try runtime_execution_memory.prepareToolExecutionOutput(arena, config, original_call, execution, null); runtime_execution_memory.applyToolResultMemory( &prepared.memory, execution.tool_result_memory, ); // The parallel run's attempts are deinitialized when this scope exits, // but history keeps prepared.memory. ArenaAllocator.free genuinely // reclaims most-recent and dedicated-chunk allocations, so the // attempt-owned slices must be transferred into the history arena // before that deinit runs. prepared.memory = try types.dupeToolResultMemory(arena, prepared.memory); try runtime_execution_memory.retainToolImages(arena, config, original_call, &prepared); const safe_tool_output = prepared.model_output; if (precomputed == null) { runtime_parallel_execution.reportInnerToolUsage(hooks, original_call.name, execution); _ = try provisional_statuses.finishExecutedCall( hooks, provisional_alloc, arena, turn_id, original_call, parallel_status_started[original_index], null, execution, safe_tool_output, prepared.memory, null, advertised_dynamic_tool_names, ); debug_trace.eventf("tool", "after_tool_execution", step_ctx, "call_id={s} name={s} result_kind={s} model_output_bytes={d}", .{ original_call.id, original_call.name, runtime_telemetry.toolExecutionResultKind(execution), safe_tool_output.len }); if (hooks.tool_activity_recorder) |recorder| { const phase: runtime_deps.ToolActivityPhase = if (execution.status == .success) .succeeded else .failed; recorder.record(original_call.id, original_call.name, phase) catch |err| { debug_trace.eventf( "subagent", "tool_activity_projection_lag", step_ctx, "call_id={s} tool_name={s} phase={s} outcome={s}", .{ original_call.id, original_call.name, @tagName(phase), @errorName(err), }, ); }; } } else if (original_call.argument_integrity != .valid) { _ = try provisional_statuses.finishExecutedCall( hooks, provisional_alloc, arena, turn_id, original_call, parallel_status_started[original_index], null, execution, safe_tool_output, prepared.memory, null, advertised_dynamic_tool_names, ); debug_trace.eventf( "tool", "argument_integrity_rejected", step_ctx, "call_id={s} name={s} failure={s} provenance=fx_local", .{ original_call.id, original_call.name, @tagName(original_call.argument_integrity) }, ); try runtime_tool_admission.recordRejectedToolCall( hooks, arena, original_call, safe_tool_output, null, ); } else if (!parallel_status_terminalized[original_index]) { _ = try provisional_statuses.finishExecutedCall( hooks, provisional_alloc, arena, turn_id, original_call, parallel_status_started[original_index], null, execution, safe_tool_output, prepared.memory, null, advertised_dynamic_tool_names, ); } for (execution.context_notices) |notice| { try hooks.pushContextNotice(notice); } debug_trace.eventf("tool", "execution_result", step_ctx, "call_id={s} name={s} result_kind={s} model_output_bytes={d}", .{ original_call.id, original_call.name, runtime_telemetry.toolExecutionResultKind(execution), safe_tool_output.len }); try appendToolResultContent( arena, within_turn_suffix, completed_tool_names, batch, original_call, safe_tool_output, prepared.memory, .{ .increment_error = execution.status == .failure or tool_result_errors.isToolOutputError(safe_tool_output), .record_completion = execution.status == .success, .status = runtime_execution_memory.persistedStatusForCurrentFxLocalResult( execution.status, safe_tool_output, ), }, ); } return cancelled_call;}
pub fn processCommittedFileResult( hooks: *const AgentRuntimeDeps, presentation_allocator: Allocator, history_allocator: Allocator, within_turn_suffix: *std.ArrayList(ChatMessage), completed_tool_names: *std.ArrayList([]u8), batch: *StepBatchState, tool_call: ToolCall, execution_call: ToolCall, execution: ToolExecutionResult, model_output: []const u8, result_memory: types.ToolResultMemory, committed_file_tool_name: []u8, status_started: bool, display_target: ?[]const u8, is_file_mutation: bool, turn_id: u64, advertised_dynamic_tool_names: []const []const u8, step_ctx: TraceContext,) !void { if (!is_file_mutation) { return error.InvalidCommittedFileExecutionResult; } const handoff = execution.committed_file_handoff orelse unreachable; if (hooks.usage) |usage| { usage.recordCommittedLines( handoff.preview.additions, handoff.preview.deletions, ) catch |err| { debug_trace.logf( "session", "usage code accounting failed additions={d} deletions={d} reason={s}", .{ handoff.preview.additions, handoff.preview.deletions, @errorName(err), }, ); usage.markCodeIncomplete(); }; if (!usage.persistCheckpoint()) usage.markCodeIncomplete(); } const committed_contract_degraded = execution.status != .success or !execution.tool_result_memory_prepared or execution.tool_result_memory == null or execution.diff_entry != null or execution.finish_turn; if (committed_contract_degraded) { debug_trace.eventf( "tool", "committed_result_contract_degraded", step_ctx, "call_id={s} name={s} status={s} memory={s} diff={s} finish_turn={s}", .{ tool_call.id, tool_call.name, @tagName(execution.status), if (execution.tool_result_memory_prepared and execution.tool_result_memory != null) "true" else "false", if (execution.diff_entry != null) "true" else "false", if (execution.finish_turn) "true" else "false", }, ); if (execution.diff_entry) |payload| { diff.freeDiffEntryPayload(std.heap.c_allocator, payload); } }
var prepared_memory = result_memory; prepared_memory.committed_file_presentation = runtime_execution_memory.captureCommittedFilePresentation( history_allocator, handoff, ) catch |err| blk: { debug_trace.logf( "session", "committed file resume presentation unavailable call_id={s} err={s}", .{ tool_call.id, @errorName(err) }, ); break :blk null; }; within_turn_suffix.appendAssumeCapacity(.{ .role = .tool, .content = model_output, .tool_call_id = tool_call.id, .tool_name = tool_call.name, .tool_result_status = runtime_execution_memory.persistedStatusForCurrentFxLocalResult( execution.status, model_output, ), .tool_result_memory = prepared_memory, }); debug_trace.eventf( "tool", "committed_result_appended", step_ctx, "call_id={s} name={s} model_output_bytes={d}", .{ tool_call.id, tool_call.name, model_output.len }, );
const publication = hooks.publish_committed_file_handoff( hooks.ctx, handoff, ); debug_trace.eventf( "tool", "committed_secondary_publication", step_ctx, "call_id={s} name={s} diff={s} tracker={s}", .{ tool_call.id, tool_call.name, @tagName(publication.diff), @tagName(publication.tracker), }, ); var reporting_degraded = publication.degraded() or committed_contract_degraded; runtime_tool_presentation.finishCommittedFileStatus( hooks, presentation_allocator, turn_id, execution_call, status_started, display_target, handoff.preview, advertised_dynamic_tool_names, ) catch |err| { reporting_degraded = true; debug_trace.eventf( "tool", "committed_status_publication_failed", step_ctx, "call_id={s} name={s} err={s}", .{ tool_call.id, tool_call.name, @errorName(err) }, ); }; if (execution.deferred_tool_completion) |deferred_completion| { if (hooks.publish_deferred_tool_completion) |publish_deferred| { const outcome = publish_deferred( hooks.ctx, deferred_completion, ); if (outcome == .failed) reporting_degraded = true; debug_trace.eventf( "tool", "deferred_tool_completion_publication", step_ctx, "call_id={s} name={s} outcome={s}", .{ tool_call.id, tool_call.name, @tagName(outcome) }, ); } else { reporting_degraded = true; debug_trace.eventf( "tool", "deferred_tool_completion_publication", step_ctx, "call_id={s} name={s} outcome=missing_publisher", .{ tool_call.id, tool_call.name }, ); } } if (reporting_degraded) { hooks.push_system_notice( hooks.ctx, "File change committed, but some secondary reporting failed.", ) catch |err| { debug_trace.eventf( "tool", "committed_reporting_warning_failed", step_ctx, "call_id={s} name={s} err={s}", .{ tool_call.id, tool_call.name, @errorName(err) }, ); }; }
debug_trace.eventf( "tool", "after_tool_execution", step_ctx, "call_id={s} name={s} result_kind=committed_file model_output_bytes={d}", .{ tool_call.id, tool_call.name, model_output.len }, ); debug_trace.eventf( "tool", "execution_result", step_ctx, "call_id={s} name={s} result_kind=committed_file model_output_bytes={d}", .{ tool_call.id, tool_call.name, model_output.len }, ); batch.step_total_count += 1; batch.step_had_writes = true; completed_tool_names.appendAssumeCapacity(committed_file_tool_name);}
pub fn appendOrdinaryExecutedResult( tool_registry: tool_dispatch.Registry, arena: Allocator, within_turn_suffix: *std.ArrayList(ChatMessage), completed_tool_names: *std.ArrayList([]u8), batch: *StepBatchState, tool_call: ToolCall, model_output: []const u8, memory: types.ToolResultMemory, execution: ToolExecutionResult,) !void { const activity = runtime_tool_presentation.activityKindForCall(arena, tool_registry, tool_call); try appendToolResultContent( arena, within_turn_suffix, completed_tool_names, batch, tool_call, model_output, memory, .{ .increment_error = execution.status == .failure or tool_result_errors.isToolOutputError(model_output), .record_completion = true, .mark_write = activity == .write or activity == .edit, .status = runtime_execution_memory.persistedStatusForCurrentFxLocalResult( execution.status, model_output, ), }, );}
pub fn appendReviewContinuationSuffix( review_enabled: bool, arena: Allocator, within_turn_suffix: *std.ArrayList(ChatMessage), batch: *const StepBatchState,) !void { if (review_enabled and batch.step_had_writes) { const review_prompt = "Review the changes you just made. Re-read any modified files and briefly note any issues (syntax errors, missing imports, logic bugs). If everything looks correct, say so."; try within_turn_suffix.append(arena, .{ .role = .user, .content = review_prompt }); }}
test "appendPermissionFeedback marks typed approval feedback" { const alloc = std.testing.allocator; var batch = StepBatchState{}; defer { for (batch.pending_user_suffix.items) |message| { if (message.content) |text| alloc.free(@constCast(text)); } batch.pending_user_suffix.deinit(alloc); } try appendPermissionFeedback( alloc, &batch, "call_permission", &.{ "", "read it after writing" }, );
try std.testing.expectEqual(@as(usize, 1), batch.pending_user_suffix.items.len); try std.testing.expect(batch.pending_user_suffix.items[0].permission_feedback); try std.testing.expectEqualStrings( "call_permission", batch.pending_user_suffix.items[0].tool_call_id.?, );}
test "drained batch feedback follows all tool results and keeps its source call" { const alloc = std.testing.allocator; var batch = StepBatchState{}; var suffix: std.ArrayList(ChatMessage) = .empty; defer { for (suffix.items) |message| { if (message.role == .user) { if (message.content) |text| alloc.free(@constCast(text)); } } suffix.deinit(alloc); batch.pending_user_suffix.deinit(alloc); } var completed_tool_names: std.ArrayList([]u8) = .empty; defer completed_tool_names.deinit(alloc); const calls = [_]ToolCall{ .{ .id = "call_first", .name = "run_command", .arguments_json = "{}" }, .{ .id = "call_second", .name = "run_command", .arguments_json = "{}" }, };
try appendAssistantToolCallStep(alloc, &suffix, null, &calls, null); try appendToolResultContent( alloc, &suffix, &completed_tool_names, &batch, calls[0], "first command completed", null, .{}, ); try appendPermissionFeedback( alloc, &batch, calls[0].id, &.{"first command feedback marker"}, ); try appendToolResultContent( alloc, &suffix, &completed_tool_names, &batch, calls[1], "second command completed", null, .{}, ); try drainPendingUserSuffix(alloc, &batch, &suffix);
try std.testing.expectEqual(@as(usize, 4), suffix.items.len); try std.testing.expectEqual(.assistant, suffix.items[0].role); try std.testing.expectEqual(.tool, suffix.items[1].role); try std.testing.expectEqual(.tool, suffix.items[2].role); try std.testing.expectEqual(.user, suffix.items[3].role); try std.testing.expectEqualStrings("call_first", suffix.items[3].tool_call_id.?); try std.testing.expect(suffix.items[3].permission_feedback);}
test "repeated identical failures escalate from the second failure on" { const backing = std.testing.allocator; var arena_state = std.heap.ArenaAllocator.init(backing); defer arena_state.deinit(); const alloc = arena_state.allocator(); var suffix: std.ArrayList(ChatMessage) = .empty; var completed_tool_names: std.ArrayList([]u8) = .empty; const call: ToolCall = .{ .id = "edit-1", .name = "edit_file", .arguments_json = "{\"path\":\"strategy.ts\",\"old_string\":\" trajectory?: Trajectory;\\n}\",\"new_string\":\"}\"}", }; const failure_output = "edit_file failed: old_string not found in file";
var escalation: runtime_tool_admission.IdenticalFailureEscalationState = .{};
var step: usize = 0; while (step < 3) : (step += 1) { var batch: StepBatchState = .{ .identical_failure_escalation = &escalation, }; try appendToolResultContent( alloc, &suffix, &completed_tool_names, &batch, call, failure_output, null, .{ .increment_error = true }, ); }
try std.testing.expectEqual(@as(usize, 3), suffix.items.len); // The first failure is delivered plain; the second and third carry the // running count, and every delivery preserves the original failure text. try std.testing.expectEqualStrings(failure_output, suffix.items[0].content.?); try std.testing.expect(std.mem.find(u8, suffix.items[1].content.?, "already failed 2 times this turn") != null); try std.testing.expect(std.mem.find(u8, suffix.items[2].content.?, "already failed 3 times this turn") != null); for (suffix.items) |message| { try std.testing.expect(std.mem.find(u8, message.content.?, failure_output) != null); }}
test "identical failures stay plain when no escalation tracker is attached" { const backing = std.testing.allocator; var arena_state = std.heap.ArenaAllocator.init(backing); defer arena_state.deinit(); const alloc = arena_state.allocator(); var suffix: std.ArrayList(ChatMessage) = .empty; var completed_tool_names: std.ArrayList([]u8) = .empty; const call: ToolCall = .{ .id = "edit-1", .name = "edit_file", .arguments_json = "{\"path\":\"strategy.ts\"}", };
var step: usize = 0; while (step < 2) : (step += 1) { var batch: StepBatchState = .{}; try appendToolResultContent( alloc, &suffix, &completed_tool_names, &batch, call, "edit_file failed: old_string not found in file", null, .{ .increment_error = true }, ); }
for (suffix.items) |message| { try std.testing.expectEqualStrings( "edit_file failed: old_string not found in file", message.content.?, ); }}
test "malformed argument rejections never escalate through the identical failure tracker" { const backing = std.testing.allocator; var arena_state = std.heap.ArenaAllocator.init(backing); defer arena_state.deinit(); const alloc = arena_state.allocator(); var suffix: std.ArrayList(ChatMessage) = .empty; var completed_tool_names: std.ArrayList([]u8) = .empty; const malformed: ToolCall = .{ .id = "call-1", .name = "read_file", .arguments_json = "{}", .argument_integrity = .malformed_json, }; const rejected_output = "Tool execution failed: malformed arguments";
var escalation: runtime_tool_admission.IdenticalFailureEscalationState = .{};
var step: usize = 0; while (step < 2) : (step += 1) { var batch: StepBatchState = .{ .identical_failure_escalation = &escalation, }; try appendToolResultContent( alloc, &suffix, &completed_tool_names, &batch, malformed, rejected_output, null, .{ .increment_error = true }, ); }
for (suffix.items) |message| { try std.testing.expectEqualStrings(rejected_output, message.content.?); } try std.testing.expectEqual(@as(usize, 0), escalation.counts.items.len);}