feat(sources): protocol-based polling source for async pipelines (PLT-1430) by kurodo3[bot] · Pull Request #139 · nauticalab/orcapod-python

kurodo3 · 2026-05-20T17:39:56Z

Summary

Adds DynamicSourceProtocol[T] — a @runtime_checkable protocol with six methods: identity(), to_config(), from_config(), poll(), fetch(), and close(); the framework handles all scheduling, cursor management, caching, and error handling
Adds Cursor[T] (with Cursor.now() convenience classmethod), PollingConfig (frozen dataclass with validated fields), and CursorInvalidatedError shared types
Adds PollingSource[T] — a RootSource wrapping any DynamicSourceProtocol; sync access uses a lazy-initialized accumulated cache; async access uses a fixed-interval polling loop with backpressure, error backoff, overrun guards, and clean shutdown
source_id is derived at construction from impl.identity() — stable and implementation-defined, not data-dependent
identity_structure() delegates to impl.identity() so the implementer controls what distinguishes two sources
Optional tag_schema / data_schema constructor params enable schema-aware operation before any data is fetched; every fetched batch is validated against declared schemas — raises InputValidationError on mismatch
Schema combining is strictly validated between consecutive batches (column set and types must match exactly) — raises InputValidationError rather than logging a warning
to_config() / from_config() use dataclasses.asdict() for PollingConfig and delegate to impl.to_config() / impl_class.from_config() for impl-specific state
Gives StreamBase.async_iter_data() a real default (wrapping iter_data()), simplifying SourceNode.async_execute to always use the async path

Test plan

56 tests in tests/test_channels/test_polling_source.py covering: types/protocol conformance, Cursor.now(), sync snapshot caching, cursor threading, last_modified tracking, async polling loop (positive/negative poll, pre-seeding, accumulation, cancellation, duration, indefinite mode), error backoff/retry, max-error termination, CursorInvalidatedError terminal handling, overrun detection, declared-schema short-circuit (output_schema/keys without fetch), schema compatibility on first fetch, combining-schema mismatch detection (column set and type drift)
Full suite: 3246 passed, 56 skipped, 0 failures

Closes PLT-1430

🤖 Generated with Claude Code

codecov · 2026-05-20T17:47:30Z

Codecov Report

❌ Patch coverage is 92.04152% with 23 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/orcapod/core/sources/polling_source.py	93.16%	16 Missing ⚠️
src/orcapod/protocols/core_protocols/sources.py	63.15%	7 Missing ⚠️

📢 Thoughts on this report? Let us know!

eywalker

We should only use Google style dictating and never mix in ReST

eywalker · 2026-05-20T17:46:30Z

    ) -> None:
        """Push all (tag, data) pairs from the wrapped stream to the output channel.

+        Delegates to :meth:`~orcapod.core.streams.base.StreamBase.async_iter_data`


Use google style docsyring — don’t use ReST

Fixed — replaced :meth:``async_iter_data`` and :class:PollingSource with plain async_iter_data and PollingSource in the async_execute docstring.

eywalker · 2026-05-20T17:47:24Z

        """Async iterator over (tag, data) pairs.

-        Subclasses should override this to provide true async iteration.
+        Default implementation wraps :meth:`iter_data` as an async generator.


Use Google style doc only — add this to Claude md as well

Fixed — replaced :meth:``iter_data`` and :class:PollingSource with plain iter_data and PollingSource in the async_iter_data docstring. Also updated CLAUDE.md and .zed/rules to explicitly prohibit ReST cross-reference roles (:meth:, :class:, :exc:, etc.) and require plain backtick notation instead.

kurodo3 · 2026-05-20T17:53:44Z

Review round summary

Addressed all docstring comments — replaced every ReST cross-reference role with plain backtick notation throughout the codebase.

Changes made (single commit `16ec04d`, plus `9a92f94` for pre-existing instances)

Files changed in this PR (ReST refs introduced by PLT-1430 work):

src/orcapod/core/streams/base.py — :meth:\iter_data`→ `` ``iter_data`` ``,:class:`PollingSource`` → PollingSource`` ``
src/orcapod/core/nodes/source_node.py — :meth:\async_iter_data`and:class:`PollingSource`` → plain backtick forms
src/orcapod/core/sources/polling_source.py — module docstring and class/method docstrings: all :class:, :exc: roles replaced
src/orcapod/types.py — :class:\Cursor`, :class:`PollingSource`` → plain backtick forms
src/orcapod/protocols/core_protocols/sources.py — :exc:\CursorInvalidatedError`` (×2) → CursorInvalidatedError`` ``

Pre-existing instances in the rest of the codebase (also fixed):

src/orcapod/core/streams/base.py:330 — :meth:\iter_data``
src/orcapod/pipeline/graph.py:612 — :meth:\Pipeline.load``

CLAUDE.md and .zed/rules updated to explicitly prohibit ReST cross-reference roles and document the correct plain backtick convention.

No ReST cross-reference roles (:meth:, :class:, :exc:, :attr:, :func:, etc.) remain anywhere in src/.

Copilot

Pull request overview

This PR introduces a protocol-based dynamic polling source to support continuously streaming data into async pipelines, along with shared cursor/config types and a default async iteration path for existing streams.

Changes:

Added Cursor[T], PollingConfig, CursorInvalidatedError, and DynamicSourceProtocol[T] to standardize dynamic source contracts.
Implemented PollingSource[T] (sync snapshot via lazy cache + async fixed-interval polling loop with backoff/overrun handling and clean shutdown).
Made StreamBase.async_iter_data() default to wrapping iter_data(), and updated SourceNode.async_execute to always use the async path; added a comprehensive test suite for the new behavior.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
tests/test_channels/test_polling_source.py	New tests covering protocol/type conformance and PollingSource sync/async behavior.
superpowers/specs/2026-05-20-polling-source-design.md	Design doc for the new polling source architecture and behavior.
superpowers/plans/2026-05-20-polling-source.md	Implementation plan/playbook for the feature and test rollout.
src/orcapod/types.py	Adds `T`, `Cursor[T]`, and `PollingConfig`.
src/orcapod/errors.py	Adds `CursorInvalidatedError`.
src/orcapod/protocols/core_protocols/sources.py	Adds `DynamicSourceProtocol[T]` definition.
src/orcapod/protocols/core_protocols/init.py	Re-exports `DynamicSourceProtocol`.
src/orcapod/core/sources/polling_source.py	Implements `PollingSource` (sync cache + async polling loop).
src/orcapod/core/sources/init.py	Exports `PollingSource` from core sources package.
src/orcapod/core/streams/base.py	Implements default `async_iter_data()` wrapper over `iter_data()`.
src/orcapod/core/nodes/source_node.py	Switches async execution to always iterate via `async_iter_data()`.
src/orcapod/pipeline/graph.py	Docstring tweak removing ReST role usage.
CLAUDE.md	Adds guidance to avoid ReST cross-reference roles in docs/docstrings.
.zed/rules	Mirrors the “no ReST roles” guidance for editor rules.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

kurodo3 · 2026-05-20T19:07:24Z

+def _run_sync(coro):
+    """Run *coro* synchronously, safe even when called from within a running loop."""
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(coro)
+    else:
+        return _get_sync_executor().submit(lambda: asyncio.run(coro)).result()
+
+


Fixed. _run_sync now accepts (async_fn, *args, **kwargs) and creates the coroutine inside the executor thread via asyncio.run(async_fn(*args, **kwargs)). This ensures the coroutine is always owned by the event loop running it, matching the data_function.py pattern. All three call sites updated.

kurodo3 · 2026-05-20T19:07:31Z

+        """Build an ``ArrowTableStream`` from raw data, returning ``None`` for empty data.
+
+        Returns:
+            ``ArrowTableStream`` if data has rows and columns, ``None`` otherwise.
+        """
+        df = pl.DataFrame(data)
+        if df.is_empty() or len(df.columns) == 0:
+            logger.debug(
+                "PollingSource %r: fetch returned empty data — skipping stream build",


Fixed. Removed the df.is_empty() guard — _try_build_stream now only returns None when len(df.columns) == 0. Zero-row DataFrames are valid (ArrowTableStream supports them) and should produce an empty-but-valid stream rather than being silently discarded.

kurodo3 · 2026-05-20T19:07:38Z

+            logger.debug("PollingSource %r: first sync access — fetching", self._source_id)
+            new_cursor, data = _run_sync(self._impl.fetch(cursor=None))
+            new_stream = self._try_build_stream(data)
+            if new_stream is not None:
+                self._schema_stream = new_stream
+                self._update_last_modified_from_cursor(new_cursor)
+            self._cursor = new_cursor
+        else:
+            # Have cache — poll for updates
+            has_new = _run_sync(self._impl.poll(cursor=self._cursor))
+            if has_new:
+                logger.debug("PollingSource %r: sync poll found new data — fetching", self._source_id)
+                new_cursor, data = _run_sync(self._impl.fetch(cursor=self._cursor))
+                new_stream = self._try_build_stream(data)
+                if new_stream is not None:
+                    self._schema_stream = self._combine(self._schema_stream, new_stream)
+                    self._update_last_modified_from_cursor(new_cursor)
+                self._cursor = new_cursor


Fixed. _update_last_modified_from_cursor(new_cursor) is now called unconditionally after every fetch() — moved outside the if new_stream is not None block in both the first-access and cache-refresh branches of _get_latest_stream.

kurodo3 · 2026-05-20T19:07:40Z

+                        new_cursor, data = await self._impl.fetch(cursor=self._cursor)
+                        new_stream = self._try_build_stream(data)
+                        if new_stream is not None:
+                            if self._schema_stream is None:
+                                self._schema_stream = new_stream
+                            else:
+                                self._schema_stream = self._combine(self._schema_stream, new_stream)
+                            self._cursor = new_cursor
+                            self._update_last_modified_from_cursor(new_cursor)
+
+                            # Emit new rows from just this fetch
+                            rows = list(new_stream.iter_data())
+                            logger.debug(
+                                "PollingSource %r: emitting %d row(s)",
+                                self._source_id,
+                                len(rows),
+                            )
+                            for item in rows:
+                                yield item
+                        else:
+                            self._cursor = new_cursor
+                    else:


Fixed. Same treatment as the sync path: self._cursor = new_cursor and self._update_last_modified_from_cursor(new_cursor) are now both called unconditionally after every successful fetch() in the async polling loop, before the if new_stream is not None branch.

kurodo3 · 2026-05-20T19:07:48Z

+            ],
+            promote_options="default",
+        )
+        return ArrowTableStream(table=combined, tag_columns=existing._tag_columns)


Fixed. _combine now uses self._tag_columns (the tuple stored on PollingSource) instead of existing._tag_columns (a private attribute of ArrowTableStream). The tag columns are constant for the lifetime of a PollingSource instance so self._tag_columns is always the correct source.

kurodo3 · 2026-05-20T19:07:49Z

+    def __init__(
+        self,
+        impl: DynamicSourceProtocol[T],
+        tag_columns: str | Collection[str],
+        polling_config: PollingConfig = PollingConfig(),
+        source_id: str | None = None,
+        label: str | None = None,
+        data_context: str | Any | None = None,
+        config: Any | None = None,
+    ) -> None:


Fixed. PollingSource.__init__ now validates all five PollingConfig fields before storing them: interval > 0, duration >= 0, max_missed_intervals >= 1, max_consecutive_errors >= 1, error_backoff_base > 0. Each raises a descriptive ValueError with the offending value.

kurodo3 · 2026-05-20T19:07:57Z

 from __future__ import annotations

-from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Any, Generic, Protocol, runtime_checkable


Fixed. Removed Generic from the from typing import ... line in sources.py. The DynamicSourceProtocol[T] parameterisation is provided by Protocol[T], so Generic is not needed here.

kurodo3 · 2026-05-20T19:07:59Z

+
+```python
+T = TypeVar("T")
+"""Generic cursor value type for :class:`Cursor` and :class:`DynamicSourceProtocol`."""


Fixed. Replaced all ReST cross-reference roles in the plan doc (:class:, :meth:, :exc:) with plain backtick notation throughout the file. The plan doc no longer contains any ReST markup.

kurodo3 · 2026-05-20T19:08:00Z

+
+@dataclass(frozen=True)
+class PollingConfig:
+    """Configuration for a :class:`~orcapod.core.sources.PollingSource`.


Fixed along with the previous instance — both occurrences were replaced as part of a sweep of all ReST roles in the plan doc.

kurodo3 · 2026-05-20T19:08:11Z

Review round 2 — all Copilot comments addressed

Commit e8e62c4 implements all fixes in a single commit. Summary:

Comment	Fix
`_run_sync` creates coroutine in caller thread	Signature changed to `(async_fn, args, kwargs)`; coroutine created inside the executor thread via `asyncio.run(async_fn(args, **kwargs))`
`_try_build_stream` treats empty DataFrame as "no data"	Removed `df.is_empty()` guard — only `len(df.columns) == 0` skips stream construction; zero-row tables are valid
Sync `last_modified` only updated when `new_stream is not None`	`_update_last_modified_from_cursor` moved outside the `if new_stream is not None` block in `_get_latest_stream`
Async `last_modified` same issue	Same fix in `async_iter_data`; cursor and timestamp now updated unconditionally after every successful `fetch()`
`_combine` reaches into `existing._tag_columns` (private attr)	Now uses `self._tag_columns` — constant for the lifetime of the instance
`PollingConfig` values not validated	`__init__` now validates all five fields with descriptive `ValueError` messages
Unused `Generic` import in `sources.py`	Removed
ReST markup in plan doc (two instances)	All `:class:`, `:meth:`, `:exc:` roles replaced with plain backtick notation throughout the plan doc

eywalker · 2026-05-20T20:00:30Z

+        self,
+        impl: DynamicSourceProtocol[T],
+        tag_columns: str | Collection[str],
+        polling_config: PollingConfig = PollingConfig(),


I suppose PollingConfig here is immutable? Otherwise, it'd be a bad practice to let the default argument be something that's mutable.

Confirmed — PollingConfig is @dataclass(frozen=True, slots=True), so the default argument is a singleton immutable value. No mutable-default-argument hazard.

eywalker · 2026-05-20T20:01:11Z

+    access and results are served from an accumulated in-memory cache.
+
+    Args:
+        impl: User-supplied protocol implementation.


Add a more detailed docstring, It should for example explain the default behavior for polling_config

Expanded the class docstring to cover: what each constructor argument does, the default PollingConfig() behaviour (1-second interval, indefinite duration, 5 missed-interval tolerance, 3 error retries, 1-second backoff), the sync-vs-async execution trade-off, and the unbounded in-memory cache note.

Fixed in 85a8ecd. Expanded the class docstring with: (1) a prose paragraph explaining the default PollingConfig() behavior (1 s interval, indefinite, 5-miss tolerance, 3-error retry with 1 s backoff, counter reset on clean tick), and (2) a full Example:: block showing a minimal impl and construction with explicit PollingConfig arguments.

eywalker · 2026-05-20T20:02:51Z

+            raise ValueError(
+                f"PollingConfig.interval must be > 0, got {polling_config.interval}"
+            )
+        if polling_config.duration < 0:


If these validations should be applied to PollingConfig regardless of where the PollingConfig would be used, the validation logic ought to be associated directly with the implementation of PollingConfig

Moved to PollingConfig.__post_init__ in types.py. All five constraints (interval > 0, duration >= 0, max_missed_intervals >= 1, max_consecutive_errors >= 1, error_backoff_base > 0) now fire wherever PollingConfig is constructed, not only inside PollingSource.__init__. The redundant validation block in PollingSource.__init__ has been removed.

Already addressed. Validation of all five PollingConfig fields (interval > 0, duration >= 0, max_missed_intervals >= 1, max_consecutive_errors >= 1, error_backoff_base > 0) lives entirely in PollingConfig.__post_init__ in types.py. PollingSource.__init__ has no duplicate validation — it just stores the config.

eywalker · 2026-05-20T20:05:37Z

+    def to_config(self, db_registry: Any = None) -> dict[str, Any]:
+        """Return a non-reconstructable descriptor for this source."""
+        return {
+            "source_type": "polling_source",


Actually it should NOT be assumed that it's not reconstructable. Just like in data_function, if impl is well-defined in a package/module, the function is reconstructable and subsequently so would be polling_source.

Fixed. to_config now serializes the impl class location (impl_module + impl_class qualname) and all PollingConfig fields. from_config imports the impl class via importlib.import_module and instantiates it with no arguments — the same pattern as PythonDataFunction.from_config. Dotted qualnames (nested classes) are handled by walking the attribute chain. If the impl requires constructor arguments, the docstring instructs callers to subclass PollingSource and override both methods.

Fixed in 85a8ecd. to_config() has always stored impl_module/impl_class for reconstruction — the old 'non-reconstructable' language was just a stale docstring artifact. The docstring now explicitly states that reconstruction is always attempted: from_config() calls impl_class.from_config(impl_config) when impl_config is non-None, and impl_class() (no-arg constructor) otherwise.

eywalker · 2026-05-20T20:06:05Z

+
+    def identity_structure(self) -> Any:
+        """Schema-independent identity (schema is unknown until first fetch)."""
+        return (self.__class__.__name__, self._tag_columns, self._source_id or "")


this should reflect the identity of the impl

Updated. identity_structure now returns (class_name, impl.__module__, impl.__qualname__, tag_columns). Two PollingSource instances wrapping different impl classes produce different hashes; same impl class with same tag columns produce the same hash — matching the PythonDataFunction pattern.

Fixed in 85a8ecd. identity_structure() now returns (self._impl.identity(), self._tag_columns), removing the self.__class__.__name__ prefix. The impl's identity() is the real differentiator; tag columns are included because the same impl with different tag selections produces logically different sources.

eywalker · 2026-05-20T20:06:45Z

+
+    @classmethod
+    def from_config(cls, config: dict[str, Any], db_registry: Any = None) -> PollingSource:
+        """Not supported — ``PollingSource`` cannot be reconstructed from config.


We should provide support for trying to reconstruct.

Done alongside the previous fix — from_config now attempts to reconstruct by importing the impl class. Added a test_from_config_reconstructs_no_arg_impl round-trip test using a NoArgSource fixture class at module level. Raises KeyError (missing key), ImportError (bad module), AttributeError (bad class name), or TypeError (non-callable) with clear messages.

Already supported. from_config() imports the impl via importlib.import_module(impl_module) and reconstructs via impl_class.from_config(impl_config) (or impl_class() if impl_config is None). This matches the DataFunction pattern — if the impl class is importable, the PollingSource is fully reconstructable.

eywalker · 2026-05-20T20:09:06Z

 """A dict-like structure mapping field names to ``DataType`` entries.
 Accepted wherever a ``Schema`` is expected so callers can pass plain dicts."""

+T = TypeVar("T")


Is there a reason for this Generic variable to be shared outside of this module? In PollingSource, can it not just use it's own generic variable without importing it from types? I find that to be a potentially unnecessary bindings?

Agreed — there's no good reason to import T from types.py in downstream modules. Each module now defines its own T = TypeVar("T") locally: one in polling_source.py and one in sources.py. The T in types.py is now scoped purely to Cursor and its docstring updated accordingly.

Fixed in 85a8ecd. Renamed T to _T in types.py (underscore prefix signals it's module-internal, not part of the public API). PollingSource and DynamicSourceProtocol both define their own T = TypeVar("T") locally — neither imports T from types.py. The renamed _T is used only within types.py itself for the Cursor[_T] generic parameterisation.

kurodo3 · 2026-05-20T20:17:03Z

Review round 3 — all eywalker comments addressed

Commit c6f8b4f implements all fixes. Summary:

Comment	Fix
Is `PollingConfig` immutable?	Confirmed frozen — `@dataclass(frozen=True, slots=True)` makes the default safe. No code change.
Add more detailed docstring for `polling_config` default	Expanded class docstring: documents each arg, spells out `PollingConfig()` defaults, sync/async trade-offs, and cache semantics
Validation belongs in `PollingConfig` directly	Moved all 5 constraints to `PollingConfig.__post_init__` in `types.py`; removed redundant block from `PollingSource.__init__`
`identity_structure` should reflect `impl` identity	Returns `(class_name, impl.__module__, impl.__qualname__, tag_columns)` — different impl classes → different hashes
Don't assume non-reconstructable	`to_config` now serializes `impl_module` + `impl_class` + `polling_config` fields; `from_config` imports the impl class via `importlib` and instantiates with no args (same pattern as `PythonDataFunction`)
Support trying to reconstruct	`from_config` implemented; `NoArgSource` fixture class added for a round-trip test; raises `KeyError`/`ImportError`/`AttributeError`/`TypeError` with clear messages
`T` TypeVar shouldn't be imported from `types.py`	`polling_source.py` and `sources.py` each define their own `T = TypeVar("T")` locally; `types.py` `T` is now scoped to `Cursor` only

All 47 tests pass (7 new).

eywalker · 2026-05-21T02:18:00Z

Please update the PR description to reflect passing tests

eywalker · 2026-05-21T02:19:40Z

+
+    async def fetch(
+        self, cursor: Cursor[T] | None = None
+    ) -> tuple[Cursor[T], "FrameInitTypes"]:


drop quotes as we are using annotations

Fixed — return type changed from to . With in effect all annotations are lazily evaluated strings already, so the explicit string quotes are redundant and now removed.

Also dropped the quotes around FrameInitTypes in the DynamicSourceProtocol.fetch return type annotation — same reason: from future import annotations makes explicit string quoting redundant.

eywalker · 2026-05-21T02:20:55Z

+
+
+@dataclass
+class Cursor(Generic[T]):


consider adding convenience method to update the modified time

Added Cursor.now(value) classmethod that creates a cursor with modified_at set to datetime.now(tz=timezone.utc). This is a convenience shorthand for the common pattern of creating a cursor from the current wall clock.

eywalker · 2026-05-21T02:34:43Z

+        impl_type = type(self._impl)
+        return (
+            self.__class__.__name__,
+            impl_type.__module__,


Rather, we should include identifying information as part of the protocol so that we can defer the choice of identity entirely to the implementer. Otherwise, it won't be clear what aspect of the object's state should be included towards its identity.

Done. DynamicSourceProtocol now includes an identity() method that returns any hashable value of the implementer's choosing. PollingSource.identity_structure() delegates entirely to self._impl.identity(), and source_id is also derived from str(impl.identity()) at construction time. The implementer decides what constitutes identity — class name, URL, tuple of config fields, UUID, etc.

eywalker · 2026-05-21T02:36:46Z

+            "impl_module": impl_type.__module__,
+            "impl_class": impl_type.__qualname__,
+            "tag_columns": list(self._tag_columns),
+            "polling_config": {


why are we expanding "polling_config" when we are literally just taking the content of the data class?
Also for impl, we should add to the protocol methods for saving & loading config. Consider allowing the config to indicate if it can/cannot save/load and gracefully deal with the failure to save/load.

Both parts addressed. (1) to_config() now uses dataclasses.asdict(self._polling_config) instead of manually expanding each field. (2) DynamicSourceProtocol gains to_config() -> dict | None and from_config(config) classmethod — None signals the impl cannot be serialized. PollingSource.to_config() stores the result under 'impl_config'; from_config() calls impl_class.from_config(impl_config) when non-None, falls back to no-arg constructor otherwise.

eywalker · 2026-05-21T02:38:34Z

+        """
+        import importlib
+
+        module = importlib.import_module(config["impl_module"])


We should include static factory function as part of the protocol. We should only go as far as importing/using the class but perform instantiation/initialization using the impl's saved config that we pass into the factory function.

Done. DynamicSourceProtocol now declares from_config(config) as a classmethod. PollingSource.from_config() imports the impl class, then calls impl_class.from_config(impl_config) when impl_config is non-None (i.e. the impl declared it's serializable), or falls back to impl_class() when impl_config is None. Instantiation is therefore always driven by the impl's own factory, not hardcoded no-arg construction.

eywalker · 2026-05-21T02:39:21Z

+
+    def _get_latest_stream(self) -> ArrowTableStream:
+        """Return the current accumulated stream, fetching/polling as needed."""
+        if self._schema_stream is None:


I believe the name schema_stream is remnant of old implementation. Change to a more appropriate name unless there is a good reason to keep this schema_stream

Renamed to _accumulated_stream throughout — this name reflects what the attribute actually holds: the accumulated/concatenated stream of all rows fetched so far.

eywalker · 2026-05-21T02:41:33Z

+            else:
+                logger.debug("PollingSource %r: sync poll — cache still valid", self._source_id)
+
+        if self._schema_stream is None:


Consider taking in the schema for tag + packet explicitly (in addition to the default system that tries to infer from the first fetch). This would allow for the stream to function even if it contains no data initially.

Added tag_schema and data_schema optional constructor parameters. When both are provided and no data has been fetched yet, output_schema() and keys() return the declared schemas directly without triggering a fetch — so schema inspection works on an empty or not-yet-started source. Additionally, each fetched batch is validated against the declared schemas and raises InputValidationError on any field missing or type mismatch.

eywalker · 2026-05-21T02:44:20Z

+            source_id=self._source_id,
+        )
+
+        if self._source_id is None:


Not super sure if this is appropriate -- the source_id would only reflect the content of the first batch. Rather source_id should then be reflected from id trait of the impl which I've propsed should implement protocol with a new method for identity.

Fixed. source_id is now derived at construction time from str(impl.identity()) rather than being lazily assigned from the first batch result. The auto-assignment in _build_stream_from_df has been removed. This makes source_id stable and implementation-defined — consistent with the identity() protocol method introduced in this round.

eywalker · 2026-05-21T02:45:48Z

+        # Pre-seed from cache (yield first, count after — avoids double iteration)
+        if self._schema_stream is not None:
+            pre_seed_count = 0
+            for item in self._schema_stream.iter_data():


why not just use enumerate?

Done. Replaced the manual counter pattern with enumerate(self._accumulated_stream.iter_data(), start=1). pre_seed_count is initialized to 0 before the loop so it's always defined, and after the loop it holds the total count of yielded items.

kurodo3 · 2026-05-21T04:16:03Z

Round 4 review — changes applied (commit `967d7a7`)

All nine comments addressed in a single commit. Summary by topic:

Protocol extension (DynamicSourceProtocol)

Added identity() -> Any — implementer-defined hashable identity value; used by PollingSource to derive source_id and identity_structure()
Added to_config() -> dict | None — None signals the impl is not serializable
Added from_config(config) classmethod — factory used by PollingSource.from_config() when impl_config is non-None

PollingSource identity & source_id

identity_structure() now delegates entirely to impl.identity() — the class no longer hardcodes __module__/__qualname__
source_id is set at construction from str(impl.identity()); the lazy auto-assignment from the first batch has been removed

Serialization

to_config() uses dataclasses.asdict(self._polling_config) instead of manual field expansion; stores impl.to_config() under impl_config
from_config() calls impl_class.from_config(impl_config) when impl_config is non-None, falls back to no-arg constructor otherwise

Schema improvements (added beyond the review)

New tag_schema / data_schema constructor params; output_schema() and keys() short-circuit when declared schemas are present and no data has been fetched yet
Each fetched batch is validated against declared schemas — raises InputValidationError on missing field or type mismatch
Combining validation between consecutive batches now raises InputValidationError instead of logging a warning (column set mismatch or type drift both raise immediately; InputValidationError is not caught by the backoff retry handler)

Naming & style

_schema_stream renamed to _accumulated_stream
Pre-seed counter uses enumerate() with start=1

Cursor.now()

Added Cursor.now(value) classmethod for convenient wall-clock cursor creation (modified_at=datetime.now(tz=timezone.utc))

SourceProtocol.from_config return type

String quotes removed (-> SourceProtocol: instead of -> "SourceProtocol":)

Test coverage

Updated _MinimalImpl, FakeDynamicSource, NoArgSource, TransientImpl, DriftingImpl with identity()/to_config()/from_config()
Replaced test_schema_drift_emits_warning with test_schema_mismatch_raises_on_column_change
Added TestCursorNow (2 tests) and TestPollingSourceSchemaValidation (7 tests)
Full suite: 3246 passed, 56 skipped

Copilot

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 6 comments.

kurodo3 · 2026-05-21T04:35:00Z

+                            # Emit new rows from just this fetch
+                            rows = list(new_stream.iter_data())
+                            logger.debug(
+                                "PollingSource %r: emitting %d row(s)",
+                                self._source_id,
+                                len(rows),
+                            )
+                            for item in rows:
+                                yield item


Fixed in ef1221f. Changed to iterate new_stream.iter_data() directly and count with a running emitted variable. The yield now happens inside the loop, so channel backpressure suspends the generator at the yield point rather than after a full materialized list.

kurodo3 · 2026-05-21T04:35:05Z

+        with pytest.raises(InputValidationError, match="schema"):
+            asyncio.run(src.async_iter_data().__anext__())
+            # Drain the rest to trigger the second fetch
+            asyncio.run(_drain_async(src.async_iter_data()))
+


Fixed in ef1221f. Rewrote as a single @pytest.mark.asyncio async test that drives one generator instance with async for _ in src.async_iter_data(): pass. The two-asyncio.run() form created separate generator instances, causing close() to be called after the first run before the second fetch could trigger the combining-schema check.

kurodo3 · 2026-05-21T04:35:10Z

+
+## Goals & Success Criteria
+
+- `DynamicSourceProtocol[T]` is defined with `poll`, `fetch`, and `close`.


Fixed in ef1221f. Updated the Goals section to list all six protocol methods: identity, to_config, from_config, poll, fetch, and close. Also updated the test coverage line to mention serialization round-trip.

kurodo3 · 2026-05-21T04:35:19Z

+Out of scope:
+
+- Database-backed dynamic source (follow-up issue).
+- Framework-level deduplication or row-identity tracking.
+- Schema enforcement across `fetch()` calls (drift is warned, not rejected).
+- Pipeline serialization / `from_config` for `PollingSource` (non-reconstructable,
+  same as `DataFrameSource`).


Fixed in ef1221f. Moved schema enforcement and serialization from Out of scope to In scope. Also removed the two stale out-of-scope bullets (drift-is-warned and non-reconstructable from_config) since both are now implemented and described in the design.

kurodo3 · 2026-05-21T04:35:20Z

+        """Return a non-reconstructable descriptor for this source."""
+        return {
+            "source_type": "polling_source",
+            "tag_columns": list(self._tag_columns),
+            "source_id": self._source_id,
+        }
+
+    @classmethod
+    def from_config(cls, config: dict[str, Any], db_registry: Any = None) -> PollingSource:
+        """Not supported — ``PollingSource`` cannot be reconstructed from config.
+
+        Raises:
+            NotImplementedError: Always.
+        """
+        raise NotImplementedError(
+            "PollingSource cannot be reconstructed from config — "
+            "the DynamicSourceProtocol implementation is not serializable."


Fixed in ef1221f. Updated the plan's to_config() sketch to reflect the actual reconstructable implementation: it now delegates to impl.to_config(), stores impl_module/impl_class/impl_config, and from_config() uses importlib to reconstruct the impl. The non-reconstructable path (when impl.to_config() returns None) is also shown.

kurodo3 · 2026-05-21T04:35:26Z

+        """Append *new_data* rows to *existing*, warning on schema drift."""
+        from orcapod.core.streams.arrow_table_stream import ArrowTableStream
+
+        new_stream = self._build_stream(new_data)
+
+        # Schema drift check (user-facing columns only)
+        old_tag_keys, old_data_keys = existing.keys()
+        new_tag_keys, new_data_keys = new_stream.keys()
+        old_cols = set(old_tag_keys) | set(old_data_keys)
+        new_cols = set(new_tag_keys) | set(new_data_keys)
+        if old_cols != new_cols:
+            logger.warning(
+                "PollingSource %r: schema drift detected — added: %r, removed: %r",
+                self._source_id,
+                sorted(new_cols - old_cols),
+                sorted(old_cols - new_cols),
+            )
+
+        combined = pa.concat_tables(
+            [
+                existing.as_table(all_info=True),
+                new_stream.as_table(all_info=True),
+            ],
+            promote_options="default",


Fixed in ef1221f. Updated the plan's _combine() sketch to match the actual implementation: signature now takes new_stream: ArrowTableStream (already built), delegates to _validate_combining_schemas() which raises InputValidationError on column-set or type mismatch, and removes the logger.warning call entirely.

kurodo3 · 2026-05-21T04:35:44Z

Round 5 review addressed (`ef1221f`)

Six comments resolved across code, tests, and documentation:

Code fix (comment 3278649166)
async_iter_data() no longer materializes each batch into a list() before yielding. Rows are now emitted one at a time inside the for item in new_stream.iter_data() loop, preserving channel backpressure. A running emitted counter feeds the debug log.

Test fix (comment 3278649191)
test_combining_type_mismatch_raises is now a single @pytest.mark.asyncio async test that drives one generator instance with async for _ in src.async_iter_data(): pass. The two-asyncio.run() form was wrong because each call creates a fresh generator; close() fired after the first run before the second fetch could trigger the combining-schema validation.

Spec fixes (comments 3278649217, 3278649233)

Goals: DynamicSourceProtocol[T] now lists all six methods (identity, to_config, from_config, poll, fetch, close); schema-mismatch language updated from "drift warning" to "raises InputValidationError"; serialization round-trip added to test coverage list.
Scope: moved schema enforcement and serialization from Out of scope to In scope; removed the two stale bullets (drift-is-warned, non-reconstructable from_config).
"Schema Drift Warning" section replaced with "Schema Validation" describing both declared-schema and combining-schema checks that raise immediately.
to_config()/from_config() section updated to describe the actual reconstructable implementation.
DynamicSourceProtocol code example updated to include identity, to_config, and from_config methods.
Polling loop pseudocode updated: removed warn_on_schema_drift() call, added backpressure-safe emit loop, added except InputValidationError: raise guard.

Plan fixes (comments 3278649257, 3278649284)

to_config() sketch updated to reflect the reconstructable implementation that stores impl_module/impl_class/impl_config; from_config() sketch uses importlib to reconstruct the impl.
_combine() sketch updated to take new_stream: ArrowTableStream (pre-built), delegate to _validate_combining_schemas() (raises on mismatch), and drop the logger.warning call entirely.

All 3246 tests pass.

kurodo3 · 2026-05-21T05:43:56Z

Round 6 review addressed (`85a8ecd`)

Six eywalker comments resolved:

Comment 3276805496 — more detailed docstring
Expanded the PollingSource class docstring with: (1) a prose paragraph explaining the default PollingConfig() behavior (1 s interval, indefinite run, 5-miss tolerance, 3-error retry with 1 s exponential backoff, counter reset on clean tick), and (2) a full Example:: block showing a minimal DynamicSourceProtocol impl and construction with explicit PollingConfig arguments.

Comment 3276815935 — validation in PollingConfig (already addressed)
All five field validations (interval > 0, duration >= 0, max_missed_intervals >= 1, max_consecutive_errors >= 1, error_backoff_base > 0) live entirely in PollingConfig.__post_init__ in types.py. PollingSource.__init__ has no duplicate validation.

Comment 3276834171 — do not assume non-reconstructable
Updated to_config() docstring to positively describe the reconstruction path: it always stores impl_module/impl_class/impl_config, and from_config() always attempts reconstruction — calling impl_class.from_config(impl_config) when config is available, or impl_class() (no-arg constructor) otherwise. The stale "non-reconstructable" language is gone.

Comment 3276837198 — identity_structure should reflect impl's identity
identity_structure() now returns (self._impl.identity(), self._tag_columns), dropping the redundant self.__class__.__name__ prefix. The impl's identity() is the primary differentiator; tag columns are included because the same impl with different tag selections produces logically different sources.

Comment 3276841041 — support trying to reconstruct (already addressed)
from_config() already reconstructs via importlib — same pattern as DataFunction. No code change needed; the docstring now makes this explicit.

Comment 3276856102 — T TypeVar unnecessary binding
Renamed T → _T in types.py (underscore prefix signals module-internal). PollingSource and DynamicSourceProtocol both define their own local T = TypeVar("T") and neither imports T from types.py. _T is only used within types.py for Cursor[_T].

All 3246 tests pass.

…onsibilities poll() now returns bool (pure availability check); fetch() returns tuple[Cursor[T], FrameInitTypes] so cursor advancement is always tied to a successful data read. Updates sync-mode logic, polling loop pseudocode, example implementation, and test table accordingly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- Add INFO/DEBUG/WARNING/ERROR log calls at all critical lifecycle points in the polling loop (start, new data, no data, cancellation, cursor invalidated, errors with backoff, overrun, duration expiry, close) - Replace misleading 'out of scope' backpressure bullet with an explicit note that channel send() provides natural backpressure and suspends the polling loop at yield points - Add backpressure to Dependencies & Risks as a positive note Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

… polling source

…_execute StreamBase.async_iter_data now has a default implementation that wraps iter_data() as an async generator instead of raising NotImplementedError. SourceNode.async_execute is simplified to delegate entirely to async_iter_data(), removing the sync iter_data() fallback path so that dynamic sources (e.g. PollingSource) stream continuously without special-casing in the node. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…lity

Implements PollingSource[T], a RootSource wrapping DynamicSourceProtocol, with cursor tracking, lazy fetch-on-first-access, poll+cache logic for sync mode, empty-data guard, schema drift warning, and full async polling loop. Adds FakeDynamicSource test fixture and 11 passing sync mode tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…_index field

…onditional assertion, differentiate shutdown tests

…r PollingSource Appends TestPollingSourceErrorHandling with 5 tests covering exponential backoff retries on transient errors, clean termination on max consecutive errors, immediate termination on CursorInvalidatedError, overrun threshold enforcement, and schema drift warning emission.

Move all mid-file import statements scattered across task sections to the top-level import block, organised per isort/PEP 8 (stdlib → third-party → orcapod). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…orce in CLAUDE.md

…ings

- _try_build_stream: remove spurious df.is_empty() guard; only skip when the DataFrame has zero columns (0-row tables are valid) - _get_latest_stream / async_iter_data: update last_modified on every successful fetch(), not only when new rows are produced - _run_sync: accept (async_fn, *args, **kwargs) instead of a pre-built coroutine so the coroutine is created inside the executor thread - _combine: use self._tag_columns instead of reaching into the private existing._tag_columns attribute - PollingSource.__init__: validate all PollingConfig fields at construction (interval > 0, duration >= 0, max_* >= 1, error_backoff_base > 0) - sources.py: remove unused Generic from typing import - plan doc: replace all :class:/:meth:/:exc: ReST roles with plain backtick notation per project docstring conventions Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- PollingConfig: add __post_init__ validation for all five fields (interval > 0, duration >= 0, max_* >= 1, error_backoff_base > 0) so that constraints apply wherever PollingConfig is constructed, not only at PollingSource construction time; remove redundant validation from PollingSource.__init__ - PollingSource class docstring: expand to document polling_config defaults, sync/async execution trade-offs, and cache semantics - identity_structure: include type(impl).__module__ and __qualname__ so different impl classes produce different pipeline hashes - to_config: serialize impl_module + impl_class + polling_config fields so the source can be reconstructed when the impl has a no-arg ctor - from_config: import impl class via importlib and instantiate with no args, following the same pattern as PythonDataFunction.from_config; raises KeyError/ImportError/TypeError with clear messages on failure - T TypeVar: define locally in polling_source.py and sources.py rather than importing from types.py; types.py T is now internal to Cursor - tests: add PollingConfig validation tests, update identity/config tests, add NoArgSource fixture class for from_config round-trip test Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…config round-trip - DynamicSourceProtocol gains identity(), to_config(), from_config() methods - PollingSource derives source_id from impl.identity() at construction - identity_structure() delegates to impl.identity() instead of class qualname - Add tag_schema/data_schema params: output_schema() and keys() short-circuit when declared schemas are present and no data has been fetched yet - Add _assert_schema_match() module helper; schema validation raises InputValidationError instead of logging a warning on mismatch — covers both declared-schema checks on every fetched batch and combining-schema checks between consecutive batches - InputValidationError propagates immediately in async_iter_data (not retried) - to_config() uses dataclasses.asdict() for polling_config and stores impl.to_config() output under impl_config key - from_config() calls impl_class.from_config(impl_config) when impl_config is non-None, falls back to no-arg constructor otherwise - Rename _schema_stream → _accumulated_stream throughout - Fix TypeVar T placement (was between import lines) - Use enumerate() for pre-seed counter in async_iter_data - Add Cursor.now() classmethod for convenience cursor creation - Update all test doubles (_MinimalImpl, FakeDynamicSource, NoArgSource, TransientImpl, DriftingImpl) with identity()/to_config()/from_config() - Replace test_schema_drift_emits_warning with test_schema_mismatch_raises_on_column_change - Add TestCursorNow and TestPollingSourceSchemaValidation test classes (7 new tests covering declared schema, type mismatch, column drift) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…turn type With from __future__ import annotations all annotations are lazily evaluated, so explicit string quoting is unnecessary. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- Yield items directly in async_iter_data() instead of buffering via list(), preserving channel backpressure - Rewrite test_combining_type_mismatch_raises as a single @pytest.mark.asyncio async test driving one generator instance - Update spec: DynamicSourceProtocol now has 6 methods (identity, to_config, from_config, poll, fetch, close); schema mismatches raise InputValidationError (not warn); serialization round-trip is in scope; polling loop pseudocode reflects validation + backpressure-safe emit - Update plan: to_config/from_config sketch reflects reconstructable implementation; _combine() sketch raises on schema mismatch Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- Rename T to _T in types.py to make the Cursor TypeVar module-internal; PollingSource and DynamicSourceProtocol already define their own T locally - Change identity_structure() to return (impl.identity(), tag_columns), removing the redundant class name prefix and letting the impl fully control the identity - Expand PollingSource class docstring with default polling_config semantics and a full Example:: block showing a minimal impl and construction - Rewrite to_config() docstring to positively describe the reconstruction path (always stores impl_module/impl_class/impl_config for from_config), removing any suggestion of a non-reconstructable fallback Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

eywalker · 2026-05-22T02:31:18Z

+
+    Example::
+
+        class MyDBSource:


include init to make this example complete

Fixed in 6e63980. Added __init__(self, db_url) that sets self._db_url = db_url — the constructor that identity, to_config, and the instantiation on the last line all depend on.

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

kurodo3 · 2026-05-22T02:54:42Z

Round 7 review addressed (`6e63980`)

Single comment resolved:

Comment 3285470260 — include __init__ to make the example complete
Added __init__(self, db_url) to MyDBSource in the class docstring's Example:: block. The method stores db_url as self._db_url, which identity(), to_config(), and the instantiation MyDBSource("postgresql://...") all depend on. Without it the example class appeared to reference an attribute it never sets.

eywalker · 2026-05-22T19:31:14Z

+                f"{field!r}: declared {expected_type!r}, got {actual[field]!r}"
+            )
+    if mismatches:
+        raise InputValidationError(


Shall we throw a more specific error like SchemaInconsistencyError?

Done in f6a2959. Added SchemaInconsistencyError(InputValidationError) to errors.py. It subclasses InputValidationError so the except InputValidationError: raise guard in the async polling loop (which intentionally bypasses the error-backoff handler) continues to work unchanged. Both raise sites (_assert_schema_match and _validate_combining_schemas) now raise SchemaInconsistencyError. All five schema-specific test assertions updated from InputValidationError to SchemaInconsistencyError.

eywalker · 2026-05-22T19:32:05Z

+
+    The default ``PollingConfig()`` polls every 1 second, runs indefinitely
+    (``duration=0`` means run until cancelled), allows up to 5 consecutive
+    missed intervals before terminating, retries up to 3 consecutive


what does "missed" mean?

Fixed in f6a2959. Changed the class docstring from "missed intervals" to "overrun intervals" and added a definition: an overrun interval is one that was consumed while the previous poll iteration was still executing (i.e. the loop fell behind its scheduled tick). This matches the actual implementation — consecutive_misses is incremented by floor((now - next_tick) / cfg.interval) when the loop takes longer than one interval to complete, not when poll() returns False.

…gy in docstring - Add SchemaInconsistencyError(InputValidationError) to errors.py for precise schema-mismatch catching in PollingSource - Switch _assert_schema_match and _validate_combining_schemas to raise SchemaInconsistencyError instead of InputValidationError - Update all docstrings (Raises sections, Args, class docstring) to reference SchemaInconsistencyError - Clarify "missed intervals" → "overrun intervals" in class docstring, with a definition: intervals consumed while the previous poll iteration was still executing - Update tests to assert SchemaInconsistencyError on all schema checks Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

kurodo3 · 2026-05-22T19:39:41Z

Round 8 review addressed (`f6a2959`)

Two comments resolved:

Comment 3290831303 — what does "missed" mean?
The class docstring now says "overrun intervals" instead of "missed intervals", with an inline definition: an overrun interval is one that was consumed while the previous poll iteration was still executing — i.e. the loop fell behind its scheduled tick. This matches the implementation exactly: consecutive_misses is incremented by floor((now - next_tick) / cfg.interval) when a single loop iteration takes longer than one polling interval to complete. Also updated "miss counters" → "overrun counters" in the final clause of that sentence.

Comment 3290827788 — use SchemaInconsistencyError
Added SchemaInconsistencyError(InputValidationError) to errors.py. The subclass relationship means the existing except InputValidationError: raise guard in the async polling loop (which bypasses error-backoff for non-transient schema errors) continues to work without modification. Changes:

_assert_schema_match now raises SchemaInconsistencyError
_validate_combining_schemas now raises SchemaInconsistencyError
All docstrings (Raises sections, Args, class docstring) updated
All five schema-specific test assertions updated from InputValidationError to SchemaInconsistencyError

56 tests pass.

eywalker · 2026-05-22T19:51:36Z

+                except asyncio.CancelledError:
+                    raise
+
+                except CursorInvalidatedError:


shouldn't this information somehow be propagated to the user? Right now it emits log entry but would silently terminate.

Fixed in 6128ffb. Changed return to raise in the except CursorInvalidatedError handler. The log entry is still emitted for observability, but now the exception propagates to the caller instead of silently ending the stream. The finally block (which calls impl.close()) runs in all cases — whether we return normally, raise, or get cancelled — so cleanup is unaffected.

Updated the CursorInvalidatedError docstring, the async_iter_data docstring, and the test (test_cursor_invalidated_error_propagates_to_caller) to assert the error now propagates with pytest.raises(CursorInvalidatedError).

…tly returning Silent return left callers with no indication of why the stream ended. Now the error is logged and then re-raised so callers receive the exception. The finally block still calls impl.close() in all cases. Update CursorInvalidatedError docstring, async_iter_data docstring, and the corresponding test to assert the error propagates. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

kurodo3 · 2026-05-22T19:54:46Z

Round 9 review addressed (`6128ffb`)

Comment 3290909755 — CursorInvalidatedError should propagate to the caller

Changed return to raise in the except CursorInvalidatedError handler inside async_iter_data(). Previously the source logged the error and silently returned, making the stream look like a normal end-of-iteration. Now the exception is re-raised so the caller receives a CursorInvalidatedError and can distinguish abnormal termination from a clean shutdown.

The finally block that calls await self._impl.close() runs unconditionally in all exit paths (normal return, raise, or CancelledError), so cleanup is not affected by the change.

Updated:

async_iter_data() docstring: notes that CursorInvalidatedError is re-raised rather than suppressed
CursorInvalidatedError docstring in errors.py: updated to say the error is re-raised after logging
Test renamed to test_cursor_invalidated_error_propagates_to_caller and now asserts the error propagates via pytest.raises(CursorInvalidatedError)

56 tests pass.

eywalker requested changes May 20, 2026

View reviewed changes

eywalker assigned Copilot May 20, 2026

eywalker requested review from brian-arnold and Copilot May 20, 2026 18:49

Copilot started reviewing on behalf of eywalker May 20, 2026 18:50 View session

Copilot AI reviewed May 20, 2026

View reviewed changes

eywalker requested changes May 20, 2026

View reviewed changes

eywalker requested changes May 21, 2026

View reviewed changes

kurodo3 Bot changed the base branch from dev to main May 21, 2026 04:21

eywalker requested a review from Copilot May 21, 2026 04:21

Copilot started reviewing on behalf of eywalker May 21, 2026 04:22 View session

Copilot AI reviewed May 21, 2026

View reviewed changes

eywalker approved these changes May 21, 2026

View reviewed changes

kurodo3 Bot and others added 9 commits May 22, 2026 01:42

docs(specs): add polling source design spec for PLT-1430

6acca7c

docs(plans): add polling source implementation plan for PLT-1430

b74c607

feat(types): add Cursor[T], PollingConfig, CursorInvalidatedError for…

10eebcb

… polling source

fix(types): correct import order and add slots=True to PollingConfig

e809080

feat(protocols): add DynamicSourceProtocol[T] for polling data sources

209319c

test(streams): strengthen async_iter_data test to verify content equa…

f34163f

…lity

kurodo3 Bot and others added 14 commits May 22, 2026 01:46

refactor(sources): remove dead _build_stream wrapper and stale _batch…

0a18923

…_index field

test(sources): add async polling loop tests for PollingSource

c736b51

test(sources): strengthen async tests — document timing params, fix c…

93e6751

…onditional assertion, differentiate shutdown tests

chore(sources): consolidate test imports to top-of-file

16d504a

Move all mid-file import statements scattered across task sections to the top-level import block, organised per isort/PEP 8 (stdlib → third-party → orcapod). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

docs: replace ReST cross-references with Google-style docstrings; enf…

e4dc714

…orce in CLAUDE.md

docs: remove remaining ReST cross-references from pre-existing docstr…

9fedbbe

…ings

fix(sources): drop redundant quotes around FrameInitTypes in fetch re…

8e66018

…turn type With from __future__ import annotations all annotations are lazily evaluated, so explicit string quoting is unnecessary. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

kurodo3 Bot force-pushed the kurodo/plt-1430-dynamic-data-source-protocol-based-polling-source branch from 85a8ecd to c79c21e Compare May 22, 2026 01:49

eywalker requested changes May 22, 2026

View reviewed changes

docs(polling_source): add __init__ to MyDBSource example in class doc…

6e63980

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

eywalker requested changes May 22, 2026

View reviewed changes

eywalker approved these changes May 22, 2026

View reviewed changes

eywalker merged commit cf4800d into main May 22, 2026
11 checks passed

eywalker deleted the kurodo/plt-1430-dynamic-data-source-protocol-based-polling-source branch May 22, 2026 19:59


		## Goals & Success Criteria

		- `DynamicSourceProtocol[T]` is defined with `poll`, `fetch`, and `close`.

Uh oh!

Conversation

kurodo3 Bot commented May 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Test plan

Uh oh!

codecov Bot commented May 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

eywalker left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot commented May 20, 2026

Review round summary

Changes made (single commit 16ec04d, plus 9a92f94 for pre-existing instances)

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot commented May 20, 2026

Review round 2 — all Copilot comments addressed

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 21, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kurodo3 Bot May 20, 2026

Choose a reason for hiding this comment

kurodo3 Bot commented May 20, 2026 •

edited

Loading

codecov Bot commented May 20, 2026 •

edited

Loading

Changes made (single commit `16ec04d`, plus `9a92f94` for pre-existing instances)