feat(operators): add GroupBy for many→one reduction, with logical-type and provenance fixes (NPIPE-204)

[brian-arnold]

Summary

Adds GroupBy, orcapod's first many→one operator, and fixes two bugs found along the way —
one already shipped, one that made the new operator unusable on the column type it was built
for.

GroupBy(by=["subject", "date"]) collapses N rows sharing a tag tuple into one packet: group
keys stay scalar tags, every other column becomes list-valued. This is what lets a downstream
pod receive a whole recording session at once — every existing pipeline fans out, and nothing
reduced.

grouped = op.operators.GroupBy(by=["subject", "date"]).process(sync_out)
common_clock_op.pod(grouped)

⚠️ API deviates from the issue

NPIPE-204 specifies Batch(group_by=[...]). This PR adds a separate GroupBy(by=[...]).

Batch exists for throughput and pipelining — its own async_execute docstring says batching
lets "downstream consumers start processing before all input is consumed" — and its git history
contains only refactors. Folding reduction into it would give one class two output contracts
selected by which kwarg you passed. Splitting keeps one contract per class, leaves Batch's
streaming path untouched, and makes GroupBy's barrier structural rather than a conditional
inside a streaming operator.

The Linear issue still shows the old API. Full reasoning in
superpowers/specs/2026-08-07-npipe-204-batch-group-by-design.md.

Two bugs fixed

1. Data could not represent list-valued provenance — already shipped and silently broken.
_ensure_source_info_table hard-coded large_string for every _source_* column, so any
operator emitting a list-valued token crashed:

pyarrow.lib.ArrowTypeError: Expected bytes, got a 'list' object
  core/datagrams/tag_data.py:342  _ensure_source_info_table

MergeJoin already does this (merge_join.py:262) and was failing inside job.run() on
966d759a. Reproduced independently of Batch; now covered by a job-level regression test.
Resolves the tag_data.py half of DESIGN_ISSUES §U1.

2. Aggregating operators dropped logical element types. Batch and GroupBy built their
schema with pa.list_(field.type), which raises ArrowNotImplementedError on any
extension-typed column — Arrow cannot embed an extension type inside a list value field
(§ET1/ET2). A pod annotated -> Path emits extension<orcapod.path>, so grouping the column
the motivating consumer actually needs was impossible. ITL-173 (#251) fixed the
type-declaration layer with ListLogicalType but never touched the operator layer, so
declaring list[Path] worked while grouping it still failed.

Both operators now build through arrow_utils.build_aggregated_table, which constructs the
list over the element's storage type and wraps it in the outer list[<element>] extension
type. Verified end to end:

grouped col type:   extension<list[orcapod.path]>
grouped py schema:  {'result_path': list[pathlib.Path]}
pod received:       [['PosixPath','PosixPath'], ['PosixPath']]

Downstream pods can take list[Path] directly — no Path → str conversion pod needed.

Design decisions

Decision	Rationale
Source-info types derived from the value	Fixes MergeJoin for free; keeps per-member provenance readable. No pipeline-DB schema bump — a node's source-column type is fixed by its own output schema.
System tags fold to a scalar digest	_build_record_id_preimage hashes them directly, so they must stay scalar. Folding rather than dropping preserves the Merkle link, so an upstream recompute yielding identical data still invalidates.
uuid5 + combine_hashes, never hash()	The digest is a cache key. A per-process value looks correct in a single-process test and misses the cache on every new driver run. Pinned by a subprocess test.
Members sorted by non-key tags, record_id as tiebreaker	Tags are unique within a stream, so this is a total order needing no assumption about which column holds a path. Upstream emission order is not stable (Ray scheduling, DB fetch order).
Groups emitted in key order	Output is byte-identical under input reordering.
One shared list-wrapping helper	Batch and GroupBy had the same defect; fixing it once keeps them from drifting.

Testing

tests/: 4742 passed, 93 skipped, 2 xfailed, 0 failures
test-objective/: 567 passed, 7 xfailed — note this is a separate suite with its own
workflow and is not covered by pytest tests/.

New job-level coverage in tests/test_pipeline/test_aggregation_job.py closes the gap that let
bug 1 ship — every prior aggregating-operator test stopped at op.process() + as_table(),
which never reaches _materialize_to_stream. Against a real DeltaTableDatabase:

• Two identical runs → zero pod invocations on the second, both rows still returned
• One member changed → only that group recomputes; the untouched group stays cached
• A control test points the second run at a fresh store, proving the cache-hit assertion can fail
• MergeJoin regression test for bug 1

Two test-objective provenance tests were updated: they asserted the old rule that Batch
list-wraps system tags. That changed deliberately, so the tests were wrong, not the code. They
now assert both halves — user columns do list-wrap, system tags stay scalar and gain a
name-extension block.

⚠️ Reviewer attention: DESIGN_ISSUES §O5

A reduction can persist a result computed from an incomplete member set. After an upstream
change, the first job.run() emits only the recomputed rows. For a per-row pod that is
complete; for a reduction it is partial — the group executes and its result is persisted with
nothing marking it incomplete, then self-corrects on the next run.

CONTROL (per-row)  changed run1  pod(sync_99)          <- converged in one run
GROUPED            changed run1  GROUP ['sync_99']     <- ONE member, persisted
                   changed run2  GROUP ['sync_99','sync_1']

Batch behaves identically, so this is pre-existing pipeline semantics surfaced by reduction —
but it matters more here, because GroupBy exists so a pod can reason over a complete group.
For common_clock_op an intermediate run can write an alignment.json built from one of a
session's probes.

This is not fixable inside the operator — an operator sees whatever its upstream emits and
cannot know whether a group is complete. A fix needs a completeness signal at the node level.

I'd value a reviewer's judgment on whether this gates the downstream rev bump; that depends on
how common_clock_op's output is consumed, which the sync-and-qc side knows better than I do.

Other known limitations, logged not fixed

• §O3 — _materialize_to_stream applies row 0's source_info() to every row, so rows 1+
  are mis-attributed. Pre-existing and equally wrong for scalars, but this PR makes it more
  visible: the list case previously crashed loudly and now completes with wrong tokens.
• §O4 — GroupBy member order is undefined when the input has duplicate tag tuples. Root
  cause is that DuplicateTagError is defined but never raised, so duplicates are never
  prevented. Does not affect the common_clock_op wiring, which groups on a strict subset.
  Now tracked in ITL-630 (below).
• §O2 — operators uniformly discard a non-default _context_key. Pre-existing, layer-wide.
• §U1's arrow_utils.add_source_info_to_table() half remains open. The dead
  polars_data_utils.add_source_info (a third site, with a latent shadowing bug) was deleted.

Follow-up tracked: ITL-630

Two design questions raised in review are deferred rather than settled here, and are tracked in
ITL-630
(Tools / Orcapod Python v0.2 Feature Sprint, alongside ITL-173 and ITL-627):

• Member and group ordering. That the order is deterministic is a requirement — the
  emitted lists are hashed into the memoization key, and upstream emission order is not stable
  across runs. Which order, and whether the operator promises one publicly, is open.
• Empty-input behaviour. "Zero rows in, zero groups out" is the current default but was
  never designed; emit-nothing, emit-empty-group, and raise were not weighed.

ITL-630 also absorbs §O4, since duplicate-tag ordering is the same "what do we promise"
question. §O5 is noted there as related but out of scope — its cause is pipeline propagation,
not operator semantics.

TODO(ITL-630) comments mark both sites in tests/test_core/operators/test_group_by.py. The
tests pin current behaviour so a change is deliberate rather than accidental.

The third review note — the source-info type-derivation shortcut in core/datagrams/tag_data.py
— carries its own TODO(NPIPE-204) and is not part of ITL-630.

Usability gotcha worth knowing

A pod function nested inside another function, in a module using
from __future__ import annotations, cannot resolve non-builtin annotations — Path arrives as
the string 'Path' with no resolvable scope. Module-level pods are fine. This bit the new test
fixture and would bite any pipeline module written the same way.

Follow-up

Rebased onto current main, so ListLogicalType (#251) and GroupBy are in one tree. Merging
requires a coordinated rev bump in both orcapod-sync-and-qc and orcapod-spikesorting — they
are deliberately kept on the same rev because they target the same Ray cluster.

Relates to NPIPE-204.

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 98.69281% with 2 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/orcapod/core/datagrams/tag_data.py	94.73%	1 Missing ⚠️
src/orcapod/core/operators/batch.py	96.15%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

[Copilot]

Pull request overview

Adds a first-class many→one reduction operator (GroupBy) to the operator layer, along with underlying provenance + logical-type aggregation fixes needed for job.run() correctness and cache stability.

Changes:

• Introduces GroupBy(by=[...]) and registers it for serialization and the stream fluent API.
• Fixes list-valued provenance handling by widening SourceInfoValue and deriving _source_* Arrow/Python types from stored values.
• Adds shared aggregation helpers (fold_system_tag_values, build_aggregated_table) and updates Batch to fold system tags to scalar digests while preserving list-valued member/source columns; adds job-level regression coverage.

Reviewed changes

Copilot reviewed 22 out of 22 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
tests/test_utils/test_arrow_utils.py	Adds pinned + cross-process tests for system-tag folding digests.
tests/test_pipeline/test_aggregation_job.py	Adds job-level Delta-backed aggregation tests covering memoization + MergeJoin regression.
tests/test_core/operators/test_operators.py	Extends Batch tests to assert scalar system tags and list-valued source columns.
tests/test_core/operators/test_group_by.py	Adds comprehensive operator-level tests for GroupBy shape, ordering, validation, schema prediction, and logical types.
tests/test_core/datagrams/test_data_source_info_types.py	Adds unit tests for list-valued _source_* provenance typing/round-tripping in Data.
test-objective/integration/test_provenance.py	Updates objective provenance assertions for “reducing” ops (system tags scalar; user cols list-wrapped).
superpowers/specs/2026-08-07-npipe-204-batch-group-by-design.md	Adds design spec documenting API deviation and implementation details.
superpowers/plans/2026-08-07-npipe-204-groupby-operator.md	Adds implementation plan / checklist for the work.
src/orcapod/utils/polars_data_utils.py	Removes dead add_source_info implementation.
src/orcapod/utils/arrow_utils.py	Adds fold_system_tag_values and build_aggregated_table to support many→one ops + logical-type-safe list aggregation.
src/orcapod/types.py	Introduces SourceInfoValue type alias for recursive list-valued provenance tokens.
src/orcapod/protocols/core_protocols/datagrams.py	Updates DataProtocol source-info typing to SourceInfoValue.
src/orcapod/pipeline/serialization.py	Registers GroupBy in the operator registry for (de)serialization.
src/orcapod/core/streams/base.py	Adds .group_by(...) fluent method to StreamBase.
src/orcapod/core/streams/arrow_table_stream.py	Widens source_info typing to SourceInfoValue.
src/orcapod/core/operators/group_by.py	Implements the new GroupBy operator (partition, deterministic ordering, system-tag folding, schema prediction).
src/orcapod/core/operators/batch.py	Updates Batch aggregation semantics to fold system tags scalar + use shared aggregated-table builder.
src/orcapod/core/operators/init.py	Exports GroupBy from the operators package.
src/orcapod/core/datagrams/tag_data.py	Makes Data derive _source_* Arrow/Python types from stored provenance values (including recursive lists).
DESIGN_ISSUES.md	Logs/updates operator-layer design issues and marks the tag_data.py half of U1 as resolved.
CLAUDE.md	Updates architecture guidance for reducing ops, GroupBy, and aggregated logical types.
.zed/rules	Mirrors the same documentation updates for Zed AI rules.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[brian-arnold]

Fixed in 3ce1be6 — explicit PYTHONPATH, prepended so an existing value survives.

On severity: it was not failing. The venv has an editable install whose .pth points at src, so sys.executable resolves orcapod regardless of cwd — verified from /tmp with PYTHONPATH unset, and CI was green on 3.11/3.12. Still right in principle, since the test depended on install mode rather than the suite path config.

[brian-arnold]

Review + one fix pushed (7baafcf6)

Reviewed the full diff against origin/main. Unit suite passes (4767 / 93 skipped / 2 xfailed), predicted-vs-materialized schema equality checked for both aggregating operators across {}, source, system_tags, all_info, and the extension-type paths (Path, np.ndarray, nested list[list[Path]]), empty input, drop_partial_batch, join-then-group, group-then-group, group-then-batch, and the job/Delta path all exercised. The DESIGN_ISSUES notes (O1–O5, U1) accurately capture most of the known gaps.

One finding fixed in 7baafcf6; the rest are left for the author to judge.

Fixed — group_by.py: the member-order tiebreaker only consulted the first record_id column

A joined stream carries one record_id per canonical input position, and the broadcast side of a fan-out repeats its id on every row it was joined into — so it cannot separate those rows; only the fanned-out side's record_id can. next((c for c in system_tag_columns if c.startswith(SYSTEM_TAG_RECORD_ID_PREFIX)), None) took one column, so whenever the broadcast input sorted to canonical position 0 the sort key tied on exactly the rows the tiebreaker exists for, and members.sort fell back to emission order. DB fetch order and Ray scheduling make that nondeterministic, so a reducing pod's input hash can change between runs and trigger a spurious recompute.

Now appends every record_id column, sorted by name so the key order does not itself depend on column order in the input table.

Worth noting how narrow the reachable case is:

• It needs a group with ≥2 members sharing identical non-key tags — i.e. duplicate tag tuples in the stream (§O4 territory). With unique tags a non-key tag column already breaks the tie and record_id is never reached.
• It also needs the broadcast input at canonical position 0. When the fanned-out input sorts first, ordering was already deterministic. Two of my test attempts landed on that benign arrangement, so the literals in the new test are load-bearing.
• The exposure is on materialized / DB-backed streams, where record_ids are fixed and fetch order varies. Permuting an in-memory source is not a valid repro: record_id there is derived per row position, not from row content, so permuting the source permutes the ids too — which the class docstring already declares undefined.

So it is not a merge blocker on its own, but it makes the tiebreaker actually function for joined streams instead of silently tying, and it makes the determinism claim in the class docstring (lines 53–58) and in the §O4 note true rather than false. New test test_join_fan_out_ties_broken_by_every_record_id_column fails with [10, 20] != [20, 10] without the fix.

Not fixed — for your call

arrow_utils.py:1279 — build_aggregated_table builds 32-bit pa.list_ while the rest of the PR uses large_list. _source_info_arrow_type (tag_data.py:263) emits large_list, ListLogicalType's storage is large_list, and add_system_tag_columns uses large_string; only the non-extension member branch uses pa.list_(field.type). Observed: GroupBy(...).unary_static_process(src).as_table() gives path: list<item: string> while the same node through PipelineJob.run() gives path: large_list<item: large_string>. Any pa.concat_tables/union across the two, or any schema comparison / arrow hash, sees a type mismatch. Suggest pa.large_list(field.type).

group_by.py:68 — a bare string by is silently expanded per character. tuple(by) with str satisfying Collection[str] means GroupBy(by="subject") type-checks and produces ('s','u','b','j','e','c','t'), failing later with InputValidationError: ['s','u','b','j','e','c','t'] are not tag columns. Other APIs here accept str | Collection[str] (e.g. PythonDataFunction(output_keys="n")), so this is a likely user mistake — normalize or reject explicitly.

arrow_utils.py:1222 — the non-record_id fold concatenates members with no delimiter and maps None to "". combine_hashes is "".join, so ["ab","c"] and ["a","bc"] fold identically, and [None,"abcd"], ["abcd",None], ["","abcd"], ["abcd"] are indistinguishable. Latent today since source_id values are fixed-length hex — but join.sort_system_tag_values substitutes [None] * num_rows for absent paired columns, so nulls do reach the fold. Interpose constants.BLOCK_SEPARATOR, as the record_id branch already does.

tag_data.py:263 — list element type inferred from value[0] alone. [None, ["tok"]] returns large_list(large_string) and _ensure_source_info_table then raises a confusing ArrowTypeError. Not constructible from current operators, so robustness rather than a live defect; scanning for the first non-None element removes the sharp edge.

types.py:50 — the SourceInfoValue docstring calls MergeJoin a many-to-one operator. It is a row-preserving binary join that merges colliding columns into list[T]; CLAUDE.md and GroupBy's own docstring both state GroupBy is the only many→one operator. Suggest "aggregating operators such as GroupBy and Batch, and MergeJoin's merged columns".

Notes

• ruff format --check would reformat group_by.py, arrow_utils.py, and three new test files, but no workflow gates on ruff and arrow_utils.py was already unformatted on main — not a CI break.
• No regressions in the highest-risk areas: Batch's rewritten batching loop is behaviourally identical to the old index-counter version (including batch_size <= 0 and drop_partial_batch), the extension re-wrap in build_aggregated_table handles the zero-chunk empty-table case, and the new top-level from orcapod.hashing.hash_utils import combine_hashes introduces no import cycle.
• On §O5 — I agree it is not fixable inside the operator; a completeness signal has to live at the node level.

🤖 Generated with Claude Code

[brian-arnold]

Review: consistency with the rest of the codebase

Reviewed specifically for whether this matches the conventions and rigor of the surrounding code. It largely does; one real gap, which I've now fixed on the branch, plus a few small items left for you to judge.

Conforms

• Operator boilerplate is complete. to_config, identity_structure, registration in pipeline/serialization.py, export from core/operators/__init__.py, and a fluent StreamBase.group_by(). Nothing in that set was skipped, which is more than some existing operators manage.
• Barrier semantics come from the base class. GroupBy correctly does not override async_execute, inheriting UnaryOperator's collect-then-process path — and test_does_not_override_async_execute pins that, so a future streaming override can't land silently.
• Schema prediction is verified the right way. TestGroupByOutputSchema compares against unary_static_process(stream).output_schema(...), not process(...) (circular) and not as_table() (legitimately different) — exactly the trap CLAUDE.md warns about.
• Google docstrings with no ReST roles, no backward-compat shims, and DESIGN_ISSUES.md / CLAUDE.md / .zed/rules all updated per the project rules.
• Test structure (class-grouped, one behavior per test) matches test_operators.py. The touched test files pass locally: 226 passed.

Gap — fixed in 6f26a20

User-facing docs did not include the operator. CLAUDE.md and .zed/rules were updated, but docs/ — the actual library documentation — was not:

• docs/api/operators.md carries one ::: orcapod.core.operators.X block per operator; GroupBy was absent.
• docs/concepts/operators.md carries a prose section per operator (### Batch at line 90); GroupBy was absent.
• docs/reference/hashing.md was made stale by this branch. Site 6's "Known exclusions" cell still read "Batch changes the column type from str to list[str] but preserves the column name" — which this PR deliberately reverses. The branch also introduces a genuinely new hash site (arrow_utils.fold_system_tag_values) that the doc's exhaustive site index did not list.

I pushed 6f26a20 covering all three: the mkdocstrings entry, a ### GroupBy concepts section (both code examples verified against a real run rather than written from the docstring), the many-to-one caveat under the operator/function-pod boundary table, and a new Site 6b — Aggregating operator system tag fold entry replacing the stale Site 6 note. mkdocs build is clean — no new warnings, and the 11 remaining ones are pre-existing in files this branch doesn't touch.

Minor, left to your judgment

1. Not ruff-formated. .pre-commit-config.yaml runs ruff-format pinned at v0.14.4; under that exact version group_by.py, arrow_utils.py, test_group_by.py, and test_aggregation_job.py all fail --check. Caveat that makes this low-stakes: roughly 106 of 185 files under src/ fail repo-wide, so the hook plainly isn't enforced today. Trivial to fix if you want the new files clean.
2. Dead import block, group_by.py:17-20 — the TYPE_CHECKING / LazyModule("pyarrow") pair is present but pa is never referenced anywhere in the file. Ruff won't flag it, since it's an assignment rather than an import.
3. build_aggregated_table(type_converter: Any) (arrow_utils.py:1231) — the same module already types this parameter as TypeConverterProtocol in make_empty_table (line 20).
4. The two branches of fold_system_tag_values disagree on separators. The record_id branch joins member hex with BLOCK_SEPARATOR; the other delegates to combine_hashes, which concatenates with "" and is therefore length-ambiguous. Not a live bug — the values are fixed-width digests — but it's an asymmetry in a function whose docstring makes a point of digest stability.

Not a regression

pyright reports 2 errors in group_by.py (stream.data_context is not on StreamProtocol). batch.py has the identical two at lines 99 and 110, so this is an inherited protocol gap, not one this branch introduces.

On the §O5 caveat in the description: I agree it's a node-level problem rather than an operator-level one. An operator genuinely cannot see whether its upstream handed it a complete group, so there's nothing to fix inside GroupBy.

🤖 Generated with Claude Code

[eywalker]

The PR looks amazing and implemented extremely well and aligned to the rest of the codebase --- thank you! I just had a few notes mostly asking to put TODO and/or create follow up issues