|
|
|
|
@@ -4,6 +4,10 @@
|
|
|
|
|
# Checks provider health, pipeline progress, token budget, and interactive load.
|
|
|
|
|
# Starts the highest-priority incomplete pipeline that can run.
|
|
|
|
|
#
|
|
|
|
|
# FIX #650: Pipeline states are date-aware. A "complete" or "failed" state from
|
|
|
|
|
# a previous day is treated as stale (not_started) so pipelines can re-run daily.
|
|
|
|
|
# Running states older than 6 hours are also treated as stale (likely crashed).
|
|
|
|
|
#
|
|
|
|
|
# Usage:
|
|
|
|
|
# ./scripts/nightly-pipeline-scheduler.sh # Normal run
|
|
|
|
|
# ./scripts/nightly-pipeline-scheduler.sh --dry-run # Show what would start
|
|
|
|
|
@@ -50,6 +54,67 @@ ensure_dirs() {
|
|
|
|
|
|
|
|
|
|
log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" | tee -a "$LOG_FILE"; }
|
|
|
|
|
|
|
|
|
|
# --- FIX #650: Staleness detection ---
|
|
|
|
|
#
|
|
|
|
|
# A pipeline state is "stale" if:
|
|
|
|
|
# - complete/failed: state was set on a different calendar day
|
|
|
|
|
# - running: state was set more than 6 hours ago (likely crashed)
|
|
|
|
|
#
|
|
|
|
|
# Stale states are treated as not_started, allowing the pipeline to re-run.
|
|
|
|
|
today_date() { date +%Y-%m-%d; }
|
|
|
|
|
|
|
|
|
|
state_is_stale() {
|
|
|
|
|
local pipeline="$1"
|
|
|
|
|
python3 -c "
|
|
|
|
|
import json, os, sys
|
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
|
|
|
|
path = '$STATE_FILE'
|
|
|
|
|
today = '$(today_date)'
|
|
|
|
|
|
|
|
|
|
if not os.path.exists(path):
|
|
|
|
|
sys.exit(0) # no state file = not stale (not_started)
|
|
|
|
|
|
|
|
|
|
with open(path) as f:
|
|
|
|
|
d = json.load(f)
|
|
|
|
|
|
|
|
|
|
entry = d.get('$pipeline', {})
|
|
|
|
|
state = entry.get('state', 'not_started')
|
|
|
|
|
updated = entry.get('updated', '')
|
|
|
|
|
|
|
|
|
|
if state == 'not_started':
|
|
|
|
|
sys.exit(0) # not stale
|
|
|
|
|
|
|
|
|
|
if not updated:
|
|
|
|
|
sys.exit(1) # no timestamp = treat as stale
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
state_date = updated[:10] # YYYY-MM-DD from ISO timestamp
|
|
|
|
|
state_time = datetime.fromisoformat(updated.replace('Z', '+00:00'))
|
|
|
|
|
except (ValueError, IndexError):
|
|
|
|
|
sys.exit(1) # unparseable = stale
|
|
|
|
|
|
|
|
|
|
if state in ('complete', 'failed'):
|
|
|
|
|
# Stale if not from today
|
|
|
|
|
if state_date != today:
|
|
|
|
|
print(f'STALE: {state} from {state_date} (today is {today})', file=sys.stderr)
|
|
|
|
|
sys.exit(1)
|
|
|
|
|
sys.exit(0) # today's state is fresh
|
|
|
|
|
|
|
|
|
|
if state == 'running':
|
|
|
|
|
# Stale if older than 6 hours (likely crashed)
|
|
|
|
|
now = datetime.now(state_time.tzinfo)
|
|
|
|
|
age_hours = (now - state_time).total_seconds() / 3600
|
|
|
|
|
if age_hours > 6:
|
|
|
|
|
print(f'STALE: running for {age_hours:.1f}h (max 6h)', file=sys.stderr)
|
|
|
|
|
sys.exit(1)
|
|
|
|
|
sys.exit(0) # recently started
|
|
|
|
|
|
|
|
|
|
sys.exit(0)
|
|
|
|
|
" 2>/dev/null
|
|
|
|
|
return $?
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
get_budget_used_today() {
|
|
|
|
|
if [[ -f "$BUDGET_FILE" ]]; then
|
|
|
|
|
local today=$(date +%Y-%m-%d)
|
|
|
|
|
@@ -113,9 +178,13 @@ with open(path, 'w') as f:
|
|
|
|
|
"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# FIX #650: is_pipeline_complete checks staleness
|
|
|
|
|
is_pipeline_complete() {
|
|
|
|
|
local pipeline="$1"
|
|
|
|
|
python3 -c "
|
|
|
|
|
# If stale, it's not complete
|
|
|
|
|
if ! state_is_stale "$pipeline" 2>/dev/null; then
|
|
|
|
|
# Fresh state — check if actually complete
|
|
|
|
|
python3 -c "
|
|
|
|
|
import json, os
|
|
|
|
|
path = '$STATE_FILE'
|
|
|
|
|
if not os.path.exists(path):
|
|
|
|
|
@@ -126,11 +195,16 @@ else:
|
|
|
|
|
state = d.get('$pipeline', {}).get('state', 'not_started')
|
|
|
|
|
print('true' if state == 'complete' else 'false')
|
|
|
|
|
" 2>/dev/null || echo false
|
|
|
|
|
else
|
|
|
|
|
echo false # Stale = not complete
|
|
|
|
|
fi
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# FIX #650: is_pipeline_running checks staleness
|
|
|
|
|
is_pipeline_running() {
|
|
|
|
|
local pipeline="$1"
|
|
|
|
|
python3 -c "
|
|
|
|
|
if ! state_is_stale "$pipeline" 2>/dev/null; then
|
|
|
|
|
python3 -c "
|
|
|
|
|
import json, os
|
|
|
|
|
path = '$STATE_FILE'
|
|
|
|
|
if not os.path.exists(path):
|
|
|
|
|
@@ -141,6 +215,9 @@ else:
|
|
|
|
|
state = d.get('$pipeline', {}).get('state', 'not_started')
|
|
|
|
|
print('true' if state == 'running' else 'false')
|
|
|
|
|
" 2>/dev/null || echo false
|
|
|
|
|
else
|
|
|
|
|
echo false # Stale = not running
|
|
|
|
|
fi
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
check_dependency() {
|
|
|
|
|
@@ -272,6 +349,57 @@ with open(path, 'w') as f:
|
|
|
|
|
fi
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# FIX #650: Daily reset — purge stale states at the start of each run
|
|
|
|
|
reset_stale_states() {
|
|
|
|
|
if [[ ! -f "$STATE_FILE" ]]; then
|
|
|
|
|
return
|
|
|
|
|
fi
|
|
|
|
|
python3 -c "
|
|
|
|
|
import json, os, sys
|
|
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
|
|
path = '$STATE_FILE'
|
|
|
|
|
today = '$(today_date)'
|
|
|
|
|
|
|
|
|
|
with open(path) as f:
|
|
|
|
|
d = json.load(f)
|
|
|
|
|
|
|
|
|
|
changed = False
|
|
|
|
|
cleaned = []
|
|
|
|
|
for name, entry in list(d.items()):
|
|
|
|
|
state = entry.get('state', '')
|
|
|
|
|
updated = entry.get('updated', '')
|
|
|
|
|
|
|
|
|
|
if state in ('complete', 'failed') and updated:
|
|
|
|
|
state_date = updated[:10]
|
|
|
|
|
if state_date != today:
|
|
|
|
|
del d[name]
|
|
|
|
|
changed = True
|
|
|
|
|
cleaned.append(name)
|
|
|
|
|
|
|
|
|
|
elif state == 'running' and updated:
|
|
|
|
|
try:
|
|
|
|
|
state_time = datetime.fromisoformat(updated.replace('Z', '+00:00'))
|
|
|
|
|
now = datetime.now(state_time.tzinfo)
|
|
|
|
|
age_hours = (now - state_time).total_seconds() / 3600
|
|
|
|
|
if age_hours > 6:
|
|
|
|
|
del d[name]
|
|
|
|
|
changed = True
|
|
|
|
|
cleaned.append(f'{name}(stale-running)')
|
|
|
|
|
except (ValueError, IndexError):
|
|
|
|
|
del d[name]
|
|
|
|
|
changed = True
|
|
|
|
|
cleaned.append(f'{name}(bad-timestamp)')
|
|
|
|
|
|
|
|
|
|
if changed:
|
|
|
|
|
with open(path, 'w') as f:
|
|
|
|
|
json.dump(d, f, indent=2)
|
|
|
|
|
print(f'Reset {len(cleaned)} stale pipelines: {', '.join(cleaned)}')
|
|
|
|
|
else:
|
|
|
|
|
print('No stale pipeline states')
|
|
|
|
|
" 2>>"$LOG_FILE"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
# --- Main ---
|
|
|
|
|
main() {
|
|
|
|
|
local mode="${1:-run}"
|
|
|
|
|
@@ -279,6 +407,9 @@ main() {
|
|
|
|
|
|
|
|
|
|
log "=== Pipeline Scheduler ($mode) ==="
|
|
|
|
|
|
|
|
|
|
# FIX #650: Reset stale states first
|
|
|
|
|
reset_stale_states
|
|
|
|
|
|
|
|
|
|
# Check 1: Is inference available?
|
|
|
|
|
if ! check_inference_available; then
|
|
|
|
|
log "No inference provider available. Skipping all pipelines."
|
|
|
|
|
@@ -327,11 +458,20 @@ else:
|
|
|
|
|
print(d.get('$name', {}).get('state', 'not_started'))
|
|
|
|
|
" 2>/dev/null || echo "not_started")
|
|
|
|
|
|
|
|
|
|
# Check staleness for display
|
|
|
|
|
if [[ "$state" == "complete" || "$state" == "failed" || "$state" == "running" ]]; then
|
|
|
|
|
if ! state_is_stale "$name" 2>/dev/null; then
|
|
|
|
|
: # fresh
|
|
|
|
|
else
|
|
|
|
|
state="${state} (stale)"
|
|
|
|
|
fi
|
|
|
|
|
fi
|
|
|
|
|
|
|
|
|
|
local color=$NC
|
|
|
|
|
case "$state" in
|
|
|
|
|
running) color=$YELLOW ;;
|
|
|
|
|
complete) color=$GREEN ;;
|
|
|
|
|
failed) color=$RED ;;
|
|
|
|
|
running*) color=$YELLOW ;;
|
|
|
|
|
complete*) color=$GREEN ;;
|
|
|
|
|
failed*) color=$RED ;;
|
|
|
|
|
esac
|
|
|
|
|
printf " %-25s %b%s%b (max: %s tokens, dep: %s)\n" "$name" "$color" "$state" "$NC" "$max_tokens" "$dep"
|
|
|
|
|
done
|
|
|
|
|
@@ -346,7 +486,7 @@ else:
|
|
|
|
|
for entry in "${PIPELINES[@]}"; do
|
|
|
|
|
IFS='|' read -r name script max_tokens dep <<< "$entry"
|
|
|
|
|
|
|
|
|
|
# Skip if already running or complete
|
|
|
|
|
# Skip if already running or complete (staleness already handled above)
|
|
|
|
|
if [[ "$(is_pipeline_running $name)" == "true" ]]; then
|
|
|
|
|
log "SKIP $name: already running"
|
|
|
|
|
continue
|
|
|
|
|
|