commit ffccea59b4d4877ec18fbcbac7f41c955416b635 from: Isaac Meerleo date: Wed Jul 15 17:44:34 2026 UTC integrate canonical generation into lead commit - 11cb7bb8d4c70c4973e2647be12cfd08cadbc8df commit + ffccea59b4d4877ec18fbcbac7f41c955416b635 blob - 8d47d8a44543aa47353f9c8c12bcf6c1c688ded6 blob + d0b7d44f39adbf0768582fe28b838c4a6bf5b273 --- regress/print/jsoncheck.c +++ regress/print/jsoncheck.c @@ -34,6 +34,9 @@ struct events { int tool_call; int tool_result; int usage; + int64_t usage_input; + int64_t usage_output; + int ndjson_stage; int error; int end; int lines; @@ -153,6 +156,14 @@ event_line(struct events *ev, const u_char *data, size if (json_obj_len(&j, root) != 2 || (s = get_string(&j, root, "text", &slen)) == NULL) goto done; + if (ev->assistant == 0 && ev->ndjson_stage == 0 && + slen == 7 && memcmp(s, "working", 7) == 0) + ev->ndjson_stage = 1; + else if (ev->assistant == 1 && ev->ndjson_stage == 3 && + slen == 4 && memcmp(s, "done", 4) == 0) + ev->ndjson_stage = 4; + else + ev->ndjson_stage = -1; ev->assistant++; } else if (strcmp(type, "tool_call") == 0) { if (json_obj_len(&j, root) != 4 || @@ -167,19 +178,31 @@ event_line(struct events *ev, const u_char *data, size s = get_string(&j, root, "id", &slen); if (s == NULL || slen == 0) goto done; + if (ev->tool_call == 0 && ev->ndjson_stage == 1) + ev->ndjson_stage = 2; + else + ev->ndjson_stage = -1; ev->tool_call++; } else if (strcmp(type, "tool_result") == 0) { if (json_obj_len(&j, root) != 4 || json_obj_get(&j, root, "is_error") == -1 || (s = get_string(&j, root, "content", &slen)) == NULL) goto done; + if (ev->tool_result == 0 && ev->ndjson_stage == 2) + ev->ndjson_stage = 3; + else + ev->ndjson_stage = -1; ev->tool_result++; } else if (strcmp(type, "usage") == 0) { if (json_obj_len(&j, root) != 3 || get_number(&j, root, "input_tokens", &n1) == -1 || - get_number(&j, root, "output_tokens", &n2) == -1) + get_number(&j, root, "output_tokens", &n2) == -1 || + n1 < 0 || n2 < 0 || n1 > INT64_MAX - ev->usage_input || + n2 > INT64_MAX - ev->usage_output) goto done; ev->usage++; + ev->usage_input += n1; + ev->usage_output += n2; } else if (strcmp(type, "turn_error") == 0) { if (json_obj_len(&j, root) != 2 || (s = get_string(&j, root, "message", &slen)) == NULL) @@ -229,8 +252,10 @@ check_events(const struct buf *b, const char *mode) if (begin != b->len || ev.end != 1 || !ev.end_was_last) return (0); if (strcmp(mode, "ndjson") == 0) - return (ev.assistant >= 2 && ev.tool_call == 1 && - ev.tool_result == 1 && ev.usage >= 2 && ev.error == 0 && + return (ev.assistant == 2 && ev.tool_call == 1 && + ev.tool_result == 1 && ev.usage == 4 && ev.error == 0 && + ev.usage_input == 8 && ev.usage_output == 6 && + ev.ndjson_stage == 4 && ev.lines == 9 && ev.exit_status == 0 && ev.end_tool_count == 1); if (strcmp(mode, "cap") == 0) return (ev.tool_call == 50 && ev.tool_result == 50 && blob - 449dfa376277c752cbcbc4975e4e561268c783fd blob + 4e2bf6deaa28c9216497a9e3260ad6c005c3619a --- regress/print/run.sh +++ regress/print/run.sh @@ -296,7 +296,7 @@ write_conf run_fugu -p "use one tool" -o ndjson ok "ndjson: tool turn exits 0" $([ "$rc" -eq 0 ]; echo $?) "$obj/jsoncheck" ndjson "$dir/out" -ok "ndjson: assistant/tool/result/usage/end taxonomy is stable" $? +ok "ndjson: taxonomy and aggregate 8/6 usage are stable" $? stop_stub # Print mode uses the slash-command and Skill dispatcher. blob - c878964fdd6faef25d24fdb12904056668e8e318 blob + a43f2699f1adfc209698717aa02f85701a373fab --- regress/turn/run.sh +++ regress/turn/run.sh @@ -191,6 +191,135 @@ grep -q '"outcome":"abandoned"' "$sessions"/*.ndjson 2 ok "empty reply: journal records the abandoned turn" $? stop_stub +# --- 5a: canonical Lead Generation stop policies -------------------- +# A partial length-limited reply remains useful final text when it carries no +# tool call. This is a successful, replayable Turn rather than an empty reply. +rm -rf "$HOME/.fugu" +start_stub "LIMITTEXT" +write_conf "$stubport" +run_fugu -p "return as much as fits" +ok "lead length text: exit 0" $([ "$rc" -eq 0 ]; echo $?) +grep -qx 'LIMIT-TEXT-PARTIAL' "$dir/out" +ok "lead length text: partial reply is the final answer once" $? +journal=$(ls "$sessions"/*.ndjson 2>/dev/null | head -1) +textn=$(grep -c 'LIMIT-TEXT-PARTIAL' "$journal" 2>/dev/null) +assistantn=$(grep -c '"role":"assistant"' "$journal" 2>/dev/null) +okn=$(grep -c '"outcome":"ok"' "$journal" 2>/dev/null) +ok "lead length text: exactly one assistant and outcome commit" \ + $([ "$textn" -eq 1 ] && [ "$assistantn" -eq 1 ] && \ + [ "$okn" -eq 1 ]; echo $?) +stop_stub + +# A length stop does not make a complete tool_use dangling. The Lead executes +# it, pairs its result, and continues to the next Generation. +rm -rf "$HOME/.fugu" +start_stub "LIMITCALL OK" +write_conf "$stubport" +run_fugu -p "use the requested tool and continue" +ok "lead length tool: exit 0 after continuation" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q 'Hello, world.' "$dir/out" +ok "lead length tool: following Generation supplies the final answer" $? +grep -q 'LIMIT-CALL-PARTIAL' "$dir/reqout" && \ + grep -q '"tool_use_id":"toolu_dangle"' "$dir/reqout" +ok "lead length tool: canonical call and result reach the continuation" $? +journal=$(ls "$sessions"/*.ndjson 2>/dev/null | head -1) +assistantn=$(grep -c '"role":"assistant"' "$journal" 2>/dev/null) +usen=$(grep -c '"bt":"tool_use"' "$journal" 2>/dev/null) +resultn=$(grep -c '"bt":"tool_result"' "$journal" 2>/dev/null) +okn=$(grep -c '"outcome":"ok"' "$journal" 2>/dev/null) +ok "lead length tool: exact assistant pair and outcome commit" \ + $([ "$assistantn" -eq 2 ] && [ "$usen" -eq 1 ] && \ + [ "$resultn" -eq 1 ] && [ "$okn" -eq 1 ]; echo $?) +stop_stub + +# The Generation budget is the Turn's remaining budget, not a fresh 4 MiB. +# Leave eight bytes after a large piped prompt, then send an 18-byte delta. +rm -rf "$HOME/.fugu" +dd if=/dev/zero bs=4194275 count=1 2>/dev/null | \ + tr '\000' a > "$dir/nearbound" +start_stub "LIMITTEXT" +write_conf "$stubport" +run_fugu_in "$dir/nearbound" -p "budget gate" +ok "lead remaining budget: overrun exits 1" \ + $([ "$rc" -eq 1 ]; echo $?) +grep -q 'turn data exceeds 4 MiB' "$dir/errout" +ok "lead remaining budget: overrun is conspicuous" $? +grep -q 'LIMIT-TEXT-PARTIAL' "$sessions"/*.ndjson 2>/dev/null +accepted=$? +n=$(grep -c '"outcome":"abandoned"' "$sessions"/*.ndjson 2>/dev/null) +ok "lead remaining budget: partial assistant is abandoned" \ + $([ "$accepted" -ne 0 ] && [ "$n" -eq 1 ]; echo $?) +stop_stub + +# A length stop with no wire content receives the canonical placeholder. It +# crosses each public presentation/persistence seam exactly once. +rm -rf "$HOME/.fugu" +start_stub "LIMITEMPTY" +write_conf "$stubport" +run_fugu -p "do not exceed the generation limit" +ok "lead empty length: exit 0" $([ "$rc" -eq 0 ]; echo $?) +grep -qx 'Generation stopped before emitting content.' "$dir/out" +ok "lead empty length: synthesized answer is printed exactly once" $? +n=$(grep -c 'Generation stopped before emitting content.' \ + "$sessions"/*.ndjson 2>/dev/null) +ok "lead empty length: synthesized answer is journaled exactly once" \ + $([ "$n" -eq 1 ]; echo $?) +grep -q '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null +ok "lead empty length: synthesized answer commits" $? +stop_stub + +# JSON returns the same synthesized assistant and the actual Provider stop, +# once, in its one final object. +rm -rf "$HOME/.fugu" +start_stub "LIMITEMPTY" +write_conf "$stubport" +run_fugu -p "return a bounded JSON answer" -o json +ok "lead empty length JSON: exit 0" $([ "$rc" -eq 0 ]; echo $?) +n=$(wc -l < "$dir/out") +ok "lead empty length JSON: exactly one final object" \ + $([ "$n" -eq 1 ]; echo $?) +grep -q '"assistant_text":"Generation stopped before emitting content\."' \ + "$dir/out" && grep -q '"stop_reason":"max_tokens"' "$dir/out" +ok "lead empty length JSON: synthesized assistant and stop are exact" $? +stop_stub + +# NDJSON emits one accepted assistant event and one terminal event; usage +# events may surround them but cannot duplicate the synthesized content. +rm -rf "$HOME/.fugu" +start_stub "LIMITEMPTY" +write_conf "$stubport" +run_fugu -p "return a bounded event stream" -o ndjson +ok "lead empty length NDJSON: exit 0" $([ "$rc" -eq 0 ]; echo $?) +n=$(grep -c '"type":"assistant_text"' "$dir/out") +ok "lead empty length NDJSON: exactly one assistant event" \ + $([ "$n" -eq 1 ]; echo $?) +n=$(grep -c '"type":"turn_end"' "$dir/out") +ok "lead empty length NDJSON: exactly one terminal event" \ + $([ "$n" -eq 1 ]; echo $?) +grep -q '"type":"assistant_text","text":"Generation stopped before emitting content\."' \ + "$dir/out" && grep -q '"type":"turn_end".*"stop_reason":"max_tokens"' \ + "$dir/out" +ok "lead empty length NDJSON: synthesized assistant and stop are exact" $? +stop_stub + +# The Lead and Subagent Adapters share canonical invalid-Generation policy: +# complete non-object arguments fail before execution or Journal acceptance. +rm -rf "$HOME/.fugu" +start_stub "BADARGS" +write_conf "$stubport" +run_fugu -p "reject malformed tool arguments" +ok "lead malformed arguments: exit 1" $([ "$rc" -eq 1 ]; echo $?) +grep -q 'malformed tool arguments' "$dir/errout" +ok "lead malformed arguments: canonical reason is visible" $? +grep -q '"bt":"tool_use"' "$sessions"/*.ndjson 2>/dev/null +ok "lead malformed arguments: invalid assistant is never journaled" \ + $([ $? -ne 0 ]; echo $?) +grep -q '^fugu: read$' "$dir/errout" 2>/dev/null +ok "lead malformed arguments: invalid call is never executed" \ + $([ $? -ne 0 ]; 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. @@ -628,6 +757,43 @@ ok "failed compact: no compact event was recorded" \ ok "failed compact: Projection and cursor survive unchanged" $? stop_stub +# Compaction accepts only a complete normal text Generation. A partial length +# stop must preserve both the Projection and its acknowledged cache cursor. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqout" +start_stub "OK LIMITTEXT OK" +write_conf "$stubport" +printf 'first question\n/compact\nsecond question\n/quit\n' > "$dir/lin" +run_fugu_in "$dir/lin" +ok "length compact: line session continues" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q '"t":"compact"' "$sessions"/*.ndjson 2>/dev/null +ok "length compact: no compact event was recorded" \ + $([ $? -ne 0 ]; echo $?) +"$obj/reqcheck" "$dir/reqout" +ok "length compact: Projection and cursor survive unchanged" $? +stop_stub + +# A hostile Provider may emit a tool despite the tool-free request. Even with +# partial text it is not a valid summary, is not executed, and changes nothing. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqout" +start_stub "OK COMPACTTOOL OK" +write_conf "$stubport" +printf 'first question\n/compact\nsecond question\n/quit\n' > "$dir/lin" +run_fugu_in "$dir/lin" +ok "tool compact: line session continues" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q '"t":"compact"' "$sessions"/*.ndjson 2>/dev/null +ok "tool compact: no compact event was recorded" \ + $([ $? -ne 0 ]; echo $?) +"$obj/reqcheck" "$dir/reqout" +ok "tool compact: Projection and cursor survive unchanged" $? +grep -q '^fugu: read$' "$dir/errout" 2>/dev/null +ok "tool compact: hostile tool call is never executed" \ + $([ $? -ne 0 ]; echo $?) +stop_stub + # --- 26: /context accounts for the compaction summary (behavior.md 3) rm -rf "$HOME/.fugu" start_stub "OK SUMMARY" blob - 66ec0c8fa2bdac78fb1dc2dd24b423fb2e70e562 blob + 01d02235089b7d2e6ffede3f2830fb03d409ec19 --- regress/turn/stub.c +++ regress/turn/stub.c @@ -37,8 +37,9 @@ * (two calls queued through a one-slot wave), AGENTROUND/AGENTCONCLUDE/ * BOUNDSYNTH (the 25-round bound and tool-free conclusion), and AGENTHANG * (cancellation of an active local child). OVERLAPA/OVERLAPB hold two - * requests in flight and complete them in reverse order. LIMITTEXT/LIMITCALL - * exercise token stops; RETAINLIMIT/RETAINCONCLUDE and RETAINTOOL/ERROR + * requests in flight and complete them in reverse order. LIMITTEXT/LIMITCALL/ + * LIMITEMPTY exercise token stops; COMPACTTOOL emits normal text plus a + * hostile tool call. RETAINLIMIT/RETAINCONCLUDE and RETAINTOOL/ERROR * preserve a capped report's conclusion and terminal diagnostic tails; * ARGBOUND/RESULTBOUND exercise child data accounting; * WEBPROVA/WEBPROVB/WEBREPORT prove concurrent children are serialized through @@ -65,7 +66,7 @@ #include #define STUB_TURN_DATA_MAX (4U * 1024 * 1024) -#define STUB_REQ_MAX (2U * 1024 * 1024) +#define STUB_REQ_MAX (8U * 1024 * 1024) #define STUB_RETAIN_TEXT (1536U * 1024) static const char body_ok[] = @@ -474,7 +475,7 @@ send_poolcall(struct tls *conn) } static void -send_limit(struct tls *conn, int with_tool) +send_limit(struct tls *conn, int with_tool, const char *stop) { char sse[4096]; int n; @@ -509,7 +510,7 @@ send_limit(struct tls *conn, int with_tool) "\"partial_json\":\"{\\\"path\\\":\\\"never-read\\\"}\"}}\n\n" "event: content_block_stop\n" "data: {\"type\":\"content_block_stop\",\"index\":1}\n\n" : "", - with_tool ? "length" : "max_tokens"); + stop); if (n < 0 || (size_t)n >= sizeof(sse)) errx(1, "limit fixture too large"); (void)tls_write_all(conn, body_ok, strlen(body_ok)); @@ -912,6 +913,17 @@ static const char sse_empty[] = "event: message_stop\n" "data: {\"type\":\"message_stop\"}\n\n"; +/* A length stop before the Provider emits either text or a tool call. */ +static const char sse_limit_empty[] = +"event: message_start\n" +"data: {\"type\":\"message_start\",\"message\":{\"usage\":" + "{\"input_tokens\":11,\"output_tokens\":0}}}\n\n" +"event: message_delta\n" +"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":" + "\"max_tokens\"},\"usage\":{\"output_tokens\":0}}\n\n" +"event: message_stop\n" +"data: {\"type\":\"message_stop\"}\n\n"; + /* A complete tool call whose argument bytes are not a JSON object. */ static const char sse_badargs[] = "event: message_start\n" @@ -1617,9 +1629,11 @@ main(int argc, char *argv[]) } else if (strcmp(argv[i], "POOLSYNTH") == 0) { send_text(conn, "POOL-SYNTHESIS-MARKER"); } else if (strcmp(argv[i], "LIMITTEXT") == 0) { - send_limit(conn, 0); + send_limit(conn, 0, "max_tokens"); } else if (strcmp(argv[i], "LIMITCALL") == 0) { - send_limit(conn, 1); + send_limit(conn, 1, "length"); + } else if (strcmp(argv[i], "COMPACTTOOL") == 0) { + send_limit(conn, 1, "tool_use"); } else if (strcmp(argv[i], "LIMITCONCLUDE") == 0) { send_text(conn, "LIMIT-CONCLUSION-TEXT"); } else if (strcmp(argv[i], "LIMITSYNTH") == 0) { @@ -1693,6 +1707,12 @@ main(int argc, char *argv[]) write_chunk(conn, sse_empty, strlen(sse_empty)); (void)tls_write_all(conn, "0\r\n\r\n", 5); + } else if (strcmp(argv[i], "LIMITEMPTY") == 0) { + (void)tls_write_all(conn, body_ok, + strlen(body_ok)); + write_chunk(conn, sse_limit_empty, + strlen(sse_limit_empty)); + (void)tls_write_all(conn, "0\r\n\r\n", 5); } else if (strcmp(argv[i], "BADARGS") == 0) { (void)tls_write_all(conn, body_ok, strlen(body_ok)); blob - ed92b0e4be6d6b4f5aa68f39e28f6d85e962b64b blob + bc10088984800846b201b22dafd89a6d3331ad87 --- src/fugu/coord.c +++ src/fugu/coord.c @@ -37,6 +37,7 @@ #include "log.h" #include "xmalloc.h" #include "buf.h" +#include "generation.h" #include "json.h" #include "model_window.h" #include "msg.h" @@ -56,7 +57,7 @@ #include "turn_txn_private.h" #define TURN_ITER_MAX 50 /* behavior.md section 5 */ -#define MAX_TOOL_CALLS 64 /* tool_use blocks per assistant msg */ +#define MAX_TOOL_CALLS GENERATION_CALL_MAX #define MODELS_RELAY_MAX 4096 /* picker entries per listing (I10) */ #define AGENT_WEB_MEMORY_MAX (64 * 1024 * 1024) #define CAPTURE_TEXT_MAX (8 * 1024) /* one bounded CTX_APPEND frame */ @@ -163,7 +164,6 @@ struct coord { struct a_usage turn_usage; /* all provider calls this turn */ int64_t turn_tool_calls; size_t turn_data; /* hostile text/args/results this turn */ - size_t response_tool_args; int turn_data_exceeded; struct buf input; /* a submitted line from the tty */ int have_input; /* an unconsumed UI_SUBMIT */ @@ -203,15 +203,18 @@ struct coord { uint32_t web_busy_id; size_t agent_web_memory; - /* per-turn stream accumulation */ + /* one canonical Lead Generation and its presentation/execution adapters */ + struct generation *generation; + int generation_terminal; + int generation_present; + + /* accepted Generation views used by the existing presentation/tool seams */ struct buf text; struct a_usage usage; struct tool_call tcalls[MAX_TOOL_CALLS]; int ntcalls; char stop_reason[FUGU_STOP_REASON_MAX]; enum coord_wait wait; - int done; - int failed; char err[FUGU_ERROR_MAX]; /* one tool call's result, filled while WAIT_TOOL or WAIT_WEB */ @@ -817,18 +820,6 @@ send_hello(struct coord *c, enum fugu_role role) } } -/* Find (or open) the tool_call being assembled at a stream index. */ -static struct tool_call * -tcall_at(struct coord *c, int index) -{ - int i; - - for (i = 0; i < c->ntcalls; i++) - if (c->tcalls[i].index == index) - return (&c->tcalls[i]); - return (NULL); -} - /* * Output sink (behavior.md 2.1, 2.2). In line/print mode the coordinator * writes stdout/stderr directly; in the curses interface it streams the @@ -1084,15 +1075,6 @@ usage_account(struct coord *c, const struct a_usage *u return (0); } -static void -stream_turn_fail(struct coord *c, const char *reason) -{ - c->failed = 1; - strlcpy(c->err, reason, sizeof(c->err)); - c->wait = WAIT_NONE; - event_loopexit(NULL); -} - /* * One typed configuration transaction: fixed settings, bounded Skill * names, then the terminal frame that makes the snapshot visible. Keeping @@ -1135,10 +1117,16 @@ send_ui_config(struct coord *c) static void stream_event(struct coord *c, struct imsg *imsg) { - struct imsgev_payload payload; - const u_char *p; - size_t len; - uint32_t type = imsg_get_type(imsg); + struct a_usage au; + struct generation_event ev; + struct generation_usage gu; + struct imsgev_payload payload; + const void *feed_data; + const u_char *p; + size_t feed_len, len; + uint32_t type = imsg_get_type(imsg); + enum generation_feed_status status; + enum generation_kind kind; /* * Mutual distrust (I10): stream events count only while pumping @@ -1151,6 +1139,8 @@ 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) + fatalx("lead provider event without a Generation"); if (imsgev_get_payload(imsg, &payload) == -1) fatal("get provider stream payload"); @@ -1159,147 +1149,89 @@ stream_event(struct coord *c, struct imsg *imsg) switch (type) { case FUGU_IMSG_A_TEXT: - if (p != NULL) { - if (c->turn_data_exceeded) - break; - if (c->text.len > FUGU_RESPONSE_MAX || - len > FUGU_RESPONSE_MAX - c->text.len) { - stream_turn_fail(c, "provider response exceeds 16 MiB"); - break; - } - if (len > FUGU_TURN_DATA_MAX - c->turn_data) { - /* Keep draining this API response to its terminal so - * uncorrelated frames cannot bleed into a later Turn. */ - c->turn_data_exceeded = 1; - break; - } - buf_add(&c->text, p, len); - c->turn_data += len; - emit_delta(c, (const char *)p, len); /* stream (UI) */ - } + kind = GENERATION_TEXT; break; case FUGU_IMSG_A_USAGE: - if (p == NULL || len != sizeof(c->usage)) - fatalx("malformed provider usage"); - else { - struct a_usage u; - - memcpy(&u, p, sizeof(u)); - if (c->turn_data_exceeded) - break; - if (usage_account(c, &u) == -1) { - /* This is a whole-Turn aggregate failure. Drain the - * current uncorrelated response through its terminal. */ - c->turn_data_exceeded = 1; - break; - } - machine_usage(c, &u); - } + kind = GENERATION_USAGE; break; - case FUGU_IMSG_A_TOOL_BEGIN: { - struct a_tool_begin hdr; - struct tool_call *tc; - const char *id, *name; - size_t idlen, namelen; - - /* struct a_tool_begin, then "id\0name" -- name is NOT - * NUL-terminated, so it is copied length-bounded */ - if (p == NULL || len < sizeof(hdr) || - c->ntcalls >= MAX_TOOL_CALLS) - fatalx("malformed or excessive tool call"); - memcpy(&hdr, p, sizeof(hdr)); - id = (const char *)p + sizeof(hdr); - idlen = strnlen(id, len - sizeof(hdr)); - if (idlen == 0 || idlen == len - sizeof(hdr) || - idlen >= sizeof(tc->id)) - fatalx("malformed tool call id"); - name = id + idlen + 1; - namelen = len - sizeof(hdr) - idlen - 1; - if (hdr.index < 0 || tcall_at(c, hdr.index) != NULL || - namelen == 0 || namelen >= sizeof(tc->name) || - memchr(name, '\0', namelen) != NULL) - fatalx("malformed tool call name or index"); - - tc = &c->tcalls[c->ntcalls++]; - tc->index = hdr.index; - memcpy(tc->id, id, idlen + 1); - memcpy(tc->name, name, namelen); - tc->name[namelen] = '\0'; - buf_init(&tc->input); - buf_init(&tc->result); - tc->result_ready = 0; - tc->result_error = 0; + case FUGU_IMSG_A_TOOL_BEGIN: + kind = GENERATION_TOOL_BEGIN; break; - } - case FUGU_IMSG_A_TOOL_INPUT: { - struct tool_call *tc; - int idx; - - /* int index prefix, then partial input JSON bytes */ - if (c->turn_data_exceeded) - break; - if (p == NULL || len < sizeof(idx)) - fatalx("malformed tool input"); - memcpy(&idx, p, sizeof(idx)); - if ((tc = tcall_at(c, idx)) == NULL || - len - sizeof(idx) > FUGU_TOOL_ARG_MAX - tc->input.len || - len - sizeof(idx) > - FUGU_TOOL_ARGS_TOTAL_MAX - c->response_tool_args) { - stream_turn_fail(c, "excessive provider tool arguments"); - break; - } - if (len - sizeof(idx) > FUGU_TURN_DATA_MAX - c->turn_data) { - c->turn_data_exceeded = 1; - break; - } - buf_add(&tc->input, p + sizeof(idx), len - sizeof(idx)); - c->response_tool_args += len - sizeof(idx); - c->turn_data += len - sizeof(idx); + case FUGU_IMSG_A_TOOL_INPUT: + kind = GENERATION_TOOL_INPUT; break; - } case FUGU_IMSG_A_RETRY: + kind = GENERATION_RETRY; + break; + case FUGU_IMSG_A_DONE: + kind = GENERATION_DONE; + break; + case FUGU_IMSG_A_ERROR: + kind = GENERATION_ERROR; + break; + default: + fatalx("unexpected lead provider imsg %u", type); + } + + feed_data = p; + feed_len = len; + if (kind == GENERATION_USAGE && p != NULL && len == sizeof(au)) { + memcpy(&au, p, sizeof(au)); + memset(&gu, 0, sizeof(gu)); + gu.input_tokens = au.input_tokens; + gu.output_tokens = au.output_tokens; + gu.cache_read = au.cache_read; + gu.cache_write = au.cache_write; + feed_data = &gu; + feed_len = sizeof(gu); + } + status = generation_feed(c->generation, kind, feed_data, feed_len, + &ev); + if (status == GENERATION_PROTOCOL) { + const char *why = generation_error(c->generation); + + fatalx("lead provider protocol: %s", + why != NULL ? why : "invalid Generation event"); + } + if (status == GENERATION_MORE && kind == GENERATION_TEXT) { + if (c->generation_present && !c->turn_data_exceeded) + emit_delta(c, ev.payload.data, ev.payload.len); + } else if (status == GENERATION_MORE && kind == GENERATION_USAGE) { + memset(&au, 0, sizeof(au)); + au.input_tokens = ev.usage.input_tokens; + 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; + else + machine_usage(c, &au); + } + } else if (status == GENERATION_MORE && kind == GENERATION_RETRY) { /* one retry notice per attempt (behavior.md 8); the turn is - * still in flight. In UI mode it also drives the status bar - * into its retry state. */ - if (p != NULL && len > 0) { - if (c->ui) { - struct ui_status s; + * still in flight. In UI mode it also drives the status bar. */ + if (c->ui) { + struct ui_status s; - ui_send(c, FUGU_IMSG_UI_NOTE, p, len); - memset(&s, 0, sizeof(s)); - s.state = UI_RETRY; - s.have_usage = 1; - s.usage = c->usage; - ui_send(c, FUGU_IMSG_UI_STATUS, &s, sizeof(s)); - } else - fprintf(stderr, "fugu: %.*s\n", (int)len, p); - } - break; - case FUGU_IMSG_A_DONE: - if (p == NULL || len == 0 || len >= sizeof(c->stop_reason) || - memchr(p, '\0', len) != NULL) { - stream_turn_fail(c, "malformed provider stop reason"); - break; - } - c->done = 1; - c->wait = WAIT_NONE; - memcpy(c->stop_reason, p, len); - c->stop_reason[len] = '\0'; - event_loopexit(NULL); - break; - case FUGU_IMSG_A_ERROR: - if (p == NULL || len == 0 || len >= sizeof(c->err) || - memchr(p, '\0', len) != NULL) { - stream_turn_fail(c, "malformed provider error"); - break; - } - c->failed = 1; - c->wait = WAIT_NONE; - memcpy(c->err, p, len); - c->err[len] = '\0'; - event_loopexit(NULL); - break; + ui_send(c, FUGU_IMSG_UI_NOTE, ev.payload.data, + ev.payload.len); + memset(&s, 0, sizeof(s)); + s.state = UI_RETRY; + s.have_usage = 1; + s.usage = c->usage; + ui_send(c, FUGU_IMSG_UI_STATUS, &s, sizeof(s)); + } else + fprintf(stderr, "fugu: %.*s\n", (int)ev.payload.len, + (const char *)ev.payload.data); } + if (status == GENERATION_MORE || status == GENERATION_DRAIN) + return; + if (status != GENERATION_TERMINAL) + fatalx("lead Generation returned an unknown feed status"); + c->generation_terminal = 1; + c->wait = WAIT_NONE; + event_loopexit(NULL); } /* Results streamed back from fugu-tool while wait == WAIT_TOOL. */ @@ -2663,39 +2595,129 @@ send_turn_request(struct coord *c) return (0); } -/* Pump the API stream to its terminal, resetting the accumulators. */ static void -pump_stream(struct coord *c) +generation_result_dispose(struct generation_result *result) { + if (result->assistant != NULL) + msg_free(result->assistant); + memset(result, 0, sizeof(*result)); +} + +/* + * Collect one Lead response through the canonical Generation module. This + * Adapter observes accepted usage and presentation events, but the module is + * the sole owner of text, tool assembly, bounds, and terminal policy. + */ +static int +pump_generation(struct coord *c, size_t data_available, int present, + struct generation_result *result) +{ + const char *reason; + + if (c->generation != NULL) + fatalx("nested Lead Generation"); + memset(result, 0, sizeof(*result)); reset_tcalls(c); buf_reset(&c->text); memset(&c->usage, 0, sizeof(c->usage)); - c->response_tool_args = 0; c->stop_reason[0] = '\0'; - c->done = c->failed = 0; c->err[0] = '\0'; + c->generation = generation_new(data_available); + c->generation_terminal = 0; + c->generation_present = present; c->wait = WAIT_STREAM; - event_dispatch(); + if (event_dispatch() == -1) + fatal("event_dispatch Lead Generation"); c->wait = WAIT_NONE; + c->generation_present = 0; + + if (!c->generation_terminal) { + generation_free(c->generation); + c->generation = NULL; + return (-1); + } + if (generation_take(c->generation, result) == -1) + fatalx("could not take terminal Lead Generation"); + if (result->data_used > data_available) + fatalx("Lead Generation violated its data budget"); + reason = result->reason; + if (result->outcome == GENERATION_NORMAL || + result->outcome == GENERATION_LENGTH) + strlcpy(c->stop_reason, reason, sizeof(c->stop_reason)); + else + strlcpy(c->err, reason, sizeof(c->err)); + result->reason = NULL; + generation_free(c->generation); + c->generation = NULL; + return (0); } -/* Assemble the streamed assistant message: text then tool_use blocks. */ -static struct msg * -build_assistant_msg(struct coord *c) +/* Borrow the canonical assistant text for presentation and compaction. */ +static struct generation_span +lead_assistant_text(const struct msg *assistant) { - struct msg *am; - int i; + struct generation_span text; + struct block *b; + int found = 0; - am = msg_new(ROLE_ASSISTANT); - if (c->text.len > 0) - msg_add_text(am, (const char *)c->text.data, c->text.len); - for (i = 0; i < c->ntcalls; i++) { - struct tool_call *tc = &c->tcalls[i]; + memset(&text, 0, sizeof(text)); + if (assistant == NULL || assistant->role != ROLE_ASSISTANT) + fatalx("invalid canonical Lead assistant"); + TAILQ_FOREACH(b, &assistant->blocks, entry) { + if (b->type == BLOCK_TEXT) { + if (found) + fatalx("canonical Lead assistant has duplicate text"); + text.data = b->text; + text.len = b->textlen; + found = 1; + } else if (b->type != BLOCK_TOOL_USE) + fatalx("canonical Lead assistant has an invalid block"); + } + return (text); +} - msg_add_tool_use(am, tc->id, tc->name, - (const char *)tc->input.data, tc->input.len); +/* Copy accepted canonical views into the existing presentation/tool seams. */ +static void +lead_generation_adapt(struct coord *c, + const struct generation_result *result) +{ + struct generation_span text; + struct tool_call *tc; + struct block *b; + + if (result->assistant == NULL || + result->tool_calls > MAX_TOOL_CALLS) + fatalx("invalid successful Lead Generation"); + text = lead_assistant_text(result->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) { + if (b->type != BLOCK_TOOL_USE) + continue; + if (c->ntcalls >= MAX_TOOL_CALLS || b->tool_id == NULL || + b->tool_name == NULL || + strlcpy(c->tcalls[c->ntcalls].id, b->tool_id, + sizeof(c->tcalls[c->ntcalls].id)) >= + sizeof(c->tcalls[c->ntcalls].id) || + strlcpy(c->tcalls[c->ntcalls].name, b->tool_name, + sizeof(c->tcalls[c->ntcalls].name)) >= + sizeof(c->tcalls[c->ntcalls].name)) + fatalx("invalid canonical Lead tool call"); + tc = &c->tcalls[c->ntcalls]; + tc->index = c->ntcalls; + buf_init(&tc->input); + buf_init(&tc->result); + if (b->tool_input_len > 0) + buf_add(&tc->input, b->tool_input, b->tool_input_len); + tc->result_ready = 0; + tc->result_error = 0; + c->ntcalls++; } - return (am); + if ((size_t)c->ntcalls != result->tool_calls) + fatalx("canonical Lead tool-call count mismatch"); } /* @@ -3233,8 +3255,10 @@ fail_turn(struct coord *c, int status, const char *mes static int run_turn(struct coord *c, const char *prompt, size_t promptlen) { - struct msg *um, *am; - int iter; + struct generation_result generation; + struct msg *um; + size_t available; + int iter, pump_rc; if (promptlen > FUGU_TURN_DATA_MAX) { emit_error(c, "fugu: prompt exceeds the 4 MiB turn-data bound"); @@ -3266,87 +3290,77 @@ run_turn(struct coord *c, const char *prompt, size_t p return (fail_turn(c, 1, "request serialization failed")); } - pump_stream(c); + 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)) + if (cancel_checkpoint(c)) { + generation_result_dispose(&generation); return (cancel_turn(c)); - if (c->turn_data_exceeded) { - int status = abandon_turn(c, 1); - - emit_error(c, "fugu: turn data exceeds 4 MiB"); - machine_error(c, "turn data exceeds 4 MiB"); - machine_finish(c, 1); - return (status); } /* 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); return (fail_turn(c, 1, "turn abandoned by session switch")); } - if (c->failed) { + if (pump_rc == -1) { + generation_result_dispose(&generation); + emit_error(c, + "fugu: provider stream ended before its terminal"); + 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); emit_error(c, "fugu: %s", why); machine_error(c, why); machine_finish(c, 1); return (status); } - /* neither text nor a tool call is empty content the - * provider would reject on the next request (I14) */ - if (c->text.len == 0 && c->ntcalls == 0) { - int status = abandon_turn(c, 1); - - emit_error(c, "fugu: provider sent an empty reply"); - machine_error(c, "provider sent an empty reply"); - machine_finish(c, 1); - return (status); - } - /* Every provider dialect must yield one complete JSON object per - * tool call before those bytes reach the Journal or an executor. - * Anthropic represents a no-argument streaming call with no - * partial_json deltas, which canonically means {}. */ - { - int i, bad = 0; - - for (i = 0; i < c->ntcalls; i++) { - if (c->tcalls[i].input.len == 0) { - if (2 > FUGU_TURN_DATA_MAX - c->turn_data) { - bad = 1; - break; - } - buf_addstr(&c->tcalls[i].input, "{}"); - c->turn_data += 2; - } - if (!json_valid_object(c->tcalls[i].input.data, - c->tcalls[i].input.len)) { - bad = 1; - break; - } - } - if (bad) { - int status = abandon_turn(c, 1); - - emit_error(c, - "fugu: provider sent malformed tool arguments"); - machine_error(c, - "provider sent malformed tool arguments"); - machine_finish(c, 1); - return (status); - } - } - if (cancel_checkpoint(c)) + 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)); + } - am = build_assistant_msg(c); - turn_txn_accept(c->turn, am); + 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)) + if (cancel_checkpoint(c)) { + generation_result_dispose(&generation); return (cancel_turn(c)); + } /* * Any tool_use blocks MUST be answered with tool_result on @@ -3357,6 +3371,7 @@ run_turn(struct coord *c, const char *prompt, size_t p */ 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); @@ -3371,6 +3386,7 @@ run_turn(struct coord *c, const char *prompt, size_t p 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 @@ -3449,9 +3465,13 @@ cmd_clear(struct coord *c) static int cmd_compact(struct coord *c) { + struct generation_result generation; + struct generation_span text; struct turn_txn_projection projection; struct msg *instr; struct buf body; + const char *why; + int pump_rc; turn_txn_projection(c->turn, &projection); if (projection.message_count == 0) { @@ -3478,17 +3498,44 @@ cmd_compact(struct coord *c) buf_free(&body); c->turn_data = 0; c->turn_data_exceeded = 0; - pump_stream(c); + pump_rc = pump_generation(c, FUGU_TURN_DATA_MAX, 1, &generation); - if (cancel_checkpoint(c)) + if (cancel_checkpoint(c)) { + generation_result_dispose(&generation); return (3); - if (c->failed || c->turn_data_exceeded || c->text.len == 0) { - emit_error(c, "fugu: compaction failed: %s", - c->err[0] != '\0' ? c->err : "no summary produced"); + } + if (pump_rc == -1) { + generation_result_dispose(&generation); + emit_error(c, "fugu: compaction failed: provider stream ended " + "before its terminal"); return (1); } - if (cancel_checkpoint(c)) + c->turn_data += generation.data_used; + if (generation.outcome == GENERATION_INVALID && + generation.invalid_cause == GENERATION_INVALID_DATA_BOUND) + c->turn_data_exceeded = 1; + memset(&text, 0, sizeof(text)); + if (generation.outcome == GENERATION_NORMAL && + generation.invalid_cause == GENERATION_INVALID_NONE && + generation.assistant != NULL && generation.tool_calls == 0) + text = lead_assistant_text(generation.assistant); + if (c->turn_data_exceeded || generation.outcome != GENERATION_NORMAL || + generation.invalid_cause != GENERATION_INVALID_NONE || + generation.assistant == NULL || generation.tool_calls != 0 || + text.len == 0) { + why = c->err[0] != '\0' ? c->err : + "summary was incomplete or contained tool calls"; + generation_result_dispose(&generation); + emit_error(c, "fugu: compaction failed: %s", why); + return (1); + } + if (cancel_checkpoint(c)) { + generation_result_dispose(&generation); return (3); + } + buf_reset(&c->text); + buf_add(&c->text, text.data, text.len); + generation_result_dispose(&generation); /* The durable summary replaces the Projection and becomes context. */ turn_txn_compact(c->turn, c->text.data, c->text.len);