commit bc50b132dba26f836a2a47e083934d55aa50d9f9 from: Isaac Meerleo date: Sat Jul 18 16:45:36 2026 UTC Prove Artifact cancellation race ordering commit - 31d2568cf13643602ba7bc0dd80bfb086b763bdc commit + bc50b132dba26f836a2a47e083934d55aa50d9f9 blob - a5983bde67263c1d8fe2f3a5033646e035903189 blob + 98e6139ae3aaffc0f149947dcdf6ed08e1dc99ab --- regress/turn/run.sh +++ regress/turn/run.sh @@ -2565,5 +2565,154 @@ else fi stop_stub +# A ROTATE frame is ownership state in its own right. Publish it with a +# blocking selftest send, then cancel before the ordinary Tool result. The +# background job is still pending in this Turn, so rollback must consume the +# frame without trying to release committed job ownership. +rm -rf "$HOME/.fugu" +rm -f "$dir/out" "$dir/errout" "$dir/reqout" +start_stub "BGTOOL JOBOUTPUT" +write_conf "$stubport" +FUGU_SELFTEST=1 FUGU_ARTIFACT_FRAME0_DELAY_MS=3000 \ + FUGU_CONF=$dir/fugu.conf FUGU_LIBEXEC=$dir/libexec/fugu \ + FUGU_CA_FILE=$dir/cert.pem FUGU_RETRY_BASE_MS=50 \ + "$fugu" -d -p "cancel pending Artifact rotation" "$dir/out" 2>"$dir/errout" & +fugupid=$! +i=0 +while ! grep -Fq 'selftest frame0 Artifact rotation published' \ + "$dir/errout" 2>/dev/null && [ "$i" -lt 200 ] && \ + kill -0 "$fugupid" 2>/dev/null; do + sleep 0.05 + i=$((i + 1)) +done +grep -Fq 'selftest frame0 Artifact rotation published' \ + "$dir/errout" 2>/dev/null +reached=$? +if [ "$reached" -eq 0 ]; then + /bin/kill -INT "$fugupid" 2>/dev/null +else + /bin/kill -TERM "$fugupid" 2>/dev/null +fi +wait "$fugupid" +rc=$? +fugupid='' +ok "background rotation cancel: pending frame publishes before Ctrl-C" \ + "$reached" +ok "background rotation cancel: pending Turn exits 3" \ + $([ "$rc" -eq 3 ]; echo $?) +n=$(grep -h -c '"outcome":"abandoned"' \ + "$sessions"/*.ndjson 2>/dev/null | awk '{ n += $1 } END { print n + 0 }') +ok "background rotation cancel: pending Turn is abandoned once" \ + $([ "$n" -eq 1 ]; echo $?) +n=$(grep -h -c '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null | \ + awk '{ n += $1 } END { print n + 0 }') +ok "background rotation cancel: pending job never commits" \ + $([ "$n" -eq 0 ]; echo $?) +n=$(grep -Fc 'selftest frame0 Artifact rotation published' \ + "$dir/errout" 2>/dev/null) +ok "background rotation cancel: pending frame publishes exactly once" \ + $([ "$n" -eq 1 ]; echo $?) +if grep -Eq 'selftest Artifact RELEASE|bad Artifact|mismatched Artifact|stale Artifact|missing cancelled Artifact' \ + "$dir/errout" 2>/dev/null; then + fail "background rotation cancel: pending ownership rolls back cleanly" +else + pass "background rotation cancel: pending ownership rolls back cleanly" +fi +stop_stub + +# The same cancellation after the launch Turn commits must preserve the +# background job but release the frozen side of the published rotation. +# Curses routes Ctrl-C through the normal UI cancellation handshake and exits +# only after Coordinator has settled that ownership frame. +if [ -x "$obj/ptydrive" ]; then + rm -rf "$HOME/.fugu" + rm -f "$dir/reqout" "$dir/reqlog" "$dir/rotation-cursesout" + start_stub "BGTOOL OK JOBOUTPUT" + write_conf "$stubport" + printf 'launch committed background\r\000\000\000poll committed background\r\000\003' | \ + TERM=vt100 HOME=$HOME FUGU_SELFTEST=1 \ + FUGU_ARTIFACT_FRAME0_DELAY_MS=3000 \ + FUGU_CONF=$dir/fugu.conf FUGU_LIBEXEC=$dir/libexec/fugu \ + FUGU_CA_FILE=$dir/cert.pem FUGU_RETRY_BASE_MS=50 \ + "$obj/ptydrive" "$fugu" -d \ + >"$dir/rotation-cursesout" 2>&1 + rc=$? + ok "background rotation cancel: curses exits after Ctrl-C" \ + $([ "$rc" -eq 0 ]; echo $?) + n=$(grep -h -c '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null | \ + awk '{ n += $1 } END { print n + 0 }') + ok "background rotation cancel: launch Turn remains committed" \ + $([ "$n" -eq 1 ]; echo $?) + n=$(grep -h -c '"outcome":"abandoned"' \ + "$sessions"/*.ndjson 2>/dev/null | \ + awk '{ n += $1 } END { print n + 0 }') + ok "background rotation cancel: poll Turn is abandoned once" \ + $([ "$n" -eq 1 ]; echo $?) + n=$(grep -Fc 'selftest frame0 Artifact rotation published' \ + "$dir/rotation-cursesout" 2>/dev/null) + ok "background rotation cancel: committed frame publishes once" \ + $([ "$n" -eq 1 ]; echo $?) + n=$(grep -Fc 'selftest Artifact RELEASE job=1' \ + "$dir/rotation-cursesout" 2>/dev/null) + ok "background rotation cancel: committed frozen spool releases once" \ + $([ "$n" -eq 1 ]; echo $?) + if grep -Eq 'selftest Artifact CANCEL_ACK|bad Artifact|mismatched Artifact|stale Artifact|missing cancelled Artifact' \ + "$dir/rotation-cursesout" 2>/dev/null; then + fail "background rotation cancel: committed ownership settles cleanly" + else + pass "background rotation cancel: committed ownership settles cleanly" + fi + stop_stub +else + fail "background rotation cancel: ptydrive helper not built" +fi + +# Force a completed committed job into normal archival while its Artifact +# terminal is delayed and converted to STORE_ERROR. DROP is already queued +# on the Tool socket when Coordinator sends CANCEL; the ordered Tool output +# must therefore be DROP followed by the exact job-wide CANCEL_ACK. +rm -rf "$HOME/.fugu" +rm -f "$dir/out" "$dir/errout" "$dir/reqout" +start_stub "BGFAST OK RETRY429 OK JOBOUTPUT OK" +write_conf "$stubport" +printf 'launch short background\nwait for store failure\nprobe cancelled background\n/quit\n' \ + > "$dir/lin" +FUGU_SELFTEST=1 FUGU_ARTIFACT_ARCHIVE_COMPLETED=1 \ + FUGU_ARTIFACT_DROP_DELAY_MS=1500 \ + FUGU_ARTIFACT_STORE_ERROR_DELAY_MS=500 \ + FUGU_CONF=$dir/fugu.conf FUGU_LIBEXEC=$dir/libexec/fugu \ + FUGU_CA_FILE=$dir/cert.pem FUGU_RETRY_BASE_MS=50 \ + "$fugu" -d <"$dir/lin" >"$dir/out" 2>"$dir/errout" +rc=$? +ok "background retirement race: session completes" \ + $([ "$rc" -eq 0 ]; echo $?) +n=$(grep -h -c '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null | \ + awk '{ n += $1 } END { print n + 0 }') +ok "background retirement race: all three Turns commit" \ + $([ "$n" -eq 3 ]; echo $?) +grep -q 'shell_output: no such job 1' "$dir/reqout" 2>/dev/null +ok "background retirement race: failed job ownership is gone" $? +drop_line=$(grep -n -m 1 'selftest Artifact DROP queued' \ + "$dir/errout" 2>/dev/null | cut -d: -f1) +error_line=$(grep -n -m 1 'selftest STORE_ERROR injected' \ + "$dir/errout" 2>/dev/null | cut -d: -f1) +ack_line=$(grep -n -m 1 'selftest Artifact CANCEL_ACK job=1' \ + "$dir/errout" 2>/dev/null | cut -d: -f1) +[ -n "$drop_line" ] && [ -n "$error_line" ] && [ -n "$ack_line" ] && \ + [ "$drop_line" -lt "$error_line" ] && [ "$error_line" -lt "$ack_line" ] +ok "background retirement race: DROP precedes STORE_ERROR and CANCEL_ACK" $? +n=$(grep -Ec 'selftest Artifact DROP queued|selftest STORE_ERROR injected|selftest Artifact CANCEL_ACK job=1' \ + "$dir/errout" 2>/dev/null) +ok "background retirement race: each protocol trace occurs exactly once" \ + $([ "$n" -eq 3 ]; echo $?) +if grep -Eq 'bad Artifact|mismatched Artifact|stale Artifact|unexpected Artifact|worker .* exited' \ + "$dir/errout" 2>/dev/null; then + fail "background retirement race: exact-key composition stays valid" +else + pass "background retirement race: exact-key composition stays valid" +fi +stop_stub + echo "$checks checks, $fails failures" [ "$fails" -eq 0 ] blob - fbccbbe31d6cad6698801d3e6f1ba7f266ad0810 blob + 98050bde55120ecfd0bf3740de52ea3a9dcadcb2 --- regress/turn/stub.c +++ regress/turn/stub.c @@ -29,6 +29,7 @@ * 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), + * BGFAST (shell_bg immediate output), * 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 @@ -1079,6 +1080,28 @@ static const char sse_bgshort[] = "event: message_stop\n" "data: {\"type\":\"message_stop\"}\n\n"; +/* Complete comfortably before the retrying next Turn so its current spool + * can be archived without a model-requested shell_output poll. */ +static const char sse_bgfast[] = +"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_fast\"," + "\"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\\\":\\\"echo fast-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. */ @@ -2060,6 +2083,12 @@ main(int argc, char *argv[]) write_chunk(conn, sse_bgshort, strlen(sse_bgshort)); (void)tls_write_all(conn, "0\r\n\r\n", 5); + } else if (strcmp(argv[i], "BGFAST") == 0) { + (void)tls_write_all(conn, body_ok, + strlen(body_ok)); + write_chunk(conn, sse_bgfast, + strlen(sse_bgfast)); + (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 - 31bf64155a0b478d0b4f6aaf0b53133f2934f272 blob + 99c942ff5464de8407bbe7eb2c5268172ce75424 --- src/fugu-artifact/main.c +++ src/fugu-artifact/main.c @@ -46,6 +46,28 @@ struct drain_event { static struct artifact_state *cleanup_state; static void +artifact_store_error_test(struct artifact_terminal *terminal) +{ + static int injected; + const char *value, *errstr; + struct timespec delay; + long long ms; + + if (injected || issetugid() || getenv("FUGU_SELFTEST") == NULL || + (value = getenv("FUGU_ARTIFACT_STORE_ERROR_DELAY_MS")) == NULL) + return; + ms = strtonum(value, 1, 5000, &errstr); + if (errstr != NULL) + return; + injected = 1; + delay.tv_sec = (time_t)(ms / 1000); + delay.tv_nsec = (long)(ms % 1000) * 1000000L; + (void)nanosleep(&delay, NULL); + log_warnx("fugu-artifact: selftest STORE_ERROR injected"); + terminal->status = ARTIFACT_STORE_ERROR; +} + +static void artifact_read_test_delay(void) { const char *value, *errstr; @@ -111,6 +133,7 @@ drain_stream(int fd, short events, void *arg) rc = artifact_pending_drain(de->pending, &terminal); if (rc == 0) return; + artifact_store_error_test(&terminal); event_del(&de->ev); drain_unlink(de); imsgev_send(&de->state->worker->iev, FUGU_IMSG_ARTIFACT_TERMINAL, blob - 690148883bf6750cb315d158868210944f9fb89a blob + ba537a15eb22ee38f21810488368ceb153bbbf93 --- src/fugu-tool/job.c +++ src/fugu-tool/job.c @@ -407,6 +407,14 @@ job_set_cap(struct joblist *jl, int cap) jl->record_limit = (size_t)cap + JOB_COMPLETED_RECORD_MAX; } +void +job_selftest_set_ownership_limit(struct joblist *jl, size_t limit) +{ + if (jl == NULL || limit == 0 || limit > JOB_OWNERSHIP_MAX) + abort(); + jl->ownership_limit = limit; +} + static struct job * job_find(struct joblist *jl, int id) { @@ -545,6 +553,23 @@ job_prune_completed_one(struct joblist *jl) (*oldest_link)->artifact_managed)); } +int +job_selftest_archive_completed(struct joblist *jl) +{ + struct job *j; + + if (jl == NULL) + abort(); + for (j = jl->head; j != NULL; j = j->next) { + if (j->pending || j->artifact_settling || j->fd != -1 || + j->artifact_fd != -1 || j->pending_write.len != 0) + continue; + job_stop(j); + break; + } + return (job_prune_completed_one(jl)); +} + static int job_make_record_room(struct joblist *jl) { blob - 796e2ecc1bb8cb70f469ead0ccee52e97306427a blob + 8e3b9e77a89f6c8b6d7171161b8dbd1721937fad --- src/fugu-tool/job.h +++ src/fugu-tool/job.h @@ -40,6 +40,9 @@ void job_list_free(struct joblist *); /* the max_bg_jobs gate; 0 (the default here) means unconfigured */ void job_set_cap(struct joblist *, int cap); +/* Regression-only bound selected by fugu-tool under guarded selftest mode. */ +void job_selftest_set_ownership_limit(struct joblist *, size_t); +int job_selftest_archive_completed(struct joblist *); void job_start(struct joblist *, struct tool_ctx *, const struct json *, int obj, struct buf *out, int *is_error); blob - 89929b783364ee5513ff07fe97e5a87e7337572c blob + 20e8d5ffe78a9765287b58c24e371c50dba899ef --- src/fugu-tool/main.c +++ src/fugu-tool/main.c @@ -39,6 +39,7 @@ #include #include #include +#include #include #include "log.h" @@ -54,6 +55,11 @@ #include "artifact_scope.h" static int readonly; /* -r: the subagent role, always read-only */ +static long long selftest_rotation_delay_ms; +static long long selftest_drop_delay_ms; +static int selftest_archive_pending; +static int selftest_rotation_trace; +static int selftest_drop_trace; struct tool_state { struct worker *worker; @@ -75,7 +81,50 @@ struct tool_state { * a clean shutdown that would otherwise orphan them into init. */ static struct joblist *jobs_atexit; +static long long +selftest_delay(const char *name) +{ + const char *value, *errstr; + long long ms; + + if (issetugid() || getenv("FUGU_SELFTEST") == NULL || + (value = getenv(name)) == NULL) + return (0); + ms = strtonum(value, 1, 5000, &errstr); + if (errstr != NULL) + fatalx("fugu-tool: invalid %s", name); + return (ms); +} + static int +selftest_flag(const char *name) +{ + const char *value; + + if (issetugid() || getenv("FUGU_SELFTEST") == NULL || + (value = getenv(name)) == NULL) + return (0); + if (strcmp(value, "1") != 0) + fatalx("fugu-tool: invalid %s", name); + return (1); +} + +static void +selftest_pause(long long *milliseconds, const char *marker) +{ + struct timespec delay; + long long ms; + + if ((ms = *milliseconds) == 0) + return; + *milliseconds = 0; + log_warnx("%s", marker); + delay.tv_sec = (time_t)(ms / 1000); + delay.tv_nsec = (long)(ms % 1000) * 1000000L; + (void)nanosleep(&delay, NULL); +} + +static int spool_drop_id_valid(const char id[FUGU_ARTIFACT_ID_BUFSZ]) { static const char zero[FUGU_ARTIFACT_ID_BUFSZ]; @@ -102,6 +151,8 @@ send_spool_drops(struct worker *w, struct tool_state * fatalx("fugu-tool: invalid Artifact spool drop"); imsgev_send(&w->iev, FUGU_IMSG_TOOL_ARTIFACT_DROP, correlation, &drop, sizeof(drop)); + selftest_pause(&selftest_drop_delay_ms, + "fugu-tool: selftest Artifact DROP queued"); } } @@ -135,6 +186,9 @@ pump_jobs(int fd, short ev, void *arg) (void)ev; if (ts->ctx != NULL) { job_pump(ts->ctx->jobs); + if (selftest_archive_pending && + job_selftest_archive_completed(ts->ctx->jobs) == 1) + selftest_archive_pending = 0; send_spool_drops(ts->worker, ts); } } @@ -377,6 +431,8 @@ tool_lockdown(struct worker *w, struct imsg *imsg) ts->ctx = tool_ctx_new(write_ok, net_ok); ts->worker = w; job_set_cap(ts->ctx->jobs, (int)ts->cap); + if (selftest_rotation_delay_ms > 0) + job_selftest_set_ownership_limit(ts->ctx->jobs, 1); /* kill our background jobs when we exit, and keep their output * draining between polls (event_init ran before this dispatch) */ @@ -439,6 +495,13 @@ send_spool_rotation(struct worker *w, const struct too if (ts->ctx->artifact_spool_adopted && ts->ctx->artifact_available) strlcpy(rotate.replacement_id, ts->artifact.id, sizeof(rotate.replacement_id)); + if (selftest_rotation_delay_ms > 0) { + imsgev_send_blocking(&w->iev, FUGU_IMSG_TOOL_ARTIFACT_ROTATE, id, + &rotate, sizeof(rotate)); + selftest_pause(&selftest_rotation_delay_ms, + "fugu-tool: selftest frame0 Artifact rotation published"); + return; + } imsgev_send(&w->iev, FUGU_IMSG_TOOL_ARTIFACT_ROTATE, id, &rotate, sizeof(rotate)); } @@ -666,6 +729,12 @@ tool_dispatch(struct worker *w, struct imsg *imsg) sizeof(ack.artifact_id)); imsgev_send(&w->iev, FUGU_IMSG_TOOL_ARTIFACT_CANCEL_ACK, job_id, &ack, sizeof(ack)); + if (selftest_drop_trace) + log_warnx("fugu-tool: selftest Artifact CANCEL_ACK " + "job=%u", job_id); + } else if (selftest_rotation_trace) { + log_warnx("fugu-tool: selftest Artifact RELEASE job=%u", + job_id); } return (0); } @@ -837,6 +906,13 @@ main(int argc, char *argv[]) buf_init(&ts.arg); ts.artifact_fd = -1; agent_boot_init(&ts.boot); + selftest_rotation_delay_ms = selftest_delay( + "FUGU_ARTIFACT_FRAME0_DELAY_MS"); + selftest_drop_delay_ms = selftest_delay("FUGU_ARTIFACT_DROP_DELAY_MS"); + selftest_archive_pending = selftest_flag( + "FUGU_ARTIFACT_ARCHIVE_COMPLETED"); + selftest_rotation_trace = selftest_rotation_delay_ms > 0; + selftest_drop_trace = selftest_drop_delay_ms > 0; /* * Transitional pledge. It is a superset of the steady promises;