Compare commits

...

1 Commits

Author SHA1 Message Date
acf1ea0d78 fix: pipeline_state.json daily reset — timestamp-based staleness (#650)
Some checks failed
Architecture Lint / Lint Repository (pull_request) Blocked by required conditions
Validate Config / Python Test Suite (pull_request) Blocked by required conditions
Architecture Lint / Linter Tests (pull_request) Failing after 19s
Smoke Test / smoke (pull_request) Failing after 17s
Validate Config / YAML Lint (pull_request) Failing after 17s
Validate Config / JSON Validate (pull_request) Successful in 27s
Validate Config / Python Syntax & Import Check (pull_request) Failing after 2m17s
PR Checklist / pr-checklist (pull_request) Failing after 4m3s
Validate Config / Cron Syntax Check (pull_request) Successful in 8s
Validate Config / Shell Script Lint (pull_request) Failing after 48s
Validate Config / Deploy Script Dry Run (pull_request) Successful in 8s
Validate Config / Playbook Schema Validation (pull_request) Successful in 16s
Pipeline states from previous days are now treated as stale:
- complete/failed from yesterday → not_started (allows re-run)
- running for >6h → not_started (likely crashed)

Changes:
- Added state_is_stale() function checking state date vs today
- Added reset_stale_states() called at start of each scheduler run
- is_pipeline_complete() and is_pipeline_running() check staleness
- --status shows '(stale)' indicator for outdated states
- Running states auto-expire after 6 hours (crash recovery)
2026-04-15 01:29:18 +00:00

View File

@@ -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