commit 0b367cd2488449362936e643c49ecfa371c49495 from: Isaac Meerleo date: Thu Jul 16 13:11:45 2026 UTC finish canonical turn and imsgev ownership commit - e233607aa644e0431ee41a6b3f1ecff807ce0072 commit + 0b367cd2488449362936e643c49ecfa371c49495 blob - 12829ba9ecd954ba6f7469ad64678f1a2ff6585e blob + 702eb7739317c611462144ccdc6713b8016835a8 --- regress/imsgev/imsgev_regress.c +++ regress/imsgev/imsgev_regress.c @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -33,12 +34,11 @@ #include "imsgev_regress.h" static struct { - enum imsgev_regress_op fail_op; - unsigned int fail_at; + unsigned int fail_at[IMSGEV_REGRESS_OPS]; + int fail_errno[IMSGEV_REGRESS_OPS]; unsigned int calls[IMSGEV_REGRESS_OPS]; int trace_fd; } regress_state = { - .fail_op = IMSGEV_REGRESS_OPS, .trace_fd = -1 }; @@ -46,19 +46,29 @@ void imsgev_regress_reset(void) { memset(regress_state.calls, 0, sizeof(regress_state.calls)); - regress_state.fail_op = IMSGEV_REGRESS_OPS; - regress_state.fail_at = 0; + memset(regress_state.fail_at, 0, sizeof(regress_state.fail_at)); + memset(regress_state.fail_errno, 0, + sizeof(regress_state.fail_errno)); regress_state.trace_fd = -1; } void imsgev_regress_fail(enum imsgev_regress_op op, unsigned int call) { - regress_state.fail_op = op; - regress_state.fail_at = call; + imsgev_regress_fail_errno(op, call, EIO); } void +imsgev_regress_fail_errno(enum imsgev_regress_op op, unsigned int call, + int error) +{ + if (op >= IMSGEV_REGRESS_OPS) + return; + regress_state.fail_at[op] = call; + regress_state.fail_errno[op] = error; +} + +void imsgev_regress_trace(int fd) { regress_state.trace_fd = fd; @@ -74,9 +84,9 @@ static int regress_fails(enum imsgev_regress_op op) { regress_state.calls[op]++; - if (regress_state.fail_op == op && - regress_state.calls[op] == regress_state.fail_at) { - errno = EIO; + if (regress_state.fail_at[op] != 0 && + regress_state.calls[op] == regress_state.fail_at[op]) { + errno = regress_state.fail_errno[op]; return (1); } return (0); @@ -109,6 +119,22 @@ regress_imsgbuf_flush(struct imsgbuf *ibuf) } static int +regress_imsgbuf_write(struct imsgbuf *ibuf) +{ + if (regress_fails(IMSGEV_REGRESS_WRITE)) + return (-1); + return (imsgbuf_write(ibuf)); +} + +static int +regress_poll(struct pollfd *fds, nfds_t nfds, int timeout) +{ + if (regress_fails(IMSGEV_REGRESS_POLL)) + return (-1); + return (poll(fds, nfds, timeout)); +} + +static int regress_event_add(struct event *ev, const struct timeval *tv) { if (regress_fails(IMSGEV_REGRESS_EVENT_ADD)) @@ -125,13 +151,17 @@ regress_event_del(struct event *ev) } #define imsg_compose regress_imsg_compose +#define imsgbuf_write regress_imsgbuf_write #define imsgbuf_flush regress_imsgbuf_flush +#define poll regress_poll #define event_add regress_event_add #define event_del regress_event_del #include "../../src/common/imsgev.c" #undef imsg_compose +#undef imsgbuf_write #undef imsgbuf_flush +#undef poll #undef event_add #undef event_del blob - 5feb4bd80968781b2a2b5843315aae47adfcd4fd blob + b795c79d140127c87f046bfdada35d32dfb5852a --- regress/imsgev/imsgev_regress.h +++ regress/imsgev/imsgev_regress.h @@ -19,7 +19,9 @@ enum imsgev_regress_op { IMSGEV_REGRESS_COMPOSE, + IMSGEV_REGRESS_WRITE, IMSGEV_REGRESS_FLUSH, + IMSGEV_REGRESS_POLL, IMSGEV_REGRESS_EVENT_ADD, IMSGEV_REGRESS_EVENT_DEL, IMSGEV_REGRESS_OPS @@ -27,6 +29,8 @@ enum imsgev_regress_op { void imsgev_regress_reset(void); void imsgev_regress_fail(enum imsgev_regress_op, unsigned int); +void imsgev_regress_fail_errno(enum imsgev_regress_op, unsigned int, + int); void imsgev_regress_trace(int); unsigned int imsgev_regress_calls(enum imsgev_regress_op); blob - 76378954702734712cbd8768607882c56c6553d8 blob + 47b07536ab67377893e6c4c0af6cdc2a99b82a27 --- regress/imsgev/imsgev_test.c +++ regress/imsgev/imsgev_test.c @@ -146,6 +146,57 @@ check_chunks(size_t len) free(data); } +static void +check_blocking_chunks(size_t len) +{ + struct imsgev_payload payload; + struct imsgbuf receiver; + struct imsgev iev; + struct imsg imsg; + u_char *data; + size_t frames, i, n, off; + int fds[2], space = 128 * 1024; + + data = malloc(len); + if (data == NULL) + err(1, "malloc blocking chunk test"); + for (i = 0; i < len; i++) + data[i] = (u_char)((i * 29 + 17) & 0xff); + frames = 1 + (len - 1) / IMSGEV_CHUNK_MAX; + channel_init(&iev, &receiver, fds); + if (setsockopt(fds[0], SOL_SOCKET, SO_SNDBUF, &space, + sizeof(space)) == -1 || + setsockopt(fds[1], SOL_SOCKET, SO_RCVBUF, &space, + sizeof(space)) == -1) + err(1, "setsockopt blocking chunk test"); + imsgev_regress_reset(); + imsgev_send_chunks_blocking(&iev, TEST_DATA, 0x87654321, data, len); + CHECK(imsgbuf_queuelen(&iev.ibuf) == 0); + CHECK(iev.events == EV_READ); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_COMPOSE) == frames); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_WRITE) >= frames); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_EVENT_DEL) == 0); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_EVENT_ADD) == 0); + for (i = off = 0; i < frames; i++, off += n) { + while (imsgbuf_get(&receiver, &imsg) == 0) + if (imsgbuf_read(&receiver) != 1) + errx(1, "imsgbuf_read blocking chunks"); + n = len - off; + if (n > IMSGEV_CHUNK_MAX) + n = IMSGEV_CHUNK_MAX; + CHECK(imsg.hdr.type == TEST_DATA && + imsg.hdr.peerid == 0x87654321); + CHECK(imsgev_get_payload(&imsg, &payload) == 0); + CHECK(payload.len == n && payload.data != NULL && + memcmp(payload.data, data + off, n) == 0); + imsg_free(&imsg); + } + CHECK(off == len); + CHECK(imsgbuf_get(&receiver, &imsg) == 0); + channel_clear(&iev, &receiver, fds); + free(data); +} + static int dies(void (*fn)(void)) { @@ -233,6 +284,32 @@ die_clear_del(void) imsgev_clear(&iev); } +static void +die_blocking_after_queued_send(void) +{ + struct imsgbuf receiver; + struct imsgev iev; + int fds[2]; + + channel_init(&iev, &receiver, fds); + imsgev_send(&iev, TEST_DATA, 0, NULL, 0); + imsgev_send_blocking(&iev, TEST_END, 0, NULL, 0); +} + +static void +die_blocking_after_stale_write_interest(void) +{ + struct imsgbuf receiver; + struct imsgev iev; + int fds[2]; + + channel_init(&iev, &receiver, fds); + imsgev_send(&iev, TEST_DATA, 0, NULL, 0); + if (imsgbuf_flush(&iev.ibuf) == -1) + err(1, "imsgbuf_flush stale write interest"); + imsgev_send_blocking(&iev, TEST_END, 0, NULL, 0); +} + static int failed_exchange(enum imsgev_regress_op op, unsigned int fail_at, int null_payload, uint32_t *trace, size_t *trace_len) @@ -280,6 +357,59 @@ failed_exchange(enum imsgev_regress_op op, unsigned in } } +static int +failed_blocking_exchange(enum imsgev_regress_op op, unsigned int fail_at, + int error, int null_payload, uint32_t *trace, size_t *trace_len) +{ + u_char data[2 * IMSGEV_CHUNK_MAX + 1]; + struct imsgbuf receiver; + struct imsgev iev; + ssize_t n; + size_t off; + pid_t pid; + int devnull, fds[2], pfd[2], status; + + if (pipe(pfd) == -1) + err(1, "pipe"); + fflush(stdout); + switch (pid = fork()) { + case -1: + err(1, "fork"); + case 0: + close(pfd[0]); + if ((devnull = open("/dev/null", O_WRONLY)) != -1) + (void)dup2(devnull, STDERR_FILENO); + channel_init(&iev, &receiver, fds); + imsgev_regress_reset(); + imsgev_regress_trace(pfd[1]); + if (null_payload) + imsgev_send_chunks_blocking(&iev, TEST_DATA, 0, NULL, 1); + memset(data, 0x5a, sizeof(data)); + if (op == IMSGEV_REGRESS_POLL) { + imsgev_regress_fail_errno(IMSGEV_REGRESS_WRITE, 1, + EAGAIN); + imsgev_regress_fail_errno(op, fail_at, error); + } else + imsgev_regress_fail_errno(op, fail_at, error); + imsgev_send_chunks_blocking(&iev, TEST_DATA, 0, data, + sizeof(data)); + imsgev_send_blocking(&iev, TEST_END, 0, NULL, 0); + _exit(0); + default: + close(pfd[1]); + off = 0; + while (off < 4 * sizeof(*trace) && + (n = read(pfd[0], (u_char *)trace + off, + 4 * sizeof(*trace) - off)) > 0) + off += (size_t)n; + close(pfd[0]); + if (waitpid(pid, &status, 0) == -1) + err(1, "waitpid"); + *trace_len = off / sizeof(*trace); + return (WIFEXITED(status) && WEXITSTATUS(status) == 1); + } +} + int main(void) { @@ -319,6 +449,62 @@ main(void) check_chunks(2 * IMSGEV_CHUNK_MAX); check_chunks(2 * IMSGEV_CHUNK_MAX + 1); + /* Blocking sends drain their exact frame without event-loop re-arming. */ + channel_init(&iev, &receiver, fds); + imsgev_regress_reset(); + imsgev_send_blocking(&iev, TEST_DATA, 11, binary, sizeof(binary)); + CHECK(imsgbuf_queuelen(&iev.ibuf) == 0); + CHECK(iev.events == EV_READ); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_COMPOSE) == 1); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_WRITE) >= 1); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_EVENT_DEL) == 0); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_EVENT_ADD) == 0); + CHECK(imsgbuf_read(&receiver) == 1); + CHECK(imsgbuf_get(&receiver, &imsg) == 1); + CHECK(imsg.hdr.type == TEST_DATA && imsg.hdr.peerid == 11); + CHECK(imsgev_get_payload(&imsg, &payload) == 0); + CHECK(payload.len == sizeof(binary) && + memcmp(payload.data, binary, sizeof(binary)) == 0); + imsg_free(&imsg); + channel_clear(&iev, &receiver, fds); + + /* Blocking chunk emission owns every exact boundary and byte. */ + check_blocking_chunks(1); + check_blocking_chunks(IMSGEV_CHUNK_MAX); + check_blocking_chunks(IMSGEV_CHUNK_MAX + 1); + check_blocking_chunks(2 * IMSGEV_CHUNK_MAX + 1); + + /* A transient full socket waits and retries without arming EV_WRITE. */ + channel_init(&iev, &receiver, fds); + imsgev_regress_reset(); + imsgev_regress_fail_errno(IMSGEV_REGRESS_WRITE, 1, EAGAIN); + imsgev_send_blocking(&iev, TEST_DATA, 13, binary, sizeof(binary)); + CHECK(imsgbuf_queuelen(&iev.ibuf) == 0); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_WRITE) >= 2); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_POLL) == 1); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_EVENT_DEL) == 0); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_EVENT_ADD) == 0); + CHECK(imsgbuf_read(&receiver) == 1); + CHECK(imsgbuf_get(&receiver, &imsg) == 1 && + imsg.hdr.type == TEST_DATA && imsg.hdr.peerid == 13); + imsg_free(&imsg); + channel_clear(&iev, &receiver, fds); + + /* An interrupted wait returns to the drain loop. */ + channel_init(&iev, &receiver, fds); + imsgev_regress_reset(); + imsgev_regress_fail_errno(IMSGEV_REGRESS_WRITE, 1, EAGAIN); + imsgev_regress_fail_errno(IMSGEV_REGRESS_POLL, 1, EINTR); + imsgev_send_blocking(&iev, TEST_DATA, 15, binary, sizeof(binary)); + CHECK(imsgbuf_queuelen(&iev.ibuf) == 0); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_WRITE) >= 2); + CHECK(imsgev_regress_calls(IMSGEV_REGRESS_POLL) == 1); + CHECK(imsgbuf_read(&receiver) == 1); + CHECK(imsgbuf_get(&receiver, &imsg) == 1 && + imsg.hdr.type == TEST_DATA && imsg.hdr.peerid == 15); + imsg_free(&imsg); + channel_clear(&iev, &receiver, fds); + make_imsg(NULL, 0, &receiver, &imsg); payload.data = (const u_char *)1; payload.len = 1; @@ -380,6 +566,10 @@ main(void) CHECK(dies(die_rearm_add)); CHECK(dies(die_clear_del)); + /* Blocking takeover may not inherit queued or stale event output. */ + CHECK(dies(die_blocking_after_queued_send)); + CHECK(dies(die_blocking_after_stale_write_interest)); + /* NULL/nonzero is rejected before even the first composition. */ CHECK(failed_exchange(IMSGEV_REGRESS_OPS, 0, 1, trace, &trace_len)); CHECK(trace_len == 0); @@ -412,6 +602,22 @@ main(void) CHECK(trace_len == 2 && trace[0] == TEST_DATA && trace[1] == TEST_DATA); + /* Blocking failures cannot fall through to a later terminal frame. */ + CHECK(failed_blocking_exchange(IMSGEV_REGRESS_OPS, 0, EIO, 1, + trace, &trace_len)); + CHECK(trace_len == 0); + CHECK(failed_blocking_exchange(IMSGEV_REGRESS_COMPOSE, 2, EIO, 0, + trace, &trace_len)); + CHECK(trace_len == 2 && trace[0] == TEST_DATA && + trace[1] == TEST_DATA); + CHECK(failed_blocking_exchange(IMSGEV_REGRESS_WRITE, 2, EIO, 0, + trace, &trace_len)); + CHECK(trace_len == 2 && trace[0] == TEST_DATA && + trace[1] == TEST_DATA); + CHECK(failed_blocking_exchange(IMSGEV_REGRESS_POLL, 1, EIO, 0, + trace, &trace_len)); + CHECK(trace_len == 1 && trace[0] == TEST_DATA); + /* Teardown is synchronous, best effort, and never re-arms events. */ channel_init(&iev, &receiver, fds); imsgev_regress_reset(); blob - f08725e30d20ea6b7831cbda22421c098dea831d blob + 2022d436db9ff6edd8bc07970951ef59b1c5713e --- src/common/imsgev.c +++ src/common/imsgev.c @@ -19,6 +19,7 @@ #include #include #include +#include #include #include "log.h" @@ -85,6 +86,48 @@ imsgev_send_chunks(struct imsgev *iev, uint32_t type, } } +void +imsgev_send_blocking(struct imsgev *iev, uint32_t type, uint32_t peerid, + const void *data, size_t datalen) +{ + struct pollfd pfd; + + if (imsgbuf_queuelen(&iev->ibuf) > 0 || + (iev->events & EV_WRITE) != 0) + fatalx("blocking send mixed with event output"); + if (imsg_compose(&iev->ibuf, type, peerid, -1, -1, data, + datalen) == -1) + fatal("blocking imsg_compose"); + while (imsgbuf_queuelen(&iev->ibuf) > 0) { + if (imsgbuf_write(&iev->ibuf) == 0) + continue; + if (errno != EAGAIN && errno != EINTR) + fatal("blocking imsgbuf_write"); + pfd.fd = iev->ibuf.fd; + pfd.events = POLLOUT; + pfd.revents = 0; + if (poll(&pfd, 1, -1) == -1 && errno != EINTR) + fatal("blocking output poll"); + } +} + +void +imsgev_send_chunks_blocking(struct imsgev *iev, uint32_t type, + uint32_t peerid, const void *data, size_t datalen) +{ + const u_char *p = data; + size_t off, n; + + if (data == NULL && datalen != 0) + fatalx("NULL blocking chunk payload"); + for (off = 0; off < datalen; off += n) { + n = datalen - off; + if (n > IMSGEV_CHUNK_MAX) + n = IMSGEV_CHUNK_MAX; + imsgev_send_blocking(iev, type, peerid, p + off, n); + } +} + /* * Teardown has no event-loop future in which queued output can drain. * Attempt one synchronous empty frame and flush, but never turn cleanup blob - 1ceb3f1eda3dd126f60dcacb3cb3e8137bf036e9 blob + 663871fbe45c36f1c7124f77341dcf4795e586c4 --- src/common/imsgev.h +++ src/common/imsgev.h @@ -58,6 +58,17 @@ void imsgev_send(struct imsgev *, uint32_t, uint32_t, */ void imsgev_send_chunks(struct imsgev *, uint32_t, uint32_t, const void *, size_t); +/* + * Blocking variants are a synchronous takeover: the channel must have no + * queued event-driven output or EV_WRITE interest. They drain their complete + * frame without re-arming the event, wait indefinitely for a full socket, and + * fail closed on composition or transport errors. Chunk emission drains each + * frame before composing the next one. + */ +void imsgev_send_blocking(struct imsgev *, uint32_t, uint32_t, + const void *, size_t); +void imsgev_send_chunks_blocking(struct imsgev *, uint32_t, uint32_t, + const void *, size_t); /* Teardown attempts one synchronous empty frame without re-arming events. */ void imsgev_send_teardown(struct imsgev *, uint32_t); int imsgev_get_payload(struct imsg *, struct imsgev_payload *); blob - 8abfa6306d29944cac454e3872a47d78c06a6dbe blob + 96f26a9a5cdab0c9f448a8f33a38a90f4a50eb76 --- src/fugu/coord.c +++ src/fugu/coord.c @@ -2866,14 +2866,12 @@ lead_assistant_text(const struct msg *assistant) return (text); } -/* Copy accepted canonical views into the existing presentation/tool seams. */ +/* Copy the accepted assistant text into the existing presentation seam. */ static void -lead_generation_adapt(struct coord *c, const struct msg *assistant, +lead_generation_present(struct coord *c, const struct msg *assistant, size_t tool_calls) { struct generation_span text; - struct tool_call *tc; - struct block *b; if (assistant == NULL || tool_calls > MAX_TOOL_CALLS) fatalx("invalid successful Lead Generation"); @@ -2881,32 +2879,6 @@ lead_generation_adapt(struct coord *c, const struct ms buf_reset(&c->text); if (text.len > 0) buf_add(&c->text, text.data, text.len); - - reset_tcalls(c); - TAILQ_FOREACH(b, &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++; - } - if ((size_t)c->ntcalls != tool_calls) - fatalx("canonical Lead tool-call count mismatch"); } static void @@ -2916,7 +2888,7 @@ lead_assistant_accepted(void *arg, const struct turn_a if (event->stop_reason == NULL) fatalx("canonical Lead assistant has no stop reason"); - lead_generation_adapt(c, event->message, event->tool_calls); + lead_generation_present(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, @@ -3319,6 +3291,47 @@ run_pending_agents(struct coord *c) buf_free(&system); } +/* Populate the existing execution view from the canonical batch Interface. */ +static int +lead_tool_batch_adapt(struct coord *c, const struct turn_tool_batch *batch, + char *error, size_t errorsz) +{ + const struct turn_tool_call *call; + struct tool_call *tc; + size_t i; + + reset_tcalls(c); + if (batch == NULL || batch->calls == NULL || batch->count == 0 || + batch->count > MAX_TOOL_CALLS) { + (void)snprintf(error, errorsz, "%s", + "invalid canonical Lead tool batch"); + return (-1); + } + for (i = 0; i < batch->count; i++) { + call = &batch->calls[i]; + tc = &c->tcalls[i]; + if (call->id == NULL || call->name == NULL || + (call->input_len > 0 && call->input == NULL) || + strlcpy(tc->id, call->id, sizeof(tc->id)) >= sizeof(tc->id) || + strlcpy(tc->name, call->name, sizeof(tc->name)) >= + sizeof(tc->name)) { + (void)snprintf(error, errorsz, "%s", + "invalid canonical Lead tool call"); + reset_tcalls(c); + return (-1); + } + tc->index = (int)i; + buf_init(&tc->input); + buf_init(&tc->result); + c->ntcalls++; + if (call->input_len > 0) + buf_add(&tc->input, call->input, call->input_len); + tc->result_ready = 0; + tc->result_error = 0; + } + return (0); +} + /* 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 @@ -3332,25 +3345,8 @@ lead_execute(void *arg, const struct turn_tool_batch * 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"); + if (lead_tool_batch_adapt(c, batch, error, errorsz) == -1) 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); blob - b4ca1248a69ef13a77f5368c3257a52e0cf2a6e5 blob + e82e6f24aa952e27ca87a4ce61e4bb3a3a48ff58 --- src/fugu-tool/agent.c +++ src/fugu-tool/agent.c @@ -219,41 +219,6 @@ agent_boot_prompt(struct agent_boot *b, const void *da return (0); } -static void -agent_send(struct agent_state *a, uint32_t type, const void *data, size_t len) -{ - struct pollfd pfd; - - if (imsg_compose(&a->w->iev.ibuf, type, 0, -1, -1, data, len) == -1) - fatal("agent imsg_compose"); - while (imsgbuf_queuelen(&a->w->iev.ibuf) > 0) { - if (imsgbuf_write(&a->w->iev.ibuf) == 0) - continue; - if (errno != EAGAIN && errno != EINTR) - fatal("agent imsgbuf_write"); - memset(&pfd, 0, sizeof(pfd)); - pfd.fd = a->w->iev.ibuf.fd; - pfd.events = POLLOUT; - if (poll(&pfd, 1, -1) == -1 && errno != EINTR) - fatal("agent output poll"); - } -} - -static void -agent_send_bytes(struct agent_state *a, uint32_t type, const void *data, - size_t len) -{ - const u_char *p = data; - size_t off, n; - - for (off = 0; off < len; off += n) { - n = len - off; - if (n > IMSGEV_CHUNK_MAX) - n = IMSGEV_CHUNK_MAX; - agent_send(a, type, p + off, n); - } -} - 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 *); @@ -280,7 +245,8 @@ agent_progress(struct agent_state *a, const char *fmt, for (i = 0; i < n; i++) if ((u_char)buf[i] < 0x20 || (u_char)buf[i] > 0x7e) buf[i] = '?'; - agent_send(a, FUGU_IMSG_AGENT_PROGRESS, buf, (size_t)n); + imsgev_send_blocking(&a->w->iev, FUGU_IMSG_AGENT_PROGRESS, 0, buf, + (size_t)n); } /* Blocking only inside the ephemeral -r worker. Provider/web deadlines live @@ -528,8 +494,10 @@ agent_generate(void *arg, const struct turn_generation return (TURN_IO_ERROR); } 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); + imsgev_send_chunks_blocking(&a->w->iev, FUGU_IMSG_AGENT_REQUEST, 0, + body.data, body.len); + imsgev_send_blocking(&a->w->iev, FUGU_IMSG_AGENT_REQUEST_END, 0, + NULL, 0); buf_free(&body); if (turn_generation_sent(io) == -1) { (void)snprintf(error, errorsz, "%s", @@ -581,7 +549,8 @@ agent_generate(void *arg, const struct turn_generation au.output_tokens = ev.usage.output_tokens; au.cache_read = ev.usage.cache_read; au.cache_write = ev.usage.cache_write; - agent_send(a, FUGU_IMSG_AGENT_USAGE, &au, sizeof(au)); + imsgev_send_blocking(&a->w->iev, FUGU_IMSG_AGENT_USAGE, 0, + &au, sizeof(au)); } else if (status == GENERATION_MORE && kind == GENERATION_RETRY) agent_progress(a, "provider retry"); @@ -626,10 +595,11 @@ agent_web(struct agent_state *a, const struct turn_too memset(&req, 0, sizeof(req)); 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->input, - call->input_len); - agent_send(a, FUGU_IMSG_AGENT_WEB_RUN, NULL, 0); + imsgev_send_blocking(&a->w->iev, FUGU_IMSG_AGENT_WEB_REQ, 0, &req, + sizeof(req)); + imsgev_send_chunks_blocking(&a->w->iev, FUGU_IMSG_AGENT_WEB_ARG, 0, + call->input, call->input_len); + imsgev_send_blocking(&a->w->iev, FUGU_IMSG_AGENT_WEB_RUN, 0, NULL, 0); buf_reset(out); *is_error = 1; while (agent_next(a, &imsg)) { @@ -785,12 +755,14 @@ agent_finish(struct agent_state *a, int is_error) "subagent completed without text"); if (a->report.len > FUGU_TOOL_OUT_MAX) agent_mark_truncated(a); - agent_send_bytes(a, FUGU_IMSG_AGENT_OUT, a->report.data, a->report.len); + imsgev_send_chunks_blocking(&a->w->iev, FUGU_IMSG_AGENT_OUT, 0, + a->report.data, a->report.len); memset(&r, 0, sizeof(r)); r.is_error = is_error; r.truncated = a->truncated; r.iterations = a->iterations; - agent_send(a, FUGU_IMSG_AGENT_RESULT, &r, sizeof(r)); + imsgev_send_blocking(&a->w->iev, FUGU_IMSG_AGENT_RESULT, 0, &r, + sizeof(r)); exit(0); }