#!/usr/bin/env bash
# Runtime capture driver for PR #23579 (webhook deliveries survive restart).
# Usage: driver.sh <checkout> <home-dir> <port> <row: R1|R2|R3> [label]
#   Rows R1 and R2 also run R4 (replay after the restart) at the end.
# Prints a transcript to stdout. Writes only under <home-dir> and /tmp/ev-23579/capture.
# Isolation: the server runs as `node apps/server/src/bin.ts --base-dir <home-dir>`. bin.ts has no
# --home-dir/--no-seed flags (those belong to `vp run dev`); a bare server never seeds from ~/.t3.
set -uo pipefail
for v in $(env | grep ^T3CODE_ | cut -d= -f1); do unset "$v"; done
export PATH="$HOME/.bun/bin:$HOME/.local/bin:$HOME/.local/share/vite-plus/bin:$PATH"

CHECKOUT=$1; HOME_DIR=$2; PORT=$3; ROW=$4; LABEL=${5:-$(basename "$CHECKOUT")}
CAP=/tmp/ev-23579/capture
case "$HOME_DIR" in /tmp/ev-23579/*) ;; *) echo "refusing home outside /tmp/ev-23579" >&2; exit 2 ;; esac
case "$PORT" in 479[1-2][0-9]) ;; *) echo "port must be 47910-47929" >&2; exit 2 ;; esac
rm -rf "$HOME_DIR"; mkdir -p "$HOME_DIR"
DB=$HOME_DIR/userdata/statev2.sqlite
PROJ=$HOME_DIR/proj; ORIGIN=$HOME_DIR/proj-origin.git; FLAG=$HOME_DIR/hold.flag
ORIGIN_URL=http://127.0.0.1:$PORT
STARTED_PIDS=(); SERVER_PID=

ts() { date -u +%FT%T.%3NZ; }
say() { echo "[$(ts)] $*"; }
q() { sqlite3 -readonly -header -column "$DB" "$@" 2>&1; }

cleanup() {
  if [ -n "$SERVER_PID" ] && kill -0 "$SERVER_PID" 2>/dev/null; then
    kill -9 "$SERVER_PID"; say "cleanup: SIGKILL server pid $SERVER_PID"
  fi
  rm -f "$FLAG"
}
trap cleanup EXIT

start_server() {
  local log=$HOME_DIR/server-$1.log
  (cd "$HOME_DIR" && PATH="$CAP/gitwrap:$PATH" EV_HOLD_FLAG="$FLAG" EV_HOLD_ROOT="$(cd "$PROJ" && pwd -P)" exec node "$CHECKOUT/apps/server/src/bin.ts" --base-dir "$HOME_DIR" --port "$PORT" \
      --host 127.0.0.1 --no-browser > "$log" 2>&1) &
  SERVER_PID=$!; STARTED_PIDS+=("$SERVER_PID")
  say "started server pid=$SERVER_PID ($1) log=$log"
  for _ in $(seq 120); do grep -q "Listening on" "$log" && break; sleep 0.5; done
  grep -q "Listening on" "$log" || { say "server did not start"; tail -30 "$log"; exit 1; }
  for _ in $(seq 60); do grep -q "orchestration recovery completed" "$log" && break; sleep 0.5; done
  say "server listening on $ORIGIN_URL"
}
sigkill_server() {
  kill -9 "$SERVER_PID"; wait "$SERVER_PID" 2>/dev/null
  say "SIGKILL server pid=$SERVER_PID (exited: $(kill -0 "$SERVER_PID" 2>/dev/null && echo no || echo yes))"
  SERVER_PID=
}
stop_server() {
  kill "$SERVER_PID"; for _ in $(seq 20); do kill -0 "$SERVER_PID" 2>/dev/null || break; sleep 0.5; done
  kill -0 "$SERVER_PID" 2>/dev/null && kill -9 "$SERVER_PID"
  wait "$SERVER_PID" 2>/dev/null; say "stopped server pid=$SERVER_PID"; SERVER_PID=
}

MINT_DIR=$HOME_DIR/mint
mint() {
  mkdir -p "$MINT_DIR"; cp "$CAP/mint-mcp-token.mjs" "$MINT_DIR/"
  ln -sfn "$CHECKOUT/apps/server/node_modules" "$MINT_DIR/node_modules"
  TOKEN=$(node "$MINT_DIR/mint-mcp-token.mjs" "$CHECKOUT" "$HOME_DIR" 2>"$HOME_DIR/mint.err" | tail -1)
  [ ${#TOKEN} -gt 40 ] || { say "token mint failed"; cat "$HOME_DIR/mint.err"; exit 1; }
}
mcp() { # mcp <tool> <json>
  (cd "$CHECKOUT/apps/server" && T3_ACP_MCP_ENDPOINT=$ORIGIN_URL/mcp T3_ACP_MCP_AUTHORIZATION="Bearer $TOKEN" \
    node src/bin.ts acp-mcp-call "$1" "$2")
}
post() { # post <tag> <json-body>
  local out; out=$(curl -s -m 30 -w ' HTTP=%{http_code}' -X POST -H 'content-type: application/json' \
    --data "$2" "$ORIGIN_URL/api/hooks/$TASK_ID/$HOOK_TOKEN")
  say "POST /api/hooks/<task>/<token> [$1] body=$2 -> $out"
}

dump() {
  say "---- DB dump: $1 ----"
  echo "# scheduled_task_webhook_deliveries"
  q "select delivery_id, received_at, outcome, error, json_extract(body,'$.n') as n from scheduled_task_webhook_deliveries order by received_at"
  echo "# scheduled_task_webhook_pending"
  q "select delivery_id, coalesce_key, due_at, state, delivery_ids_json from scheduled_task_webhook_pending order by due_at"
  echo "# scheduled_task_webhook_replay_keys"
  q "select * from scheduled_task_webhook_replay_keys" | cut -c1-200
  echo "# threads launched for the task's project (user message -> thread -> run status)"
  q "select substr(m.message_id, instr(m.message_id,'webhook:')+8) as delivery, m.thread_id, r.status as run_status, r.requested_at, r.completed_at
     from orchestration_v2_projection_messages m left join orchestration_v2_projection_runs r on r.run_id = m.run_id
     where m.role='user' and m.thread_id in (select thread_id from orchestration_v2_projection_threads where project_id='$PROJECT_ID') order by r.requested_at"
  echo "# threads per marker (each delivery must map to exactly one thread)"
  q "select json_extract(d.body,'$.n') as n, d.delivery_id,
       (select count(distinct m.thread_id) from orchestration_v2_projection_messages m
        where m.role='user' and m.message_id like '%webhook:' || d.delivery_id) as threads_for_delivery,
       (select count(distinct m.thread_id) from orchestration_v2_projection_messages m
        where m.role='user' and m.payload_json like '%marker ' || json_extract(d.body,'$.n') || '%') as threads_with_marker
     from scheduled_task_webhook_deliveries d order by d.received_at"
  echo "# project thread total: $(sqlite3 -readonly "$DB" "select count(*) from orchestration_v2_projection_threads where project_id='$PROJECT_ID'")"
  echo "# task: $(sqlite3 -readonly "$DB" "select last_run_status||' runs='||run_count||' err='||ifnull(last_run_error,'') from scheduled_tasks where task_id='$TASK_ID'")"
}
run_status_for() { # delivery id -> run status or empty
  sqlite3 -readonly "$DB" "select r.status from orchestration_v2_projection_messages m join orchestration_v2_projection_runs r on r.run_id=m.run_id where m.message_id like '%webhook:$1' limit 1" 2>/dev/null
}
wait_settled() { # wait_settled <max-seconds> <delivery ids...>
  local max=$1; shift; local t0=$SECONDS
  while [ $((SECONDS - t0)) -lt "$max" ]; do
    local done_all=1
    for d in "$@"; do
      case "$(run_status_for "$d")" in completed|failed|interrupted|cancelled) ;; *) done_all=0 ;; esac
    done
    [ $done_all = 1 ] && { say "settled after $((SECONDS - t0))s: all runs terminal"; return 0; }
    sleep 2
  done
  say "NOT settled after ${max}s"; for d in "$@"; do say "  $d run_status='$(run_status_for "$d")'"; done
}
delivery_id_of() { sqlite3 -readonly "$DB" "select delivery_id from scheduled_task_webhook_deliveries where json_extract(body,'$.n')='$1' order by received_at limit 1"; }

say "=== PR #23579 capture: row=$ROW side=$LABEL checkout=$CHECKOUT sha=$(git -C "$CHECKOUT" rev-parse HEAD)"
say "env: $(uname -sr) node=$(node --version) home=$HOME_DIR port=$PORT"

# Scratch project: a git repo with a local bare origin (task worktrees start from origin/main).
git init -q -b main "$PROJ" && echo ev > "$PROJ/README" && git -C "$PROJ" add README \
  && git -C "$PROJ" -c user.email=ev@x -c user.name=ev commit -q -m init
git clone -q --bare "$PROJ" "$ORIGIN" && git -C "$PROJ" remote add origin "$ORIGIN" && git -C "$PROJ" fetch -q origin

start_server first
mint
PROJECT_ID=$(mcp t3_project_create "{\"title\":\"ev-$ROW\",\"workspaceRoot\":\"$PROJ\"}" | node -pe 'JSON.parse(require("fs").readFileSync(0)).structuredContent.id')
[ -n "$PROJECT_ID" ] && [ "$PROJECT_ID" != undefined ] || { say "project create failed"; exit 1; }
mcp t3_project_update "$(node -e 'console.log(JSON.stringify({projectId:process.argv[1],defaultModelSelection:{instanceId:"claudeAgent",model:"claude-haiku-5-5"}}))' "$PROJECT_ID")" > /dev/null
PROMPT='Webhook marker {{body.n}}. Reply with the single word OK. Do not use any tools and do not edit files.'
if [ "$ROW" = R3 ]; then
  DEDUPE='{"coalesceKey":"{{body.k}}","coalesceWindowSeconds":10}'
else
  DEDUPE='{"replayKey":"{{body.id}}"}'
fi
TASK_JSON=$(mcp schedule_task "$(node -e 'console.log(JSON.stringify({projectId:process.argv[1],title:"ev task",prompt:process.argv[2],schedule:{type:"webhook",dedupe:JSON.parse(process.argv[3])},bindToCurrentThread:false}))' "$PROJECT_ID" "$PROMPT" "$DEDUPE")")
TASK_ID=$(echo "$TASK_JSON" | node -pe 'JSON.parse(require("fs").readFileSync(0)).structuredContent.scheduledTaskId')
[ -n "$TASK_ID" ] && [ "$TASK_ID" != undefined ] || { say "schedule_task failed: $TASK_JSON"; exit 1; }
HOOK_TOKEN=$(sqlite3 -readonly "$DB" "select webhook_token from scheduled_tasks where task_id='$TASK_ID'")
say "project=$PROJECT_ID task=$TASK_ID schedule=$(sqlite3 -readonly "$DB" "select schedule_json from scheduled_tasks where task_id='$TASK_ID'")"
say "provider=claudeAgent model=claude-haiku-5-5 workspace=$(sqlite3 -readonly "$DB" "select workspace_strategy_json from scheduled_tasks where task_id='$TASK_ID'")"

if [ "$ROW" = R3 ]; then
  post coalesced '{"n":"r3-c","k":"same"}'
  sleep 1; dump "at SIGKILL (coalescing window still open)"
  sigkill_server
  start_server restart
  D=$(delivery_id_of r3-c)
  wait_settled 120 "$D"
  dump "after restart settled"
  stop_server
  say "PIDs started: ${STARTED_PIDS[*]} (all stopped)"
  exit 0
fi

# A real server only queues a delivery while the task's previous launch command is in flight
# (threadLaunch.launch returns once the thread is created; worktree prep runs afterwards).
# burst.mjs posts the in-flight delivery first, then the queued one(s), and SIGKILLs the server
# the instant the last 202 arrives. The DB dump right after the kill shows which were still queued.
if [ "$ROW" = R1 ]; then MARKERS=(r1-a); else MARKERS=(r2-a r2-b); fi
BODIES=('{"id":"hold","n":"hold"}'); for m in "${MARKERS[@]}"; do BODIES+=("{\"id\":\"$m\",\"n\":\"$m\"}"); done
node "$CAP/burst.mjs" "$ORIGIN_URL/api/hooks/$TASK_ID/$HOOK_TOKEN" "$SERVER_PID" "${BODIES[@]}" | while read -r l; do say "$l"; done
wait "$SERVER_PID" 2>/dev/null; SERVER_PID=
HOLD_D=$(delivery_id_of hold)
dump "at SIGKILL (right after the last 202)"
start_server restart
IDS=("$HOLD_D"); for m in "${MARKERS[@]}"; do IDS+=("$(delivery_id_of "$m")"); done
wait_settled 120 "${IDS[@]}"
dump "after restart settled"

say "=== R4: replay the same deliveries (same replay key body.id) after the restart"
for m in "${MARKERS[@]}"; do post "replay-$m" "{\"id\":\"$m\",\"n\":\"$m\"}"; done
sleep 15
dump "after replay (+15 s)"
stop_server
say "PIDs started: ${STARTED_PIDS[*]} (all stopped)"