diff --git a/.claude/agents/test-runner.md b/.claude/agents/test-runner.md index 144bfd245..ab646b724 100644 --- a/.claude/agents/test-runner.md +++ b/.claude/agents/test-runner.md @@ -213,6 +213,7 @@ The test suite covers: - **Locking** (scheduler_tests/test_locking.py) - Redis lock acquisition and renewal - **Cron** (scheduler_tests/test_cron.py) - Cron expression parsing, next run calculation - **Config** (scheduler_tests/test_config.py) - Scheduler configuration loading +- **Sync Loop** (scheduler_tests/test_sync_loop.py) - Periodic schedule sync idempotence; `updated_at` invariant for run-time writes (#420) ## Performance Notes (2026-02-05) @@ -227,9 +228,9 @@ The test suite covers: ## Test Suite Statistics -**Total Tests**: ~2,207 tests across 120 test files +**Total Tests**: ~2,211 tests across 121 test files **Smoke Tests**: ~578 tests (fast, no agent creation) -**Unit Tests**: ~86 tests (no backend needed, rate limit detection, watchdog logic, context formula, OTel trace logging, file upload, voice transcription, inter-agent timeout) +**Unit Tests**: ~90 tests (no backend needed, rate limit detection, watchdog logic, context formula, OTel trace logging, file upload, voice transcription, inter-agent timeout, scheduler sync loop) **Core Tests (not slow)**: ~2,076 tests **Slow Tests**: ~89 tests (chat execution, fleet ops, system agent ops, execution termination) **WebSocket Tests**: ~10 tests (web terminal, execution streaming) @@ -263,8 +264,18 @@ Use these thresholds to assess test health (based on **executed** tests, not inc | Test File | Description | Tests Added | |-----------|-------------|-------------| +| `scheduler_tests/test_sync_loop.py` | Scheduler sync loop idempotence — regression test for self-triggering `updated_at` bump (#420) | 4 tests | | `test_inter_agent_timeout_unit.py` | Inter-agent timeout honors per-agent config (#418) | 7 tests | +**Scheduler Sync Loop (#420)** (`scheduler_tests/test_sync_loop.py`): + +- `test_update_run_times_does_not_bump_updated_at` — `update_schedule_run_times` must leave `updated_at` unchanged +- `test_update_process_run_times_does_not_bump_updated_at` — same invariant for process schedules +- `test_sync_is_noop_when_db_unchanged` — three consecutive `_sync_agent_schedules` ticks with no DB edits produce zero `_add_job` / `_remove_job` calls +- `test_sync_detects_legitimate_config_change` — a cron edit that bumps `updated_at` still triggers remove-and-re-add (guards against over-correcting) + +Direct DB + in-process `SchedulerService` tests using the existing `db_with_data` and `mock_lock_manager` fixtures. No backend container required. + **Inter-Agent Timeout Fix (#418)** (`test_inter_agent_timeout_unit.py`): - `test_fan_out_request_allows_omitted_timeout` — FanOutRequest accepts no `timeout_seconds`, defaults to None diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md index 2912ea238..cc74035af 100644 --- a/docs/memory/feature-flows.md +++ b/docs/memory/feature-flows.md @@ -11,6 +11,7 @@ | Date | ID | Feature | Flow | |------|-----|---------|------| +| 2026-04-20 | #420 | Scheduler sync loop fix — `update_schedule_run_times` no longer bumps `updated_at`, stopping the self-triggering re-register of every schedule per tick | [scheduler-service.md](feature-flows/scheduler-service.md) | | 2026-04-20 | #418 | Inter-agent timeout honors per-agent `execution_timeout_seconds` — removed 600s hardcoded defaults in MCP `chat_with_agent`/`fan_out` tools and fan-out service; HTTP client ceiling bumped to platform max (7200s) | [fan-out.md](feature-flows/fan-out.md), [mcp-orchestration.md](feature-flows/mcp-orchestration.md), [parallel-headless-execution.md](feature-flows/parallel-headless-execution.md) | | 2026-04-19 | #211 | Auto-propagate global GitHub PAT to running agents on update — per-agent PAT holders and agents without `GITHUB_PAT` in `.env` are skipped; delete does NOT propagate | [github-sync.md](feature-flows/github-sync.md), [platform-settings.md](feature-flows/platform-settings.md) | | 2026-04-19 | #378 | Cleanup service Phase 3 just-in-time re-verify + parallel per-agent fan-out — eliminates phantom stale-slot failures for still-running tasks; adds residual-race observability log | [cleanup-service.md](feature-flows/cleanup-service.md) | diff --git a/docs/memory/feature-flows/scheduler-service.md b/docs/memory/feature-flows/scheduler-service.md index da4a27f59..2ee0e86ac 100644 --- a/docs/memory/feature-flows/scheduler-service.md +++ b/docs/memory/feature-flows/scheduler-service.md @@ -556,6 +556,10 @@ async def _sync_agent_schedules(self): Identical pattern to agent schedule sync but operates on `_process_schedule_snapshot` and uses `_add_process_job()` / `_remove_process_job()`. Reads from `process_schedules` table via `db.list_all_process_schedules()`. +**Invariant — run-time writes must not bump `updated_at`** (Issue #420): + +The sync loop compares `(enabled, updated_at)` to detect config changes. `update_schedule_run_times()` and `update_process_schedule_run_times()` therefore write `last_run_at` / `next_run_at` only — they must NOT touch `updated_at`. Bumping it produced a self-triggering loop where each sync tick saw its own previous `_add_job` write, flagged every schedule as "updated", and re-registered all N jobs once per tick. Legitimate config edits still bump `updated_at` via `update_schedule()` / `set_schedule_enabled()` in the backend, so user-initiated changes are still detected. + --- ## Flow 7: Manual Trigger (via Dedicated Scheduler) @@ -758,7 +762,7 @@ On startup, `_recover_pending_retries()` queries executions with `status='pendin | Method | Line | SQL | Purpose | |--------|------|-----|---------| -| `update_schedule_run_times()` | 166-190 | `UPDATE agent_schedules SET last_run_at, next_run_at` | Track execution times | +| `update_schedule_run_times()` | 184-215 | `UPDATE agent_schedules SET last_run_at, next_run_at` (does NOT touch `updated_at` — Issue #420) | Track execution times | ### Agent Execution Operations @@ -781,7 +785,7 @@ On startup, `_recover_pending_retries()` queries executions with `status='pendin | `list_all_process_schedules()` | 488-496 | `SELECT * FROM process_schedules` | Sync detection | | `list_process_schedules(process_id)` | 498-506 | `SELECT ... WHERE process_id = ?` | Per-process list | | `create_process_schedule()` | 508-558 | `INSERT INTO process_schedules` | Create schedule | -| `update_process_schedule_run_times()` | 560-584 | `UPDATE process_schedules SET last_run_at, next_run_at` | Track run times | +| `update_process_schedule_run_times()` | 685-715 | `UPDATE process_schedules SET last_run_at, next_run_at` (does NOT touch `updated_at` — Issue #420) | Track run times | | `delete_process_schedule()` | 586-592 | `DELETE FROM process_schedules WHERE id = ?` | Delete single | | `delete_process_schedules_for_process()` | 594-600 | `DELETE ... WHERE process_id = ?` | Delete all for process | | `create_process_schedule_execution()` | 602-639 | `INSERT INTO process_schedule_executions` | Create execution record | diff --git a/src/backend/db/schedules.py b/src/backend/db/schedules.py index ce8d396c8..4557fac5b 100644 --- a/src/backend/db/schedules.py +++ b/src/backend/db/schedules.py @@ -461,11 +461,20 @@ def set_schedule_enabled(self, schedule_id: str, enabled: bool) -> bool: return cursor.rowcount > 0 def update_schedule_run_times(self, schedule_id: str, last_run_at: datetime = None, next_run_at: datetime = None) -> bool: - """Update schedule run timestamps.""" + """Update schedule run timestamps. + + Does NOT bump ``updated_at`` — that column signals config changes and + is watched by the scheduler service's sync loop. Bumping it here caused + a self-triggering loop that re-registered every schedule once per tick + (Issue #420). + """ + if last_run_at is None and next_run_at is None: + return False + with get_db_connection() as conn: cursor = conn.cursor() - updates = ["updated_at = ?"] - params = [utc_now_iso()] + updates = [] + params = [] if last_run_at: updates.append("last_run_at = ?") diff --git a/src/scheduler/database.py b/src/scheduler/database.py index a4f427f07..718c1de11 100644 --- a/src/scheduler/database.py +++ b/src/scheduler/database.py @@ -187,11 +187,20 @@ def update_schedule_run_times( last_run_at: datetime = None, next_run_at: datetime = None ) -> bool: - """Update schedule run timestamps.""" + """Update schedule run timestamps. + + Does NOT bump ``updated_at`` — that column signals config changes and + is watched by the sync loop in ``SchedulerService._sync_agent_schedules``. + Bumping it on every run would cause a self-triggering sync loop that + re-registers every schedule once per tick (Issue #420). + """ + if last_run_at is None and next_run_at is None: + return False + with self.get_connection() as conn: cursor = conn.cursor() - updates = ["updated_at = ?"] - params = [datetime.utcnow().isoformat()] + updates = [] + params = [] if last_run_at: updates.append("last_run_at = ?") @@ -688,11 +697,18 @@ def update_process_schedule_run_times( last_run_at: datetime = None, next_run_at: datetime = None ) -> bool: - """Update process schedule run timestamps.""" + """Update process schedule run timestamps. + + Does NOT bump ``updated_at`` — same rationale as + :meth:`update_schedule_run_times` (Issue #420). + """ + if last_run_at is None and next_run_at is None: + return False + with self.get_connection() as conn: cursor = conn.cursor() - updates = ["updated_at = ?"] - params = [datetime.utcnow().isoformat()] + updates = [] + params = [] if last_run_at: updates.append("last_run_at = ?") diff --git a/tests/scheduler_tests/test_sync_loop.py b/tests/scheduler_tests/test_sync_loop.py new file mode 100644 index 000000000..93bc2bcc3 --- /dev/null +++ b/tests/scheduler_tests/test_sync_loop.py @@ -0,0 +1,175 @@ +""" +Regression tests for the scheduler sync loop (Issue #420). + +The periodic sync loop must be idempotent: if nothing has changed in the DB, +consecutive sync ticks must not re-register jobs. The original bug was a +self-triggering loop where `_add_job` wrote a fresh `updated_at` via +`update_schedule_run_times`, which the next sync tick then interpreted as a +config change and re-added the job — amplifying to N jobs/minute × fleet size. +""" + +# Path setup must happen before scheduler imports +import sys +from pathlib import Path +_this_file = Path(__file__).resolve() +_src_path = str(_this_file.parent.parent.parent / 'src') +if _src_path not in sys.path: + sys.path.insert(0, _src_path) + +from datetime import datetime +from unittest.mock import MagicMock, patch + +import pytest + +from scheduler.service import SchedulerService +from scheduler.database import SchedulerDatabase +from scheduler.locking import LockManager + + +class TestSyncLoopIdempotence: + """Sync loop must not re-register schedules when nothing has changed.""" + + @pytest.mark.asyncio + async def test_update_run_times_does_not_bump_updated_at( + self, db_with_data: SchedulerDatabase + ): + """ + `update_schedule_run_times` must not bump `updated_at`. That column + signals config changes and is watched by `_sync_agent_schedules`; + bumping it on every run caused the sync loop to fire perpetually. + """ + before = db_with_data.get_schedule("schedule-1") + assert before is not None + + db_with_data.update_schedule_run_times( + "schedule-1", + last_run_at=datetime.utcnow(), + next_run_at=datetime(2099, 1, 1, 9, 0, 0), + ) + + after = db_with_data.get_schedule("schedule-1") + assert after.updated_at == before.updated_at, ( + "update_schedule_run_times must not modify updated_at — doing so " + "triggers a self-reinforcing sync loop (Issue #420)" + ) + assert after.last_run_at is not None + assert after.next_run_at is not None + + @pytest.mark.asyncio + async def test_update_process_run_times_does_not_bump_updated_at( + self, initialized_db: str + ): + """Same invariant for process schedules.""" + db = SchedulerDatabase(database_path=initialized_db) + db.ensure_process_schedules_table() + + import sqlite3 + now_iso = datetime.utcnow().isoformat() + conn = sqlite3.connect(initialized_db) + try: + conn.execute( + """ + INSERT INTO process_schedules ( + id, process_id, process_name, trigger_id, + cron_expression, enabled, timezone, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ("ps-1", "proc-1", "test-process", "trig-1", + "0 * * * *", 1, "UTC", now_iso, now_iso), + ) + conn.commit() + finally: + conn.close() + + before = db.get_process_schedule("ps-1") + assert before is not None + + db.update_process_schedule_run_times( + "ps-1", + last_run_at=datetime.utcnow(), + next_run_at=datetime(2099, 1, 1, 0, 0, 0), + ) + + after = db.get_process_schedule("ps-1") + assert after.updated_at == before.updated_at, ( + "update_process_schedule_run_times must not modify updated_at " + "(Issue #420)" + ) + + @pytest.mark.asyncio + async def test_sync_is_noop_when_db_unchanged( + self, db_with_data: SchedulerDatabase, mock_lock_manager: LockManager + ): + """ + After `initialize()` registers jobs, calling `_sync_agent_schedules` + repeatedly with no DB edits must not call `_add_job` or `_remove_job` + again. This is the direct regression test for Issue #420. + """ + service = SchedulerService( + database=db_with_data, + lock_manager=mock_lock_manager, + ) + service.initialize() + + try: + # Reset spies AFTER init so we only observe sync-tick behavior + with patch.object(service, "_add_job") as add_spy, \ + patch.object(service, "_remove_job") as remove_spy: + await service._sync_agent_schedules() + await service._sync_agent_schedules() + await service._sync_agent_schedules() + + assert add_spy.call_count == 0, ( + f"Sync re-registered jobs {add_spy.call_count} times with " + "no DB changes — self-triggering loop regression (#420)" + ) + assert remove_spy.call_count == 0 + finally: + service.shutdown() + + @pytest.mark.asyncio + async def test_sync_detects_legitimate_config_change( + self, db_with_data: SchedulerDatabase, mock_lock_manager: LockManager + ): + """ + A real config edit (one that bumps `updated_at` via SQL) must still + trigger the sync loop's update branch. This guards against over- + correcting the fix. + """ + service = SchedulerService( + database=db_with_data, + lock_manager=mock_lock_manager, + ) + service.initialize() + + try: + # Simulate a user editing cron_expression via the backend router, + # which bumps updated_at. + import sqlite3 + conn = sqlite3.connect(db_with_data.database_path) + try: + conn.execute( + "UPDATE agent_schedules SET cron_expression = ?, " + "updated_at = ? WHERE id = ?", + ("30 9 * * *", datetime.utcnow().isoformat(), "schedule-1"), + ) + conn.commit() + finally: + conn.close() + + with patch.object(service, "_add_job") as add_spy, \ + patch.object(service, "_remove_job") as remove_spy: + await service._sync_agent_schedules() + + # Sync should remove + re-add exactly the edited schedule + add_calls = [c for c in add_spy.call_args_list + if c.args and c.args[0].id == "schedule-1"] + remove_calls = [c for c in remove_spy.call_args_list + if c.args and c.args[0] == "schedule-1"] + assert len(add_calls) == 1, ( + f"Edited schedule not re-added (got {len(add_calls)} calls)" + ) + assert len(remove_calls) == 1 + finally: + service.shutdown()