commit d120f420c9b56088098e2f370bc6e1919ca8564d from: Isaac Meerleo date: Sat Jul 18 15:47:34 2026 UTC Exercise Artifact paging and retirement end to end commit - 2440fa356713a6a5e24841131db5ecafa143a0ed commit + d120f420c9b56088098e2f370bc6e1919ca8564d blob - db3d3686c376acc3355626e9a0e505ae5ad71eaa blob + a5983bde67263c1d8fe2f3a5033646e035903189 --- regress/turn/run.sh +++ regress/turn/run.sh @@ -35,7 +35,9 @@ ok() { if [ "$2" -eq 0 ]; then pass "$1"; else fail dir=$(mktemp -d /tmp/fugu-turn.XXXXXXXXXX) test_home=$(mktemp -d /tmp/fugu-turn-home.XXXXXXXXXX) stubpid='' +fugupid='' cleanup() { + [ -n "$fugupid" ] && { kill "$fugupid" 2>/dev/null; wait "$fugupid" 2>/dev/null; } [ -n "$stubpid" ] && { kill "$stubpid" 2>/dev/null; wait "$stubpid" 2>/dev/null; } rm -rf "$dir" "$test_home" } @@ -610,7 +612,7 @@ stop_stub # --- 13-artifact: abbreviated raw output stays path-free ------------- rm -rf "$HOME/.fugu" -start_stub "ARTIFACTTOOL OK" +start_stub "ARTIFACTTOOL OUTPUTREAD OUTPUTREADDEFAULT OK" write_conf "$stubport" run_fugu -p "produce retained output" ok "artifact tool loop: abbreviated result completes" \ @@ -628,8 +630,67 @@ ok "artifact tool loop: no private storage path crosse $([ $? -ne 0 ]; echo $?) grep -q '"t":"artifact"' "$journal" ok "artifact tool loop: the structured Artifact record commits" $? +request_log_entry 3 "$dir/output-read-request" +grep -Fq '"name":"output_read"' "$dir/output-read-request" && + grep -Fq '"offset":65536' "$dir/output-read-request" && + grep -Fq '"limit":4' "$dir/output-read-request" && + grep -Fq '"encoding":"base64"' "$dir/output-read-request" +ok "output_read loop: Provider arguments cross the Coordinator seam" $? +grep -Fq '\"offset\":65536' "$dir/output-read-request" && + grep -Fq '\"returned_bytes\":4' "$dir/output-read-request" && + grep -Fq '\"next_offset\":65540' "$dir/output-read-request" && + grep -Fq '\"state\":\"pending\"' "$dir/output-read-request" && + grep -Fq '\"data\":\"QQD/Qg==\"' \ + "$dir/output-read-request" +ok "output_read loop: pending base64 page reaches the next Generation" $? +request_log_entry 4 "$dir/output-read-default-request" +grep -Eq '"id":"toolu_output_read_default","name":"output_read",'\ +'"input":\{"id":"[0-9a-f]{32}"\}' "$dir/output-read-default-request" && + grep -Fq '\"offset\":0' "$dir/output-read-default-request" && + grep -Fq '\"returned_bytes\":65536' \ + "$dir/output-read-default-request" && + grep -Fq '\"next_offset\":65536' \ + "$dir/output-read-default-request" && + grep -Fq '\"encoding\":\"text\"' \ + "$dir/output-read-default-request" && + grep -Fq '\"state\":\"pending\"' \ + "$dir/output-read-default-request" +ok "output_read loop: id-only call applies text and 64 KiB defaults" $? stop_stub +# A read response remains an exact bounded transaction after Ctrl-C. The +# regression-only Role delay widens the cancellation window; Coordinator must +# drain DATA/READ_RESULT before Artifact Turn abort reuses the same channel. +rm -rf "$HOME/.fugu" +start_stub "ARTIFACTTOOL OUTPUTREAD OK" +write_conf "$stubport" +FUGU_CONF=$dir/fugu.conf FUGU_LIBEXEC=$dir/libexec/fugu \ + FUGU_CA_FILE=$dir/cert.pem FUGU_RETRY_BASE_MS=50 \ + FUGU_ARTIFACT_READ_DELAY_MS=1000 \ + "$fugu" -p "cancel retained output read" "$dir/out" 2>"$dir/errout" & +fugupid=$! +i=0 +while ! grep -q '^--- OUTPUTREAD ---$' "$dir/reqlog" 2>/dev/null && + [ "$i" -lt 100 ]; do + sleep 0.05 + i=$((i + 1)) +done +sleep 0.1 +/bin/kill -INT "$fugupid" 2>/dev/null +wait "$fugupid" +rc=$? +fugupid='' +ok "output_read cancel: interrupted read exits 3" \ + $([ "$rc" -eq 3 ]; echo $?) +if grep -Eq 'bad Artifact read|unexpected Artifact event' \ + "$dir/errout" 2>/dev/null; then + fail "output_read cancel: exact delayed response drains before rollback" +else + pass "output_read cancel: exact delayed response drains before rollback" +fi +stop_stub + # --- 13a: a full brokered web loop over the real mesh (M8) ---------- # The development-only private-address seam lets the isolated fixture act # as both provider and web origin; installed/setgid binaries ignore it. @@ -2447,5 +2508,62 @@ else fi stop_stub +# A committed short job retires its current available spool while the Lead is +# waiting through a Provider retry. DROP is asynchronous to Tool results; +# Coordinator must discard only that exact current spool and ACK it once. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqout" +start_stub "BGSHORT OK RETRY429 OK" +write_conf "$stubport" +printf 'launch short background\nwait for available retirement\n/quit\n' \ + > "$dir/lin" +run_fugu_in "$dir/lin" +ok "background retirement: available DROP round-trip completes" \ + $([ "$rc" -eq 0 ]; echo $?) +n=$(grep -c '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null) +ok "background retirement: both available-DROP Turns commit" \ + $([ "$n" = 2 ]; echo $?) +if grep -Eq 'bad Artifact|mismatched|stale Artifact' \ + "$dir/errout" 2>/dev/null; then + fail "background retirement: available key is acknowledged once" +else + pass "background retirement: available key is acknowledged once" +fi +stop_stub + +# Eight detached logical reservations fill the 128 MiB Session quota without +# writing 128 MiB of physical data. The ninth job receives an unavailable +# zero-id reservation, commits, and later retires it over the same exact +# DROP/ACK transaction without waiting for an Artifact terminal. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqout" +scenario='' +: > "$dir/lin" +i=0 +while [ "$i" -lt 8 ]; do + scenario="$scenario BGTOOL OK" + printf 'reserve background spool %s\n' "$i" >> "$dir/lin" + i=$((i + 1)) +done +scenario="$scenario BGSHORT OK RETRY429 OK" +printf 'launch unavailable short background\nwait for unavailable retirement\n/quit\n' \ + >> "$dir/lin" +start_stub "$scenario" +write_conf "$stubport" +printf 'max_bg_jobs 9\n' >> "$dir/fugu.conf" +run_fugu_in "$dir/lin" +ok "background retirement: unavailable DROP round-trip completes" \ + $([ "$rc" -eq 0 ]; echo $?) +n=$(grep -c '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null) +ok "background retirement: all unavailable-DROP Turns commit" \ + $([ "$n" = 10 ]; echo $?) +if grep -Eq 'bad Artifact|mismatched|stale Artifact' \ + "$dir/errout" 2>/dev/null; then + fail "background retirement: zero-id key is acknowledged once" +else + pass "background retirement: zero-id key is acknowledged once" +fi +stop_stub + echo "$checks checks, $fails failures" [ "$fails" -eq 0 ] blob - d389e6fc626ecf334df56134b3f539986c0e8638 blob + fbccbbe31d6cad6698801d3e6f1ba7f266ad0810 --- regress/turn/stub.c +++ regress/turn/stub.c @@ -26,8 +26,10 @@ * stream with no content blocks), BADARGS (a complete but malformed * 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 + * WEBFETCH (a brokered web_fetch), OUTPUTREAD (an explicit brokered Artifact + * page), OUTPUTREADDEFAULT (a defaults-only Artifact page), + * BGTOOL (shell_bg sleep 300), BGSHORT (shell_bg sleep 1), + * ERROR_BG (the provider failure after * that launch), JOBOUTPUT (shell_output job 1), DOUBLE (a hostile error event * appended after a complete OK stream), AGENTCALL (lead delegates to a * labelled subagent), AGENTREAD (the subagent reads a fixture), AGENTREPORT @@ -217,7 +219,9 @@ static const char sse_artifact_tool[] = "event: content_block_delta\n" "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":" "{\"type\":\"input_json_delta\",\"partial_json\":" - "\"{\\\"command\\\":\\\"perl -e 'print q(ARTIFACT-MARKER) " + "\"{\\\"command\\\":\\\"perl -e 'print q(ARTIFACT-MARKER); " + "print q(x) x 65521; print pack(q(C*),65,0,255,66); " + "print q(ARTIFACT-MARKER) " "x 24000'\\\"}\"}}\n\n" "event: content_block_stop\n" "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" @@ -858,7 +862,108 @@ send_webfetch(struct tls *conn, int port) send_webfetch_path(conn, port, "web"); } +static int +hex_lower(int ch) +{ + return ((ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f')); +} + static void +output_read_id(const char *req, char id[33]) +{ + static const char marker[] = "\\\"id\\\":\\\""; + const char *p; + size_t i; + + p = strstr(req, marker); + if (p == NULL) + errx(1, "OUTPUTREAD fixture missing Artifact id"); + p += sizeof(marker) - 1; + for (i = 0; i < 32; i++) { + if (!hex_lower((unsigned char)p[i])) + errx(1, "OUTPUTREAD fixture malformed Artifact id"); + id[i] = p[i]; + } + id[32] = '\0'; + if (p[32] != '\\' || p[33] != '"') + errx(1, "OUTPUTREAD fixture unterminated Artifact id"); +} + +/* Recover the opaque id from the preceding generated Artifact result, then + * ask the Coordinator-local output_read tool for an exact binary page. */ +static void +send_output_read(struct tls *conn, const char *req) +{ + char id[33], sse[2048]; + int n; + + output_read_id(req, id); + + n = snprintf(sse, sizeof(sse), + "event: message_start\n" + "data: {\"type\":\"message_start\",\"message\":{\"usage\":" + "{\"input_tokens\":9,\"output_tokens\":1}}}\n\n" + "event: content_block_start\n" + "data: {\"type\":\"content_block_start\",\"index\":0," + "\"content_block\":{\"type\":\"tool_use\",\"id\":" + "\"toolu_output_read\",\"name\":\"output_read\"}}\n\n" + "event: content_block_delta\n" + "data: {\"type\":\"content_block_delta\",\"index\":0," + "\"delta\":{\"type\":\"input_json_delta\"," + "\"partial_json\":\"{\\\"id\\\":\\\"%s\\\"," + "\\\"offset\\\":65536,\\\"limit\\\":4," + "\\\"encoding\\\":\\\"base64\\\"}\"}}\n\n" + "event: content_block_stop\n" + "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" + "event: message_delta\n" + "data: {\"type\":\"message_delta\",\"delta\":{" + "\"stop_reason\":\"tool_use\"},\"usage\":{" + "\"output_tokens\":12}}\n\n" + "event: message_stop\n" + "data: {\"type\":\"message_stop\"}\n\n", id); + if (n < 0 || (size_t)n >= sizeof(sse)) + errx(1, "OUTPUTREAD fixture too large"); + (void)tls_write_all(conn, body_ok, strlen(body_ok)); + write_chunk(conn, sse, (size_t)n); + (void)tls_write_all(conn, "0\r\n\r\n", 5); +} + +/* Exercise Coordinator defaults with id as the only supplied argument. */ +static void +send_output_read_default(struct tls *conn, const char *req) +{ + char id[33], sse[2048]; + int n; + + output_read_id(req, id); + n = snprintf(sse, sizeof(sse), + "event: message_start\n" + "data: {\"type\":\"message_start\",\"message\":{\"usage\":" + "{\"input_tokens\":9,\"output_tokens\":1}}}\n\n" + "event: content_block_start\n" + "data: {\"type\":\"content_block_start\",\"index\":0," + "\"content_block\":{\"type\":\"tool_use\",\"id\":" + "\"toolu_output_read_default\",\"name\":\"output_read\"}}\n\n" + "event: content_block_delta\n" + "data: {\"type\":\"content_block_delta\",\"index\":0," + "\"delta\":{\"type\":\"input_json_delta\"," + "\"partial_json\":\"{\\\"id\\\":\\\"%s\\\"}\"}}\n\n" + "event: content_block_stop\n" + "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" + "event: message_delta\n" + "data: {\"type\":\"message_delta\",\"delta\":{" + "\"stop_reason\":\"tool_use\"},\"usage\":{" + "\"output_tokens\":12}}\n\n" + "event: message_stop\n" + "data: {\"type\":\"message_stop\"}\n\n", id); + if (n < 0 || (size_t)n >= sizeof(sse)) + errx(1, "OUTPUTREADDEFAULT fixture too large"); + (void)tls_write_all(conn, body_ok, strlen(body_ok)); + write_chunk(conn, sse, (size_t)n); + (void)tls_write_all(conn, "0\r\n\r\n", 5); +} + +static void send_wave_webfetch(struct tls *conn, int port, const char *req) { if (strstr(req, "WAVE-PROMPT-A") != NULL) @@ -951,6 +1056,29 @@ static const char sse_bgtool[] = "event: message_stop\n" "data: {\"type\":\"message_stop\"}\n\n"; +/* A short job completes after its launching Turn can commit. A later + * retrying request keeps the Coordinator event loop alive long enough to + * exercise asynchronous current-spool retirement. */ +static const char sse_bgshort[] = +"event: message_start\n" +"data: {\"type\":\"message_start\",\"message\":{\"usage\":" + "{\"input_tokens\":9,\"output_tokens\":1}}}\n\n" +"event: content_block_start\n" +"data: {\"type\":\"content_block_start\",\"index\":0," + "\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_short\"," + "\"name\":\"shell_bg\"}}\n\n" +"event: content_block_delta\n" +"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":" + "{\"type\":\"input_json_delta\",\"partial_json\":" + "\"{\\\"command\\\":\\\"sleep 1; echo short-done\\\"}\"}}\n\n" +"event: content_block_stop\n" +"data: {\"type\":\"content_block_stop\",\"index\":0}\n\n" +"event: message_delta\n" +"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":" + "\"tool_use\"},\"usage\":{\"output_tokens\":12}}\n\n" +"event: message_stop\n" +"data: {\"type\":\"message_stop\"}\n\n"; + /* Probe the exact id allocated by the preceding BGTOOL response. If the * failed turn was rolled back, shell_output returns "no such job 1" and * that tool error appears in the following request. */ @@ -1583,6 +1711,10 @@ main(int argc, char *argv[]) write_chunk(conn, sse_artifact_tool, strlen(sse_artifact_tool)); (void)tls_write_all(conn, "0\r\n\r\n", 5); + } else if (strcmp(argv[i], "OUTPUTREAD") == 0) { + send_output_read(conn, req); + } else if (strcmp(argv[i], "OUTPUTREADDEFAULT") == 0) { + send_output_read_default(conn, req); } else if (strcmp(argv[i], "AGENTCALL") == 0) { (void)tls_write_all(conn, body_ok, strlen(body_ok)); @@ -1922,6 +2054,12 @@ main(int argc, char *argv[]) write_chunk(conn, sse_bgtool, strlen(sse_bgtool)); (void)tls_write_all(conn, "0\r\n\r\n", 5); + } else if (strcmp(argv[i], "BGSHORT") == 0) { + (void)tls_write_all(conn, body_ok, + strlen(body_ok)); + write_chunk(conn, sse_bgshort, + strlen(sse_bgshort)); + (void)tls_write_all(conn, "0\r\n\r\n", 5); } else if (strcmp(argv[i], "ERROR_BG") == 0) { (void)tls_write_all(conn, body_ok, strlen(body_ok)); blob - 2da225f2bbc881261a17aacf484fa9d5c0208709 blob + 31bf64155a0b478d0b4f6aaf0b53133f2934f272 --- src/fugu-artifact/main.c +++ src/fugu-artifact/main.c @@ -15,6 +15,7 @@ #include #include #include +#include #include #include "buf.h" @@ -45,6 +46,24 @@ struct drain_event { static struct artifact_state *cleanup_state; static void +artifact_read_test_delay(void) +{ + const char *value, *errstr; + struct timespec delay; + long long ms; + + if (issetugid() || + (value = getenv("FUGU_ARTIFACT_READ_DELAY_MS")) == NULL) + return; + ms = strtonum(value, 1, 5000, &errstr); + if (errstr != NULL) + return; + delay.tv_sec = (time_t)(ms / 1000); + delay.tv_nsec = (long)(ms % 1000) * 1000000L; + (void)nanosleep(&delay, NULL); +} + +static void drain_unlink(struct drain_event *de) { struct drain_event *item; @@ -256,6 +275,7 @@ artifact_dispatch(struct worker *w, struct imsg *imsg) if (id == 0 || p == NULL || len != sizeof(struct artifact_read_req)) fatalx("fugu-artifact: bad read"); + artifact_read_test_delay(); buf_init(&out); if (artifact_store_read(state->store, (const struct artifact_read_req *)p, &out, &result) == -1)