ops(scripts): dispatch, queue gating, heartbeat, quota probe

dispatch-turn.sh (headless turn launcher with TURN_MODEL routing), queue-after.sh (serial gate chain + quota-wall guard), pmo-heartbeat.sh (PMO wake loop), quota-probe-then-chain.sh (post-wall relaunch).
2026-08-29 05:22:43 -05:00
parent 458d92831e
commit 79d93b6c49
18 changed files with 825 additions and 0 deletions
@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# source: /home/_crossfeed/tooling/agent-stack @ HEAD
+# semaphore.sh — org-wide agent-turn concurrency gate (z.ai plan: 3
+# simultaneous API calls; we approximate by limiting concurrent ACTIVE
+# agent processes: headless work-turns + freshly-dispatched turns).
+#
+# Usage:
+#   semaphore.sh status            -> print active count / max
+#   semaphore.sh try <tag> <secs>  -> exit 0 + hold slot if free, else exit 1
+#   semaphore.sh release <tag>     -> release slot held by <tag>
+#   semaphore.sh clean             -> drop stale holds (>MaxAge, default 30m)
+#
+# Slot dir: /home/_crossfeed/metrics/active/ (one file per active turn,
+# filename = <tag>.pid; content = ISO start). Not a hard API-call gate —
+# it bounds AGENT TURN concurrency, the controllable unit.
+set -uo pipefail
+
+METRICS=${METRICS:-/home/_crossfeed/metrics}
+ACTIVE="$METRICS/active"
+MAX=${MAX_CONCURRENT:-2}   # headroom of 1 below z.ai cap for interactive PMO use
+STALE_SECS=${STALE_SECS:-1800}
+
+mkdir -p "$ACTIVE" 2>/dev/null || true
+# graceful degrade: no writable metrics dir -> gate unavailable (exit 99)
+[ -w "$ACTIVE" ] || { [ "${1:-}" = status ] && echo "gate=unavailable"; exit 99; }
+
+active_count() {
+  local n=0 f
+  for f in "$ACTIVE"/*; do
+    [ -f "$f" ] || continue
+    if [ "$STALE_SECS" -gt 0 ]; then
+      local age=$(( $(date +%s) - $(stat -c %Y "$f") ))
+      [ "$age" -gt "$STALE_SECS" ] && continue
+    fi
+    n=$((n+1))
+  done
+  echo "$n"
+}
+
+cmd=${1:-status}
+case "$cmd" in
+  status)
+    echo "active=$(active_count)/$MAX"
+    [ "$(active_count)" -lt "$MAX" ] && exit 0 || exit 1
+    ;;
+  try)
+    tag=${2:?tag}; [ "$#" -ge 3 ] || { echo "usage: try <tag> <max-wait-secs>" >&2; exit 2; }
+    wait_secs=$3; waited=0
+    while true; do
+      if [ "$(active_count)" -lt "$MAX" ]; then
+        date -Is > "$ACTIVE/$tag.pid" 2>/dev/null && { echo "acquired: $tag"; exit 0; }
+        echo "ERROR: cannot write $ACTIVE (perms?)" >&2; exit 2
+      fi
+      [ "$waited" -ge "$wait_secs" ] && { echo "timeout: no slot after ${wait_secs}s (active=$(active_count)/$MAX)" >&2; exit 1; }
+      sleep 10; waited=$((waited+10))
+    done
+    ;;
+  release)
+    tag=${2:?tag}
+    rm -f "$ACTIVE/$tag.pid" && echo "released: $tag"
+    ;;
+  clean)
+    for f in "$ACTIVE"/*; do
+      [ -f "$f" ] || continue
+      age=$(( $(date +%s) - $(stat -c %Y "$f") ))
+      [ "$age" -gt "$STALE_SECS" ] && { rm -f "$f"; echo "dropped stale: $(basename "$f")"; }
+    done
+    ;;
+  *)
+    echo "usage: semaphore.sh {status|try <tag> <secs>|release <tag>|clean}" >&2; exit 2
+    ;;
+esac