test(scripts): mollifier challenge suite for manual API verification

14 scenario scripts covering every customer-visible behaviour added in
Phases A-E. Each script is independently runnable; bash + curl + jq
only (script 12 also needs redis-cli; script 14 also needs a session
cookie + slugs).

Coverage map:
- 01 burst baseline + sanity-check
- 02 every read endpoint on a buffered run (A1-A6)
- 03 each mutation patch type with read-back (B3, C2, C3 replace+ops, C4)
- 04 idempotency SETNX collision in burst (B6a)
- 05 drainer round-trip — mutations survive materialisation (B2 + drainer)
- 06 cancel bifurcation — buffered cancel → CANCELED PG row (C1, Q4)
- 07 replay creates fresh distinct runId, original untouched (C5)
- 08 listing merges buffered + PG, sorts correctly, pagination dedupes (E)
- 09 50-way concurrent metadata increments — CAS retry loop (C3 atomicity)
- 10 idempotency-key reset clears both stores (B6b reset)
- 11 parent/root metadata operations fan out from buffered child (C3)
- 12 state-3 replay (Q2 — direct Redis HSET status=FAILED)
- 13 triggerAndWait + idempotencyKey skips buffer lookup (B6b guard)
- 14 dashboard cancel/replay/idempotencyKey-reset via session cookies (D1-D3)

README documents prerequisites, run order, drainer-on vs drainer-off
scripts, and an explicit "not covered" list (busy path, buffer outage,
forward-compat skew, CI invocation) with rationale.

Author: d-cs

scripts/mollifier-challenge/00-lib.sh

@@ -0,0 +1,132 @@
+#!/usr/bin/env bash
+# Shared helpers for the mollifier challenge suite. Source this from each
+# scenario script: `source "$(dirname "$0")/00-lib.sh"`.
+
+set -uo pipefail
+
+: "${API_BASE:=http://localhost:3030}"
+: "${TASK_ID:=hello-world}"
+: "${BURST_SIZE:=30}"
+: "${VERBOSE:=0}"
+
+if [[ -z "${API_KEY:-}" ]]; then
+  echo "ERROR: API_KEY env var is required" >&2
+  exit 2
+fi
+
+if ! command -v jq >/dev/null 2>&1; then
+  echo "ERROR: jq is required" >&2
+  exit 2
+fi
+
+if [[ -t 1 ]]; then
+  C_OK=$'\033[32m'; C_FAIL=$'\033[31m'; C_WARN=$'\033[33m'
+  C_DIM=$'\033[2m'; C_BOLD=$'\033[1m'; C_RESET=$'\033[0m'
+else
+  C_OK=; C_FAIL=; C_WARN=; C_DIM=; C_BOLD=; C_RESET=
+fi
+
+# Per-script work directory, auto-cleaned on exit.
+WORK=$(mktemp -d)
+trap 'rm -rf "$WORK"' EXIT
+
+# pass_count + fail_count accumulators. Use `pass`, `fail`, and `summary`.
+PASS_COUNT=0
+FAIL_COUNT=0
+declare -a FAILURES=()
+
+pass() {
+  printf "  %s✓%s %s\n" "$C_OK" "$C_RESET" "$1"
+  PASS_COUNT=$((PASS_COUNT + 1))
+}
+
+fail() {
+  printf "  %s✗%s %s\n" "$C_FAIL" "$C_RESET" "$1"
+  FAILURES+=( "$1" )
+  FAIL_COUNT=$((FAIL_COUNT + 1))
+}
+
+info() {
+  printf "  %s%s%s\n" "$C_DIM" "$1" "$C_RESET"
+}
+
+header() {
+  printf "\n%s==>%s %s%s%s\n" "$C_DIM" "$C_RESET" "$C_BOLD" "$1" "$C_RESET"
+}
+
+summary() {
+  printf "\n%s==>%s Summary\n" "$C_DIM" "$C_RESET"
+  printf "  passed: %d\n" "$PASS_COUNT"
+  if (( FAIL_COUNT > 0 )); then
+    printf "  %sfailed: %d%s\n" "$C_FAIL" "$FAIL_COUNT" "$C_RESET"
+    for f in "${FAILURES[@]}"; do
+      printf "    %s- %s%s\n" "$C_FAIL" "$f" "$C_RESET"
+    done
+    exit 1
+  fi
+  printf "  %sall scenarios pass%s\n" "$C_OK" "$C_RESET"
+  exit 0
+}
+
+# api METHOD PATH [DATA] → echoes "STATUS BODY"
+# Stores body in $WORK/last.body, status in $WORK/last.status.
+api() {
+  local method=$1 path=$2 data=${3:-}
+  local body_file=$WORK/last.body
+  local status_file=$WORK/last.status
+  local args=( -s -o "$body_file" -w "%{http_code}" -X "$method"
+    -H "Authorization: Bearer $API_KEY" )
+  if [[ -n "$data" ]]; then
+    args+=( -H "Content-Type: application/json" -d "$data" )
+  fi
+  args+=( "$API_BASE$path" )
+  local status
+  status=$(curl "${args[@]}")
+  echo "$status" > "$status_file"
+  if [[ "$VERBOSE" == "1" ]]; then
+    info "$method $path → $status"
+    info "  $(head -c 200 "$body_file")"
+  fi
+  printf "%s" "$status"
+}
+
+# Returns 0 if last status is 2xx.
+last_status_ok() {
+  [[ "$(cat "$WORK/last.status" 2>/dev/null)" =~ ^2 ]]
+}
+
+# Read last body or empty.
+last_body() {
+  cat "$WORK/last.body" 2>/dev/null || echo ""
+}
+
+# Returns 0 if the body matches a jq filter.
+body_matches() {
+  local filter=$1
+  jq -e "$filter" "$WORK/last.body" >/dev/null 2>&1
+}
+
+# Trigger a burst, return one buffered runId on stdout (or empty if none).
+# Side effect: also writes burst responses to $WORK/burst/.
+capture_buffered_run_id() {
+  local task=${1:-$TASK_ID}
+  local size=${2:-$BURST_SIZE}
+  local payload=${3:-'{"message":"burst"}'}
+  local burst_dir=$WORK/burst
+  mkdir -p "$burst_dir"
+  for i in $(seq 1 "$size"); do
+    curl -s -X POST \
+      -H "Authorization: Bearer $API_KEY" \
+      -H "Content-Type: application/json" \
+      -d "{\"payload\":$payload}" \
+      "$API_BASE/api/v1/tasks/$task/trigger" \
+      -o "$burst_dir/$i.json" &
+  done
+  wait
+  for f in "$burst_dir"/*.json; do
+    if jq -e '.notice.code == "mollifier.queued"' "$f" >/dev/null 2>&1; then
+      jq -r '.id' "$f"
+      return 0
+    fi
+  done
+}

scripts/mollifier-challenge/01-burst-baseline.sh

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+# 01 — fire a burst, confirm the gate mollifies at least one trigger,
+# capture the buffered runId, sanity-check the response shape.
+# Required: drainer OFF.
+
+source "$(dirname "$0")/00-lib.sh"
+
+header "Burst baseline"
+
+info "firing $BURST_SIZE concurrent triggers against $TASK_ID"
+BUFFERED_ID=$(capture_buffered_run_id)
+
+if [[ -z "$BUFFERED_ID" ]]; then
+  fail "no mollifier.queued response across $BURST_SIZE triggers"
+  info "check: TRIGGER_MOLLIFIER_ENABLED=1, org flag on, threshold low, drainer OFF"
+  summary
+fi
+pass "captured buffered runId: $BUFFERED_ID"
+
+# Inspect via /api/v3/runs/{id} — should resolve via the buffer read-fallback
+# even though the run isn't in PG.
+api GET "/api/v3/runs/$BUFFERED_ID"
+if last_status_ok; then
+  pass "retrieve returns 2xx for the buffered run"
+else
+  fail "retrieve returned $(cat "$WORK/last.status") (expected 2xx)"
+fi
+
+if body_matches '.id == "'"$BUFFERED_ID"'"'; then
+  pass "retrieve body carries the right runId"
+else
+  fail "retrieve body missing runId"
+fi
+
+if body_matches '.status == "PENDING" or .status == "QUEUED" or .status == "DELAYED"'; then
+  pass "retrieve status is QUEUED-equivalent ($(last_body | jq -r .status))"
+else
+  fail "retrieve status unexpected: $(last_body | jq -r .status)"
+fi
+
+# Sanity: control trigger with a long delay should be in PG, not mollified.
+header "Control sanity"
+api POST "/api/v1/tasks/$TASK_ID/trigger" '{"payload":{"control":true},"options":{"delay":"10m"}}'
+if last_status_ok; then
+  CONTROL_ID=$(last_body | jq -r '.id')
+  if [[ -n "$CONTROL_ID" && "$CONTROL_ID" != "null" ]]; then
+    if last_body | jq -e '.notice.code == "mollifier.queued"' >/dev/null 2>&1; then
+      fail "control trigger with delay was mollified — check threshold / hold settings"
+    else
+      pass "control trigger landed in PG (delayed), runId: $CONTROL_ID"
+    fi
+  else
+    fail "control trigger response missing id"
+  fi
+else
+  fail "control trigger returned $(cat "$WORK/last.status")"
+fi
+
+summary

scripts/mollifier-challenge/02-reads-on-buffered.sh

@@ -0,0 +1,85 @@
+#!/usr/bin/env bash
+# 02 — read endpoints all behave correctly on a buffered run.
+# Required: drainer OFF.
+
+source "$(dirname "$0")/00-lib.sh"
+
+header "Read endpoints on a buffered run"
+
+BUFFERED_ID=$(capture_buffered_run_id)
+if [[ -z "$BUFFERED_ID" ]]; then
+  fail "could not buffer a run (rerun 01 to debug)"
+  summary
+fi
+info "using buffered runId: $BUFFERED_ID"
+
+# /api/v3/runs/{id}
+api GET "/api/v3/runs/$BUFFERED_ID"
+if last_status_ok && body_matches '.id and .taskIdentifier and .status'; then
+  pass "GET /api/v3/runs/{id} — 2xx with id+taskIdentifier+status"
+else
+  fail "GET /api/v3/runs/{id} — status=$(cat "$WORK/last.status") body=$(last_body | head -c 100)"
+fi
+
+# /api/v1/runs/{id}/trace
+api GET "/api/v1/runs/$BUFFERED_ID/trace"
+if last_status_ok && body_matches '.trace and .trace.traceId'; then
+  pass "GET /trace — 2xx with trace.traceId"
+else
+  fail "GET /trace — status=$(cat "$WORK/last.status") body=$(last_body | head -c 100)"
+fi
+
+# /api/v1/runs/{id}/events
+api GET "/api/v1/runs/$BUFFERED_ID/events"
+if last_status_ok && body_matches '.events | type == "array"'; then
+  pass "GET /events — 2xx, events is an array"
+else
+  fail "GET /events — status=$(cat "$WORK/last.status") body=$(last_body | head -c 100)"
+fi
+
+# /api/v1/runs/{id}/attempts
+api GET "/api/v1/runs/$BUFFERED_ID/attempts"
+if last_status_ok && body_matches '.attempts | type == "array" and length == 0'; then
+  pass "GET /attempts — 2xx, attempts is empty array"
+else
+  fail "GET /attempts — status=$(cat "$WORK/last.status") body=$(last_body | head -c 100)"
+fi
+
+# /api/v1/runs/{id}/metadata (loader)
+api GET "/api/v1/runs/$BUFFERED_ID/metadata"
+if last_status_ok && body_matches 'has("metadata") and has("metadataType")'; then
+  pass "GET /metadata — 2xx with { metadata, metadataType }"
+else
+  fail "GET /metadata — status=$(cat "$WORK/last.status") body=$(last_body | head -c 100)"
+fi
+
+# /api/v1/runs/{id}/result — expected 404 (run not finished)
+api GET "/api/v1/runs/$BUFFERED_ID/result"
+status=$(cat "$WORK/last.status")
+if [[ "$status" == "404" ]]; then
+  pass "GET /result — 404 (run not finished, expected contract)"
+else
+  fail "GET /result — expected 404, got $status"
+fi
+
+# Spans endpoint — buffered run only has the queued span; 404 for any other.
+SPAN_ID=$(api GET "/api/v3/runs/$BUFFERED_ID" >/dev/null; last_body | jq -r '.spanId // empty')
+if [[ -n "$SPAN_ID" ]]; then
+  api GET "/api/v1/runs/$BUFFERED_ID/spans/$SPAN_ID"
+  if last_status_ok; then
+    pass "GET /spans/{spanId} — 2xx for the queued span"
+  else
+    fail "GET /spans/{spanId} — expected 2xx, got $(cat "$WORK/last.status")"
+  fi
+
+  api GET "/api/v1/runs/$BUFFERED_ID/spans/nonexistent_span_xyz"
+  if [[ "$(cat "$WORK/last.status")" == "404" ]]; then
+    pass "GET /spans/{unknown} — 404"
+  else
+    fail "GET /spans/{unknown} — expected 404, got $(cat "$WORK/last.status")"
+  fi
+else
+  info "skipping spans probe — no spanId on retrieve response"
+fi
+
+summary

scripts/mollifier-challenge/03-mutations-on-buffered.sh

@@ -0,0 +1,87 @@
+#!/usr/bin/env bash
+# 03 — each mutation lands on the snapshot (verified by follow-up read).
+# Cancel is left for 06-cancel-bifurcation.sh because it terminates the
+# snapshot. Required: drainer OFF.
+
+source "$(dirname "$0")/00-lib.sh"
+
+header "Mutations land on the buffered snapshot"
+
+BUFFERED_ID=$(capture_buffered_run_id)
+if [[ -z "$BUFFERED_ID" ]]; then
+  fail "could not buffer a run"
+  summary
+fi
+info "using buffered runId: $BUFFERED_ID"
+
+# --- tags ---
+header "tags-add → readback"
+api POST "/api/v1/runs/$BUFFERED_ID/tags" '{"tags":["challenge-tag-a","challenge-tag-b"]}'
+if last_status_ok; then
+  pass "POST /tags returned 2xx"
+else
+  fail "POST /tags status=$(cat "$WORK/last.status")"
+fi
+api GET "/api/v3/runs/$BUFFERED_ID"
+if body_matches '.runTags // [] | (any(. == "challenge-tag-a") and any(. == "challenge-tag-b"))'; then
+  pass "retrieve shows both new tags on the snapshot"
+else
+  fail "retrieve runTags=$(last_body | jq -c '.runTags // []')"
+fi
+
+# Idempotent dedup
+api POST "/api/v1/runs/$BUFFERED_ID/tags" '{"tags":["challenge-tag-a"]}'
+api GET "/api/v3/runs/$BUFFERED_ID"
+tag_count=$(last_body | jq '.runTags // [] | map(select(. == "challenge-tag-a")) | length')
+if [[ "$tag_count" == "1" ]]; then
+  pass "duplicate tag deduplicated by mutateSnapshot Lua"
+else
+  fail "duplicate tag landed $tag_count times (expected 1)"
+fi
+
+# --- metadata-put replace ---
+header "metadata-put (replace) → readback"
+api PUT "/api/v1/runs/$BUFFERED_ID/metadata" '{"metadata":{"phase":"challenge","attempt":1}}'
+if last_status_ok; then
+  pass "PUT /metadata returned 2xx"
+else
+  fail "PUT /metadata status=$(cat "$WORK/last.status") body=$(last_body | head -c 200)"
+fi
+api GET "/api/v1/runs/$BUFFERED_ID/metadata"
+if body_matches '(.metadata // "" | tostring) | (contains("\"phase\":\"challenge\"") and contains("\"attempt\":1"))'; then
+  pass "GET /metadata reflects PUT"
+else
+  fail "metadata readback=$(last_body | head -c 200)"
+fi
+
+# --- metadata-put operations (increment) ---
+header "metadata operations (increment) → readback"
+api PUT "/api/v1/runs/$BUFFERED_ID/metadata" \
+  '{"operations":[{"type":"increment","key":"counter","value":5}]}'
+if last_status_ok; then
+  pass "PUT /metadata (increment by 5) returned 2xx"
+else
+  fail "PUT /metadata increment status=$(cat "$WORK/last.status") body=$(last_body | head -c 200)"
+fi
+api PUT "/api/v1/runs/$BUFFERED_ID/metadata" \
+  '{"operations":[{"type":"increment","key":"counter","value":3}]}'
+api GET "/api/v1/runs/$BUFFERED_ID/metadata"
+if body_matches '(.metadata // "" | tostring) | contains("\"counter\":8")'; then
+  pass "two increments produce counter=8 (CAS retry not losing deltas)"
+else
+  fail "counter after 5+3 = $(last_body | head -c 200)"
+fi
+
+# --- reschedule ---
+header "reschedule → readback"
+api POST "/api/v1/runs/$BUFFERED_ID/reschedule" '{"delay":"10m"}'
+if last_status_ok; then
+  pass "POST /reschedule returned 2xx"
+else
+  fail "POST /reschedule status=$(cat "$WORK/last.status") body=$(last_body | head -c 200)"
+fi
+# Reschedule applies set_delay on the snapshot — no direct read-back via
+# the public API (the snapshot delay is internal until materialise).
+# This is by design; we accept the 2xx as the contract here.
+
+summary