commit - 5ffcba5dccaf9389ab199d516bc4b7d9d03cfd1a
commit + 5420dda02627f052ebf902f19bf168af2e27547b
blob - 63a17da473cac7714cc5324c24c36a51637ea12b
blob + 1ce70ab953bb2f623c36c6b3c7be8703268a1997
--- regress/turn/agent.sh
+++ regress/turn/agent.sh
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 </dev/null ) \
+ >"$dir/out" 2>"$dir/errout"
+ rc=$?
+}
+
request_section() {
awk -v name="$1" '
$0 == "--- " name " ---" { found = 1; next }
$([ "$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
$([ $? -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
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
#include "skills.h"
#include "tooldefs.h"
#include "agentcfg.h"
+#include "turn_mechanics.h"
#include "turn_txn.h"
#include "turn_txn_private.h"
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 */
/* 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;
(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)
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 *
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)
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");
}
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);
}
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;
}
}
-/*
- * 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
/* 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 ||
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
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)));
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;
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
{
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);
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");
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 */
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)) {
