## Running commands

Always run Python commands via `uv run`, e.g.:

  uv run pytest tests/
  uv run python -c "..."

Never use `python`, `pytest`, or `python3` directly.

## Updating agent instructions

When adding or changing any instruction, update BOTH:
- CLAUDE.md (for Claude Code)
- .zed/rules (for Zed AI)

## Design issues log

DESIGN_ISSUES.md at the project root is the canonical log of known design problems, bugs, and
code quality issues.

When fixing a bug or addressing a design problem:
1. Check DESIGN_ISSUES.md first — if a matching issue exists, update its status to
   "in progress" while working and "resolved" once done, adding a brief Fix: note.
2. If no matching issue exists, ask the user whether it should be added before proceeding.
   If yes, add it (status "open" or "in progress" as appropriate).

When discovering a new issue that won't be fixed immediately, ask the user whether it should be
logged in DESIGN_ISSUES.md before adding it.

## Superpowers artifacts

Place all superpowers-related artifacts (design specs, plans, etc.) in the superpowers/
directory at the project root — NOT under docs/. The docs/ directory is reserved for
actual library documentation.

- Specs go in superpowers/specs/

## Backward compatibility

This is a greenfield project pre-v0.1.0. Do NOT add backward-compatibility shims,
re-exports, aliases, or deprecation wrappers when making design or implementation changes.
Just change the code and update all references directly.

## No sys.modules hacks

Never manipulate sys.modules directly (e.g. sys.modules.setdefault). If a subpackage
import path doesn't work, create a proper re-export package with an __init__.py instead.

## Docstrings

Use Google style (https://google.github.io/styleguide/pyguide.html#38-comments-and-docstrings)
Python docstrings everywhere.

Never mix in ReST markup. Do not use :meth:, :class:, :exc:, :attr:, :func:, or any
other ReST cross-reference roles. Use plain ``backtick`` notation to refer to names
(e.g. ``PollingSource``, ``iter_data``, ``CursorInvalidatedError``).

## Linear issue tracking

All work must be linked to a Linear issue. Before starting any feature, bug fix, or
refactor:

1. Check for an existing issue — search Linear for a corresponding issue.
2. If none exists — ask the developer whether to create one. Do not proceed without
   either a linked issue or explicit approval to skip.
3. When starting work on an issue — update its Linear status to "In Progress".
4. When a new issue is discovered during development (bug, design problem, deferred
   work), create a corresponding Linear issue using the template below.

When creating Linear issues, always use this template for the description:

  ## Overview
  What is this project about? Describe the problem space and the high-level approach.

  ## Goals & Success Criteria
  * Specific, measurable outcomes.

  ## Scope & Boundaries
  (Optional — remove if not needed.)
  In scope:
  * ...
  Out of scope:
  * ...

  ## Dependencies & Risks
  (Optional — remove if none.)
  * ...

  ## Resources & References
  (Optional — remove if none.)
  * ...

  ## Milestones
  (Optional — only for projects longer than ~4 weeks. Remove for shorter projects.)
  * ...

Remove any optional sections that don't apply rather than leaving them empty.

### Branches and PRs

When working on a feature, create and checkout a git branch using the gitBranchName
returned by the primary Linear issue (e.g. eywalker/plt-911-add-documentation-for-orcapod-python).

Feature branch PRs always target "main". Create a feature branch from "main" and open PRs against "main".

If a feature branch / PR corresponds to multiple Linear issues, list all of them in the
PR description body so that Linear's GitHub integration auto-tracks the PR against each
issue. Use the format "Fixes PLT-123" or "Closes PLT-123" (GitHub magic words) for issues
that the PR fully resolves, and simply mention "PLT-456" for issues that are related but
not fully resolved by the PR.

## Responding to PR reviews

When asked to respond to PR reviewer comments:

1. **Fetch and present** — Read all review comments, then present a response plan as a table:
   each comment, its severity, whether to fix or explain, and the proposed action.
2. **Wait for approval** — Let the user approve the plan before making changes.
3. **Fix, then reply** — Make all fixes in a single commit, then post replies to each
   reviewer comment explaining what was done (or why it was declined).

Never make fixes silently or skip the plan step.

## Git commits

Always use Conventional Commits style (https://www.conventionalcommits.org/):

  <type>(<optional scope>): <short description>

Common types: feat, fix, refactor, test, docs, chore, perf, ci.

Examples:
- feat(schema): add optional_fields to Schema
- fix(data_function): reject variadic parameters at construction
- test(function_pod): add schema validation tests
- refactor(schema_utils): use Schema.optional_fields directly

---

## Project layout

src/orcapod/
  types.py                    — Schema, ColumnConfig, ContentHash
  system_constants.py         — Column prefixes and separators
  errors.py                   — InputValidationError, DuplicateTagError, FieldNotResolvableError
  config.py                   — Config dataclass
  contexts/                   — DataContext (semantic_hasher, arrow_hasher, type_converter)
  protocols/
    hashing_protocols.py      — PipelineElementProtocol, ContentIdentifiableProtocol
    core_protocols/           — StreamProtocol, PodProtocol, SourceProtocol,
                                DataFunctionProtocol, DatagramProtocol, TagProtocol,
                                DataProtocol, TrackerProtocol
  core/
    base.py                   — ContentIdentifiableBase, PipelineElementBase, TraceableBase
    static_output_pod.py      — StaticOutputPod (operator base), DynamicPodStream
    function_pod.py           — FunctionPod, FunctionPodStream, FunctionNode
    data_function.py        — DataFunctionBase, PythonDataFunction, CachedDataFunction
    operator_node.py          — OperatorNode (DB-backed operator execution)
    tracker.py                — Invocation tracking
    datagrams/
      datagram.py             — Datagram (unified dict/Arrow backing, lazy conversion)
      tag_data.py           — Tag (+ system tags), Data (+ source info)
    sources/
      base.py                 — RootSource (abstract, no upstream)
      arrow_table_source.py   — Core source — all other sources delegate to it
      derived_source.py       — DerivedSource (backed by FunctionNode/OperatorNode DB)
      csv_source.py, dict_source.py, list_source.py,
        data_frame_source.py, delta_table_source.py — Delegating wrappers
      source_registry.py      — SourceRegistry for provenance resolution
    streams/
      base.py                 — StreamBase (abstract)
      arrow_table_stream.py   — ArrowTableStream (concrete, immutable)
    operators/
      base.py                 — UnaryOperator, BinaryOperator, NonZeroInputOperator
      join.py                 — Join (N-ary inner join, commutative)
      merge_join.py           — MergeJoin (binary, colliding cols → sorted list[T])
      semijoin.py             — SemiJoin (binary, non-commutative)
      batch.py                — Batch (group rows, types become list[T])
      column_selection.py     — Select/Drop Tag/Data columns
      mappers.py              — MapTags, MapData (rename columns)
      filters.py              — PolarsFilter
  hashing/
    semantic_hashing/         — BaseSemanticHasher, type handlers
  semantic_types/             — Type conversion (Python ↔ Arrow)
  databases/                  — ArrowDatabaseProtocol implementations (Delta Lake, in-memory)
  utils/
    arrow_data_utils.py       — System tag manipulation, source info, column helpers
    arrow_utils.py            — Arrow table utilities
    schema_utils.py           — Schema extraction, union, intersection, compatibility
    lazy_module.py            — LazyModule for deferred heavy imports

tests/
  test_core/
    datagrams/                — Lazy conversion, dict/Arrow round-trip
    sources/                  — Source construction, protocol conformance, DerivedSource
    streams/                  — ArrowTableStream behavior
    function_pod/             — FunctionPod, FunctionNode, pipeline hash integration
    operators/                — All operators, OperatorNode, MergeJoin
    data_function/          — DataFunction, CachedDataFunction
  test_hashing/               — Semantic hasher, hash stability
  test_databases/             — Delta Lake, in-memory, no-op databases
  test_semantic_types/        — Type converter tests

---

## Architecture overview

See orcapod-design.md at the project root for the full design specification.

### Core data flow

  RootSource → ArrowTableStream → [Operator / FunctionPod] → ArrowTableStream → ...

Every stream is an immutable sequence of (Tag, Data) pairs backed by a PyArrow Table.
Tag columns are join keys and metadata; data columns are the data payload.

### Core abstractions

Datagram (core/datagrams/datagram.py) — immutable data container with lazy dict ↔ Arrow
conversion. Two specializations:
- Tag — metadata columns + hidden system tag columns for provenance tracking
- Data — data columns + per-column source info provenance tokens

Stream (core/streams/arrow_table_stream.py) — immutable (Tag, Data) sequence.
Key methods: output_schema(), keys(), iter_data(), as_table().

Source (core/sources/) — produces a stream from external data. ArrowTableSource is the core
implementation; CSV/Delta/DataFrame/Dict/List sources all delegate to it internally. Each
source adds source-info columns and a system tag column. DerivedSource wraps a
FunctionNode/OperatorNode's DB records as a new source.

Function Pod (core/function_pod.py) — wraps a DataFunction that transforms individual
data. Never inspects tags. Two execution models:
- FunctionPod → FunctionPodStream: lazy, in-memory
- FunctionNode: DB-backed, two-phase (yield cached results first, then compute missing)

Operator (core/operators/) — structural pod transforming streams without synthesizing new
data values. All subclass StaticOutputPod:
- UnaryOperator — 1 input (Batch, Select/Drop columns, Map, Filter)
- BinaryOperator — 2 inputs (MergeJoin, SemiJoin)
- NonZeroInputOperator — 1+ inputs (Join)

OperatorNode (core/operator_node.py) — DB-backed operator execution, analogous to
FunctionNode.

### Strict operator / function pod boundary

Operators: inspect tags (never data content), can rename columns, cannot synthesize values.
Function Pods: inspect data content (never tags), synthesize new values, cached by content.

### Two identity chains

Every pipeline element has two parallel hashes:

1. content_hash() — data-inclusive. Changes when data changes. Used for deduplication.
2. pipeline_hash() — schema + topology only. Ignores data content. Used for DB path scoping
   so that different sources with identical schemas share database tables.

Base case: RootSource.pipeline_identity_structure() returns (tag_schema, data_schema).
Each downstream node's pipeline hash commits to its own identity plus upstream pipeline
hashes, forming a Merkle chain.

### Column naming conventions

  __ prefix       — System metadata (ColumnConfig meta)
  _source_ prefix — Source info provenance (ColumnConfig source)
  _tag:: prefix   — System tag (ColumnConfig system_tags)
  _context_key    — Data context (ColumnConfig context)

Prefixes are computed from SystemConstant in system_constants.py.

### System tag evolution rules

1. Name-preserving — single-stream ops. Column name/value pass through unchanged.
2. Name-extending — multi-input ops. System tag column name gets
   ::{pipeline_hash}:{canonical_position} appended. Commutative operators sort by
   pipeline_hash and sort system tag values per row.
3. Type-evolving — aggregation ops. Column type changes from str to list[str].

### Key patterns

- LazyModule("pyarrow") — deferred import for heavy deps. Used in
  if TYPE_CHECKING: / else: blocks.
- Argument symmetry — operators return frozenset (commutative) or tuple (ordered).
- StaticOutputPod.process() → DynamicPodStream — wraps static_process() with staleness
  detection and automatic recomputation.
- Source delegation — CSVSource, DictSource, etc. create an internal ArrowTableSource.

### Important implementation details

- ArrowTableSource raises ValueError if any tag_columns are not in the table.
- ArrowTableStream requires at least one data column; raises ValueError otherwise.
- FunctionNode Phase 1 returns ALL records in the shared pipeline_path DB table.
  Phase 2 skips inputs whose hash is already in the DB.
- Empty data → ArrowTableSource raises ValueError("Table is empty").
- DerivedSource before run() → raises ValueError (no computed records).
- Join requires non-overlapping data columns; raises InputValidationError on collision.
- MergeJoin requires colliding columns to have identical types; merges into sorted list[T].
- Operators predict output schema (including system tag names) without computation.
