scheduler: honor pipeline_settings (extraction_enabled, window, max_items)

The dev-side scheduler now snapshots the singleton pipeline_settings row
at the top of run() and applies three DB-driven gates:

- extraction_enabled = false → skip the entire run with a single log line.
  Lets Travis pause extraction from the dashboard without touching the
  config file or restarting anything.
- extraction_active_hours_start/_end → fast-fail if outside the window
  AND (when set) override the static settings.toml [scheduler].window_*
  bounds so the inner loop also respects the dashboard's choice. Either
  side being NULL means "no constraint on that side" — matches the form's
  "leave blank = always active" semantics.
- extraction_max_items_per_run → hard cap on processed items per
  invocation. 0/null means unlimited (existing behavior).

Snapshot semantics are deliberate: a mid-run toggle doesn't half-apply,
same way max_calls_per_hour is tracked locally. Workers re-read on the
next invocation.
This commit is contained in:
Travis Herbranson 2026-05-25 08:17:44 -04:00
parent c92a40bce1
commit 7eafdf3e03

View File

@ -25,6 +25,7 @@ from datetime import datetime, time as dtime
from second_brain.config import Config
from second_brain.database import get_database
from second_brain.models import Source, SourceStatus, utcnow
from second_brain.settings_store import get_settings, in_active_window
class Scheduler:
@ -51,6 +52,38 @@ class Scheduler:
from second_brain.models import Extraction
db = get_database()
# Pull the DB-side runtime knobs once at the top of the run. The
# dashboard edits these; settings.toml is the static fallback. The
# values are snapshotted for the duration of this run so a mid-run
# toggle doesn't half-apply (consistent with the way max_calls is
# tracked locally).
with db.session() as sess:
settings = get_settings(sess)
extraction_enabled = settings.extraction_enabled
db_window = (
settings.extraction_active_hours_start,
settings.extraction_active_hours_end,
)
max_items = int(settings.extraction_max_items_per_run or 0)
if not extraction_enabled:
print("[Scheduler] extraction_enabled=false in pipeline_settings. Skipping run.")
return
# DB window (if set) overrides the static settings.toml window. If
# only one side is set, treat the other as unbounded (no constraint).
if db_window[0] is not None or db_window[1] is not None:
self.window_start = db_window[0] or dtime(0, 0)
self.window_end = db_window[1] or dtime(23, 59)
if not in_active_window(datetime.now().time(), db_window[0], db_window[1]):
print(
f"[Scheduler] outside extraction active window "
f"{_fmt_opt(db_window[0])}{_fmt_opt(db_window[1])}. Skipping run."
)
return
assembler = ContextAssembler(self.config)
ex_cfg = self.config.extractor
engine = ExtractionEngine(
@ -74,10 +107,14 @@ class Scheduler:
print(
f"[Scheduler] backend={self.backend} window "
f"{_fmt(self.window_start)}{_fmt(self.window_end)} cap {cap_desc} "
f"gap {self.min_gap_seconds}s"
f"gap {self.min_gap_seconds}s max_items={max_items or ''}"
)
while self._in_window():
# DB-side hard cap on items per run (0 = unlimited).
if max_items and processed >= max_items:
print(f"[Scheduler] reached extraction_max_items_per_run={max_items}. Done.")
break
# Reset hourly buckets
elapsed = time.time() - hour_start
if elapsed >= 3600:
@ -191,4 +228,8 @@ def _fmt(t: dtime) -> str:
return t.strftime("%H:%M")
def _fmt_opt(t: dtime | None) -> str:
return t.strftime("%H:%M") if t is not None else ""
__all__ = ["Scheduler"]