commit 5420dda02627f052ebf902f19bf168af2e27547b from: Isaac Meerleo date: Wed Jul 15 19:15:13 2026 UTC integrate canonical turn mechanics into lead commit - 5ffcba5dccaf9389ab199d516bc4b7d9d03cfd1a commit + 5420dda02627f052ebf902f19bf168af2e27547b blob - 63a17da473cac7714cc5324c24c36a51637ea12b blob + 1ce70ab953bb2f623c36c6b3c7be8703268a1997 --- regress/turn/agent.sh +++ regress/turn/agent.sh @@ -131,6 +131,15 @@ run_fugu_ndjson() { rc=$? } +run_fugu_json() { + ( cd "$dir/work" && \ + FUGU_CONF=$dir/fugu.conf FUGU_LIBEXEC=$dir/libexec/fugu \ + FUGU_CA_FILE=$dir/cert.pem FUGU_RETRY_BASE_MS=50 \ + "$fugu" -p "$1" -o json "$dir/out" 2>"$dir/errout" + rc=$? +} + request_section() { awk -v name="$1" ' $0 == "--- " name " ---" { found = 1; next } @@ -570,6 +579,23 @@ ok "agent usage: provider events are forwarded exactly $([ "$n" -eq 6 ]; echo $?) stop_stub +# The final aggregate contains both Lead Generations and the nested child +# Generation exactly once: 13+21+25 input and (1+14)+(1+7)+(1+8) output. +rm -rf "$HOME/.fugu" +start_stub AGENTCALL AGENTREPORT AGENTSYNTH +write_conf +run_fugu_json "exercise subagent aggregate usage" +ok "agent usage: JSON turn completes" $([ "$rc" -eq 0 ]; echo $?) +n=$(wc -l < "$dir/out") +grep -q '"assistant_text":"LEAD-SYNTHESIS-MARKER"' "$dir/out" && \ + grep -q '"stop_reason":"end_turn"' "$dir/out" && \ + grep -q '"tool_call_count":1' "$dir/out" && [ "$n" -eq 1 ] +ok "agent usage: final JSON has the delegated Turn shape" $? +grep -q '"input_tokens":59' "$dir/out" && \ + grep -q '"output_tokens":32' "$dir/out" +ok "agent usage: nested usage enters the final aggregate exactly once" $? +stop_stub + # A token-limited text reply is not mistaken for a completed report. Its # partial text is preserved, followed by one tool-free conclusion request. rm -rf "$HOME/.fugu" blob - a43f2699f1adfc209698717aa02f85701a373fab blob + ad425bd327f6fed1b27f8091d0960f111bd718cf --- regress/turn/run.sh +++ regress/turn/run.sh @@ -320,6 +320,54 @@ ok "lead malformed arguments: invalid call is never ex $([ $? -ne 0 ]; echo $?) stop_stub +# A canonical result-batch overflow abandons the whole transaction. Line mode +# must remain usable, and neither the results accepted before the overflowing +# ordinal nor a follow-up Projection may escape the abandoned Turn. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqlog" +start_stub "RESULTBOUND OK" +write_conf "$stubport" +printf 'overflow the tool results\nrecover after overflow\n/quit\n' > "$dir/lin" +run_fugu_in "$dir/lin" +ok "lead result bound: line session survives the abandoned batch" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q 'Hello, world.' "$dir/out" +ok "lead result bound: the next Turn completes" $? +journal=$(ls "$sessions"/*.ndjson 2>/dev/null | head -1) +abandonedn=$(grep -c '"outcome":"abandoned"' "$journal" 2>/dev/null) +okn=$(grep -c '"outcome":"ok"' "$journal" 2>/dev/null) +ok "lead result bound: one Turn abandons and one commits" \ + $([ "$abandonedn" -eq 1 ] && [ "$okn" -eq 1 ]; echo $?) +resultn=$(grep -c '"bt":"tool_result"' "$journal" 2>/dev/null) +ok "lead result bound: no partial result batch is journaled" \ + $([ "$resultn" -eq 0 ]; echo $?) +n=$(grep -c '^--- .* ---$' "$dir/reqlog") +ok "lead result bound: no follow-up Projection escapes" \ + $([ "$n" -eq 2 ]; echo $?) +stop_stub + +# A malformed nonterminal Lead Generation must drain through the raw Provider +# terminal before line mode reuses the Lead correlation id. PROTODRAIN pauses +# after an API-valid duplicate tool id; leaving the first request early races +# the following request against that still-live id in fugu-api. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqlog" +start_stub "PROTODRAIN OK" +write_conf "$stubport" +printf 'reject a duplicate tool id\ncontinue after protocol failure\n/quit\n' \ + > "$dir/lin" +run_fugu_in "$dir/lin" +ok "lead protocol drain: line session survives the delayed terminal" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q 'malformed tool begin' "$dir/errout" +ok "lead protocol drain: canonical protocol error is visible" $? +grep -q 'Hello, world.' "$dir/out" +ok "lead protocol drain: the settled correlation id is reusable" $? +n=$(grep -c '^--- .* ---$' "$dir/reqlog") +ok "lead protocol drain: exactly two Lead requests ran" \ + $([ "$n" -eq 2 ]; echo $?) +stop_stub + # --- 6: a hostile second terminal after message_stop ---------------- # Exactly one terminal event per turn: a trailing error event must not # flip a completed turn into a failure. blob - 0c9c6f59381bc6cb457c42e3e72c9324fca043dc blob + e4bfb3fa6050306c035eb09fe1ca0d61ba252be2 --- src/fugu/Makefile +++ src/fugu/Makefile @@ -1,7 +1,7 @@ PROG= fugu SRCS= main.c coord.c priv.c conf.c parse.y \ journal.c sysprompt.c tooldefs.c agentcfg.c skills.c output.c \ - turn_txn.c anthropic_req.c openai_req.c generation.c msg.c \ + turn_txn.c turn_mechanics.c anthropic_req.c openai_req.c generation.c msg.c \ json.c model_window.c buf.c imsgev.c log.c xmalloc.c MAN= fugu.1 fugu.conf.5 MANDIR= ${PREFIX}/man/man blob - bc10088984800846b201b22dafd89a6d3331ad87 blob + 4472b2143282d3cdebf4abed38f4f8e1e66f820a --- src/fugu/coord.c +++ src/fugu/coord.c @@ -53,6 +53,7 @@ #include "skills.h" #include "tooldefs.h" #include "agentcfg.h" +#include "turn_mechanics.h" #include "turn_txn.h" #include "turn_txn_private.h" @@ -165,6 +166,7 @@ struct coord { int64_t turn_tool_calls; size_t turn_data; /* hostile text/args/results this turn */ int turn_data_exceeded; + int turn_usage_exceeded; struct buf input; /* a submitted line from the tty */ int have_input; /* an unconsumed UI_SUBMIT */ int want_quit; /* the tty asked to quit */ @@ -205,8 +207,13 @@ struct coord { /* one canonical Lead Generation and its presentation/execution adapters */ struct generation *generation; + struct turn_generation_io *turn_generation_io; + struct turn_tool_results *turn_tool_results; + int turn_tool_results_rejected; int generation_terminal; + int generation_protocol; int generation_present; + int turn_interim_pending; /* accepted Generation views used by the existing presentation/tool seams */ struct buf text; @@ -450,21 +457,38 @@ post_stream_test_delay(void) (void)nanosleep(&ts, NULL); } -/* The turn-end record is the cancellation boundary. Block SIGINT while - * writing it, after atomically checking both libevent's pipe and the kernel - * pending set, so a pre-commit Ctrl-C always abandons and a post-boundary - * Ctrl-C cannot create a half-committed outcome. */ -static int +enum commit_gate { + COMMIT_GATE_COMMITTED, + COMMIT_GATE_CANCELLED, + COMMIT_GATE_QUIT, + COMMIT_GATE_SWITCH +}; + +/* + * The turn-end record is the cancellation boundary. First drain every + * queued event so UI quit/switch decisions already at the Coordinator are + * visible. In print mode, block SIGINT while atomically checking both + * libevent's pipe and the kernel pending set, so a pre-commit Ctrl-C always + * abandons and a post-boundary Ctrl-C cannot create a half-committed outcome. + */ +static enum commit_gate commit_turn(struct coord *c, enum turn_txn_commit outcome) { sigset_t block, oldmask, pending; - int interrupted = 0; + enum commit_gate gate = COMMIT_GATE_COMMITTED; + (void)event_loop(EVLOOP_NONBLOCK); if (!c->have_sigint_ev) { - turn_txn_commit(c->turn, outcome); - return (0); + if (c->cancelled) + gate = COMMIT_GATE_CANCELLED; + else if (c->want_quit) + gate = COMMIT_GATE_QUIT; + else if (c->want_switch) + gate = COMMIT_GATE_SWITCH; + else + turn_txn_commit(c->turn, outcome); + return (gate); } - (void)event_loop(EVLOOP_NONBLOCK); sigemptyset(&block); sigaddset(&block, SIGINT); if (sigprocmask(SIG_BLOCK, &block, &oldmask) == -1) @@ -474,12 +498,16 @@ commit_turn(struct coord *c, enum turn_txn_commit outc fatal("sigpending"); if (c->cancelled || sigismember(&pending, SIGINT)) { c->cancelled = 1; - interrupted = 1; - } else + gate = COMMIT_GATE_CANCELLED; + } else if (c->want_quit) + gate = COMMIT_GATE_QUIT; + else if (c->want_switch) + gate = COMMIT_GATE_SWITCH; + else turn_txn_commit(c->turn, outcome); if (sigprocmask(SIG_SETMASK, &oldmask, NULL) == -1) fatal("sigprocmask restore SIGINT"); - return (interrupted ? -1 : 0); + return (gate); } static const char * @@ -1139,7 +1167,7 @@ stream_event(struct coord *c, struct imsg *imsg) return; if (imsg_get_id(imsg) != c->lead_api_id) fatalx("lead provider correlation mismatch"); - if (c->generation == NULL) + if ((c->generation == NULL) == (c->turn_generation_io == NULL)) fatalx("lead provider event without a Generation"); if (imsgev_get_payload(imsg, &payload) == -1) @@ -1185,11 +1213,40 @@ stream_event(struct coord *c, struct imsg *imsg) feed_data = &gu; feed_len = sizeof(gu); } - status = generation_feed(c->generation, kind, feed_data, feed_len, - &ev); + /* + * Once canonical Turn collection finds a protocol error, preserve its + * first diagnostic but keep this correlation live until fugu-api sends + * the raw terminal. Returning the Lead id early would race the next + * Generation against an API request that still owns it. Compaction keeps + * its standalone fail-closed Generation behavior. + */ + if (c->turn_generation_io != NULL && c->generation_protocol) { + if (kind == GENERATION_DONE || kind == GENERATION_ERROR) { + c->generation_terminal = 1; + c->wait = WAIT_NONE; + event_loopexit(NULL); + } + return; + } + if (c->turn_generation_io != NULL) + status = turn_generation_feed(c->turn_generation_io, kind, + feed_data, feed_len, &ev); + else + status = generation_feed(c->generation, kind, feed_data, feed_len, + &ev); if (status == GENERATION_PROTOCOL) { - const char *why = generation_error(c->generation); + const char *why; + if (c->turn_generation_io != NULL) { + c->generation_protocol = 1; + if (kind == GENERATION_DONE || kind == GENERATION_ERROR) { + c->generation_terminal = 1; + c->wait = WAIT_NONE; + event_loopexit(NULL); + } + return; + } + why = generation_error(c->generation); fatalx("lead provider protocol: %s", why != NULL ? why : "invalid Generation event"); } @@ -1202,9 +1259,13 @@ stream_event(struct coord *c, struct imsg *imsg) au.output_tokens = ev.usage.output_tokens; au.cache_read = ev.usage.cache_read; au.cache_write = ev.usage.cache_write; - if (!c->turn_data_exceeded) { - if (usage_account(c, &au) == -1) - c->turn_data_exceeded = 1; + if (!c->turn_data_exceeded && !c->turn_usage_exceeded) { + if (usage_account(c, &au) == -1) { + if (c->turn_generation_io != NULL) + c->turn_usage_exceeded = 1; + else + c->turn_data_exceeded = 1; + } else machine_usage(c, &au); } @@ -2301,12 +2362,25 @@ agent_dispatch(int fd, short event, void *arg) break; case FUGU_IMSG_AGENT_USAGE: { struct a_usage u; + struct generation_usage gu; + int observer_rc; if (p == NULL || len != sizeof(u)) fatalx("bad subagent usage"); + if (c->turn_tool_results == NULL) + fatalx("subagent usage outside Lead Tool execution"); memcpy(&u, p, sizeof(u)); - if (usage_account(c, &u) == -1) - c->turn_data_exceeded = 1; + memset(&gu, 0, sizeof(gu)); + gu.input_tokens = u.input_tokens; + gu.output_tokens = u.output_tokens; + gu.cache_read = u.cache_read; + gu.cache_write = u.cache_write; + if (turn_tool_results_add_usage(c->turn_tool_results, + &gu) == -1) + c->turn_tool_results_rejected = 1; + observer_rc = usage_account(c, &u); + if (observer_rc == -1) + c->turn_usage_exceeded = 1; else machine_usage(c, &u); break; @@ -2544,55 +2618,144 @@ build_provider_request(struct coord *c, struct buf *bo } } -/* - * Build and send the request for the current conversation, offering - * the configured tool surface. Returns -1 (with a message printed) - * when the serialized body exceeds the wire cap. - */ -static int -send_turn_request(struct coord *c) +static void +lead_projection(void *arg, struct turn_projection *projection) { + struct coord *c = arg; + struct turn_txn_projection p; + + turn_txn_projection(c->turn, &p); + memset(projection, 0, sizeof(*projection)); + projection->messages = p.messages; + projection->message_count = p.message_count; + projection->cache_tail_msg = p.cache_tail_msg; + projection->receipt = p.token; +} + +static void +lead_projection_sent(void *arg, const struct turn_projection *projection) +{ + struct coord *c = arg; + struct turn_txn_projection p; + + memset(&p, 0, sizeof(p)); + p.messages = projection->messages; + p.message_count = projection->message_count; + p.cache_tail_msg = projection->cache_tail_msg; + p.token = projection->receipt; + turn_txn_projection_sent(c->turn, &p); +} + +static void +lead_accept(void *arg, struct msg *message) +{ + struct coord *c = arg; + + turn_txn_accept(c->turn, message); +} + +/* The post-accept checkpoint preserves the old cancellation-before-progress + * order for interim non-UI narration. */ +static enum turn_checkpoint +lead_checkpoint(void *arg) +{ + struct coord *c = arg; + + if (cancel_checkpoint(c) || c->want_quit || c->want_switch) + return (TURN_CHECKPOINT_INTERRUPT); + if (c->turn_interim_pending) { + if (c->text.len > 0 && !c->ui) { + fputs("fugu: ", stderr); + fwrite(c->text.data, 1, c->text.len, stderr); + fputc('\n', stderr); + } + c->turn_interim_pending = 0; + } + return (TURN_CHECKPOINT_CONTINUE); +} + +/* Build, send, and collect one provider-neutral Lead Generation. */ +static enum turn_io +lead_generate(void *arg, const struct turn_generation_request *request, + struct turn_generation_io *io, char *error, size_t errorsz) +{ + struct coord *c = arg; const struct tool_def *base, *agent; - struct turn_txn_projection projection; struct tool_def tools[20]; struct buf body; - int nbase, ntools; + int nbase, ntools = 0, rc; - base = tooldefs(c->conf, c->skills, &nbase); - if (nbase > (int)(sizeof(tools) / sizeof(tools[0]))) - fatalx("too many tool definitions"); - memcpy(tools, base, (size_t)nbase * sizeof(tools[0])); - ntools = nbase; - agent = agentcfg_tool(c->agentcfg, c->active_provider_slot, - c->conf->provider_type, c->conf->model); - if (agent != NULL) { - if (ntools >= (int)(sizeof(tools) / sizeof(tools[0]))) + buf_init(&body); + if (request->tools == TURN_TOOLS_ENABLED) { + base = tooldefs(c->conf, c->skills, &nbase); + if (nbase > (int)(sizeof(tools) / sizeof(tools[0]))) fatalx("too many tool definitions"); - tools[ntools++] = *agent; + memcpy(tools, base, (size_t)nbase * sizeof(tools[0])); + ntools = nbase; + agent = agentcfg_tool(c->agentcfg, c->active_provider_slot, + c->conf->provider_type, c->conf->model); + if (agent != NULL) { + if (ntools >= (int)(sizeof(tools) / sizeof(tools[0]))) + fatalx("too many tool definitions"); + tools[ntools++] = *agent; + } + } else if (request->tools != TURN_TOOLS_DISABLED) { + buf_free(&body); + (void)snprintf(error, errorsz, "%s", "invalid Lead tool mode"); + return (TURN_IO_ERROR); } - buf_init(&body); - turn_txn_projection(c->turn, &projection); - build_provider_request(c, &body, projection.messages, NULL, tools, - ntools, projection.cache_tail_msg); + build_provider_request(c, &body, request->projection.messages, NULL, + ntools > 0 ? tools : NULL, ntools, + request->projection.cache_tail_msg); if (body.len > FUGU_REQUEST_MAX) { - /* - * JSON escaping can expand piped context or a large tool - * result severalfold; refuse conspicuously here. To the - * custodian an oversized body is a protocol violation, - * not a user error (I13). - */ - size_t mb = body.len / (1024 * 1024); + size_t mb = body.len / (1024 * 1024); buf_free(&body); fprintf(stderr, "fugu: serialized request too large " "(%zu MB; the cap is %d MB)\n", mb, FUGU_REQUEST_MAX / (1024 * 1024)); - return (-1); + (void)snprintf(error, errorsz, "%s", + "request serialization failed"); + return (TURN_IO_ERROR); } + + reset_tcalls(c); + buf_reset(&c->text); + memset(&c->usage, 0, sizeof(c->usage)); + c->stop_reason[0] = '\0'; + c->err[0] = '\0'; + c->generation_terminal = 0; + c->generation_protocol = 0; + c->generation_present = 1; + c->turn_generation_io = io; send_request(c, &body, c->lead_api_id, c->active_provider_slot); - turn_txn_projection_sent(c->turn, &projection); buf_free(&body); - return (0); + if (turn_generation_sent(io) == -1) { + c->turn_generation_io = NULL; + c->generation_present = 0; + (void)snprintf(error, errorsz, "%s", + "could not acknowledge the Lead Projection"); + return (TURN_IO_ERROR); + } + + c->wait = WAIT_STREAM; + rc = event_dispatch(); + c->wait = WAIT_NONE; + c->turn_generation_io = NULL; + c->generation_present = 0; + post_stream_test_delay(); + if (rc == -1) + fatal("event_dispatch Lead Generation"); + if (c->cancelled || c->want_quit || c->want_switch) + return (TURN_IO_INTERRUPTED); + if (c->generation_protocol) + return (TURN_IO_OK); + if (!c->generation_terminal) { + (void)snprintf(error, errorsz, "%s", + "provider stream ended before its terminal"); + return (TURN_IO_ERROR); + } + return (TURN_IO_OK); } static void @@ -2678,23 +2841,22 @@ lead_assistant_text(const struct msg *assistant) /* Copy accepted canonical views into the existing presentation/tool seams. */ static void -lead_generation_adapt(struct coord *c, - const struct generation_result *result) +lead_generation_adapt(struct coord *c, const struct msg *assistant, + size_t tool_calls) { struct generation_span text; struct tool_call *tc; struct block *b; - if (result->assistant == NULL || - result->tool_calls > MAX_TOOL_CALLS) + if (assistant == NULL || tool_calls > MAX_TOOL_CALLS) fatalx("invalid successful Lead Generation"); - text = lead_assistant_text(result->assistant); + text = lead_assistant_text(assistant); buf_reset(&c->text); if (text.len > 0) buf_add(&c->text, text.data, text.len); reset_tcalls(c); - TAILQ_FOREACH(b, &result->assistant->blocks, entry) { + TAILQ_FOREACH(b, &assistant->blocks, entry) { if (b->type != BLOCK_TOOL_USE) continue; if (c->ntcalls >= MAX_TOOL_CALLS || b->tool_id == NULL || @@ -2716,10 +2878,27 @@ lead_generation_adapt(struct coord *c, tc->result_error = 0; c->ntcalls++; } - if ((size_t)c->ntcalls != result->tool_calls) + if ((size_t)c->ntcalls != tool_calls) fatalx("canonical Lead tool-call count mismatch"); } +static void +lead_assistant_accepted(void *arg, const struct turn_assistant_event *event) +{ + struct coord *c = arg; + + if (event->stop_reason == NULL) + fatalx("canonical Lead assistant has no stop reason"); + lead_generation_adapt(c, event->message, event->tool_calls); + strlcpy(c->stop_reason, event->stop_reason, sizeof(c->stop_reason)); + if (event->synthesized_text.len > 0 && c->ui) + emit_delta(c, event->synthesized_text.data, + event->synthesized_text.len); + machine_assistant(c); + if (event->tool_calls > 0 && c->text.len > 0 && !c->ui) + c->turn_interim_pending = 1; +} + /* * The skill tool (behavior.md section 4) is answered by the * coordinator itself -- the bodies never leave this process -- rather @@ -2762,18 +2941,6 @@ resolve_skill(struct coord *c, struct tool_call *tc, s json_done(&j); } -static int -account_tool_result(struct coord *c) -{ - if (c->turn_data_exceeded || - c->tool_out.len > FUGU_TURN_DATA_MAX - c->turn_data) { - c->turn_data_exceeded = 1; - return (-1); - } - c->turn_data += c->tool_out.len; - return (0); -} - static void agent_call_error(struct tool_call *, const char *, ...) __attribute__((__format__ (printf, 2, 3))); @@ -3125,48 +3292,80 @@ run_pending_agents(struct coord *c) buf_free(&system); } -/* Run every pending tool_use, returning a user message of results. */ -static struct msg * -run_pending_tools(struct coord *c) +/* Execute one canonical batch through the existing authority and presentation + * seams. The result builder copies and orders every result by call ordinal. */ +static enum turn_io +lead_execute(void *arg, const struct turn_tool_batch *batch, + struct turn_tool_results *results, char *error, size_t errorsz) { - struct msg *um; - int i; + struct coord *c = arg; + struct tool_call *tc; + const struct turn_tool_call *call; + size_t i; + int err, is_web; + c->turn_tool_results_rejected = 0; + + if (batch->count != (size_t)c->ntcalls) { + (void)snprintf(error, errorsz, "%s", + "canonical Lead tool batch count mismatch"); + return (TURN_IO_ERROR); + } + for (i = 0; i < batch->count; i++) { + tc = &c->tcalls[i]; + call = &batch->calls[i]; + if (strcmp(tc->id, call->id) != 0 || + strcmp(tc->name, call->name) != 0 || + tc->input.len != call->input_len || + (tc->input.len > 0 && + memcmp(tc->input.data, call->input, tc->input.len) != 0)) { + (void)snprintf(error, errorsz, "%s", + "canonical Lead tool batch mismatch"); + return (TURN_IO_ERROR); + } + } + + c->turn_tool_results = results; run_pending_agents(c); - um = msg_new(ROLE_USER); - if (c->cancelled || c->want_switch || c->want_quit) - return (um); - for (i = 0; i < c->ntcalls; i++) { - struct tool_call *tc = &c->tcalls[i]; + if (c->cancelled || c->want_switch || c->want_quit) { + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + return (TURN_IO_INTERRUPTED); + } + if (c->turn_data_exceeded) { + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + (void)snprintf(error, errorsz, "%s", + "turn data or worker result exceeded its bound"); + return (TURN_IO_ERROR); + } + if (c->turn_tool_results_rejected) { + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + return (TURN_IO_OK); + } + for (i = 0; i < batch->count; i++) { + tc = &c->tcalls[i]; + call = &batch->calls[i]; c->turn_tool_calls++; machine_tool_call(c, tc); emit_tool(c, tc->name); /* terse (S1) */ if (strcmp(tc->name, "agent") == 0) { if (!tc->result_ready) agent_call_error(tc, "agent: no result"); - buf_reset(&c->tool_out); - if (tc->result.len > 0) - buf_add(&c->tool_out, tc->result.data, tc->result.len); - if (account_tool_result(c) == -1) + if (turn_tool_result_set(results, i, call->id, + tc->result.data, tc->result.len, + tc->result_error) == -1) break; - msg_add_tool_result(um, tc->id, - (const char *)tc->result.data, tc->result.len, - tc->result_error); machine_tool_result(c, tc, &tc->result, tc->result_error); } else if (strcmp(tc->name, "skill") == 0) { - int err; - resolve_skill(c, tc, &c->tool_out, &err); - if (account_tool_result(c) == -1) + if (turn_tool_result_set(results, i, call->id, + c->tool_out.data, c->tool_out.len, err) == -1) break; - msg_add_tool_result(um, tc->id, - (const char *)c->tool_out.data, c->tool_out.len, - err); machine_tool_result(c, tc, &c->tool_out, err); } else { - int is_web; - is_web = strcmp(tc->name, "web_search") == 0 || strcmp(tc->name, "web_fetch") == 0 || strcmp(tc->name, "http_request") == 0; @@ -3174,41 +3373,39 @@ run_pending_tools(struct coord *c) run_web(c, tc); else run_tool(c, tc); - if (c->want_switch || c->cancelled) - break; /* the turn is being abandoned */ - if (account_tool_result(c) == -1) + if (c->cancelled || c->want_switch || c->want_quit) { + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + return (TURN_IO_INTERRUPTED); + } + if (c->turn_data_exceeded) { + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + (void)snprintf(error, errorsz, "%s", + "turn data or worker result exceeded its bound"); + return (TURN_IO_ERROR); + } + if (turn_tool_result_set(results, i, call->id, + c->tool_out.data, c->tool_out.len, + c->tool_is_error) == -1) break; - msg_add_tool_result(um, tc->id, - (const char *)c->tool_out.data, c->tool_out.len, - c->tool_is_error); machine_tool_result(c, tc, &c->tool_out, c->tool_is_error); - /* - * The diff is presentation only (behavior.md 2.1): it - * is shown but never joins the tool_result the model - * sees or the journal (I14). Line/print mode prints it - * plainly; the curses front end colors it. - */ + /* The diff is presentation only and never joins Projection. */ if (!is_web) emit_diff(c); } } - /* - * A steered queue entry rides with this batch as user text - * (behavior.md 2.1): journaled inside the turn at exactly this - * position, so a resumed projection is equivalent (section 7). - */ - if (c->steer.len > 0 && !c->want_switch) { - if (c->steer.len > FUGU_TURN_DATA_MAX - c->turn_data) { - c->turn_data_exceeded = 1; - return (um); - } - c->turn_data += c->steer.len; - msg_add_text(um, (const char *)c->steer.data, c->steer.len); - buf_reset(&c->steer); + /* A Steer rides in the same canonical user message as this result batch. */ + if (i == batch->count && c->steer.len > 0 && !c->want_switch) { + if (turn_tool_results_add_text(results, c->steer.data, + c->steer.len) == 0) + buf_reset(&c->steer); } - return (um); + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + return (TURN_IO_OK); } static int @@ -3216,6 +3413,10 @@ abandon_turn(struct coord *c, int status) { turn_txn_abandon(c->turn); reset_tcalls(c); + c->turn_generation_io = NULL; + c->turn_tool_results = NULL; + c->turn_tool_results_rejected = 0; + c->turn_interim_pending = 0; buf_reset(&c->steer); buf_reset(&c->steer_pend); return (status); @@ -3243,22 +3444,26 @@ fail_turn(struct coord *c, int status, const char *mes return (result); } -/* - * One turn: append the user message, then loop -- request, stream, - * run any requested tools, feed the results back -- until the model - * stops asking for tools or the iteration cap is reached. The whole - * turn is journaled transactionally (invariants.md I14); a failed - * turn contributes nothing, a completed or capped turn commits. - * Returns the print-mode exit code (0 clean, 1 error, 2 cap, - * 3 cancellation). - */ static int +lead_usage_matches(const struct a_usage *observed, + const struct generation_usage *canonical) +{ + return (observed->input_tokens == canonical->input_tokens && + observed->output_tokens == canonical->output_tokens && + observed->cache_read == canonical->cache_read && + observed->cache_write == canonical->cache_write); +} + +static int run_turn(struct coord *c, const char *prompt, size_t promptlen) { - struct generation_result generation; + struct turn_adapter adapter; + struct turn_options options; + struct turn_result result; struct msg *um; - size_t available; - int iter, pump_rc; + const char *why; + enum commit_gate gate; + int status; if (promptlen > FUGU_TURN_DATA_MAX) { emit_error(c, "fugu: prompt exceeds the 4 MiB turn-data bound"); @@ -3270,159 +3475,100 @@ run_turn(struct coord *c, const char *prompt, size_t p msg_add_text(um, prompt, promptlen); memset(&c->turn_usage, 0, sizeof(c->turn_usage)); + memset(&c->usage, 0, sizeof(c->usage)); c->turn_tool_calls = 0; c->turn_data = promptlen; c->turn_data_exceeded = 0; + c->turn_usage_exceeded = 0; + c->turn_tool_results_rejected = 0; + c->turn_interim_pending = 0; turn_txn_begin(c->turn); turn_txn_accept(c->turn, um); - if (cancel_checkpoint(c)) - return (cancel_turn(c)); - if (c->want_switch) { - int status = abandon_turn(c, 1); - machine_error(c, "turn abandoned by session switch"); + memset(&adapter, 0, sizeof(adapter)); + adapter.arg = c; + adapter.projection = lead_projection; + adapter.projection_sent = lead_projection_sent; + adapter.accept = lead_accept; + adapter.assistant_accepted = lead_assistant_accepted; + adapter.checkpoint = lead_checkpoint; + adapter.generate = lead_generate; + adapter.execute = lead_execute; + memset(&options, 0, sizeof(options)); + options.policy = TURN_POLICY_LEAD; + options.data_limit = FUGU_TURN_DATA_MAX; + options.data_used = promptlen; + options.tool_batch_limit = TURN_ITER_MAX; + if (turn_run(&adapter, &options, &result) == -1) + fatalx("invalid canonical Lead Turn configuration"); + + c->turn_data = result.data_used; + reset_tcalls(c); + c->turn_interim_pending = 0; + /* + * The presentation ledger records every direct and nested usage event as + * it crosses the Adapter. A settled canonical Turn must independently + * reach the same aggregate; never overwrite either view and mask a missed + * or duplicate bridge. + */ + if ((result.status == TURN_STATUS_DONE || + result.status == TURN_STATUS_CAP) && + !lead_usage_matches(&c->turn_usage, &result.usage)) { + status = abandon_turn(c, 1); + emit_error(c, "fugu: Lead Turn usage accounting mismatch"); + machine_error(c, "Lead Turn usage accounting mismatch"); machine_finish(c, 1); return (status); } - for (iter = 1; ; iter++) { - if (send_turn_request(c) == -1) { + switch (result.status) { + case TURN_STATUS_DONE: + gate = commit_turn(c, TURN_TXN_OK); + if (gate == COMMIT_GATE_SWITCH) return (fail_turn(c, 1, - "request serialization failed")); - } - if (c->turn_data > FUGU_TURN_DATA_MAX) - fatalx("Lead Turn exceeded its data budget"); - available = FUGU_TURN_DATA_MAX - c->turn_data; - pump_rc = pump_generation(c, available, 1, &generation); - post_stream_test_delay(); - - if (cancel_checkpoint(c)) { - generation_result_dispose(&generation); + "turn abandoned by session switch")); + if (gate != COMMIT_GATE_COMMITTED) return (cancel_turn(c)); - } - /* a drawer switch abandons the in-flight turn cleanly - * (behavior.md 2.1, section 7); its steer text dies with it */ - if (c->want_switch) { - generation_result_dispose(&generation); + emit_answer(c); + machine_finish(c, 0); + return (0); + case TURN_STATUS_CAP: + gate = commit_turn(c, TURN_TXN_CAP); + if (gate == COMMIT_GATE_SWITCH) return (fail_turn(c, 1, "turn abandoned by session switch")); - } - if (pump_rc == -1) { - generation_result_dispose(&generation); - emit_error(c, - "fugu: provider stream ended before its terminal"); + if (gate != COMMIT_GATE_COMMITTED) + return (cancel_turn(c)); + emit_error(c, "fugu: reached the %d-step tool limit for one turn", + TURN_ITER_MAX); + machine_error(c, "tool-loop iteration limit reached"); + machine_finish(c, 2); + return (2); + case TURN_STATUS_INTERRUPTED: + if (c->cancelled || c->want_quit) + return (cancel_turn(c)); + if (c->want_switch) return (fail_turn(c, 1, - "provider stream ended before its terminal")); - } - c->turn_data += generation.data_used; - if (generation.outcome == GENERATION_INVALID && - generation.invalid_cause == GENERATION_INVALID_DATA_BOUND) - c->turn_data_exceeded = 1; - if (c->turn_data_exceeded) { - int status = abandon_turn(c, 1); - - generation_result_dispose(&generation); - emit_error(c, "fugu: turn data exceeds 4 MiB"); - machine_error(c, "turn data exceeds 4 MiB"); - machine_finish(c, 1); - return (status); - } - if (generation.outcome == GENERATION_PROVIDER_ERROR || - generation.outcome == GENERATION_INVALID) { - const char *why = c->err[0] != '\0' ? c->err : - "turn failed"; - int status = abandon_turn(c, 1); - - /* a failed turn contributes nothing (I14) */ - generation_result_dispose(&generation); + "turn abandoned by session switch")); + return (fail_turn(c, 1, "turn interrupted")); + case TURN_STATUS_FAILED: + if (result.failure == TURN_FAILURE_GENERATION && + result.invalid_cause == GENERATION_INVALID_DATA_BOUND) + why = "turn data exceeds 4 MiB"; + else if (result.failure == TURN_FAILURE_DATA_BOUND) + why = "turn data or worker result exceeded its bound"; + else + why = result.error[0] != '\0' ? result.error : "turn failed"; + status = abandon_turn(c, 1); + if (strcmp(why, "request serialization failed") != 0) emit_error(c, "fugu: %s", why); - machine_error(c, why); - machine_finish(c, 1); - return (status); - } - if ((generation.outcome != GENERATION_NORMAL && - generation.outcome != GENERATION_LENGTH) || - generation.invalid_cause != GENERATION_INVALID_NONE || - generation.assistant == NULL) - fatalx("invalid successful Lead Generation result"); - if (cancel_checkpoint(c)) { - generation_result_dispose(&generation); - return (cancel_turn(c)); - } - - lead_generation_adapt(c, &generation); - turn_txn_accept(c->turn, generation.assistant); - generation.assistant = NULL; - if (generation.synthesized_text.len > 0 && c->ui) - emit_delta(c, generation.synthesized_text.data, - generation.synthesized_text.len); - machine_assistant(c); - if (cancel_checkpoint(c)) { - generation_result_dispose(&generation); - return (cancel_turn(c)); - } - - /* - * Any tool_use blocks MUST be answered with tool_result on - * the next request, whatever stop_reason the provider sent; - * keying the decision on the presence of tool calls (not on - * stop_reason) keeps a hostile or buggy provider from - * leaving a dangling tool_use committed to the journal (I14). - */ - if (c->ntcalls == 0) { - /* the model is done: commit and show its text */ - generation_result_dispose(&generation); - if (commit_turn(c, TURN_TXN_OK) == -1) - return (cancel_turn(c)); - emit_answer(c); - machine_finish(c, 0); - return (0); - } - - /* interim narration is progress, not the final answer (S1); - * in UI mode it already streamed as deltas into the block */ - if (c->text.len > 0 && !c->ui) { - fputs("fugu: ", stderr); - fwrite(c->text.data, 1, c->text.len, stderr); - fputc('\n', stderr); - } - generation_result_dispose(&generation); - - /* - * Run this round's tools and feed the results back. At the - * cap we still run and commit them, so the journaled turn - * ends on a valid tool_result rather than a dangling - * tool_use (I14), then stop with the pinned error/exit 2. - */ - um = run_pending_tools(c); - if (c->want_switch || cancel_checkpoint(c) || - c->turn_data_exceeded) { /* abandoned mid-tools */ - const char *why; - int status; - - msg_free(um); - status = c->cancelled ? 3 : 1; - why = c->cancelled ? "turn cancelled" : - c->turn_data_exceeded ? - "turn data or worker result exceeded its bound" : - "turn abandoned by session switch"; - return (fail_turn(c, status, why)); - } - turn_txn_accept(c->turn, um); - reset_tcalls(c); - if (cancel_checkpoint(c)) - return (cancel_turn(c)); - - if (iter >= TURN_ITER_MAX) { - if (commit_turn(c, TURN_TXN_CAP) == -1) - return (cancel_turn(c)); - emit_error(c, "fugu: reached the %d-step tool limit " - "for one turn", TURN_ITER_MAX); - machine_error(c, "tool-loop iteration limit reached"); - machine_finish(c, 2); - return (2); - } + machine_error(c, why); + machine_finish(c, 1); + return (status); + case TURN_STATUS_CONCLUDED: + fatalx("canonical Lead Turn returned Subagent conclusion status"); } + fatalx("canonical Lead Turn returned an unknown status"); } static int64_t est_tokens(size_t); /* defined below */ @@ -3498,6 +3644,7 @@ cmd_compact(struct coord *c) buf_free(&body); c->turn_data = 0; c->turn_data_exceeded = 0; + c->turn_usage_exceeded = 0; pump_rc = pump_generation(c, FUGU_TURN_DATA_MAX, 1, &generation); if (cancel_checkpoint(c)) {