commit - 11cb7bb8d4c70c4973e2647be12cfd08cadbc8df
commit + ffccea59b4d4877ec18fbcbac7f41c955416b635
blob - 8d47d8a44543aa47353f9c8c12bcf6c1c688ded6
blob + d0b7d44f39adbf0768582fe28b838c4a6bf5b273
--- regress/print/jsoncheck.c
+++ regress/print/jsoncheck.c
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;
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 ||
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)
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
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
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.
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
* (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
#include <unistd.h>
#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[] =
}
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;
"\"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));
"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"
} 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) {
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
#include "log.h"
#include "xmalloc.h"
#include "buf.h"
+#include "generation.h"
#include "json.h"
#include "model_window.h"
#include "msg.h"
#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 */
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 */
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 */
}
}
-/* 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
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
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
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");
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. */
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");
}
/*
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");
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
*/
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);
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
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) {
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);
