commit - e233607aa644e0431ee41a6b3f1ecff807ce0072
commit + 0b367cd2488449362936e643c49ecfa371c49495
blob - 12829ba9ecd954ba6f7469ad64678f1a2ff6585e
blob + 702eb7739317c611462144ccdc6713b8016835a8
--- regress/imsgev/imsgev_regress.c
+++ regress/imsgev/imsgev_regress.c
#include <event.h>
#include <errno.h>
#include <imsg.h>
+#include <poll.h>
#include <stdint.h>
#include <string.h>
#include <unistd.h>
#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
};
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;
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);
}
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))
}
#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
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
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
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))
{
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)
}
}
+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)
{
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;
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);
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
#include <event.h>
#include <errno.h>
#include <imsg.h>
+#include <poll.h>
#include <stdint.h>
#include "log.h"
}
}
+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
*/
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
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");
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
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,
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
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
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 *);
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
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",
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");
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)) {
"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);
}
