From b39625eb273c96e631c769dcb26f56a81626f499 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 25 Aug 2026 19:10:26 +0000 Subject: [PATCH 1/6] docs(specs): add ITL-616 design spec for _combine content_hash leak fix --- ...polling-source-combine-content-hash-fix.md | 131 ++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 superpowers/specs/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md diff --git a/superpowers/specs/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md b/superpowers/specs/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md new file mode 100644 index 00000000..1d6d3f9f --- /dev/null +++ b/superpowers/specs/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md @@ -0,0 +1,131 @@ +# ITL-616: Fix `PollingSource._combine` leaking `_content_hash` into the data schema + +**Date:** 2026-08-25 +**Issue:** [ITL-616](https://linear.app/metamorphic/issue/ITL-616) +**Branch:** `eywalker/itl-616-pollingsource_combine-leaks-_content_hash-into-the-data` + +## Overview + +`PollingSource._combine` passes `all_info=True` to `as_table()` on both streams before +concatenating them into the new accumulated stream. `all_info=True` resolves to +`ColumnConfig.all()`, which includes `content_hash=True`. In `ArrowTableStream.as_table()`, +`content_hash=True` dynamically appends a `_content_hash` column to the output table — a +synthetic, on-demand column that is not part of `ArrowTableStream._table` (internal storage). + +The concatenated table (now containing `_content_hash`) is passed directly to +`ArrowTableStream.__init__`. Since `_content_hash` has no recognized prefix (`_tag::`, +`_source_`, `_context_key`), the constructor treats it as a user data column and stores it +in `_data_columns`. + +This causes two problems: + +1. **Silent schema mutation** — after the first accumulating combine, `output_schema()` and + `keys()` report `_content_hash` as a data column, which is incorrect. +2. **Crash on the second combine** — `_validate_combining_schemas` compares the accumulated + stream (which now has `_content_hash` in `keys()`) against a fresh batch (which does not), + and raises `SchemaInconsistencyError`. A polling source emitting one new row per poll + silently corrupts its schema on the second poll and crashes on the third. + +## Audit scope + +All other `as_table(all_info=True)` call sites were audited: + +- `ArrowTableStream.identity_structure()` — result is hashed only, never fed into a new + `ArrowTableStream`. Safe. +- `function_pod.py`, `function_node.py` — call `tag.as_dict(all_info=True)` / + `data.as_dict(all_info=True)` on `Tag`/`Data` datagrams, not streams. Safe. +- All operators (`join.py`, `merge_join.py`, `static_output_pod.py`, etc.) — use explicit + `ColumnConfig` dicts that never include `content_hash`. Safe. + +The only affected site is `PollingSource._combine`. + +## Fix + +### Module-level constant + +Add a named constant at module level in `polling_source.py` (below the existing imports, +above the `PollingSource` class): + +```python +# ColumnConfig used when concatenating streams in _combine. +# Includes the provenance columns (system_tags, source, context) that +# ArrowTableStream.__init__ knows how to parse and split into their +# respective internal tables. +# content_hash is intentionally absent: it is a synthetic, on-demand +# column produced by as_table(); including it would bake it into stored +# data and corrupt the data schema on the next combine. +_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True) +``` + +### Change in `_combine` + +Replace the two `as_table(all_info=True)` calls with `as_table(columns=_STREAM_COMBINE_COLUMNS)`: + +```python +# Before +combined = pa.concat_tables( + [ + existing.as_table(all_info=True), + new_stream.as_table(all_info=True), + ], + promote_options="default", +) + +# After +combined = pa.concat_tables( + [ + existing.as_table(columns=_STREAM_COMBINE_COLUMNS), + new_stream.as_table(columns=_STREAM_COMBINE_COLUMNS), + ], + promote_options="default", +) +``` + +All other code in `_combine` is unchanged: `_validate_combining_schemas` is still called +first, and the `ArrowTableStream` constructor call at the end is unchanged. + +### Why this ColumnConfig is correct + +`_STREAM_COMBINE_COLUMNS` includes exactly the columns that `ArrowTableStream.__init__` +knows how to parse and store: + +| Flag | What it includes | Handled by | +|---|---|---| +| `system_tags=True` | `_tag::source:` columns | Detected by `_tag::` prefix in `__init__` | +| `source=True` | `_source_` provenance columns | Extracted by `prepare_prefixed_columns` in `__init__` | +| `context=True` | `_context_key` column | Split off by `split_by_column_groups` in `__init__` | +| `content_hash` (absent) | `_content_hash` (synthetic) | Would land in `_data_columns` — excluded | + +## Tests + +Two new tests in `TestPollingSourceSchemaValidation` in +`tests/test_channels/test_polling_source.py`, covering the exact failure mode (existing tests +cover only a single combine; two combines are sufficient to trigger the crash): + +### Sync test: `test_sync_three_fetches_no_content_hash_leak` + +Three `iter_data()` calls against a `FakeDynamicSource` with 3 batches: +- Assert row counts: 1, 2, 3 (accumulation working correctly) +- Assert `"_content_hash"` is not in `src.keys()[1]` (data keys) +- Assert `"_content_hash"` is not in `src.output_schema()[1]` (data schema) + +### Async test: `test_async_three_fetches_no_content_hash_leak` + +Three-batch async drain using existing `PollingConfig` timing patterns: +- Assert 3 items emitted +- Assert `"_content_hash"` is not in `src._accumulated_stream.keys()[1]` +- Assert `"_content_hash"` is not in `src._accumulated_stream.output_schema()[1]` + +## `DESIGN_ISSUES.md` + +Add entry **PS3** to the `src/orcapod/core/sources/polling_source.py` section (after PS2), +with status `in progress` during development and `resolved` on merge. See the full entry +text in the design session above. + +## What is not changed + +- `_validate_combining_schemas` — unchanged; it correctly catches real schema drift between + batches. +- `ArrowTableStream.__init__` — no defensive stripping added. The correct fix is at the call + site: `_combine` should never request `content_hash` when building a storage table. +- No Polars intermediate step — the fix stays entirely in PyArrow. From b5daaa07ba9b2600e03f8ecddce2961af6069177 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 25 Aug 2026 20:49:47 +0000 Subject: [PATCH 2/6] docs(plans): add ITL-616 implementation plan for _combine content_hash leak fix --- ...polling-source-combine-content-hash-fix.md | 331 ++++++++++++++++++ 1 file changed, 331 insertions(+) create mode 100644 superpowers/plans/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md diff --git a/superpowers/plans/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md b/superpowers/plans/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md new file mode 100644 index 00000000..91f133b5 --- /dev/null +++ b/superpowers/plans/2026-08-25-itl-616-polling-source-combine-content-hash-fix.md @@ -0,0 +1,331 @@ +# ITL-616: Fix `PollingSource._combine` `_content_hash` Leak — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use sensei:subagent-driven-development (recommended) or sensei:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Fix `PollingSource._combine` so that the synthetic `_content_hash` column is never stored in the accumulated stream, preventing a `SchemaInconsistencyError` on the third accumulating fetch. + +**Architecture:** Add a module-level `_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)` constant to `polling_source.py` and replace the two `as_table(all_info=True)` calls in `_combine` with `as_table(columns=_STREAM_COMBINE_COLUMNS)`. No other files change except `DESIGN_ISSUES.md` (new PS3 entry) and the test file (two new regression tests). + +**Tech Stack:** Python 3.11+, pytest, pytest-asyncio, pyarrow, orcapod internal classes (`PollingSource`, `ArrowTableStream`, `ColumnConfig`). + +--- + +## File Map + +| Action | Path | What changes | +|--------|------|-------------| +| Modify | `src/orcapod/core/sources/polling_source.py` | Add `_STREAM_COMBINE_COLUMNS` constant; replace `all_info=True` in `_combine` | +| Modify | `tests/test_channels/test_polling_source.py` | Add two regression tests to `TestPollingSourceSchemaValidation` | +| Modify | `DESIGN_ISSUES.md` | Add PS3 entry; update to `resolved` after fix | + +--- + +## Setup: Create the branch + +Before any task, create and check out the branch from `main`: + +```bash +cd /path/to/orcapod-python +git checkout main +git checkout -b eywalker/itl-616-pollingsource_combine-leaks-_content_hash-into-the-data +``` + +Verify: + +```bash +git branch --show-current +# eywalker/itl-616-pollingsource_combine-leaks-_content_hash-into-the-data +``` + +--- + +### Task 1: Add `DESIGN_ISSUES.md` PS3 entry + +**Files:** +- Modify: `DESIGN_ISSUES.md` + +- [ ] **Step 1: Open `DESIGN_ISSUES.md` and insert the PS3 entry** + + Locate the block that ends with PS2 (search for `### PS2`). The section looks like this: + + ```markdown + ### PS2 — Concurrent iteration over a single `PollingSource` has unguarded cursor/stream mutation + **Status:** open + ... + + **Fix:** Guard updates to `_cursor` and `_accumulated_stream` with an `asyncio.Lock` ... + + --- + + ## `src/orcapod/core/nodes/function_node.py` + ``` + + Insert the following block **between** the `---` separator and the `## src/orcapod/core/nodes/function_node.py` header: + + ```markdown + ### PS3 — `_combine` leaks `_content_hash` into the data schema on the second accumulating fetch + **Status:** in progress + **Severity:** high + **Issue:** ITL-616 + + `_combine` calls `as_table(all_info=True)` on both streams before concatenating them. + `all_info=True` resolves to `ColumnConfig.all()`, which includes `content_hash=True`. + In `ArrowTableStream.as_table()`, `content_hash=True` dynamically appends a `_content_hash` + column to the output table. This is a synthetic column — computed on demand, not stored in + `ArrowTableStream._table`. + + `pa.concat_tables` then includes `_content_hash` in the combined table, which is passed + directly to `ArrowTableStream.__init__`. Since `_content_hash` has no recognized prefix + (`_tag::`, `_source_`, `_context_key`), it lands in `_data_columns` as if it were user data. + + On the next `_combine` call, `_validate_combining_schemas` compares: + - `existing.keys()` → includes `_content_hash` in data keys (baked in from previous combine) + - `new_stream.keys()` → no `_content_hash` (freshly built from raw fetched data) + + This raises `SchemaInconsistencyError`. A polling source emitting one new row per poll will + change its data schema on the second new-data poll and crash on the third. + + **Fix:** Replace `all_info=True` with a named module-level constant + `_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)`. + `content_hash` is intentionally absent — it is a synthetic output column, never a stored one. + + --- + ``` + + Note: the `---` at the end of the PS3 entry replaces the existing `---` that was between PS2 and `## src/orcapod/core/nodes/function_node.py`. + +- [ ] **Step 2: Commit** + + ```bash + git add DESIGN_ISSUES.md + git commit -m "docs(design-issues): add PS3 entry for _combine content_hash leak (ITL-616)" + ``` + +--- + +### Task 2: Write the failing regression tests + +**Files:** +- Modify: `tests/test_channels/test_polling_source.py` + +The two tests go inside the existing `class TestPollingSourceSchemaValidation` (find it by searching for that class name). They must be placed **after** the last existing test in that class (`test_combining_column_set_mismatch_raises`). + +- [ ] **Step 1: Add the two tests to `TestPollingSourceSchemaValidation`** + + The test file already imports `pyarrow as pa`, `pytest`, `FakeDynamicSource`, `_batch`, `PollingSource`, `PollingConfig` — no new imports needed. + + Add these two methods at the end of `class TestPollingSourceSchemaValidation`: + + ```python + def test_sync_three_fetches_no_content_hash_leak(self): + """After 3 accumulating fetches (2 combines), data schema is stable and + contains no _content_hash column, and all rows are present. + + Regression test for ITL-616: _combine called as_table(all_info=True), + which injected the synthetic _content_hash column into the stored stream. + The second combine then raised SchemaInconsistencyError. + """ + fake = FakeDynamicSource( + batches=[_batch(1, 10), _batch(2, 20), _batch(3, 30)] + ) + src = PollingSource( + fake, tag_columns="id", polling_config=PollingConfig(interval=1.0) + ) + + rows1 = list(src.iter_data()) # fetch 1 — builds initial stream + rows2 = list(src.iter_data()) # fetch 2 — first combine + rows3 = list(src.iter_data()) # fetch 3 — second combine (crashed before fix) + + assert len(rows1) == 1 + assert len(rows2) == 2 + assert len(rows3) == 3 + + _, data_keys = src.keys() + assert "_content_hash" not in data_keys + + _, data_schema = src.output_schema() + assert "_content_hash" not in data_schema + + @pytest.mark.asyncio + async def test_async_three_fetches_no_content_hash_leak(self): + """After 3 async batches (2 combines), data schema is stable and contains + no _content_hash column, and all rows are accumulated. + + Regression test for ITL-616: same root cause as the sync path. + """ + fake = FakeDynamicSource( + batches=[_batch(1, 10), _batch(2, 20), _batch(3, 30)] + ) + src = PollingSource( + fake, + tag_columns="id", + polling_config=PollingConfig( + interval=0.05, duration=0.5, max_missed_intervals=50 + ), + ) + + items = [] + async for tag, data in src.async_iter_data(): + items.append((tag, data)) + + assert len(items) == 3 + assert src._accumulated_stream is not None + + _, data_keys = src._accumulated_stream.keys() + assert "_content_hash" not in data_keys + + _, data_schema = src._accumulated_stream.output_schema() + assert "_content_hash" not in data_schema + ``` + +- [ ] **Step 2: Run the tests to confirm they fail** + + ```bash + uv run pytest tests/test_channels/test_polling_source.py::TestPollingSourceSchemaValidation::test_sync_three_fetches_no_content_hash_leak tests/test_channels/test_polling_source.py::TestPollingSourceSchemaValidation::test_async_three_fetches_no_content_hash_leak -v + ``` + + Expected: both tests **FAIL**. The sync test fails with `SchemaInconsistencyError` on the third `iter_data()` call. The async test fails similarly. + +- [ ] **Step 3: Commit the failing tests** + + ```bash + git add tests/test_channels/test_polling_source.py + git commit -m "test(polling_source): add three-fetch regression tests for content_hash leak (ITL-616)" + ``` + +--- + +### Task 3: Implement the fix and close out + +**Files:** +- Modify: `src/orcapod/core/sources/polling_source.py` +- Modify: `DESIGN_ISSUES.md` + +- [ ] **Step 1: Add `_STREAM_COMBINE_COLUMNS` constant to `polling_source.py`** + + Open `src/orcapod/core/sources/polling_source.py`. The file has three module-level + functions before the class: `_get_sync_executor`, `_run_sync`, and `_assert_schema_match`. + Find the end of `_assert_schema_match` (it ends with a `raise SchemaInconsistencyError(...)` + block) and the `# PollingSource` section comment that follows it: + + ```python + # --------------------------------------------------------------------------- + # PollingSource + # --------------------------------------------------------------------------- + + + class PollingSource(RootSource, Generic[T]): + ``` + + Insert the constant in the gap between `_assert_schema_match` and that section comment: + + ```python + # ColumnConfig used when concatenating streams in _combine. + # Includes the provenance columns (system_tags, source, context) that + # ArrowTableStream.__init__ knows how to parse and split into their + # respective internal tables. + # content_hash is intentionally absent: it is a synthetic, on-demand + # column produced by as_table(); including it would bake it into stored + # data and corrupt the data schema on the next combine. + _STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True) + + + # --------------------------------------------------------------------------- + # PollingSource + # --------------------------------------------------------------------------- + ``` + + Note: `ColumnConfig` is already imported at the top of the file via + `from orcapod.types import ColumnConfig, Cursor, PollingConfig, Schema` — no new import needed. + +- [ ] **Step 2: Fix `_combine` to use `_STREAM_COMBINE_COLUMNS`** + + In the same file, locate the `_combine` method. It contains: + + ```python + combined = pa.concat_tables( + [ + existing.as_table(all_info=True), + new_stream.as_table(all_info=True), + ], + promote_options="default", + ) + ``` + + Replace with: + + ```python + combined = pa.concat_tables( + [ + existing.as_table(columns=_STREAM_COMBINE_COLUMNS), + new_stream.as_table(columns=_STREAM_COMBINE_COLUMNS), + ], + promote_options="default", + ) + ``` + + Nothing else in `_combine` changes. + +- [ ] **Step 3: Run the regression tests to confirm they now pass** + + ```bash + uv run pytest tests/test_channels/test_polling_source.py::TestPollingSourceSchemaValidation::test_sync_three_fetches_no_content_hash_leak tests/test_channels/test_polling_source.py::TestPollingSourceSchemaValidation::test_async_three_fetches_no_content_hash_leak -v + ``` + + Expected: both tests **PASS**. + +- [ ] **Step 4: Run the full polling source test suite** + + ```bash + uv run pytest tests/test_channels/test_polling_source.py tests/test_channels/test_polling_source_pipeline_integration.py -v + ``` + + Expected: all tests **PASS**. No regressions. + +- [ ] **Step 5: Update PS3 status to `resolved` in `DESIGN_ISSUES.md`** + + Locate the PS3 entry added in Task 1. Change: + + ```markdown + **Status:** in progress + ``` + + to: + + ```markdown + **Status:** resolved + ``` + + Add a **Fix:** note immediately after the status line: + + ```markdown + **Status:** resolved + **Fix:** Added `_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, + context=True)` constant and replaced `as_table(all_info=True)` with + `as_table(columns=_STREAM_COMBINE_COLUMNS)` in `_combine`. `content_hash` is + intentionally excluded — it is a synthetic output column, never a stored one. + ``` + +- [ ] **Step 6: Commit the fix** + + ```bash + git add src/orcapod/core/sources/polling_source.py DESIGN_ISSUES.md + git commit -m "fix(polling_source): exclude _content_hash from _combine column config (ITL-616) + + _combine passed all_info=True to as_table(), which triggered content_hash=True + and injected the synthetic _content_hash column into the concatenated table. + ArrowTableStream.__init__ stored it as a data column, corrupting the schema + on the second combine and raising SchemaInconsistencyError on the third fetch. + + Add _STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, + context=True) and use it in place of all_info=True. content_hash is excluded + because it is a synthetic, on-demand output — not a stored column." + ``` + +- [ ] **Step 7: Run the full test suite** + + ```bash + uv run pytest tests/ -x -q + ``` + + Expected: all tests pass. From 6d70aa02560c51a46bb9636f112e351d564d30ba Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 25 Aug 2026 20:51:08 +0000 Subject: [PATCH 3/6] docs(design-issues): add PS3 entry for _combine content_hash leak (ITL-616) --- DESIGN_ISSUES.md | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index da5784d3..95c39781 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -124,6 +124,34 @@ lock before iterating, so later writes to the stream don't affect the snapshot m --- +### PS3 — `_combine` leaks `_content_hash` into the data schema on the second accumulating fetch +**Status:** in progress +**Severity:** high +**Issue:** ITL-616 + +`_combine` calls `as_table(all_info=True)` on both streams before concatenating them. +`all_info=True` resolves to `ColumnConfig.all()`, which includes `content_hash=True`. +In `ArrowTableStream.as_table()`, `content_hash=True` dynamically appends a `_content_hash` +column to the output table. This is a synthetic column — computed on demand, not stored in +`ArrowTableStream._table`. + +`pa.concat_tables` then includes `_content_hash` in the combined table, which is passed +directly to `ArrowTableStream.__init__`. Since `_content_hash` has no recognized prefix +(`_tag::`, `_source_`, `_context_key`), it lands in `_data_columns` as if it were user data. + +On the next `_combine` call, `_validate_combining_schemas` compares: +- `existing.keys()` → includes `_content_hash` in data keys (baked in from previous combine) +- `new_stream.keys()` → no `_content_hash` (freshly built from raw fetched data) + +This raises `SchemaInconsistencyError`. A polling source emitting one new row per poll will +change its data schema on the second new-data poll and crash on the third. + +**Fix:** Replace `all_info=True` with a named module-level constant +`_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)`. +`content_hash` is intentionally absent — it is a synthetic output column, never a stored one. + +--- + ## `src/orcapod/core/nodes/function_node.py` ### FN1 — `FunctionNodeBase.as_table()` returned empty schema when no data existed From cf238d116b49298ce6a014b16e80ea9d1b8691d9 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 25 Aug 2026 20:53:42 +0000 Subject: [PATCH 4/6] test(polling_source): add three-fetch regression tests for content_hash leak (ITL-616) --- tests/test_channels/test_polling_source.py | 60 ++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/tests/test_channels/test_polling_source.py b/tests/test_channels/test_polling_source.py index 478e41cd..a47cf403 100644 --- a/tests/test_channels/test_polling_source.py +++ b/tests/test_channels/test_polling_source.py @@ -1077,6 +1077,66 @@ async def close(self): async for _ in src.async_iter_data(): pass + def test_sync_three_fetches_no_content_hash_leak(self): + """After 3 accumulating fetches (2 combines), data schema is stable and + contains no _content_hash column, and all rows are present. + + Regression test for ITL-616: _combine called as_table(all_info=True), + which injected the synthetic _content_hash column into the stored stream. + The second combine then raised SchemaInconsistencyError. + """ + fake = FakeDynamicSource( + batches=[_batch(1, 10), _batch(2, 20), _batch(3, 30)] + ) + src = PollingSource( + fake, tag_columns="id", polling_config=PollingConfig(interval=1.0) + ) + + rows1 = list(src.iter_data()) # fetch 1 — builds initial stream + rows2 = list(src.iter_data()) # fetch 2 — first combine + rows3 = list(src.iter_data()) # fetch 3 — second combine (crashed before fix) + + assert len(rows1) == 1 + assert len(rows2) == 2 + assert len(rows3) == 3 + + _, data_keys = src.keys() + assert "_content_hash" not in data_keys + + _, data_schema = src.output_schema() + assert "_content_hash" not in data_schema + + @pytest.mark.asyncio + async def test_async_three_fetches_no_content_hash_leak(self): + """After 3 async batches (2 combines), data schema is stable and contains + no _content_hash column, and all rows are accumulated. + + Regression test for ITL-616: same root cause as the sync path. + """ + fake = FakeDynamicSource( + batches=[_batch(1, 10), _batch(2, 20), _batch(3, 30)] + ) + src = PollingSource( + fake, + tag_columns="id", + polling_config=PollingConfig( + interval=0.05, duration=0.5, max_missed_intervals=50 + ), + ) + + items = [] + async for tag, data in src.async_iter_data(): + items.append((tag, data)) + + assert len(items) == 3 + assert src._accumulated_stream is not None + + _, data_keys = src._accumulated_stream.keys() + assert "_content_hash" not in data_keys + + _, data_schema = src._accumulated_stream.output_schema() + assert "_content_hash" not in data_schema + async def _drain_async(agen): """Consume all items from an async generator.""" From 7cbad19af64e252918c84822bb2a6aa73faba2ae Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 25 Aug 2026 20:57:04 +0000 Subject: [PATCH 5/6] fix(polling_source): exclude _content_hash from _combine column config (ITL-616) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _combine passed all_info=True to as_table(), which triggered content_hash=True and injected the synthetic _content_hash column into the concatenated table. ArrowTableStream.__init__ stored it as a data column, corrupting the schema on the second combine and raising SchemaInconsistencyError on the third fetch. Add _STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True) and use it in place of all_info=True. content_hash is excluded because it is a synthetic, on-demand output — not a stored column. --- DESIGN_ISSUES.md | 3 ++- src/orcapod/core/sources/polling_source.py | 14 ++++++++++++-- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 95c39781..6e267fcd 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -125,7 +125,8 @@ lock before iterating, so later writes to the stream don't affect the snapshot m --- ### PS3 — `_combine` leaks `_content_hash` into the data schema on the second accumulating fetch -**Status:** in progress +**Status:** resolved +**Fix:** Added `_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)` constant and replaced `as_table(all_info=True)` with `as_table(columns=_STREAM_COMBINE_COLUMNS)` in `_combine`. `content_hash` is intentionally excluded — it is a synthetic output column, never a stored one. **Severity:** high **Issue:** ITL-616 diff --git a/src/orcapod/core/sources/polling_source.py b/src/orcapod/core/sources/polling_source.py index 9cd4a58c..d729e051 100644 --- a/src/orcapod/core/sources/polling_source.py +++ b/src/orcapod/core/sources/polling_source.py @@ -121,6 +121,16 @@ def _assert_schema_match( ) +# ColumnConfig used when concatenating streams in _combine. +# Includes the provenance columns (system_tags, source, context) that +# ArrowTableStream.__init__ knows how to parse and split into their +# respective internal tables. +# content_hash is intentionally absent: it is a synthetic, on-demand +# column produced by as_table(); including it would bake it into stored +# data and corrupt the data schema on the next combine. +_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True) + + # --------------------------------------------------------------------------- # PollingSource # --------------------------------------------------------------------------- @@ -652,8 +662,8 @@ def _combine( combined = pa.concat_tables( [ - existing.as_table(all_info=True), - new_stream.as_table(all_info=True), + existing.as_table(columns=_STREAM_COMBINE_COLUMNS), + new_stream.as_table(columns=_STREAM_COMBINE_COLUMNS), ], promote_options="default", ) From 3669e65bd778029fb8c07446e12605b8fb68bb5c Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 25 Aug 2026 21:42:17 +0000 Subject: [PATCH 6/6] fix(test): use public API in async regression test; fix PS3 metadata order (ITL-616) Switch test_async_three_fetches_no_content_hash_leak to call src.keys() and src.output_schema() instead of accessing the private _accumulated_stream attribute. Also reorder PS3 metadata block in DESIGN_ISSUES.md to match the file-wide convention (Status, Severity, Issue then Fix at end of body). Co-Authored-By: Claude Sonnet 4.6 --- DESIGN_ISSUES.md | 5 +---- tests/test_channels/test_polling_source.py | 5 ++--- 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 6e267fcd..c2d5c812 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -126,7 +126,6 @@ lock before iterating, so later writes to the stream don't affect the snapshot m ### PS3 — `_combine` leaks `_content_hash` into the data schema on the second accumulating fetch **Status:** resolved -**Fix:** Added `_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)` constant and replaced `as_table(all_info=True)` with `as_table(columns=_STREAM_COMBINE_COLUMNS)` in `_combine`. `content_hash` is intentionally excluded — it is a synthetic output column, never a stored one. **Severity:** high **Issue:** ITL-616 @@ -147,9 +146,7 @@ On the next `_combine` call, `_validate_combining_schemas` compares: This raises `SchemaInconsistencyError`. A polling source emitting one new row per poll will change its data schema on the second new-data poll and crash on the third. -**Fix:** Replace `all_info=True` with a named module-level constant -`_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)`. -`content_hash` is intentionally absent — it is a synthetic output column, never a stored one. +**Fix:** Added `_STREAM_COMBINE_COLUMNS = ColumnConfig(system_tags=True, source=True, context=True)` constant and replaced `as_table(all_info=True)` with `as_table(columns=_STREAM_COMBINE_COLUMNS)` in `_combine`. `content_hash` is intentionally excluded — it is a synthetic output column, never a stored one. --- diff --git a/tests/test_channels/test_polling_source.py b/tests/test_channels/test_polling_source.py index a47cf403..27a545c5 100644 --- a/tests/test_channels/test_polling_source.py +++ b/tests/test_channels/test_polling_source.py @@ -1129,12 +1129,11 @@ async def test_async_three_fetches_no_content_hash_leak(self): items.append((tag, data)) assert len(items) == 3 - assert src._accumulated_stream is not None - _, data_keys = src._accumulated_stream.keys() + _, data_keys = src.keys() assert "_content_hash" not in data_keys - _, data_schema = src._accumulated_stream.output_schema() + _, data_schema = src.output_schema() assert "_content_hash" not in data_schema