commit 5ffcba5dccaf9389ab199d516bc4b7d9d03cfd1a from: Isaac Meerleo date: Wed Jul 15 18:38:17 2026 UTC integrate canonical turn mechanics into subagents commit - f82b590e4d17bcf395c594962292546a67ee7c8a commit + 5ffcba5dccaf9389ab199d516bc4b7d9d03cfd1a blob - 401a79bf7f548c9d585d90ad873ad32d7b0357fe blob + 63a17da473cac7714cc5324c24c36a51637ea12b --- regress/turn/agent.sh +++ regress/turn/agent.sh @@ -122,6 +122,15 @@ run_fugu() { rc=$? } +run_fugu_ndjson() { + ( 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 ndjson "$dir/out" 2>"$dir/errout" + rc=$? +} + request_section() { awk -v name="$1" ' $0 == "--- " name " ---" { found = 1; next } @@ -486,6 +495,27 @@ ok "malformed arguments: rejected call is never execut $([ $? -ne 0 ]; echo $?) stop_stub +# A malformed nonterminal Generation must not let the child report while its +# brokered Provider request is still in flight. The stub pauses after an API- +# valid duplicate tool id, then sends the terminal that settles correlation. +rm -rf "$HOME/.fugu" +start_stub AGENTCALL PROTODRAIN PROTODRAINSYNTH +write_conf +run_fugu "exercise a nonterminal subagent protocol error" +ok "protocol drain: lead turn survives the delayed terminal" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q 'PROTODRAIN-SYNTHESIS-MARKER' "$dir/out" +ok "protocol drain: lead synthesis runs after Provider settlement" $? +n=$(grep -c '^--- .* ---$' "$dir/reqlog") +ok "protocol drain: exactly one child Generation ran" \ + $([ "$n" -eq 3 ]; echo $?) +request_section PROTODRAINSYNTH "$dir/protodrain-lead" +grep -q 'subagent provider error: malformed tool begin' \ + "$dir/protodrain-lead" && \ + grep -q '"is_error":true' "$dir/protodrain-lead" +ok "protocol drain: canonical protocol error reaches the Lead" $? +stop_stub + # A normal provider stop with neither text nor tools is not a successful # empty report. It is an invalid Generation returned to the Lead as an error. rm -rf "$HOME/.fugu" @@ -518,8 +548,28 @@ grep -q '"id":"toolu_noargs"' "$dir/noargs-followup" & ok "absent arguments: canonical object and result are paired" $? grep -q 'agent\[reader\]: tool ls' "$dir/errout" ok "absent arguments: canonical object is executable" $? +sed -n '/^{/s/^.*,"messages"://p' "$dir/noargs-followup" \ + > "$dir/noargs-messages" +n=$(grep -o '"cache_control":{"type":"ephemeral"}' \ + "$dir/noargs-messages" | wc -l) +grep -q '^\[{"role":"user".*"cache_control":{"type":"ephemeral"}.*},'\ +'{"role":"assistant"' \ + "$dir/noargs-messages" && [ "$n" -eq 1 ] +ok "absent arguments: cache cursor marks only the prior sent tail" $? stop_stub +# Each provider usage event is relayed once. In this three-Generation turn, +# every fixture emits one start and one delta usage event: exactly six total. +rm -rf "$HOME/.fugu" +start_stub AGENTCALL AGENTREPORT AGENTSYNTH +write_conf +run_fugu_ndjson "exercise subagent usage forwarding" +ok "agent usage: NDJSON turn completes" $([ "$rc" -eq 0 ]; echo $?) +n=$(grep -c '"type":"usage"' "$dir/out") +ok "agent usage: provider events are forwarded exactly once" \ + $([ "$n" -eq 6 ]; echo $?) +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" @@ -570,6 +620,10 @@ grep -q '"tool_use_id":"toolu_dangle"' \ grep -q 'truncated and was not executed' "$dir/limit-call-conclusion" && \ grep -q '"is_error":true' "$dir/limit-call-conclusion" ok "length limit: dangling tool receives an error result" $? +roles=$(grep -o '"role":"[^"]*"' "$dir/limit-call-conclusion" | \ + paste -sd, -) +[ "$roles" = '"role":"user","role":"assistant","role":"user"' ] +ok "length limit: conclusion projection has no adjacent user roles" $? sed -n '/^{/s/,"messages":.*//p' "$dir/limit-call-conclusion" \ > "$dir/limit-call-prefix" grep -q '"tools"' "$dir/limit-call-prefix" 2>/dev/null @@ -584,6 +638,40 @@ grep -q 'LIMIT-CALL-PARTIAL' "$dir/limit-call-lead" && ok "length limit: bounded report preserves all available text" $? stop_stub +# A provider failure in the tool-free conclusion keeps the accepted partial +# report, appends a distinct diagnostic tail, and still returns a useful +# truncated result to the Lead. +rm -rf "$HOME/.fugu" +start_stub AGENTCALL LIMITTEXT ERROR LIMITSYNTH +write_conf +run_fugu "exercise a failed subagent conclusion" +ok "conclusion error: lead turn completes" $([ "$rc" -eq 0 ]; echo $?) +n=$(grep -c '^--- .* ---$' "$dir/reqlog") +ok "conclusion error: exactly one conclusion generation ran" \ + $([ "$n" -eq 4 ]; echo $?) +request_section ERROR "$dir/conclusion-error-request" +grep -q 'LIMIT-TEXT-PARTIAL' "$dir/conclusion-error-request" && \ + grep -q 'Conclude now with the most useful concise final report' \ + "$dir/conclusion-error-request" +ok "conclusion error: request retains partial report and guidance" $? +sed -n '/^{/s/,"messages":.*//p' "$dir/conclusion-error-request" \ + > "$dir/conclusion-error-prefix" +grep -q '"tools"' "$dir/conclusion-error-prefix" 2>/dev/null +ok "conclusion error: failed conclusion request offers no tools" \ + $([ $? -ne 0 ]; echo $?) +request_section LIMITSYNTH "$dir/conclusion-error-lead" +grep -q 'LIMIT-TEXT-PARTIAL' "$dir/conclusion-error-lead" +ok "conclusion error: lead receives the partial fallback" $? +grep -q 'subagent conclusion failed: overloaded_error: stub overloaded' \ + "$dir/conclusion-error-lead" +ok "conclusion error: lead receives the distinct diagnostic" $? +grep -q 'subagent report truncated by its execution bound' \ + "$dir/conclusion-error-lead" +ok "conclusion error: lead receives the truncation marker" $? +grep -q 'agent\[reader\]: completed (truncated)' "$dir/errout" +ok "conclusion error: completion progress marks truncation" $? +stop_stub + # Retained interim prose may itself exceed the 1 MiB tool-result boundary. # The bounded projection must keep a useful head and the later, much smaller # diagnostic conclusion rather than trimming the conclusion off the end. blob - 01d02235089b7d2e6ffede3f2830fb03d409ec19 blob + 9155028c505545b78aa5190d9a4026fe25079c59 --- regress/turn/stub.c +++ regress/turn/stub.c @@ -23,7 +23,8 @@ * Anthropic-style JSON error body), RETRY503/RETRY429 (retryable * statuses, 429 carrying Retry-After: 2), EMPTY (a well-formed * stream with no content blocks), BADARGS (a complete but malformed - * subagent tool input), NOARGS (a subagent tool call with absent input), + * subagent tool input), PROTODRAIN (a malformed Generation followed by a + * delayed Provider terminal), NOARGS (a subagent tool call with absent input), * WEBFETCH (a brokered web_fetch), * BGTOOL (shell_bg sleep 300), ERROR_BG (the provider failure after * that launch), JOBOUTPUT (shell_output job 1), DOUBLE (a hostile error event @@ -944,6 +945,43 @@ static const char sse_badargs[] = "event: message_stop\n" "data: {\"type\":\"message_stop\"}\n\n"; +/* The API accepts distinct indexes; Generation rejects the duplicate id. */ +static const char sse_protodrain_bad[] = +"event: message_start\n" +"data: {\"type\":\"message_start\",\"message\":{\"usage\":" + "{\"input_tokens\":11,\"output_tokens\":1}}}\n\n" +"event: content_block_start\n" +"data: {\"type\":\"content_block_start\",\"index\":0," + "\"content_block\":{\"type\":\"tool_use\",\"id\":" + "\"toolu_dup\",\"name\":\"read\"}}\n\n" +"event: content_block_stop\n" +"data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" +"event: content_block_start\n" +"data: {\"type\":\"content_block_start\",\"index\":1," + "\"content_block\":{\"type\":\"tool_use\",\"id\":" + "\"toolu_dup\",\"name\":\"ls\"}}\n\n"; + +static const char sse_protodrain_done[] = +"event: content_block_stop\n" +"data: {\"type\":\"content_block_stop\",\"index\":1}\n\n" +"event: message_delta\n" +"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":" + "\"tool_use\"},\"usage\":{\"output_tokens\":1}}\n\n" +"event: message_stop\n" +"data: {\"type\":\"message_stop\"}\n\n"; + +static void +send_protodrain(struct tls *conn) +{ + (void)tls_write_all(conn, body_ok, strlen(body_ok)); + write_chunk(conn, sse_protodrain_bad, + sizeof(sse_protodrain_bad) - 1); + usleep(250000); + write_chunk(conn, sse_protodrain_done, + sizeof(sse_protodrain_done) - 1); + (void)tls_write_all(conn, "0\r\n\r\n", 5); +} + /* An optional-argument tool call with no input_json_delta at all. */ static const char sse_noargs[] = "event: message_start\n" @@ -1414,6 +1452,8 @@ main(int argc, char *argv[]) (void)tls_write_all(conn, "0\r\n\r\n", 5); } else if (strcmp(argv[i], "BADARGSSYNTH") == 0) { send_text(conn, "BADARGS-SYNTHESIS-MARKER"); + } else if (strcmp(argv[i], "PROTODRAINSYNTH") == 0) { + send_text(conn, "PROTODRAIN-SYNTHESIS-MARKER"); } else if (strcmp(argv[i], "BADARGSUNUSED") == 0) { send_text(conn, "BADARGS-OLD-FOLLOWUP-MARKER"); } else if (strcmp(argv[i], "EMPTYSYNTH") == 0) { @@ -1719,6 +1759,8 @@ main(int argc, char *argv[]) write_chunk(conn, sse_badargs, strlen(sse_badargs)); (void)tls_write_all(conn, "0\r\n\r\n", 5); + } else if (strcmp(argv[i], "PROTODRAIN") == 0) { + send_protodrain(conn); } else if (strcmp(argv[i], "NOARGS") == 0) { (void)tls_write_all(conn, body_ok, strlen(body_ok)); blob - de55bf5a265ce25293d5f70f21fc28a557c18fe3 blob + 94e8cf9c91dc4d6ff55310800b5d65a0d4580b86 --- src/fugu-tool/Makefile +++ src/fugu-tool/Makefile @@ -1,6 +1,7 @@ PROG= fugu-tool SRCS= main.c agent.c tools.c job.c supervisor.c worker.c imsgev.c \ - anthropic_req.c openai_req.c generation.c msg.c json.c buf.c diff.c log.c xmalloc.c + anthropic_req.c openai_req.c turn_mechanics.c generation.c msg.c json.c \ + buf.c diff.c log.c xmalloc.c BINDIR= ${LIBEXECDIR}/fugu MAN= blob - fb7083b3591be50cbf6ff1e8e15d8630a1c70563 blob + b4ca1248a69ef13a77f5368c3257a52e0cf2a6e5 --- src/fugu-tool/agent.c +++ src/fugu-tool/agent.c @@ -27,6 +27,7 @@ #include #include +#include #include #include #include @@ -40,6 +41,7 @@ #include "buf.h" #include "generation.h" #include "msg.h" +#include "turn_mechanics.h" #include "anthropic.h" #include "openai.h" #include "worker.h" @@ -58,11 +60,17 @@ struct agent_state { struct tool_ctx *tools; struct agent_boot *boot; struct msglist conv; - int prev_msgs; + size_t prev_msgs; + uint64_t projection_serial; + uint64_t pending_receipt; + size_t pending_messages; + int pending_cache_tail; + int projection_pending; + enum turn_tools_mode generation_tools; + const struct tool_def *defs; + int ndefs; struct buf report; char err[FUGU_ERROR_MAX]; - size_t data_total; - int data_exceeded; int iterations; int truncated; size_t report_tail; @@ -248,6 +256,8 @@ agent_send_bytes(struct agent_state *a, uint32_t type, static void agent_progress(struct agent_state *, const char *, ...) __attribute__((__format__ (printf, 2, 3))); +static void agent_report_text(struct agent_state *, const struct msg *); +static void agent_report_tail(struct agent_state *); static void agent_progress(struct agent_state *a, const char *fmt, ...) @@ -314,8 +324,15 @@ agent_msg_count(const struct msglist *l) static void agent_build_request(struct agent_state *a, struct buf *body, - const struct tool_def *defs, int ndefs) + const struct turn_generation_request *request) { + const struct tool_def *defs = NULL; + int ndefs = 0; + + if (request->tools == TURN_TOOLS_ENABLED) { + defs = a->defs; + ndefs = a->ndefs; + } if (a->boot->start.provider_type == PROVIDER_OPENAI) { struct openai_req req; @@ -326,7 +343,7 @@ agent_build_request(struct agent_state *a, struct buf req.systemlen = a->boot->system.len; req.tools = defs; req.ntools = ndefs; - req.conv = &a->conv; + req.conv = request->projection.messages; openai_build_request(body, &req); } else { struct anthropic_req req; @@ -338,24 +355,13 @@ agent_build_request(struct agent_state *a, struct buf req.systemlen = a->boot->system.len; req.tools = defs; req.ntools = ndefs; - req.conv = &a->conv; + req.conv = request->projection.messages; req.cache = 1; - req.cache_tail_msg = a->prev_msgs > 0 ? a->prev_msgs - 1 : -1; + req.cache_tail_msg = request->projection.cache_tail_msg; anthropic_build_request(body, &req); } } -static int -agent_account(struct agent_state *a, size_t len) -{ - if (len > FUGU_TURN_DATA_MAX - a->data_total) { - a->data_exceeded = 1; - return (-1); - } - a->data_total += len; - return (0); -} - static enum generation_kind agent_generation_kind(uint32_t type) { @@ -408,14 +414,88 @@ agent_invalid_error(struct agent_state *a, strlcpy(a->err, s, sizeof(a->err)); } -/* Send one projection and collect its already-decoded provider events. */ -static int -agent_generation(struct agent_state *a, const struct tool_def *defs, - int ndefs, struct generation_result *result) +static void +agent_turn_error(struct agent_state *a, enum turn_failure failure, + enum generation_invalid_cause cause, const char *error) { + if (failure == TURN_FAILURE_GENERATION && + cause != GENERATION_INVALID_NONE) { + agent_invalid_error(a, cause); + return; + } + switch (failure) { + case TURN_FAILURE_DATA_BOUND: + strlcpy(a->err, "subagent data bound exceeded", sizeof(a->err)); + return; + case TURN_FAILURE_USAGE_BOUND: + strlcpy(a->err, "subagent usage bound exceeded", sizeof(a->err)); + return; + case TURN_FAILURE_NONE: + strlcpy(a->err, "subagent interrupted", sizeof(a->err)); + return; + case TURN_FAILURE_ADAPTER: + case TURN_FAILURE_PROVIDER: + case TURN_FAILURE_GENERATION: + case TURN_FAILURE_TOOL_RESULTS: + break; + } + strlcpy(a->err, error != NULL && error[0] != '\0' ? error : + "subagent execution failed", sizeof(a->err)); +} + +static void +agent_projection(void *arg, struct turn_projection *projection) +{ + struct agent_state *a = arg; + size_t count; + + if (a->projection_pending) + fatalx("subagent projected before its prior send receipt"); + count = (size_t)agent_msg_count(&a->conv); + if (a->prev_msgs > INT_MAX || ++a->projection_serial == 0) + fatalx("subagent Projection metadata overflow"); + memset(projection, 0, sizeof(*projection)); + projection->messages = &a->conv; + projection->message_count = count; + projection->cache_tail_msg = a->prev_msgs == 0 ? -1 : + (int)a->prev_msgs - 1; + projection->receipt = a->projection_serial; + a->pending_receipt = projection->receipt; + a->pending_messages = count; + a->pending_cache_tail = projection->cache_tail_msg; + a->projection_pending = 1; +} + +static void +agent_projection_sent(void *arg, const struct turn_projection *projection) +{ + struct agent_state *a = arg; + + if (!a->projection_pending || projection->messages != &a->conv || + projection->receipt != a->pending_receipt || + projection->message_count != a->pending_messages || + projection->cache_tail_msg != a->pending_cache_tail) + fatalx("subagent acknowledged the wrong Projection"); + a->prev_msgs = projection->message_count; + a->projection_pending = 0; +} + +static void +agent_accept(void *arg, struct msg *message) +{ + struct agent_state *a = arg; + + TAILQ_INSERT_TAIL(&a->conv, message, entry); +} + +/* Send one Projection and feed its already-decoded provider events. */ +static enum turn_io +agent_generate(void *arg, const struct turn_generation_request *request, + struct turn_generation_io *io, char *error, size_t errorsz) +{ + struct agent_state *a = arg; struct a_usage au; struct buf body; - struct generation *g; struct generation_event ev; struct generation_usage gu; struct imsg imsg; @@ -426,22 +506,36 @@ agent_generation(struct agent_state *a, const struct t uint32_t type; enum generation_feed_status status; enum generation_kind kind; + int protocol = 0; - memset(result, 0, sizeof(*result)); - a->err[0] = '\0'; + if (!a->projection_pending || request->projection.messages != &a->conv || + request->projection.receipt != a->pending_receipt || + request->projection.message_count != a->pending_messages || + request->projection.cache_tail_msg != a->pending_cache_tail) { + (void)snprintf(error, errorsz, "%s", + "subagent received an inconsistent Projection"); + return (TURN_IO_ERROR); + } + a->generation_tools = request->tools; + if (request->tools == TURN_TOOLS_DISABLED) + agent_report_tail(a); buf_init(&body); - agent_build_request(a, &body, defs, ndefs); + agent_build_request(a, &body, request); if (body.len > FUGU_REQUEST_MAX) { buf_free(&body); - strlcpy(a->err, "subagent request exceeds 16 MiB", sizeof(a->err)); - return (-1); + (void)snprintf(error, errorsz, "%s", + "subagent request exceeds 16 MiB"); + return (TURN_IO_ERROR); } - g = generation_new(FUGU_TURN_DATA_MAX - a->data_total); agent_progress(a, "requesting %s", a->boot->model); agent_send_bytes(a, FUGU_IMSG_AGENT_REQUEST, body.data, body.len); agent_send(a, FUGU_IMSG_AGENT_REQUEST_END, NULL, 0); buf_free(&body); - a->prev_msgs = agent_msg_count(&a->conv); + if (turn_generation_sent(io) == -1) { + (void)snprintf(error, errorsz, "%s", + "subagent could not acknowledge its Projection"); + return (TURN_IO_ERROR); + } while (agent_next(a, &imsg)) { type = imsg_get_type(&imsg); @@ -454,6 +548,12 @@ agent_generation(struct agent_state *a, const struct t exit(0); } kind = agent_generation_kind(type); + if (protocol) { + imsg_free(&imsg); + if (kind == GENERATION_DONE || kind == GENERATION_ERROR) + return (TURN_IO_ERROR); + continue; + } feed_data = p; feed_len = len; if (kind == GENERATION_USAGE && p != NULL && @@ -467,13 +567,13 @@ agent_generation(struct agent_state *a, const struct t feed_data = &gu; feed_len = sizeof(gu); } - status = generation_feed(g, kind, feed_data, feed_len, &ev); + status = turn_generation_feed(io, kind, feed_data, feed_len, &ev); if (status == GENERATION_PROTOCOL) { - const char *why = generation_error(g); - + protocol = 1; imsg_free(&imsg); - fatalx("subagent provider protocol: %s", - why != NULL ? why : "invalid Generation event"); + if (kind == GENERATION_DONE || kind == GENERATION_ERROR) + return (TURN_IO_ERROR); + continue; } if (status == GENERATION_MORE && kind == GENERATION_USAGE) { memset(&au, 0, sizeof(au)); @@ -490,35 +590,11 @@ agent_generation(struct agent_state *a, const struct t continue; if (status != GENERATION_TERMINAL) fatalx("subagent Generation returned an unknown feed status"); - if (generation_take(g, result) == -1) - fatalx("subagent could not take terminal Generation"); - if (result->data_used > FUGU_TURN_DATA_MAX - a->data_total) - fatalx("subagent Generation violated its data budget"); - a->data_total += result->data_used; - if (result->outcome == GENERATION_NORMAL || - result->outcome == GENERATION_LENGTH) { - if (result->assistant == NULL || - result->invalid_cause != GENERATION_INVALID_NONE) - fatalx("successful subagent Generation has no assistant"); - result->reason = NULL; - generation_free(g); - return (0); - } - if (result->assistant != NULL) - fatalx("failed subagent Generation returned an assistant"); - if (result->outcome == GENERATION_PROVIDER_ERROR) - strlcpy(a->err, result->reason, sizeof(a->err)); - else if (result->outcome == GENERATION_INVALID) - agent_invalid_error(a, result->invalid_cause); - else - fatalx("subagent Generation returned an unknown outcome"); - result->reason = NULL; - generation_free(g); - return (-1); + return (TURN_IO_OK); } - generation_free(g); - strlcpy(a->err, "coordinator closed the subagent channel", sizeof(a->err)); - return (-1); + (void)snprintf(error, errorsz, "%s", + "coordinator closed the subagent channel"); + return (TURN_IO_ERROR); } static int @@ -538,8 +614,8 @@ agent_local_allowed(const char *name) } static void -agent_web(struct agent_state *a, const struct block *call, struct buf *out, - int *is_error) +agent_web(struct agent_state *a, const struct turn_tool_call *call, + struct buf *out, int *is_error) { struct tool_req req; struct tool_result result; @@ -549,10 +625,10 @@ agent_web(struct agent_state *a, const struct block *c size_t len; memset(&req, 0, sizeof(req)); - strlcpy(req.name, call->tool_name, sizeof(req.name)); + strlcpy(req.name, call->name, sizeof(req.name)); agent_send(a, FUGU_IMSG_AGENT_WEB_REQ, &req, sizeof(req)); - agent_send_bytes(a, FUGU_IMSG_AGENT_WEB_ARG, call->tool_input, - call->tool_input_len); + agent_send_bytes(a, FUGU_IMSG_AGENT_WEB_ARG, call->input, + call->input_len); agent_send(a, FUGU_IMSG_AGENT_WEB_RUN, NULL, 0); buf_reset(out); *is_error = 1; @@ -588,98 +664,51 @@ agent_web(struct agent_state *a, const struct block *c exit(0); } -static void -agent_account_result(struct agent_state *a, struct buf *out, int *is_error) +static enum turn_io +agent_execute(void *arg, const struct turn_tool_batch *batch, + struct turn_tool_results *results, char *error, size_t errorsz) { - static const char mark[] = "subagent data bound exceeded"; - size_t room, n; + struct agent_state *a = arg; + const struct turn_tool_call *call; + struct buf out; + size_t i; + int is_error; - if (out->len > FUGU_TOOL_OUT_MAX) { - out->len = FUGU_TOOL_OUT_MAX; - *is_error = 1; - } - if (agent_account(a, out->len) == 0) - return; - room = FUGU_TURN_DATA_MAX - a->data_total; - buf_reset(out); - n = sizeof(mark) - 1; - if (n > room) - n = room; - if (n > 0) - buf_add(out, mark, n); - a->data_total += n; - *is_error = 1; -} - -static struct msg * -agent_execute_calls(struct agent_state *a, const struct msg *assistant) -{ - struct block *call; - struct msg *m; - struct buf out; - int is_error; - - m = msg_new(ROLE_USER); + (void)error; + (void)errorsz; buf_init(&out); - TAILQ_FOREACH(call, &assistant->blocks, entry) { - if (call->type != BLOCK_TOOL_USE) - continue; - if (a->data_exceeded) { - buf_reset(&out); - buf_addstr(&out, "subagent data bound exceeded"); - is_error = 1; - } - if (!a->data_exceeded && agent_local_allowed(call->tool_name)) { - agent_progress(a, "tool %s", call->tool_name); - tool_exec(a->tools, call->tool_name, call->tool_input, - call->tool_input_len, &out, &is_error); - } else if (!a->data_exceeded && agent_is_web(call->tool_name) && - ((strcmp(call->tool_name, "http_request") == 0 && + for (i = 0; i < batch->count; i++) { + call = &batch->calls[i]; + buf_reset(&out); + is_error = 1; + if (agent_local_allowed(call->name)) { + agent_progress(a, "tool %s", call->name); + tool_exec(a->tools, call->name, call->input, + call->input_len, &out, &is_error); + } else if (agent_is_web(call->name) && + ((strcmp(call->name, "http_request") == 0 && a->boot->start.http_request) || - (strcmp(call->tool_name, "http_request") != 0 && + (strcmp(call->name, "http_request") != 0 && a->boot->start.web_search))) { - agent_progress(a, "tool %s", call->tool_name); + agent_progress(a, "tool %s", call->name); agent_web(a, call, &out, &is_error); - } else if (!a->data_exceeded) { - buf_reset(&out); + } else { buf_addf(&out, "%s is unavailable to a read-only subagent", - call->tool_name); + call->name); is_error = 1; } - agent_account_result(a, &out, &is_error); - msg_add_tool_result(m, call->tool_id, (const char *)out.data, - out.len, is_error); + if (out.len > FUGU_TOOL_OUT_MAX) { + out.len = FUGU_TOOL_OUT_MAX; + is_error = 1; + } + if (turn_tool_result_set(results, i, call->id, out.data, + out.len, is_error) == -1) + break; } buf_free(&out); - return (m); + return (TURN_IO_OK); } -static struct msg * -agent_answer_dangling(struct agent_state *a, const struct msg *assistant) -{ - static const char reason[] = - "subagent tool call was truncated and was not executed"; - struct block *call; - struct msg *m; - struct buf out; - int is_error; - - m = msg_new(ROLE_USER); - buf_init(&out); - TAILQ_FOREACH(call, &assistant->blocks, entry) { - if (call->type != BLOCK_TOOL_USE) - continue; - buf_reset(&out); - buf_add(&out, reason, sizeof(reason) - 1); - is_error = 1; - agent_account_result(a, &out, &is_error); - msg_add_tool_result(m, call->tool_id, - (const char *)out.data, out.len, is_error); - } - buf_free(&out); - return (m); -} - static void agent_report_text(struct agent_state *a, const struct msg *assistant) { @@ -702,6 +731,20 @@ agent_report_tail(struct agent_state *a) } static void +agent_assistant_accepted(void *arg, const struct turn_assistant_event *event) +{ + struct agent_state *a = arg; + + if (a->generation_tools == TURN_TOOLS_ENABLED && + event->outcome == GENERATION_NORMAL && event->tool_calls == 0) + buf_reset(&a->report); + agent_report_text(a, event->message); + if (a->generation_tools == TURN_TOOLS_ENABLED && + event->outcome == GENERATION_NORMAL && event->tool_calls > 0) + a->iterations++; +} + +static void agent_mark_truncated(struct agent_state *a) { static const char mark[] = @@ -751,38 +794,14 @@ agent_finish(struct agent_state *a, int is_error) exit(0); } -static __dead void -agent_conclude(struct agent_state *a, struct msg *m) -{ - static const char conclude[] = - "You have reached the subagent tool-round bound or another subagent " - "execution bound. Do not call tools. " - "Conclude now with the most useful concise final report possible, " - "including uncertainty and unfinished checks."; - struct generation_result result; - - msg_add_text(m, conclude, sizeof(conclude) - 1); - agent_report_tail(a); - if (agent_generation(a, NULL, 0, &result) == 0) { - agent_report_text(a, result.assistant); - msg_free(result.assistant); - } else if (a->err[0] != '\0') { - if (a->report.len > 0) - buf_addstr(&a->report, "\n\n"); - buf_addf(&a->report, "subagent conclusion failed: %s", a->err); - } - agent_mark_truncated(a); - agent_finish(a, 0); -} - __dead void agent_run(struct worker *w, struct tool_ctx *tools, struct agent_boot *boot) { - struct agent_state a; - const struct tool_def *defs; - struct generation_result result; - struct msg *m; - int ndefs, round; + struct agent_state a; + struct turn_adapter adapter; + struct turn_options options; + struct turn_result result; + struct msg *m; memset(&a, 0, sizeof(a)); a.w = w; @@ -790,54 +809,64 @@ agent_run(struct worker *w, struct tool_ctx *tools, st a.boot = boot; TAILQ_INIT(&a.conv); buf_init(&a.report); - a.data_total = boot->prompt.len; m = msg_new(ROLE_USER); msg_add_text(m, (const char *)boot->prompt.data, boot->prompt.len); TAILQ_INSERT_TAIL(&a.conv, m, entry); - defs = agent_tooldefs(&a, &ndefs); + a.defs = agent_tooldefs(&a, &a.ndefs); agent_progress(&a, "started"); - for (round = 1; round <= AGENT_ITER_MAX; round++) { - if (agent_generation(&a, defs, ndefs, &result) == -1) { - agent_report_tail(&a); + memset(&adapter, 0, sizeof(adapter)); + adapter.arg = &a; + adapter.projection = agent_projection; + adapter.projection_sent = agent_projection_sent; + adapter.accept = agent_accept; + adapter.assistant_accepted = agent_assistant_accepted; + adapter.generate = agent_generate; + adapter.execute = agent_execute; + memset(&options, 0, sizeof(options)); + options.policy = TURN_POLICY_SUBAGENT; + options.data_limit = FUGU_TURN_DATA_MAX; + options.data_used = boot->prompt.len; + options.tool_batch_limit = AGENT_ITER_MAX; + if (turn_run(&adapter, &options, &result) == -1) + fatalx("invalid canonical subagent Turn configuration"); + + switch (result.status) { + case TURN_STATUS_DONE: + agent_finish(&a, 0); + case TURN_STATUS_CONCLUDED: + if (result.conclusion_failure != TURN_FAILURE_NONE) { + agent_turn_error(&a, result.conclusion_failure, + result.conclusion_invalid_cause, + result.conclusion_error); if (a.report.len > 0) buf_addstr(&a.report, "\n\n"); + buf_addf(&a.report, "subagent conclusion failed: %s", + a.err); + } + agent_mark_truncated(&a); + agent_finish(&a, 0); + case TURN_STATUS_FAILED: + agent_turn_error(&a, result.failure, result.invalid_cause, + result.error); + agent_report_tail(&a); + if (a.report.len > 0) + buf_addstr(&a.report, "\n\n"); + if (result.failure == TURN_FAILURE_DATA_BOUND) + buf_addstr(&a.report, a.err); + else buf_addf(&a.report, "subagent provider error: %s", a.err); - agent_finish(&a, 1); - } - m = result.assistant; - TAILQ_INSERT_TAIL(&a.conv, m, entry); - if (result.outcome == GENERATION_LENGTH) { - agent_report_text(&a, m); - m = agent_answer_dangling(&a, m); - TAILQ_INSERT_TAIL(&a.conv, m, entry); - agent_conclude(&a, m); - } - if (result.tool_calls == 0) { - /* A normal completion returns only its final assistant text; - * interim prose is retained solely as a cap fallback. */ - buf_reset(&a.report); - agent_report_text(&a, m); - agent_finish(&a, 0); - } - /* Preserve tool-bearing prose so an iteration cap can still return - * everything useful emitted before its one conclusion attempt. */ - agent_report_text(&a, m); - - a.iterations++; - m = agent_execute_calls(&a, m); - TAILQ_INSERT_TAIL(&a.conv, m, entry); - if (a.data_exceeded) { - agent_report_tail(&a); - if (a.report.len > 0) - buf_addstr(&a.report, "\n\n"); - buf_addstr(&a.report, "subagent data bound exceeded"); - agent_finish(&a, 1); - } - if (round == AGENT_ITER_MAX) { - agent_conclude(&a, m); - } + agent_finish(&a, 1); + case TURN_STATUS_INTERRUPTED: + agent_turn_error(&a, TURN_FAILURE_NONE, + GENERATION_INVALID_NONE, NULL); + agent_report_tail(&a); + if (a.report.len > 0) + buf_addstr(&a.report, "\n\n"); + buf_addstr(&a.report, a.err); + agent_finish(&a, 1); + case TURN_STATUS_CAP: + fatalx("canonical Subagent Turn returned Lead cap status"); } - /* NOTREACHED */ - agent_finish(&a, 1); + fatalx("canonical Subagent Turn returned an unknown status"); }