feat: add SyncScheduler for periodic incremental sync

Implements SyncScheduler that runs a daemon background thread to call
SyncEngine.sync() on all registered connectors at a configurable interval.
Provides run_once() as a synchronous helper for testing.  Disconnected
connectors are skipped each cycle; errors are logged without stopping the
loop.  4 tests covering run_once, disconnected skip, start/stop, and
per-connector chunk counts.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/openjarvis/connectors/scheduler.py

@@ -0,0 +1,149 @@
+"""SyncScheduler — background thread for periodic incremental connector syncs.
+
+Registers connectors for timed re-sync and runs them on a configurable
+interval.  Designed to be long-lived (daemon thread) inside a running
+OpenJarvis server process.
+
+Typical usage::
+
+    store = KnowledgeStore(db_path=":memory:")
+    pipeline = IngestionPipeline(store)
+    engine = SyncEngine(pipeline)
+
+    scheduler = SyncScheduler(engine, interval_seconds=3600)
+    scheduler.add(gmail_connector)
+    scheduler.add(slack_connector)
+    scheduler.start()       # background thread syncs every hour
+
+    # Later:
+    scheduler.stop()
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Dict, List, Optional
+
+from openjarvis.connectors._stubs import BaseConnector
+from openjarvis.connectors.sync_engine import SyncEngine
+
+logger = logging.getLogger(__name__)
+
+
+class SyncScheduler:
+    """Runs incremental sync for all registered connectors on a schedule.
+
+    Parameters
+    ----------
+    sync_engine:
+        The :class:`~openjarvis.connectors.sync_engine.SyncEngine` used to
+        drive each connector's sync (handles checkpointing).
+    interval_seconds:
+        How often (in seconds) to sync all connected connectors.
+        Defaults to ``3600`` (one hour).
+    """
+
+    def __init__(self, sync_engine: SyncEngine, interval_seconds: int = 3600) -> None:
+        self._engine = sync_engine
+        self._interval = interval_seconds
+        self._thread: Optional[threading.Thread] = None
+        self._stop = threading.Event()
+        self._connectors: List[BaseConnector] = []
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def add(self, connector: BaseConnector) -> None:
+        """Register a connector for scheduled sync.
+
+        Parameters
+        ----------
+        connector:
+            Any :class:`~openjarvis.connectors._stubs.BaseConnector` instance.
+            Only connected connectors are synced during each cycle.
+        """
+        self._connectors.append(connector)
+
+    def start(self) -> None:
+        """Start the background sync thread.
+
+        The thread is a daemon so it does not prevent process exit.  The
+        first sync run occurs after one full *interval_seconds* wait.
+        Calling :meth:`start` on an already-running scheduler is a no-op.
+        """
+        if self._thread is not None and self._thread.is_alive():
+            logger.debug("SyncScheduler already running; ignoring start()")
+            return
+
+        self._stop.clear()
+        self._thread = threading.Thread(
+            target=self._loop, daemon=True, name="sync_scheduler"
+        )
+        self._thread.start()
+        logger.info(
+            "SyncScheduler started (interval=%ds, connectors=%d)",
+            self._interval,
+            len(self._connectors),
+        )
+
+    def stop(self) -> None:
+        """Stop the background sync thread.
+
+        Signals the thread to exit and waits up to 5 seconds for it to
+        finish the current sync cycle.  Safe to call even when the scheduler
+        is not running.
+        """
+        self._stop.set()
+        if self._thread is not None and self._thread.is_alive():
+            self._thread.join(timeout=5)
+        self._thread = None
+        logger.info("SyncScheduler stopped")
+
+    def run_once(self) -> Dict[str, int]:
+        """Sync all connected connectors once (synchronous, non-blocking test helper).
+
+        Returns
+        -------
+        dict[str, int]
+            Mapping of ``connector_id`` → number of new chunks ingested.
+            Only connectors that are currently connected are included.
+        """
+        results: Dict[str, int] = {}
+        for conn in self._connectors:
+            if conn.is_connected():
+                try:
+                    count = self._engine.sync(conn)
+                    results[conn.connector_id] = count
+                except Exception as exc:
+                    logger.error(
+                        "run_once sync failed for %s: %s", conn.connector_id, exc
+                    )
+        return results
+
+    # ------------------------------------------------------------------
+    # Internal
+    # ------------------------------------------------------------------
+
+    def _loop(self) -> None:
+        """Background thread body: wait *interval* seconds then sync."""
+        while not self._stop.wait(timeout=self._interval):
+            for conn in self._connectors:
+                if conn.is_connected():
+                    try:
+                        count = self._engine.sync(conn)
+                        logger.debug(
+                            "Scheduled sync completed for %s (%d items)",
+                            conn.connector_id,
+                            count,
+                        )
+                    except Exception as exc:
+                        logger.error(
+                            "Scheduled sync failed for %s: %s",
+                            conn.connector_id,
+                            exc,
+                        )
+
+
+__all__ = ["SyncScheduler"]

tests/connectors/test_scheduler.py

@@ -0,0 +1,140 @@
+"""Tests for SyncScheduler — periodic incremental sync background thread."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from pathlib import Path
+from typing import Iterator, Optional
+
+import pytest
+
+from openjarvis.connectors._stubs import BaseConnector, Document, SyncStatus
+from openjarvis.connectors.pipeline import IngestionPipeline
+from openjarvis.connectors.scheduler import SyncScheduler
+from openjarvis.connectors.store import KnowledgeStore
+from openjarvis.connectors.sync_engine import SyncEngine
+
+# ---------------------------------------------------------------------------
+# Fixtures and helpers
+# ---------------------------------------------------------------------------
+
+
+class _FakeConnector(BaseConnector):
+    """A minimal connector that yields a fixed number of documents."""
+
+    display_name = "Fake Scheduler Connector"
+    auth_type = "filesystem"
+
+    def __init__(
+        self,
+        connector_id: str = "fake_sched",
+        *,
+        connected: bool = True,
+        doc_count: int = 2,
+    ) -> None:
+        self.connector_id = connector_id  # type: ignore[misc]
+        self._connected = connected
+        self._doc_count = doc_count
+
+    def is_connected(self) -> bool:
+        return self._connected
+
+    def disconnect(self) -> None:
+        self._connected = False
+
+    def sync(
+        self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
+    ) -> Iterator[Document]:
+        for i in range(self._doc_count):
+            yield Document(
+                doc_id=f"{self.connector_id}:{i}",
+                source=self.connector_id,
+                doc_type="note",
+                content=f"Scheduled sync document {self.connector_id}:{i}.",
+                title=f"Doc {i}",
+            )
+
+    def sync_status(self) -> SyncStatus:
+        return SyncStatus(state="idle", items_synced=self._doc_count)
+
+
+@pytest.fixture()
+def engine(tmp_path: Path) -> SyncEngine:
+    """Return a SyncEngine backed by in-memory stores."""
+    store = KnowledgeStore(db_path=":memory:")
+    pipeline = IngestionPipeline(store)
+    return SyncEngine(pipeline, state_db=str(tmp_path / "state.db"))
+
+
+# ---------------------------------------------------------------------------
+# Test 1: add + run_once syncs a connected connector
+# ---------------------------------------------------------------------------
+
+
+def test_run_once_syncs_connected_connector(engine: SyncEngine) -> None:
+    """run_once() calls engine.sync() for a connected connector."""
+    conn = _FakeConnector("conn_single", connected=True, doc_count=3)
+    scheduler = SyncScheduler(engine, interval_seconds=3600)
+    scheduler.add(conn)
+
+    results = scheduler.run_once()
+
+    assert conn.connector_id in results
+    assert results[conn.connector_id] == 3
+
+
+# ---------------------------------------------------------------------------
+# Test 2: run_once skips disconnected connectors
+# ---------------------------------------------------------------------------
+
+
+def test_run_once_skips_disconnected_connector(engine: SyncEngine) -> None:
+    """run_once() does not attempt to sync a disconnected connector."""
+    connected = _FakeConnector("conn_yes", connected=True, doc_count=1)
+    disconnected = _FakeConnector("conn_no", connected=False, doc_count=5)
+
+    scheduler = SyncScheduler(engine, interval_seconds=3600)
+    scheduler.add(connected)
+    scheduler.add(disconnected)
+
+    results = scheduler.run_once()
+
+    assert "conn_yes" in results
+    assert "conn_no" not in results
+
+
+# ---------------------------------------------------------------------------
+# Test 3: start/stop does not crash
+# ---------------------------------------------------------------------------
+
+
+def test_start_stop_does_not_crash(engine: SyncEngine) -> None:
+    """start() and stop() complete without error even with no connectors."""
+    scheduler = SyncScheduler(engine, interval_seconds=60)
+    scheduler.start()
+    assert scheduler._thread is not None
+    assert scheduler._thread.is_alive()
+    scheduler.stop()
+    # After stop the internal thread reference is cleared
+    assert scheduler._thread is None
+
+
+# ---------------------------------------------------------------------------
+# Test 4: run_once returns chunk counts per connector
+# ---------------------------------------------------------------------------
+
+
+def test_run_once_returns_chunk_counts(engine: SyncEngine) -> None:
+    """run_once() returns a mapping of connector_id → chunks ingested."""
+    conn_a = _FakeConnector("conn_a", connected=True, doc_count=2)
+    conn_b = _FakeConnector("conn_b", connected=True, doc_count=4)
+
+    scheduler = SyncScheduler(engine, interval_seconds=3600)
+    scheduler.add(conn_a)
+    scheduler.add(conn_b)
+
+    results = scheduler.run_once()
+
+    assert results["conn_a"] == 2
+    assert results["conn_b"] == 4
+    assert len(results) == 2