Compare commits
1 Commits
fix/600
...
fix/issue-
| Author | SHA1 | Date | |
|---|---|---|---|
| acf1ea0d78 |
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user