feat: framework logic-change detection for live components

python/cocoindex/_internal/core.pyi

@@ -313,6 +313,7 @@ class LiveComponentController:
     def read_committed_state_async(
         self, key: StableKey
     ) -> Coroutine[Any, Any, bytes | None]: ...
+    def processing_unchanged_async(self) -> Coroutine[Any, Any, bool]: ...
     def write_committed_state_async(
         self, key: StableKey, value: bytes
     ) -> Coroutine[Any, Any, None]: ...

python/cocoindex/_internal/live_component.py

@@ -438,6 +438,23 @@ async def write_committed_state(self, key: StableKey, value: Any) -> None:
         controller = self._require_controller()
         await controller.write_committed_state_async(key, serialize(value))
 
+    async def processing_unchanged(self) -> bool:
+        """Whether this component's processing logic is unchanged since its
+        last committed scan.
+
+        The framework persists the subtree's logic-dependency set after each
+        committed scan (``update_full`` / incremental ``update``); this checks
+        that set still resolves against the currently registered logic. A
+        durable connector pairs this with its own cursor to decide whether the
+        startup full scan can be skipped — ``<durable cursor> and await
+        subscriber.processing_unchanged()``.
+
+        Failure-safe: returns ``False`` when no scan has been committed yet or
+        when any dependency's code changed — either way, re-scan.
+        """
+        controller = self._require_controller()
+        return await controller.processing_unchanged_async()
+
     async def report_exception(self, exc: BaseException) -> None:
         """Route an exception raised during ``process_live`` to the parent's exception handler chain.
 
@@ -557,6 +574,16 @@ async def write_committed_state(self, key: StableKey, value: Any) -> None:
         """
         await self._operator.write_committed_state(key, value)
 
+    async def processing_unchanged(self) -> bool:
+        """Whether the processing logic is unchanged since the last scan.
+
+        Delegates to :meth:`LiveComponentOperator.processing_unchanged` so a
+        ``watch()`` implementation can gate its startup ``update_all()`` on the
+        framework-computed logic-change signal, paired with its own durable
+        cursor.
+        """
+        return await self._operator.processing_unchanged()
+
 
 class _MountEachLiveComponent:
     """Internal LiveComponent created by mount_each() for LiveMapFeed/LiveMapView items."""

python/cocoindex/connectors/oci_object_storage/_source.py

@@ -59,11 +59,6 @@
 # concrete need emerges (per AGENTS.md "Minimize API surface").
 _SKEW_TOLERANCE = timedelta(seconds=5)
 
-# Committed-state key under which the live view records the user-declared
-# ``logic_version`` after a successful full scan. Read back on a later run to
-# decide whether the startup scan can be skipped. See ``list_objects``.
-_SCAN_VERSION_KEY = "__coco_oci_scan_version__"
-
 
 def _parse_event_time(s: Any) -> datetime | None:
     """Parse an OCI event ``eventTime`` (ISO-8601). Returns ``None`` on
@@ -264,7 +259,7 @@ class OCIWalker:
     _path_matcher: file.FilePathMatcher
     _max_file_size: int | None
     _live_stream: LiveStream[bytes] | None
-    _logic_version: str | None
+    _durable_stream: bool
 
     def __init__(
         self,
@@ -276,7 +271,7 @@ def __init__(
         path_matcher: file.FilePathMatcher | None = None,
         max_file_size: int | None = None,
         live_stream: LiveStream[bytes] | None = None,
-        logic_version: str | None = None,
+        durable_stream: bool = False,
     ) -> None:
         self._client = client
         self._namespace = namespace
@@ -285,7 +280,7 @@ def __init__(
         self._path_matcher = path_matcher or file.MatchAllFilePathMatcher()
         self._max_file_size = max_file_size
         self._live_stream = live_stream
-        self._logic_version = logic_version
+        self._durable_stream = durable_stream
 
     @property
     def namespace(self) -> str:
@@ -380,12 +375,13 @@ class _LiveOCIItems:
        ``send()`` calls parked on ``_ready_complete``.
     6. Await the stream task; it runs until cancellation.
 
-    Skip-scan mode (opt-in via ``OCIWalker(logic_version=...)``): when a prior
-    run committed the same ``logic_version``, steps 1+3 are replaced — the scan
-    is skipped and ``cutoff`` is ``None`` so the durable stream's replayed
+    Skip-scan mode (opt-in via ``OCIWalker(durable_stream=True)``): when the
+    framework reports the processing logic unchanged since the last committed
+    scan (``subscriber.processing_unchanged()``), steps 1+3 are replaced — the
+    scan is skipped and ``cutoff`` is ``None`` so the durable stream's replayed
     backlog (resumed from its committed cursor) is processed rather than
-    dropped. The committed version is (re)written after step 4 on any run that
-    actually scans.
+    dropped. The logic-change signal is framework-computed; the connector only
+    asserts the stream is durable.
     """
 
     __slots__ = ("_walker", "_live_stream", "_watch_guard")
@@ -408,11 +404,13 @@ async def watch(self, subscriber: LiveMapSubscriber[str, OCIFile]) -> None:
             await self._watch(subscriber)
 
     async def _watch(self, subscriber: LiveMapSubscriber[str, OCIFile]) -> None:
-        logic_version = self._walker._logic_version
-        skip_scan = False
-        if logic_version is not None:
-            stored = await subscriber.read_committed_state(_SCAN_VERSION_KEY)
-            skip_scan = stored == logic_version
+        # Skip the startup scan only when the user asserts a durable stream AND
+        # the framework reports the processing logic unchanged since the last
+        # committed scan. The framework persists the subtree's logic-dependency
+        # set after each committed scan, so no manual version is needed.
+        skip_scan = (
+            self._walker._durable_stream and await subscriber.processing_unchanged()
+        )
 
         # With no scan, the wall-clock cutoff must not fire: the durable stream
         # (resumed from its committed cursor) replays the downtime backlog, and
@@ -425,10 +423,6 @@ async def _watch(self, subscriber: LiveMapSubscriber[str, OCIFile]) -> None:
             if not skip_scan:
                 await subscriber.update_all()
             await subscriber.mark_ready()
-            # Record the version only after mark_ready (the scan has committed),
-            # so a later run skips only when the bootstrap durably landed.
-            if not skip_scan and logic_version is not None:
-                await subscriber.write_committed_state(_SCAN_VERSION_KEY, logic_version)
             adapter.mark_ready_complete()
             await stream_task
         finally:
@@ -646,7 +640,7 @@ def list_objects(
     path_matcher: file.FilePathMatcher | None = None,
     max_file_size: int | None = None,
     live_stream: LiveStream[bytes] | None = None,
-    logic_version: str | None = None,
+    durable_stream: bool = False,
 ) -> OCIWalker:
     """List objects in an OCI bucket and yield file entries.
 
@@ -666,14 +660,15 @@ def list_objects(
             path, after prefix stripping).
         max_file_size: Skip objects larger than this size in bytes.
         live_stream: Optional ``LiveStream[bytes]`` of OCI Object Storage events.
-        logic_version: Opt into skipping the startup full scan on reruns. When
-            set, the live view records this version after a successful scan and,
-            on a later run, skips the scan if the recorded version matches —
-            relying on the durable stream to replay the downtime backlog from
-            its committed cursor. You MUST bump this whenever your processing
-            logic changes (otherwise stale state is silently kept), and the
-            stream MUST be durable (e.g. a Kafka consumer with a stable
-            ``group_id`` and committed offsets). Leave unset (``None``) to always
+        durable_stream: Opt into skipping the startup full scan on reruns. When
+            ``True``, the scan is skipped on a later run if the framework reports
+            the processing logic unchanged since the last committed scan —
+            relying on the durable stream to replay the downtime backlog from its
+            committed cursor. The logic-change check is automatic (no manual
+            version to bump). You are responsible for the durability guarantee:
+            the stream MUST resume from a committed cursor (e.g. a Kafka consumer
+            with a stable ``group_id`` and committed offsets); otherwise the
+            downtime backlog is lost. Leave ``False`` (the default) to always
             scan on startup.
     """
     return OCIWalker(
@@ -684,5 +679,5 @@ def list_objects(
         path_matcher=path_matcher,
         max_file_size=max_file_size,
         live_stream=live_stream,
-        logic_version=logic_version,
+        durable_stream=durable_stream,
     )

python/tests/connectors/test_oci_object_storage.py

@@ -201,9 +201,6 @@ def get_object(self, **kwargs: Any) -> _MockGetObjectResponse:
     list_objects,
     read,
 )
-from cocoindex.connectors.oci_object_storage._source import (  # noqa: E402
-    _SCAN_VERSION_KEY,
-)
 from cocoindex.resources.file import PatternFilePathMatcher  # noqa: E402
 
 
@@ -360,28 +357,26 @@ async def test_oci_exists_false_on_404(
 class _MockMapSubscriber:
     """Records LiveMapSubscriber calls; auto-resolved handles.
 
-    ``committed`` backs the ``read_committed_state`` / ``write_committed_state``
-    primitives; pass a shared dict to simulate state persisting across runs.
+    ``processing_unchanged`` is the framework-computed logic-change signal the
+    skip-scan path gates on; pass ``True`` to simulate a rerun whose processing
+    logic is unchanged since the last committed scan.
     """
 
-    def __init__(self, committed: dict[Any, Any] | None = None) -> None:
+    def __init__(self, processing_unchanged: bool = False) -> None:
         self.update_all_called = False
         self.mark_ready_called = False
         self.updates: list[tuple[str, OCIFile]] = []
         self.deletes: list[str] = []
-        self.committed: dict[Any, Any] = {} if committed is None else committed
+        self._processing_unchanged = processing_unchanged
 
     async def update_all(self) -> None:
         self.update_all_called = True
 
     async def mark_ready(self) -> None:
         self.mark_ready_called = True
 
-    async def read_committed_state(self, key: Any) -> Any | None:
-        return self.committed.get(key)
-
-    async def write_committed_state(self, key: Any, value: Any) -> None:
-        self.committed[key] = value
+    async def processing_unchanged(self) -> bool:
+        return self._processing_unchanged
 
     async def update(self, key: str, value: OCIFile) -> ReadyAwaitable:
         self.updates.append((key, value))
@@ -720,73 +715,70 @@ async def test_oci_live_view_cross_bucket_event_filtered(
 
 
 # ===========================================================================
-# Live view — skip-full-scan on unchanged logic (logic_version)
+# Live view — skip-full-scan on unchanged logic (durable_stream)
 # ===========================================================================
 
 
 @pytest.mark.asyncio
-async def test_oci_live_view_no_logic_version_always_scans(
+async def test_oci_live_view_not_durable_always_scans(
     oci_client: MockObjectStorageClient,
 ) -> None:
-    """Without ``logic_version`` the startup scan always runs and nothing is
-    persisted (behavior unchanged)."""
+    """Without ``durable_stream`` the startup scan always runs, even when the
+    processing logic is unchanged — the durability opt-in gates the skip."""
     oci_client.put("a.txt", b"a")
     stream = _ManualLiveStream()
     walker = list_objects(oci_client, "ns", "bucket", live_stream=stream)
 
-    sub = _MockMapSubscriber()
+    sub = _MockMapSubscriber(processing_unchanged=True)
     items = _live_items(walker)
     watch_task = asyncio.create_task(items.watch(sub))  # type: ignore[arg-type]
 
     await _drive_to_ready(stream, sub)
 
     assert sub.update_all_called
-    assert sub.committed == {}
 
     stream.end()
     await watch_task
 
 
 @pytest.mark.asyncio
-async def test_oci_live_view_first_run_scans_and_records_version(
+async def test_oci_live_view_durable_scans_when_logic_changed(
     oci_client: MockObjectStorageClient,
 ) -> None:
-    """First run with ``logic_version`` set (nothing committed yet) scans, then
-    records the version once the scan has committed."""
+    """With ``durable_stream=True`` but the logic changed (or never scanned),
+    ``processing_unchanged()`` is False, so the startup scan still runs."""
     oci_client.put("a.txt", b"a")
     stream = _ManualLiveStream()
     walker = list_objects(
-        oci_client, "ns", "bucket", live_stream=stream, logic_version="v1"
+        oci_client, "ns", "bucket", live_stream=stream, durable_stream=True
     )
 
-    sub = _MockMapSubscriber()
+    sub = _MockMapSubscriber(processing_unchanged=False)
     items = _live_items(walker)
     watch_task = asyncio.create_task(items.watch(sub))  # type: ignore[arg-type]
 
     await _drive_to_ready(stream, sub)
 
-    assert sub.update_all_called
-    assert sub.committed == {_SCAN_VERSION_KEY: "v1"}
+    assert sub.update_all_called  # logic changed (or first run) → full scan
 
     stream.end()
     await watch_task
 
 
 @pytest.mark.asyncio
-async def test_oci_live_view_skips_scan_when_version_matches(
+async def test_oci_live_view_durable_skips_scan_when_unchanged(
     oci_client: MockObjectStorageClient,
 ) -> None:
-    """A rerun whose committed version matches skips the scan, and the cutoff is
-    disabled so the replayed backlog (here a far-past event) is still processed.
-    """
+    """With ``durable_stream=True`` and the logic unchanged, the scan is skipped
+    and the cutoff is disabled so the replayed backlog (here a far-past event)
+    is still processed."""
     oci_client.put("a.txt", b"a")
     stream = _ManualLiveStream()
     walker = list_objects(
-        oci_client, "ns", "bucket", live_stream=stream, logic_version="v1"
+        oci_client, "ns", "bucket", live_stream=stream, durable_stream=True
     )
 
-    # Simulate a prior run having bootstrapped at "v1".
-    sub = _MockMapSubscriber(committed={_SCAN_VERSION_KEY: "v1"})
+    sub = _MockMapSubscriber(processing_unchanged=True)
     items = _live_items(walker)
     watch_task = asyncio.create_task(items.watch(sub))  # type: ignore[arg-type]
 
@@ -804,31 +796,6 @@ async def test_oci_live_view_skips_scan_when_version_matches(
     await watch_task
 
 
-@pytest.mark.asyncio
-async def test_oci_live_view_rescans_when_version_changed(
-    oci_client: MockObjectStorageClient,
-) -> None:
-    """A rerun whose ``logic_version`` differs from the committed one rescans and
-    records the new version."""
-    oci_client.put("a.txt", b"a")
-    stream = _ManualLiveStream()
-    walker = list_objects(
-        oci_client, "ns", "bucket", live_stream=stream, logic_version="v2"
-    )
-
-    sub = _MockMapSubscriber(committed={_SCAN_VERSION_KEY: "v1"})
-    items = _live_items(walker)
-    watch_task = asyncio.create_task(items.watch(sub))  # type: ignore[arg-type]
-
-    await _drive_to_ready(stream, sub)
-
-    assert sub.update_all_called  # logic changed → full scan
-    assert sub.committed == {_SCAN_VERSION_KEY: "v2"}
-
-    stream.end()
-    await watch_task
-
-
 @pytest.mark.asyncio
 async def test_oci_live_view_max_file_size_filters_via_size(
     oci_client: MockObjectStorageClient,