commit - 2440fa356713a6a5e24841131db5ecafa143a0ed
commit + d120f420c9b56088098e2f370bc6e1919ca8564d
blob - db3d3686c376acc3355626e9a0e505ae5ad71eaa
blob + a5983bde67263c1d8fe2f3a5033646e035903189
--- regress/turn/run.sh
+++ regress/turn/run.sh
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"
}
# --- 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" \
$([ $? -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" </dev/null \
+ >"$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.
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
* 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
"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"
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)
"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. */
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));
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
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
+#include <time.h>
#include <unistd.h>
#include "buf.h"
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;
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)
