#!/usr/bin/env bash
# Checks one job, then enqueues one Dagu run of bench/dagu/job.yaml for it
# (specs/job-submit-land.md). Runs on the development box; nothing of the job runs here.
#
#   jobs/scripts/submit.sh [--restart] <name>
#
# Refuses with one line on stderr and exit 1, without contacting the lane, unless, in order:
#   - the primary checkout is on the base branch and clean;
#   - specs/<name>.md and specs/<name>.job.json exist on the base tip, and the manifest
#     passes `job-check manifest --spec` against that spec (which refuses `feature_rules`
#     when the spec's `touches` omit docs/feature-map.json);
#   - the spec's commit sha (the last commit on base that changed it) has no tag
#     spec/<name>/<sha> (bench/land.sh writes it at landing), and the spec says
#     `enabled: true`;
#   - every `depends-on` spec has a tag spec/<dep>/*, that is, has landed;
#   - the base tip is on the forge: after `git fetch <forge> <base>` it is an ancestor of,
#     or equal to, the fetched tip. This script never pushes it;
#   - the record for <name>, if any, is not `parked`, is not `landed` with a `spec_sha`
#     equal to the spec's blob at base (or with none), has `resends` below three, and has
#     no empty `outcome` whose Dagu run is still open (queued, running, not_started or
#     waiting). That last check reads the run from Dagu, so with the lane or Dagu not
#     answering it exits 3 instead. A `landed` record whose `spec_sha` differs (a later
#     version of the spec) counts as no record: the job starts afresh, resends at 0;
#   - a `needs-person` record (the job's failed rework rounds are used up,
#     specs/job-review-rework.md) has a `spec_sha` equal to the spec's blob at base; one
#     whose spec changed is refused, naming --restart. Its resend, which follows a
#     person's fix pushed to the forge's job/<name>, neither raises `resends` nor is
#     refused for reaching three, and its line says the job resumes after a person's call;
#   - on a resend whose spec blob at the base tip equals the record's `spec_sha`, the base
#     it resumes at (below) is a commit here, on the forge's base branch, holds the spec
#     with that blob, and its manifest passes `job-check manifest --spec` against it.
#
# The base sent. A first send, or a resend whose spec blob changed since the record, goes
# at the base tip. A resend whose spec blob is unchanged goes at the record's `base_sha`,
# the base the job started at, because the lane's `prepare` resumes a job only at the
# BASE_SHA and SPEC_SHA it started with (and a fresh start at a newer base fails while the
# forge holds job/<name> from the older one). A record from before `base_sha` was kept has
# none: then JOBS_RESEND_BASE names that base, and without it the resend is refused. A
# resend with a changed spec blob is refused by `prepare` (exit 2 there) unless the run
# has RESET=yes, which this script does not send.
#
# --restart starts the job from scratch instead of resuming it. The checks above apply,
# except that `resends` may be three or more and nothing about a resend's base is checked,
# so it still refuses a parked record, a landed one with this spec, and a run of this job
# still open in Dagu. Before sending it then, in order: deletes job/<name> and
# result/<name> from the forge (those it holds) and the local refs/jobs/<name>/* copies;
# runs `<dagu> clean-jobs <name>` (the lane's jobs-work/<name>) and `<dagu> prune-mirror`;
# and renames the record to <name>.json.restarted-<UTC time>, which records.py does not
# read. Any of these failing exits 3 with nothing sent; each is safe to repeat, so running
# the restart again finishes it. It sends as a first send: at the base tip, resends 0.
#
# Then it runs, from the primary checkout,
#
#   <dagu> --lane <index> run --no-wait [--no-sync] bench/dagu/job.yaml \
#     NAME=<name> SPEC_SHA=<blob of specs/<name>.md at base> BASE_SHA=<the base sent> \
#     [MODEL=<JOBS_MODEL>] [BATCH_MODELS=<JOBS_BATCH_MODELS>]
#
# MODEL= and BATCH_MODELS= are added only when JOBS_MODEL or JOBS_BATCH_MODELS is set (see
# Settings); then the manifest at the base sent, with them applied by
# jobs/scripts/manifest-models.py, must also pass `job-check manifest --spec`, or the send
# is refused. The lane applies them only on a job's first run: a resumed run keeps the
# models its first run wrote.
#
# with --no-sync when any recorded run of `job` is still open, so the source copy under an
# earlier run's build is not replaced. With the call failing or printing no run id it exits
# 3 and writes no record; otherwise it writes the record (run id, resends raised by one when
# a record already existed, empty outcome, spec_sha the blob sent, base_sha the commit
# sent) and prints one line, which says when the job resumed at the recorded base.
#
# Job events (jobs/scripts/job-events.sh), best effort and never changing the exit: once
# every check has passed, a plan (spec, submit, queue, the manifest's stage names, publish,
# land); after a send, `spec` and `submit` passed under the Dagu run id, or, when the send
# fails, `spec` passed and `submit` failed under submit-<UTC>. A refusal sends none.
#
# Settings, each defaulting to the real collaborator so tests can replace it:
#   NEV_JOB_EVENTS_URL, NEV_JOB_EVENTS_TOKEN
#                          the job-events store and its token (default: the same keys in
#                          <repo>/.env; no URL means no posting, events still go to
#                          $JOBS_STATE_DIR/events.jsonl)
#   JOBS_REPO              the primary checkout (default: the one this repository belongs to)
#   JOBS_BASE_BRANCH       default: tracks.json `base` on HEAD, else main
#   JOBS_STATE_DIR         the job records (default $HOME/.local/state/nev-jobs)
#   JOBS_DAGU              default <repo>/bench/dagu-lane.sh
#   JOBS_FORGE_REMOTE      the forge the lane's mirror fetches from (default rig)
#   JOBS_CHECK             the job-check binary (default $JOBS_STATE_DIR/bin/job-check,
#                          else job-check on PATH; jobs/scripts/install-job-check.sh
#                          builds it on a lane and puts it in the first place)
#   NEV_DAGU_LANE          the lane's index in bench/lane.local.json (default 0)
#   JOBS_RESEND_UNCOUNTED  yes: a resend after the `run` step exited 3 (nothing ran), which
#                          neither raises `resends` nor is refused for reaching three
#   JOBS_RESEND_BASE       a commit (full or abbreviated sha): the base to resume at, for a
#                          resend whose record has a `spec_sha` but no `base_sha`; read only
#                          then
#   JOBS_MODEL             a model name: the writer's and the reviewer's model, in place of
#                          the manifest's (default: unset, the manifest's own)
#   JOBS_BATCH_MODELS      a model list in the NEV_MODELS form
#                          (model:count:concurrency[:timeout],...): the batch's models, in
#                          place of the manifest's gateway agents' `models` (default: unset,
#                          the manifest's own). Both are configuration of one send, for a lane
#                          whose models differ from the ones the manifest names, and never
#                          change specs/
set -euo pipefail

SCRIPTS="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
RECORDS=(python3 "$SCRIPTS/records.py")
WORKFLOW=bench/dagu/job.yaml

refuse() { echo "submit: $*" >&2; exit 1; }
# Once the send has been tried (ATTEMPT set), a failure to send is also a job event.
unreachable() {
  echo "submit: $*; $NAME not sent" >&2
  if [ -n "${ATTEMPT:-}" ]; then
    event_send step "$NAME" "$ATTEMPT" spec passed
    event_send step "$NAME" "$ATTEMPT" submit failed "$*"
  fi
  exit 3
}

RESTART=no
if [ "${1:-}" = --restart ]; then RESTART=yes; shift; fi
[ $# -eq 1 ] || { echo "usage: jobs/scripts/submit.sh [--restart] <name>" >&2; exit 2; }
NAME="$1"
case "$NAME" in
  ''|*..*|*[!A-Za-z0-9._-]*) refuse "unusable job name: $NAME" ;;
esac

if [ -z "${JOBS_REPO:-}" ]; then
  common="$(git -C "$SCRIPTS" rev-parse --path-format=absolute --git-common-dir)"
  JOBS_REPO="$(dirname "$common")"
fi
REPO="$JOBS_REPO"
STATE_DIR="${JOBS_STATE_DIR:-$HOME/.local/state/nev-jobs}"
DAGU="${JOBS_DAGU:-$REPO/bench/dagu-lane.sh}"
FORGE="${JOBS_FORGE_REMOTE:-rig}"
LANE="${NEV_DAGU_LANE:-0}"
UNCOUNTED="${JOBS_RESEND_UNCOUNTED:-no}"
if [ -z "${JOBS_CHECK:-}" ]; then
  if [ -x "$STATE_DIR/bin/job-check" ]; then JOBS_CHECK="$STATE_DIR/bin/job-check"
  else JOBS_CHECK="$(command -v job-check || true)"; fi
fi
if [ -z "${JOBS_BASE_BRANCH:-}" ]; then
  JOBS_BASE_BRANCH="$(git -C "$REPO" show HEAD:tracks.json 2>/dev/null |
    python3 -c 'import json,sys; print(json.load(sys.stdin).get("base") or "main")' 2>/dev/null || echo main)"
fi
BASE="$JOBS_BASE_BRANCH"

# --- the primary checkout ---------------------------------------------------------
current="$(git -C "$REPO" symbolic-ref --quiet --short HEAD || true)"
[ "$current" = "$BASE" ] || refuse "the primary checkout is on ${current:-a detached HEAD}, not $BASE"
[ -z "$(git -C "$REPO" status --porcelain)" ] || refuse "the primary checkout has uncommitted changes"
BASE_SHA="$(git -C "$REPO" rev-parse "$BASE^{commit}")"

# --- the spec and its manifest ----------------------------------------------------
SPEC="specs/$NAME.md"
MANIFEST="specs/$NAME.job.json"
git -C "$REPO" cat-file -e "$BASE_SHA:$SPEC" 2>/dev/null || refuse "$SPEC does not exist on $BASE"
git -C "$REPO" cat-file -e "$BASE_SHA:$MANIFEST" 2>/dev/null || refuse "$MANIFEST does not exist on $BASE"
[ -n "$JOBS_CHECK" ] && [ -x "$JOBS_CHECK" ] ||
  refuse "no job-check binary to check the manifest with (run jobs/scripts/install-job-check.sh, or set JOBS_CHECK)"
TMP="$(mktemp -d)"
trap 'rm -rf "$TMP"' EXIT
# check_manifest <commit>: the manifest at that commit passes `job-check manifest` against
# the spec at the same commit, so a manifest whose `feature_rules` commit a file the spec's
# `touches` omit (docs/feature-map.json, which bench/land.sh would then refuse) is refused
# before the job runs rather than at its landing.
check_manifest() {
  local at="$1" check_out
  git -C "$REPO" show "$at:$MANIFEST" > "$TMP/$NAME.job.json"
  git -C "$REPO" show "$at:$SPEC" > "$TMP/$NAME.md"
  if ! check_out="$("$JOBS_CHECK" manifest "$TMP/$NAME.job.json" --spec "$TMP/$NAME.md" 2>&1)"; then
    check_out="${check_out//$TMP\//}"
    refuse "$MANIFEST at ${at:0:12} does not pass job-check manifest: ${check_out//$'\n'/; }"
  fi
}
check_manifest "$BASE_SHA"

# Dagu's SPEC_SHA is the spec's blob at base; the landing tag names the commit.
SPEC_BLOB="$(git -C "$REPO" rev-parse "$BASE_SHA:$SPEC")"
SPEC_COMMIT="$(git -C "$REPO" log -1 --format=%H "$BASE_SHA" -- "$SPEC")"
if git -C "$REPO" rev-parse --verify -q "refs/tags/spec/$NAME/$SPEC_COMMIT" >/dev/null; then
  refuse "$SPEC at ${SPEC_COMMIT:0:12} has already landed (tag spec/$NAME/$SPEC_COMMIT)"
fi
FRONT="$(git -C "$REPO" show "$BASE_SHA:$SPEC" | python3 -c '
import re, sys
text = sys.stdin.read()
m = re.match(r"---\n(.*?)\n---", text, re.S)
fields = {}
for line in (m.group(1).splitlines() if m else []):
    key, _, value = line.partition(":")
    fields[key.strip()] = value.strip()
deps = fields.get("depends-on", "").strip("[]")
print(fields.get("enabled", ""))
print(" ".join(d.strip() for d in deps.split(",") if d.strip()))
')"
ENABLED="$(printf '%s\n' "$FRONT" | sed -n 1p)"
DEPS="$(printf '%s\n' "$FRONT" | sed -n 2p)"
[ "$ENABLED" = true ] || refuse "$SPEC is not enabled: true"
for dep in $DEPS; do
  [ -n "$(git -C "$REPO" tag -l "spec/$dep/*")" ] || refuse "depends-on $dep has not landed (no tag spec/$dep/*)"
done

# --- the base tip on the forge ----------------------------------------------------
git -C "$REPO" fetch --quiet "$FORGE" "$BASE" 2>/dev/null ||
  refuse "could not fetch $BASE from the forge remote $FORGE"
FORGE_TIP="$(git -C "$REPO" rev-parse FETCH_HEAD)"
git -C "$REPO" merge-base --is-ancestor "$BASE_SHA" "$FORGE_TIP" ||
  refuse "the base tip ${BASE_SHA:0:12} is not on the forge's $BASE; push it first"

# --- the record -------------------------------------------------------------------
PREV_RUN="$("${RECORDS[@]}" "$STATE_DIR" get "$NAME" run_id)"
PREV_OUTCOME="$("${RECORDS[@]}" "$STATE_DIR" get "$NAME" outcome)"
PREV_RESENDS="$("${RECORDS[@]}" "$STATE_DIR" get "$NAME" resends)"
PREV_RESENDS="${PREV_RESENDS:-0}"
PREV_SPEC="$("${RECORDS[@]}" "$STATE_DIR" get "$NAME" spec_sha)"
case "$PREV_OUTCOME" in
  parked) refuse "$NAME is recorded parked" ;;
  landed)
    # A landed record refuses only the spec content that landed; a later version of the
    # spec starts a fresh record. One with no spec_sha cannot tell, so it refuses.
    if [ -z "$PREV_SPEC" ] || [ "$PREV_SPEC" = "$SPEC_BLOB" ]; then
      refuse "$NAME is recorded landed with this spec (blob ${SPEC_BLOB:0:12})"
    fi
    PREV_RUN="" PREV_OUTCOME="" PREV_RESENDS=0
    ;;
esac
# A needs-person record (the job's failed rework rounds are used up) is resent once a
# person has made the call: a fix pushed to the forge's job/<name> resumes it here, without
# counting, since that is not a retry of a flaky publish. A changed spec or manifest needs
# --restart, because the lane's `prepare` resumes a job only at the spec it started with.
PERSON=no
if [ "$PREV_OUTCOME" = needs-person ] && [ "$RESTART" = no ]; then
  if [ -z "$PREV_SPEC" ] || [ "$PREV_SPEC" != "$SPEC_BLOB" ]; then
    refuse "$NAME waits for a person and $SPEC changed since it was sent (blob ${PREV_SPEC:0:12} then, ${SPEC_BLOB:0:12} now); a resend resumes only at the spec it started with, so send it with jobs/scripts/submit.sh --restart $NAME"
  fi
  PERSON=yes
fi
RESENDS=0
if [ -n "$PREV_RUN" ] && [ "$RESTART" = no ]; then
  if [ "$UNCOUNTED" = yes ] || [ "$PERSON" = yes ]; then
    RESENDS="$PREV_RESENDS"
  else
    [ "$PREV_RESENDS" -lt 3 ] || refuse "$NAME has been resent $PREV_RESENDS times"
    RESENDS=$((PREV_RESENDS + 1))
  fi
fi

# The base to send. A resend whose spec blob is still the recorded one goes at the recorded
# base, since the lane's `prepare` resumes a job only at the BASE_SHA and SPEC_SHA it started
# with; anything else, a restart included, goes at the base tip.
SEND_BASE="$BASE_SHA"
RESUME=""
if [ "$RESTART" = no ] && [ -n "$PREV_RUN" ] && [ -n "$PREV_SPEC" ] && [ "$PREV_SPEC" = "$SPEC_BLOB" ]; then
  PREV_BASE="$("${RECORDS[@]}" "$STATE_DIR" get "$NAME" base_sha)"
  if [ -z "$PREV_BASE" ]; then
    given="${JOBS_RESEND_BASE:-}"
    [ -n "$given" ] ||
      refuse "the record of $NAME has no base_sha, so the base it started at is unknown; set JOBS_RESEND_BASE=<commit>"
    case "$given" in
      *[!0-9a-f]*) refuse "JOBS_RESEND_BASE is not a commit sha: $given" ;;
    esac
    [ "${#given}" -ge 7 ] && [ "${#given}" -le 40 ] || refuse "JOBS_RESEND_BASE is not a commit sha: $given"
    PREV_BASE="$(git -C "$REPO" rev-parse --verify -q "$given^{commit}")" ||
      refuse "JOBS_RESEND_BASE $given is not a commit here"
    at="$(git -C "$REPO" rev-parse -q --verify "$PREV_BASE:$SPEC" || true)"
    [ "$at" = "$PREV_SPEC" ] ||
      refuse "JOBS_RESEND_BASE ${PREV_BASE:0:12} does not hold $SPEC as recorded (blob ${PREV_SPEC:0:12})"
    what="JOBS_RESEND_BASE"
  else
    git -C "$REPO" cat-file -e "$PREV_BASE^{commit}" 2>/dev/null ||
      refuse "the recorded base ${PREV_BASE:0:12} of $NAME is not a commit here"
    at="$(git -C "$REPO" rev-parse -q --verify "$PREV_BASE:$SPEC" || true)"
    [ "$at" = "$PREV_SPEC" ] ||
      refuse "the recorded base ${PREV_BASE:0:12} of $NAME does not hold $SPEC as recorded (blob ${PREV_SPEC:0:12})"
    what="the recorded base"
  fi
  git -C "$REPO" merge-base --is-ancestor "$PREV_BASE" "$FORGE_TIP" ||
    refuse "$what ${PREV_BASE:0:12} is no longer on the forge's $BASE"
  git -C "$REPO" cat-file -e "$PREV_BASE:$MANIFEST" 2>/dev/null ||
    refuse "$MANIFEST does not exist at $what ${PREV_BASE:0:12}"
  [ "$PREV_BASE" = "$BASE_SHA" ] || check_manifest "$PREV_BASE"
  SEND_BASE="$PREV_BASE"
  RESUME=", resumed at the recorded base"
  [ "$PERSON" = no ] || RESUME=", resumed after a person's call, at the recorded base"
  [ "$SEND_BASE" = "$BASE_SHA" ] || RESUME="$RESUME (the base tip is ${BASE_SHA:0:12})"
fi

# --- models given at submit time ----------------------------------------------------
# JOBS_MODEL and JOBS_BATCH_MODELS go to the run as MODEL= and BATCH_MODELS=, and only
# when set, so with neither the call is what it always was. The manifest at the base sent,
# with them applied by the same manifest-models.py the lane's `prepare` runs, must pass
# `job-check manifest --spec`, so a malformed list is refused here rather than on the lane.
OVERRIDES=()
OVERRIDE_NOTE=""
if [ -n "${JOBS_MODEL:-}${JOBS_BATCH_MODELS:-}" ]; then
  case "${JOBS_MODEL:-}" in
    *[!A-Za-z0-9._:/-]*) refuse "JOBS_MODEL is not a model name: $JOBS_MODEL" ;;
  esac
  case "${JOBS_BATCH_MODELS:-}" in
    *[!A-Za-z0-9._:/,-]*) refuse "JOBS_BATCH_MODELS is not a model list (model:count:concurrency[:timeout],...): $JOBS_BATCH_MODELS" ;;
  esac
  git -C "$REPO" show "$SEND_BASE:$MANIFEST" > "$TMP/manifest.src.json"
  git -C "$REPO" show "$SEND_BASE:$SPEC" > "$TMP/$NAME.md"
  if ! models_out="$(python3 "$SCRIPTS/manifest-models.py" "$TMP/manifest.src.json" "$TMP/$NAME.job.json" \
      "$NAME" "${JOBS_MODEL:-}" "${JOBS_BATCH_MODELS:-}" 2>&1)"; then
    refuse "the models given (JOBS_MODEL, JOBS_BATCH_MODELS) do not apply to $MANIFEST: ${models_out//$'\n'/; }"
  fi
  if ! check_out="$("$JOBS_CHECK" manifest "$TMP/$NAME.job.json" --spec "$TMP/$NAME.md" 2>&1)"; then
    check_out="${check_out//$TMP\//}"
    refuse "$MANIFEST with the models given (JOBS_MODEL, JOBS_BATCH_MODELS) does not pass job-check manifest: ${check_out//$'\n'/; }"
  fi
  if [ -n "${JOBS_MODEL:-}" ]; then
    OVERRIDES+=("MODEL=$JOBS_MODEL")
    OVERRIDE_NOTE="$OVERRIDE_NOTE, writer and review $JOBS_MODEL"
  fi
  if [ -n "${JOBS_BATCH_MODELS:-}" ]; then
    OVERRIDES+=("BATCH_MODELS=$JOBS_BATCH_MODELS")
    OVERRIDE_NOTE="$OVERRIDE_NOTE, batch $JOBS_BATCH_MODELS"
  fi
fi

# Every recorded run still in Dagu's hands, read once: this job's own must have ended,
# and any open one means the source copy must not be replaced (--no-sync).
mapfile -t OPEN < <("${RECORDS[@]}" "$STATE_DIR" open)
NO_SYNC=()
if [ "${#OPEN[@]}" -gt 0 ]; then
  ids=()
  for line in "${OPEN[@]}"; do ids+=("${line#* }"); done
  status_out="$(cd "$REPO" && "$DAGU" --lane "$LANE" watch --once --step run job "${ids[@]}")" ||
    unreachable "could not read the open runs from Dagu"
  while read -r id status _; do
    [ -n "${id:-}" ] || continue
    case "$status" in
      queued|running|not_started|waiting)
        [ "$id" != "$PREV_RUN" ] || [ -n "$PREV_OUTCOME" ] ||
          refuse "the run $id of $NAME is still $status"
        NO_SYNC=(--no-sync)
        ;;
    esac
  done <<<"$status_out"
fi

# --- a restart clears what a resume would pick up ---------------------------------
# In this order, each step safe to repeat, so a restart that stops partway is finished by
# running it again: the forge's job/<name> and result/<name> (a fresh start fails while the
# forge holds job/<name>); the lane's leftover directories (clean-jobs); the lane mirror's
# copies of the refs just deleted (prune-mirror); and last the record, kept under a dated
# name that records.py does not read, so until then the record still guards against a
# second send while a run is open.
KEPT=""
if [ "$RESTART" = yes ]; then
  refs="$(git -C "$REPO" ls-remote "$FORGE" "refs/heads/job/$NAME" "refs/heads/result/$NAME" 2>/dev/null)" ||
    unreachable "could not list job/$NAME and result/$NAME on the forge remote $FORGE"
  held=()
  while read -r _ ref; do
    case "$ref" in "refs/heads/job/$NAME"|"refs/heads/result/$NAME") held+=("$ref") ;; esac
  done <<<"$refs"
  if [ "${#held[@]}" -gt 0 ]; then
    git -C "$REPO" push --quiet "$FORGE" --delete "${held[@]}" >/dev/null 2>&1 ||
      unreachable "could not delete ${held[*]#refs/heads/} from the forge remote $FORGE"
  fi
  git -C "$REPO" update-ref -d "refs/jobs/$NAME/branch" 2>/dev/null || true
  git -C "$REPO" update-ref -d "refs/jobs/$NAME/result" 2>/dev/null || true
  (cd "$REPO" && "$DAGU" --lane "$LANE" clean-jobs "$NAME") >&2 ||
    unreachable "dagu-lane clean-jobs $NAME failed"
  (cd "$REPO" && "$DAGU" --lane "$LANE" prune-mirror) >&2 ||
    unreachable "dagu-lane prune-mirror failed"
  if [ -e "$STATE_DIR/$NAME.json" ]; then
    KEPT="$NAME.json.restarted-$(date -u +%Y%m%dT%H%M%SZ)"
    [ ! -e "$STATE_DIR/$KEPT" ] || refuse "$STATE_DIR/$KEPT already exists; try again in a second"
    mv "$STATE_DIR/$NAME.json" "$STATE_DIR/$KEPT"
  fi
  RESUME=", restarted from scratch"
  [ -z "$KEPT" ] || RESUME="$RESUME (the old record is $KEPT)"
fi

# --- send it ------------------------------------------------------------------------
# Job events (jobs/scripts/job-events.sh): the feature's plan, from the manifest at the base
# sent, then `spec` and `submit` under the Dagu run id once there is one, or under
# submit-<UTC> when the send fails. A refusal above sends no event.
source "$SCRIPTS/job-events.sh"
events_setup "$REPO" "$STATE_DIR"
mapfile -t STAGES < <(git -C "$REPO" show "$SEND_BASE:$MANIFEST" 2>/dev/null |
  python3 -c 'import json, sys; [print(s["name"]) for s in json.load(sys.stdin).get("stages", [])]' 2>/dev/null || true)
event_send plan "$NAME" spec submit queue "${STAGES[@]+"${STAGES[@]}"}" publish land
ATTEMPT="submit-$(date -u +%Y-%m-%dT%H:%M:%SZ)"
set +e
OUT="$(cd "$REPO" && "$DAGU" --lane "$LANE" run --no-wait "${NO_SYNC[@]}" "$WORKFLOW" \
  "NAME=$NAME" "SPEC_SHA=$SPEC_BLOB" "BASE_SHA=$SEND_BASE" "${OVERRIDES[@]+"${OVERRIDES[@]}"}")"
STATUS=$?
set -e
[ "$STATUS" -eq 0 ] || unreachable "dagu-lane run exited $STATUS"
RUN_ID="$(printf '%s\n' "$OUT" | sed -n -E 's/^dagu-lane: (enqueued|started) run job ([^ ]+).*/\2/p' | head -n 1)"
[ -n "$RUN_ID" ] || unreachable "dagu-lane run printed no run id"

MESSAGE="enqueued run $RUN_ID: spec ${SPEC_BLOB:0:12}, base ${SEND_BASE:0:12}$OVERRIDE_NOTE$RESUME"
"${RECORDS[@]}" "$STATE_DIR" set "$NAME" "run_id=$RUN_ID" "resends=$RESENDS" outcome= \
  "message=$MESSAGE" "spec_sha=$SPEC_BLOB" "base_sha=$SEND_BASE"
event_send step "$NAME" "$RUN_ID" spec passed
event_send step "$NAME" "$RUN_ID" submit passed "$MESSAGE"
echo "submitted $NAME: $MESSAGE${NO_SYNC[0]+ (no sync)}"