From f6cb89f7b95e636dd70637a3e6f556c25422530a Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 02:35:27 +0000 Subject: [PATCH 01/13] docs(spike): design spec for side-effect pods with pipeline-linked execution hash (ITL-524) Adds superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md with a complete design resolving all 10 axes from ITL-524: pod taxonomy (new SideEffectPod class), pass-through output contract, author-declared idempotency, dual-hash (invocation_signature + execution_id), SideEffectContext auto-injection, fail-loud error semantics, DAG ordering, _orcapod_side_effect_invocations invocation log, full reverse-lookup path, and canonical orcapod- artifact serialization. Includes worked patient audit example with end-to-end reverse-lookup walk-through and explicit reconciliation with ITL-523 post-run hook. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 740 ++++++++++++++++++ 1 file changed, 740 insertions(+) create mode 100644 superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md new file mode 100644 index 00000000..266704d3 --- /dev/null +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -0,0 +1,740 @@ +# ITL-524: Side-Effect Pods — First-Class Support with Pipeline-Linked Execution Hash + +## Overview + +Orcapod pipelines are built around **function pods** — nodes that transform an input +packet into an output packet. Many real pipelines also need steps whose primary purpose +is a side effect: writing a structured log line, appending a row to an external database, +emitting a metric, or sending a notification. Today those steps must be shoehorned into +function pods with fake outputs or `None` returns. + +This design introduces **side-effect pods** as a first-class pipeline concept. A side-effect +pod wraps a user function that returns `None`, passes its input stream through unchanged, +and embeds a deterministic **invocation signature** into every external artifact it +produces. That signature bridges the provenance boundary: anyone holding an artifact can +look it up in Orcapod's invocation log and trace it back to the originating pipeline run, +pod version, and input data — following the same provenance chains as any other Orcapod +node. + +--- + +## Goals & Success Criteria + +- A `SideEffectPod` class and `@side_effect_pod` decorator that: + - Accept a user function `(data: T, [ctx: SideEffectContext]) -> None`. + - Pass the input stream through unchanged as output. + - Inject a `SideEffectContext` carrying `invocation_signature`, `execution_id`, + `pod_name`, and `pipeline_run_id` when the user function declares a `ctx` + parameter typed `SideEffectContext`. + - Accept a `SideEffectConfig` controlling idempotency and error handling. +- A `SideEffectContext` dataclass with a `format_id()` helper that returns the + canonical `orcapod-{invocation_signature}` string for embedding in artifacts. +- An `_orcapod_side_effect_invocations` table in the pipeline database that records + every invocation (hash, timing, status, input hash, pod identity) as the + **near-side** of the reverse-lookup chain. +- Idempotency support: when `SideEffectConfig(idempotent=True)`, skip re-execution + if the same `invocation_signature` is already in the invocation log. +- Sync and async execution paths both supported. +- A worked end-to-end example demonstrating the full reverse-lookup walk-through. +- Explicit reconciliation with the ITL-523 post-run hook: shared hash formula, shared + `on_error` vocabulary, shared `InvocationIdentity` base type. +- A follow-up implementation issue (ITL-525) filed with concrete task breakdown. + +--- + +## First-Principles Framing + +For function pods, Orcapod holds the complete provenance loop: inputs → code → outputs +are all in-framework, and the result store ties them together natively. For side-effect +pods, the "output" lives **outside** — in Loki, a Postgres table, a Slack message, a +Prometheus metric. Orcapod cannot hold the artifact, so it cannot natively close the +provenance loop. + +The **invocation signature** bridges that gap. If the side-effect pod embeds the +signature into every artifact it produces, and Orcapod records the mapping from +signature → (pod version, input data, pipeline run) in its own invocation log, then: + +``` +external artifact (has invocation_signature) + → Orcapod invocation log (signature → pod + inputs) + → Orcapod result store (inputs → upstream lineage) + → original source rows +``` + +This is the reverse-lookup chain. The signature must be: +- **Plain text** — embeddable in log fields, DB columns, message bodies. +- **Deterministic** — stable across identical re-runs (same pod code + same inputs → + same signature), so an artifact produced months ago can still be resolved. +- **Short enough** — 32 hex characters (128 bits) is compact and unambiguous. + +--- + +## Design Axes + +### 1. Pod Taxonomy — New `SideEffectPod` class + +**Decision:** Introduce a new `SideEffectPod` class alongside `FunctionPod`. + +`SideEffectPod` shares the `_FunctionPodBase` execution infrastructure (executor +routing, async channel execution, observer integration), but has distinct semantics: +the wrapped function returns `None`, the output stream is the input stream passed +through, and invocations are always recorded in the side-effect invocation log. + +**Why not a flag (`side_effect=True`) on `FunctionPod`?** +A flag adds dead branches to every `FunctionPod` code path and spreads side-effect +semantics across a class whose core contract is `input → output`. It also makes type +signatures ambiguous — `FunctionPod.process()` always promises an output stream. + +**Why not a subclass of `FunctionPod`?** +Subclassing inherits the output-oriented contract (return type, caching logic, +`record_id_hash` on the output datagram). Overriding those creates a leaky abstraction. +A sibling class with shared base infrastructure is cleaner. + +**Class hierarchy:** +``` +TraceableBase +├── _FunctionPodBase (shared: executor routing, hooks, async, observer) +│ ├── FunctionPod (input → output; cached by default) +│ └── SideEffectPod (input → pass-through; invocation always logged) +└── StaticOutputOperatorPod (operator pod base) +``` + +### 2. Output Contract — Pass-Through + +**Decision:** Side-effect pods emit their input stream unchanged as output. The user +function returns `None`. + +**Rejected alternatives:** +- **No output (terminates branch):** Cuts the DAG at the side-effect node, preventing + downstream steps from seeing the same data. Not composable. +- **Structured record `{status, artifact_ref, timestamp}`:** Forces pod authors to define + an output schema for a conceptually void operation. Adds schema prediction complexity. + Deferred to a future `StatusEmittingPod` variant if downstream aggregation proves + necessary. + +Pass-through is the most composable choice: a side-effect pod can be inserted anywhere +in a pipeline without disrupting the data flow. + +### 3. Caching / Re-Execution Semantics — Author-Declared Idempotency + +**Decision:** Controlled by `SideEffectConfig.idempotent: bool = False`. + +```python +@dataclasses.dataclass(frozen=True) +class SideEffectConfig: + idempotent: bool = False + on_error: Literal["raise", "log"] = "raise" +``` + +- `idempotent=False` (default): always execute the side effect, even if the same + `invocation_signature` was seen before. Correct for append-only operations (log + writes, metric points). A new `execution_id` is generated each time. +- `idempotent=True`: skip execution if `invocation_signature` already present in the + invocation log. Correct for upsert-style operations where re-running with the same + inputs should not produce a duplicate artifact. A `status="skipped"` row is recorded. + +Regardless of `idempotent`, Orcapod **always persists an invocation log row** (with +appropriate status). This is required for the reverse-lookup guarantee — a hash found +in an artifact must be resolvable even if the pipeline has since been re-run. + +**Interaction with `FileHasher` / result records:** Side-effect pods do not participate +in the `ResultCache` / `FunctionNode` result table system. Their execution state is +tracked exclusively through the `_orcapod_side_effect_invocations` table. + +### 4. Pipeline Execution Hash — Two Complementary Identifiers + +**Decision:** Expose two hashes via `SideEffectContext`. + +#### `invocation_signature` (stable, deterministic) + +```python +import hashlib + +invocation_signature = hashlib.sha256( + pod.content_hash().digest + input_data.content_hash().digest +).digest()[:16].hex() # first 16 bytes → 32 lowercase hex chars, 128 bits +``` + +This is computed from the pod's code identity and the content hash of the input data +packet — the same components that determine `record_id_hash` for function pods in +ITL-523 (which uses the output datagram UUID; for deterministic functions, these +converge to the same value). For side-effect pods there is no output datagram, so the +input-based formula is the canonical choice. + +`pod.content_hash()` and `input_data.content_hash()` return `ContentHash` objects from +the existing Orcapod infrastructure. Only the `digest: bytes` field is used — the +method identifier is excluded so the formula is stable across hasher-version changes +(as long as the digest bytes are stable, which is guaranteed by each `ContentHash`'s +own versioning). + +Properties: +- **Deterministic:** same pod code + same inputs → same signature, always. +- **Stable across re-runs:** re-running the same pipeline with the same data produces + the same signature, even months later. +- **32 lowercase hex characters** (128 bits) — compact, embeddable, and practically + collision-free for any real-world invocation volume. + +`invocation_signature` is the primary key for idempotency checks and the primary +identifier embedded in external artifacts. + +#### `execution_id` (unique per invocation) + +``` +execution_id = str(uuid7()) # UUIDv7 — time-ordered, unique per actual execution +``` + +Properties: +- **Globally unique:** no two invocations share an `execution_id`, even with identical + inputs. +- **Ordered:** UUIDv7 embeds a timestamp, enabling chronological sorting of invocations + in the log. +- **36-character UUID string:** familiar format, works in every DB as a TEXT column. + +Consistent with ITL-523's `record_id_hash`, which is `str(output_data.datagram_uuid)`, +a per-execution UUID. The two hashes serve different purposes: `invocation_signature` +for deterministic provenance, `execution_id` for per-run tracing. + +**Format when embedded in artifacts:** + +``` +orcapod-{invocation_signature} → orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f +``` + +`ctx.format_id()` returns this canonical string. + +### 5. Hash Surface — `SideEffectContext` Parameter Auto-Injection + +**Decision:** The framework auto-injects a `SideEffectContext` instance when the user +function declares a parameter typed `SideEffectContext`. + +```python +@dataclasses.dataclass(frozen=True) +class SideEffectContext: + invocation_signature: str # 32-char hex; stable across identical re-runs + execution_id: str # UUID string; unique per actual execution + pod_name: str # human-readable pod label + pod_content_hash: str # pod.content_hash().to_string() + pipeline_run_id: str | None # run-level context; None for lazy pipelines + + def format_id(self) -> str: + """Canonical artifact tag: 'orcapod-{invocation_signature}'.""" + return f"orcapod-{self.invocation_signature}" +``` + +Detection is by **type annotation** (`SideEffectContext`), not by parameter name. +If no `SideEffectContext` parameter is declared, the framework calls the function +without it (zero overhead for pods that don't need the hash). + +**Why a context object, not a special parameter name?** +Type-based detection is explicit, IDE-discoverable, and carries no magic naming +convention. It mirrors how Python frameworks inject dependencies (FastAPI, pytest +fixtures with type resolution). It avoids the risk of a user accidentally declaring a +parameter named `ctx` for unrelated purposes. + +**Relationship to `InvocationIdentity` shared base:** +See Section 11 (Reconciliation with ITL-523). `SideEffectContext` embeds an +`InvocationIdentity` base — or is equivalent to one — enabling shared utilities. + +### 6. Failure Semantics — Author-Declared, Default Fail-Loud + +**Decision:** Controlled by `SideEffectConfig.on_error`: +- `"raise"` (default): exception propagates and the pipeline run fails. Consistent with + ITL-523's `HookConfig.on_error="raise"` default. +- `"log"`: exception is logged at `WARNING` level and swallowed; the input row is + passed through to downstream pods. Correct for non-critical side effects (metrics + emission) where a network hiccup should not abort the pipeline. + +No retries in v1. Retry logic (with backoff, max attempts) is deferred. + +When `on_error="log"` and an exception is raised: +- A log row with `status="error"` and `error_message=str(exc)` is recorded in the + invocation log (so the failure is visible in the near-side provenance record). +- The input row is still emitted as output (pass-through continues). + +### 7. Ordering & DAG Placement + +Side-effect pods are **synchronous barriers** in the DAG: + +1. Consume input row. +2. Execute side effect (call user function with `(data, ctx)` or `(data,)` as applicable). +3. On success: emit the same row as output. Record `status="success"` in invocation log. +4. On error with `on_error="raise"`: propagate exception; do not emit output. +5. On error with `on_error="log"`: log + record; emit input row as output anyway. + +For the **async channel execution path**, `async_execute()` must process each input +row, complete the side effect, and write to the output channel before returning. +Concurrent per-row execution within a single side-effect pod is permitted when +`PodConfig.max_concurrency > 1` and the pod is so configured — subject to the same +concurrency model as `FunctionPod`. + +**No fire-and-forget in v1.** All side effects complete (or fail) synchronously with +respect to the pipeline's progress on that row. Downstream pods cannot begin processing +a row until the side effect for that row has finished. + +### 8. Observability — `_orcapod_side_effect_invocations` Table + +**Decision:** Orcapod persists every side-effect invocation to a dedicated table in the +pipeline database. + +| Column | Type | Notes | +|--------|------|-------| +| `execution_id` | TEXT | Primary key — unique per actual execution | +| `invocation_signature` | TEXT | Indexed — stable per (pod, inputs) pair | +| `pod_name` | TEXT | Human-readable pod label | +| `pod_content_hash` | TEXT | `pod.content_hash().to_string()` — exact code version | +| `input_hash` | TEXT | `input_data.content_hash().to_string()` | +| `pipeline_run_id` | TEXT NULLABLE | `None` for lazy/non-compiled pipelines | +| `executed_at` | TIMESTAMP | UTC wall-clock time of execution | +| `status` | TEXT | `"success"` / `"error"` / `"skipped"` | +| `error_message` | TEXT NULLABLE | Set when `status = "error"` | + +**Schema location:** A shared table in the pipeline database (not scoped per-node), +allowing cross-pipeline lookup by `invocation_signature` alone. Table is created on +first invocation if it does not exist. + +This table is the **near side of the reverse-lookup chain**. Without it, a signature +extracted from an external artifact has nothing to resolve against. + +**For lazy/ephemeral pipelines** (no `PipelineJob`, no DB-backed nodes): the invocation +log still records the invocation, but `pipeline_run_id` is `None` and `input_hash` can +be used only to identify the data structure — the actual data rows are not persisted +and cannot be retrieved from the framework. + +### 9. Reverse Lookup Path + +**Given:** `invocation_signature = "d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f"` found in an +external artifact (e.g., a Postgres audit row or a log line). + +**Step 1 — Signature → invocation record:** +```sql +SELECT * +FROM _orcapod_side_effect_invocations +WHERE invocation_signature = 'd4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f' +ORDER BY executed_at +``` +Returns: `{pod_name, pod_content_hash, input_hash, pipeline_run_id, executed_at, status}`. + +Multiple rows may exist if `idempotent=False` and the pod ran multiple times with the +same inputs (each run has a distinct `execution_id`). + +**Step 2 — Input hash → input data:** +Query the pipeline's function node result tables for rows where +`_input_data_hash = input_hash`. This locates the exact `(Tag, Data)` packet that +triggered the side effect. + +*(Requires a DB-backed pipeline — `FunctionJobNode` or `OperatorJobNode`. For lazy +pipelines, the input hash identifies the data structure but rows cannot be retrieved.)* + +**Step 3 — Input data → upstream lineage:** +Follow `_tag_source_id` and `_tag_record_id` system tag columns on the retrieved input +row. These columns encode the full upstream provenance chain, back to the original +source and row number. + +**Step 4 — Pod code version:** +`pod_content_hash` identifies the exact pod function that ran. If the codebase is +version-controlled, this hash can be resolved to a specific commit and function definition. + +**Full reverse-lookup chain:** +``` +external artifact (orcapod-d4a8f3b2...) + → _orcapod_side_effect_invocations[invocation_signature] + → input_hash → function node result table → (Tag, Data) packet + → _tag_source_id / _tag_record_id → upstream nodes → root source rows + + pod_content_hash → exact pod code version +``` + +**Minimum persistence requirements:** +- `_orcapod_side_effect_invocations` table must be durable (not ephemeral). +- Input data rows must be persisted in DB-backed nodes for step 2 to work. +- For full lineage, all intermediate nodes must use DB-backed execution. + +### 10. Serialization — Prescribe Canonical Format, Provide Helpers + +**Decision:** Define a canonical artifact tag format and expose a `format_id()` helper. +Deviation is allowed but not recommended. + +**Canonical format:** `orcapod-{invocation_signature}` where `invocation_signature` is +32 lowercase hex characters (128 bits). + +**Examples:** + +Log line (structured logging): +``` +{"level": "info", "msg": "Patient processed", "orcapod_id": "orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f", "patient_id": "P-12345"} +``` + +Database column: +```sql +INSERT INTO audit_log (orcapod_record_id, patient_id, processed_at) +VALUES ('d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f', 'P-12345', NOW()) +``` +*(DB column stores the raw signature; the `orcapod-` prefix is for human-facing contexts.)* + +Slack notification body: +``` +Pipeline run complete for patient P-12345. +Trace: orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f +``` + +`SideEffectContext` provides: +- `ctx.format_id()` → `"orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f"` (full canonical tag) +- `ctx.invocation_signature` → `"d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f"` (raw hex for DB columns) + +Pod authors who deviate from the canonical format do so at their own risk. The reverse +lookup machinery works as long as `invocation_signature` appears in the invocation log — +but human discoverability and tooling support will be reduced for non-canonical embeddings. + +--- + +## Reconciliation with ITL-523 (Post-Run Hook) + +| Dimension | Post-run hook (ITL-523) | Side-effect pod (ITL-524) | +|-----------|------------------------|--------------------------| +| DAG presence | No — passive observer | Yes — first-class node | +| Triggers on | After a function pod runs | IS the pipeline step | +| Gates downstream | No | Yes | +| Composable | No | Yes | +| Hash name | `record_id_hash` | `invocation_signature` | +| Hash formula | `str(output_data.datagram_uuid)` (UUIDv7 per execution) | `content_hash(pod ∥ inputs).to_hex()` (deterministic) | +| Hash stability | Stable for cache hits; unique for fresh computations | Always deterministic | +| `on_error` vocabulary | `"raise"` / `"log"` | Same: `"raise"` / `"log"` | + +**Hash formula divergence — justified:** + +ITL-523's `record_id_hash` uses `output_data.datagram_uuid`, a UUIDv7 that is unique +per fresh execution but stable on cache hits (UUID stored in DB, restored on retrieval). +This is appropriate for function pods because the "result record ID" IS the output +datagram identity. + +Side-effect pods have no output datagram, so the equivalent must be derived from +inputs. Using `content_hash(pod ∥ inputs)` is more fundamental: it is deterministic +regardless of caching, enabling idempotency checks across re-runs. For a deterministic +function pod, the two formulas converge: `content_hash(pod ∥ inputs)` ≡ +`output_data.datagram_uuid` whenever the output is content-addressed from the same +inputs. + +**Shared `InvocationIdentity` base type:** + +To avoid duplicating the canonical `format_id()` logic and enable shared tooling +(e.g., a logging helper that works for both hook payloads and side-effect contexts), +extract a common type from `src/orcapod/hooks.py`: + +```python +@dataclasses.dataclass(frozen=True) +class InvocationIdentity: + """Shared identity carrier for post-run hooks and side-effect pods. + + Attributes: + record_id: The invocation identifier. For post-run hooks this is + ``str(output_data.datagram_uuid)``; for side-effect pods this is + ``content_hash(pod || inputs).to_hex()``. Both are embeddable + plain-text strings. + pod_name: Human-readable pod label. + pipeline_run_id: Run context; ``None`` for lazy pipelines. + """ + record_id: str + pod_name: str + pipeline_run_id: str | None + + def format_id(self) -> str: + """Canonical artifact tag: 'orcapod-{record_id}'.""" + return f"orcapod-{self.record_id}" +``` + +`SideEffectContext` may directly embed the `InvocationIdentity` fields or inherit from it +(implementation decision); it adds `execution_id` and `pod_content_hash`: + +```python +@dataclasses.dataclass(frozen=True) +class SideEffectContext: + # InvocationIdentity fields: + invocation_signature: str # = record_id for side-effect pods + pod_name: str + pipeline_run_id: str | None + + # SideEffectPod-specific fields: + execution_id: str # UUIDv7 unique per execution + pod_content_hash: str # pod.content_hash().to_string() + + def format_id(self) -> str: ... +``` + +Post-run hook `PodContext` gains `InvocationIdentity` fields or a reference in the +updated `PostRunPayload` (minor ITL-523 extension, not in scope for this ticket but +called out for the implementer). + +--- + +## Worked Example + +**Scenario:** An ML pipeline processes patient records and writes an audit row to an +external compliance database for each processed patient. The audit table must trace +back to the exact pipeline run and input data. + +### Pod definition + +```python +import orcapod as op +from orcapod import SideEffectContext, SideEffectConfig +from datetime import datetime, timezone + +@op.side_effect_pod(config=SideEffectConfig(idempotent=True, on_error="log")) +def audit_patient_processing( + data: PatientRecord, + ctx: SideEffectContext, +) -> None: + """Write a compliance audit row to the external database. + + Uses ctx.invocation_signature as the stable provenance key so that + re-running the same pipeline with the same inputs does NOT produce a + duplicate audit row (idempotent=True skips execution if the signature + is already in the invocation log, and ON CONFLICT DO NOTHING handles + any race between instances). + """ + audit_db.execute( + """ + INSERT INTO patient_audit + (orcapod_record_id, patient_id, processed_at, pipeline_run) + VALUES (%s, %s, %s, %s) + ON CONFLICT (orcapod_record_id) DO NOTHING + """, + ( + ctx.invocation_signature, + data.patient_id, + datetime.now(timezone.utc), + ctx.pipeline_run_id, + ), + ) +``` + +### Pipeline + +```python +patient_stream = op.csv_source("patients_2026.csv") +validated_stream = validate_patient_pod(patient_stream) +enriched_stream = enrich_patient_pod(validated_stream) +# Side-effect node inserted mid-pipeline; downstream sees the same data. +audited_stream = audit_patient_processing(enriched_stream) +results_stream = aggregate_results_pod(audited_stream) +``` + +### External artifact + +Row in the external `patient_audit` table: + +| orcapod_record_id | patient_id | processed_at | pipeline_run | +|---|---|---|---| +| `d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f` | P-12345 | 2026-07-14 02:30:45 UTC | run-2026-07-14-001 | + +### Reverse-lookup walk-through + +**Problem:** Six months later, an auditor finds the row for `P-12345` and asks: "What +input data produced this audit record, and what pipeline logic ran?" + +**Step 1 — Signature → invocation record:** +```sql +SELECT pod_name, pod_content_hash, input_hash, pipeline_run_id, executed_at, status +FROM _orcapod_side_effect_invocations +WHERE invocation_signature = 'd4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f' +``` +``` +pod_name: "audit_patient_processing" +pod_content_hash: "object_v0.1:c9f1a3b2..." +input_hash: "object_v0.1:8f3a4b2c..." +pipeline_run_id: "run-2026-07-14-001" +executed_at: 2026-07-14T02:30:45Z +status: "success" +``` + +**Step 2 — Input hash → input data:** +```sql +-- Query enrich_patient_pod result table (shared pipeline hash table). +-- _input_data_hash is system_constants.INPUT_DATA_HASH_COL = "_input_data_hash" +SELECT * FROM enrich_patient_pod_results +WHERE _input_data_hash = 'object_v0.1:8f3a4b2c...' +``` +``` +patient_id: "P-12345" +birth_year: 1985 +diagnosis: "T2D" +risk_score: 0.73 +_tag_source_id::...::abc... "patients_2026_csv::row47" +``` + +**Step 3 — System tags → upstream lineage:** +The `_tag_source_id` column resolves to the original source: +``` +Source: patients_2026.csv +Row: 47 +Column: patient_id → "P-12345" +``` + +**Full lineage:** +``` +patient_audit[orcapod_record_id=d4a8f3b2...] + ← audit_patient_processing (invocation_signature=d4a8f3b2..., pod_content_hash=c9f1a3b2...) + ← enrich_patient_pod(PatientRecord{patient_id=P-12345}) + ← validate_patient_pod(PatientRecord{patient_id=P-12345}) + ← patients_2026.csv : row 47 +``` + +**Pod code version:** `pod_content_hash = "object_v0.1:c9f1a3b2..."` can be resolved +to the exact Python function definition (git commit, code hash) via Orcapod's pod +registry or by running `content_hash()` against the current function definition. + +--- + +## Public API Summary + +```python +# src/orcapod/side_effects.py + +@dataclasses.dataclass(frozen=True) +class SideEffectConfig: + idempotent: bool = False + on_error: Literal["raise", "log"] = "raise" + + +@dataclasses.dataclass(frozen=True) +class SideEffectContext: + invocation_signature: str # 32-char hex; deterministic per (pod, inputs) + execution_id: str # UUID string; unique per actual execution + pod_name: str # human-readable pod label + pod_content_hash: str # pod.content_hash().to_string() + pipeline_run_id: str | None # None for lazy/non-compiled pipelines + + def format_id(self) -> str: + """Return 'orcapod-{invocation_signature}'.""" + return f"orcapod-{self.invocation_signature}" + + +class SideEffectPod(_FunctionPodBase): + """A pipeline node whose primary purpose is a side effect. + + The wrapped function returns None. The input stream is passed through + unchanged as output. Every invocation is recorded in the pipeline + database's _orcapod_side_effect_invocations table. + """ + ... + + +def side_effect_pod( + fn: SideEffectFn | None = None, + *, + config: SideEffectConfig | None = None, +) -> SideEffectPod | Callable[[SideEffectFn], SideEffectPod]: + """Decorator that wraps a function as a SideEffectPod. + + Examples: + @side_effect_pod + def emit_metric(data: MyData, ctx: SideEffectContext) -> None: + metrics.emit("my.metric", data.value, tags={"id": ctx.format_id()}) + + @side_effect_pod(config=SideEffectConfig(idempotent=True)) + def write_audit_row(data: MyData, ctx: SideEffectContext) -> None: + audit_db.execute("INSERT ...", (ctx.invocation_signature, ...)) + """ + ... +``` + +Re-exported from `orcapod.__init__`: +- `SideEffectPod` +- `SideEffectConfig` +- `SideEffectContext` +- `side_effect_pod` + +--- + +## Tests + +New test file: `tests/test_core/side_effect_pod/test_side_effect_pod.py` + +| # | Scenario | Assertion | +|---|---|---| +| T1 | Basic execution — pod runs, output equals input | Pass-through: input (Tag, Data) emitted unchanged | +| T2 | `ctx` auto-injection — function with `ctx: SideEffectContext` | `ctx.invocation_signature` is 32 hex chars; `ctx.format_id()` returns `"orcapod-{sig}"` | +| T3 | No-ctx function — function without `ctx` param | Pod runs without error; `SideEffectContext` not constructed | +| T4 | Invocation log written — DB-backed pipeline | `_orcapod_side_effect_invocations` has one row; `status="success"` | +| T5 | `idempotent=True` — same inputs re-run | Second invocation skipped; `status="skipped"` row added; side-effect function called once only | +| T6 | `idempotent=False` (default) — same inputs re-run | Side-effect function called both times; two rows in invocation log (different `execution_id`) | +| T7 | `on_error="raise"` — function raises | Exception propagates; no output row; `status="error"` in log | +| T8 | `on_error="log"` — function raises | Exception logged; input row emitted unchanged; `status="error"` in log | +| T9 | Multiple inputs — N input rows | N log rows; pass-through of all N rows | +| T10 | `invocation_signature` determinism | Re-running pod with identical inputs + identical code → identical `invocation_signature` | +| T11 | `execution_id` uniqueness | Two runs with identical inputs → same `invocation_signature`, different `execution_id` | +| T12 | Async channel execution | Pod runs correctly through `async_execute()` path; log rows written | +| T13 | Parallel execution — `max_concurrency > 1` | All invocations complete; all log rows written; no data loss | +| T14 | Pipeline composition — side-effect pod mid-pipeline | Downstream pod receives same data as upstream; side effect runs | +| T15 | `@side_effect_pod` decorator — functional form | `@side_effect_pod` without args and `@side_effect_pod(config=...)` both produce `SideEffectPod` | + +--- + +## Scope & Boundaries + +**In scope:** +- `SideEffectPod` class, `SideEffectConfig`, `SideEffectContext`, `side_effect_pod` + decorator. +- `_orcapod_side_effect_invocations` table: creation, insertion, idempotency check. +- `SideEffectContext` auto-injection by type annotation. +- Pass-through output stream. +- Sync and async execution paths. +- `InvocationIdentity` shared base type extracted into `hooks.py`. +- All re-exports from `orcapod.__init__`. +- Tests covering T1–T15. +- Documentation update: new `docs/concepts/side-effect-pods.md`. + +**Out of scope:** +- Actual implementation (this is a design spike). +- Retry logic. +- External adapter helpers (Slack notifier, structured-logging wrapper). +- Run history UI or searchable index service (downstream of this primitive; see PLT-1950). +- Pre-run or on-error hooks for side-effect pods. +- Remote/cross-process side-effect execution. +- `idempotency_key` customization (v1 uses `invocation_signature` only). + +--- + +## Dependencies & Risks + +**Dependencies:** +- ITL-523 (post-run hook, In Progress) — `hooks.py` is the insertion point for the + shared `InvocationIdentity` base type. Coordinate to avoid conflicting changes to + `_FunctionPodBase`. +- PLT-1950 (EDI provenance) — the `_orcapod_side_effect_invocations` table is a + natural data source for provenance stamping. Schema should be coordinated. + +**Risks:** +- **Framework creep:** Side-effect pods that "almost" produce output. Enforced by + the type signature (`-> None` required) and the pass-through contract. +- **Overly long hash:** 32 hex chars (128 bits) is short enough for all common embedding + contexts. Test with real log pipelines before finalizing. +- **Invocation log growth:** A high-volume pipeline calling a side-effect pod per row + will produce large invocation logs. Add a note in docs about pruning / archiving. +- **Idempotency false-positive:** If pod code changes between runs, the same inputs + produce the same `invocation_signature` (code change is not reflected). This is + correct behavior (the signature is input + code hash, so code changes produce + different signatures), but implementers should verify the `pod_content_hash` column + includes code identity, not just name. + +--- + +## Follow-Up Implementation Issue + +See ITL-525: *Implement `SideEffectPod` with invocation hash and invocation log +(ITL-524 follow-up).* + +Task breakdown for ITL-525: + +1. **`InvocationIdentity` base type** — add to `src/orcapod/hooks.py`; update + `PostRunPayload` to embed it (minor ITL-523 coordination). +2. **`src/orcapod/side_effects.py`** — new module: `SideEffectConfig`, + `SideEffectContext`, `SideEffectPod`, `side_effect_pod` decorator. +3. **`SideEffectPod` core execution** — `process_data()` / `async_process_data()` + pass-through implementation; `SideEffectContext` injection; invocation log writes. +4. **Invocation log** — `_orcapod_side_effect_invocations` table creation and insertion + in the pipeline DB; idempotency check for `idempotent=True`. +5. **`orcapod.__init__` re-exports** — `SideEffectPod`, `SideEffectConfig`, + `SideEffectContext`, `side_effect_pod`. +6. **Tests** — `tests/test_core/side_effect_pod/test_side_effect_pod.py` covering T1–T15. +7. **Documentation** — `docs/concepts/side-effect-pods.md` with overview, usage + examples, reverse-lookup walk-through, and idempotency guidance. From df01eb2b17dbf8bb1d53ac74cdf3df42159c1a8f Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:14:03 +0000 Subject: [PATCH 02/13] docs(spike): fix Overview to not pre-announce design outcomes (ITL-524) Remove premature commitment to '-> None' and pass-through contract from the Overview; those are Axis 2 outcomes argued in the Design section. Overview now describes the problem space and the provenance motivation only. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index 266704d3..aa9c349a 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -6,15 +6,19 @@ Orcapod pipelines are built around **function pods** — nodes that transform an packet into an output packet. Many real pipelines also need steps whose primary purpose is a side effect: writing a structured log line, appending a row to an external database, emitting a metric, or sending a notification. Today those steps must be shoehorned into -function pods with fake outputs or `None` returns. - -This design introduces **side-effect pods** as a first-class pipeline concept. A side-effect -pod wraps a user function that returns `None`, passes its input stream through unchanged, -and embeds a deterministic **invocation signature** into every external artifact it -produces. That signature bridges the provenance boundary: anyone holding an artifact can -look it up in Orcapod's invocation log and trace it back to the originating pipeline run, -pod version, and input data — following the same provenance chains as any other Orcapod -node. +function pods with fake outputs or breadcrumb return values. + +This design introduces **side-effect pods** as a first-class pipeline concept, with a +crucial property: any external artifact the side effect produces can be **linked back to +the specific pipeline execution** that produced it — via a deterministic **invocation +signature** embedded into the artifact. That signature bridges the provenance boundary: +anyone holding an artifact can look it up in Orcapod's invocation log and trace it back +to the originating pipeline run, pod version, and input data — following the same +provenance chains as any other Orcapod node. + +This document resolves the open design axes (taxonomy, output contract, caching +semantics, hash formula, API surface, failure handling, DAG ordering, observability, and +serialization conventions) and specifies the follow-up implementation work. --- From cb3b94648808acc7d42f641667fd343fc1cf14e3 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:28:21 +0000 Subject: [PATCH 03/13] docs(spike): reframe Goals to reflect interactive design process (ITL-524) Goals & Success Criteria now describes what the spike produces (discussed, Edgar-approved decisions per axis) rather than front-loading the design outcomes. Adds explicit statement that final calls belong to Edgar. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 39 ++++++++++--------- 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index aa9c349a..b80549d2 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -24,25 +24,26 @@ serialization conventions) and specifies the follow-up implementation work. ## Goals & Success Criteria -- A `SideEffectPod` class and `@side_effect_pod` decorator that: - - Accept a user function `(data: T, [ctx: SideEffectContext]) -> None`. - - Pass the input stream through unchanged as output. - - Inject a `SideEffectContext` carrying `invocation_signature`, `execution_id`, - `pod_name`, and `pipeline_run_id` when the user function declares a `ctx` - parameter typed `SideEffectContext`. - - Accept a `SideEffectConfig` controlling idempotency and error handling. -- A `SideEffectContext` dataclass with a `format_id()` helper that returns the - canonical `orcapod-{invocation_signature}` string for embedding in artifacts. -- An `_orcapod_side_effect_invocations` table in the pipeline database that records - every invocation (hash, timing, status, input hash, pod identity) as the - **near-side** of the reverse-lookup chain. -- Idempotency support: when `SideEffectConfig(idempotent=True)`, skip re-execution - if the same `invocation_signature` is already in the invocation log. -- Sync and async execution paths both supported. -- A worked end-to-end example demonstrating the full reverse-lookup walk-through. -- Explicit reconciliation with the ITL-523 post-run hook: shared hash formula, shared - `on_error` vocabulary, shared `InvocationIdentity` base type. -- A follow-up implementation issue (ITL-525) filed with concrete task breakdown. +This is an **interactive design spike**. Every design axis listed in this document must +be explicitly discussed with Edgar Walker before a decision is recorded. The agent's +role is to surface options, trade-offs, and codebase constraints; the final call on each +axis belongs to Edgar. The spec is a summary of those concluded discussions — not a +set of decisions made independently. + +The spike is complete when: + +- Every design axis below has been discussed and a final decision recorded, with the + chosen option and the rationale for rejecting alternatives. +- The pipeline execution hash is fully specified: its formula, format, stability + properties, and how a pod author accesses it from inside a pod body. +- At least one worked example is included: a side-effect pod definition, the external + artifact it produces, and a concrete walk-through of the reverse-lookup path from + that artifact back to the originating pipeline run and inputs. +- The design is explicitly reconciled with ITL-523 (post-run hook): where the two + concepts share primitives, that is called out; where they diverge, the reason is + stated. +- A follow-up implementation issue is filed in Linear with a concrete task breakdown, + ready to pick up. --- From 83c2b4185b2507ecb0312d5e582679e1a139de75 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:31:15 +0000 Subject: [PATCH 04/13] =?UTF-8?q?docs(spike):=20record=20Axis=201=20decisi?= =?UTF-8?q?on=20=E2=80=94=20protocol=20+=20=5FFunctionPodBase=20impl=20(IT?= =?UTF-8?q?L-524)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SideEffectPodProtocol defines the contract in protocols/; SideEffectPod implements it by extending _FunctionPodBase for implementation reuse. Mirrors existing protocol/implementation pattern throughout the codebase. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 49 +++++++++++-------- 1 file changed, 29 insertions(+), 20 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index b80549d2..4089d020 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -76,32 +76,41 @@ This is the reverse-lookup chain. The signature must be: ## Design Axes -### 1. Pod Taxonomy — New `SideEffectPod` class +### 1. Pod Taxonomy -**Decision:** Introduce a new `SideEffectPod` class alongside `FunctionPod`. +**Decision:** Introduce `SideEffectPodProtocol` as a proper protocol and `SideEffectPod` +as its concrete implementation. -`SideEffectPod` shares the `_FunctionPodBase` execution infrastructure (executor -routing, async channel execution, observer integration), but has distinct semantics: -the wrapped function returns `None`, the output stream is the input stream passed -through, and invocations are always recorded in the side-effect invocation log. +**Protocol layer — `SideEffectPodProtocol`** (in `src/orcapod/protocols/`): +Defines the contract of a side-effect pod purely in terms of its methods and type +signatures, independent of any implementation detail. This is the type that user code, +type checkers, and the framework itself reason about. It follows the same pattern as +`StreamProtocol`, `DataProtocol`, `PodProtocol`, and other protocol definitions +throughout Orcapod. -**Why not a flag (`side_effect=True`) on `FunctionPod`?** -A flag adds dead branches to every `FunctionPod` code path and spreads side-effect -semantics across a class whose core contract is `input → output`. It also makes type -signatures ambiguous — `FunctionPod.process()` always promises an output stream. +**Implementation layer — `SideEffectPod`** (in `src/orcapod/core/`): +Implements `SideEffectPodProtocol` by extending `_FunctionPodBase`. This inheritance +is purely an implementation choice — it gives `SideEffectPod` hashing, executor +routing, async channel execution, and observer wiring for free, without those +mechanisms needing to be duplicated. The public contract is determined by the protocol, +not by what `_FunctionPodBase` exposes. -**Why not a subclass of `FunctionPod`?** -Subclassing inherits the output-oriented contract (return type, caching logic, -`record_id_hash` on the output datagram). Overriding those creates a leaky abstraction. -A sibling class with shared base infrastructure is cleaner. +**Rejected alternatives:** +- **Flag on `FunctionPod`** (`side_effect=True`): creates a dead branch in every + `FunctionPod` code path and makes the type contract ambiguous at the call site. +- **Subclass of `FunctionPod`**: inherits the output-oriented contract and then must + override it, producing a leaky abstraction. `_FunctionPodBase` is the right + inheritance target — not `FunctionPod` itself. -**Class hierarchy:** +**Class structure:** ``` -TraceableBase -├── _FunctionPodBase (shared: executor routing, hooks, async, observer) -│ ├── FunctionPod (input → output; cached by default) -│ └── SideEffectPod (input → pass-through; invocation always logged) -└── StaticOutputOperatorPod (operator pod base) +protocols/ +└── SideEffectPodProtocol (contract: methods + type signatures) + +core/ +└── _FunctionPodBase (shared execution infrastructure) + ├── FunctionPod (implements FunctionPodProtocol) + └── SideEffectPod (implements SideEffectPodProtocol) ``` ### 2. Output Contract — Pass-Through From 960c71b2e797acc94f607055472741245e93eb74 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:13:22 +0000 Subject: [PATCH 05/13] =?UTF-8?q?docs(spike):=20record=20Axis=202=20?= =?UTF-8?q?=E2=80=94=20SinkPod=20(terminal)=20and=20TapPod=20(pass-through?= =?UTF-8?q?)=20(ITL-524)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two distinct pod kinds: SinkPod for completion-tracked terminal delivery to external sources, TapPod for always-fire pass-through observation. Retry policy deferred to ITL-526. Grouping under a shared SideEffectPod umbrella deferred. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 74 +++++++++++++++---- 1 file changed, 61 insertions(+), 13 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index 4089d020..d4b3d01c 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -113,21 +113,69 @@ core/ └── SideEffectPod (implements SideEffectPodProtocol) ``` -### 2. Output Contract — Pass-Through +### 2. Pod Kinds and Output Contract -**Decision:** Side-effect pods emit their input stream unchanged as output. The user -function returns `None`. +**Decision:** There are two categorically distinct kinds of side-effect pod. They differ +enough in semantics, infrastructure requirements, and output contract to warrant +separate types rather than a single class with a flag. Whether they share a common +`SideEffectPod` umbrella type is deferred — it is not necessary to decide for the +initial implementation. -**Rejected alternatives:** -- **No output (terminates branch):** Cuts the DAG at the side-effect node, preventing - downstream steps from seeing the same data. Not composable. -- **Structured record `{status, artifact_ref, timestamp}`:** Forces pod authors to define - an output schema for a conceptually void operation. Adds schema prediction complexity. - Deferred to a future `StatusEmittingPod` variant if downstream aggregation proves - necessary. - -Pass-through is the most composable choice: a side-effect pod can be inserted anywhere -in a pipeline without disrupting the data flow. +#### `SinkPod` — completion-tracked, terminal delivery + +A `SinkPod` writes to an external source with a meaningful notion of success or failure +*per input*. The intent is exactly-once delivery: each (pod, input) pair should result +in exactly one successful write. + +**Output contract: terminal.** The pod ends the DAG branch. Delivering data outward is +the purpose; what happens next in the pipeline is a separate concern expressed as a +separate branch before this node. + +**Completion tracking.** The framework records completion state per (pod, input) in an +internal log: + +- **Not yet seen** → execute the delivery. +- **Previously succeeded** → skip silently on re-run. +- **Previously failed** → re-attempt on the next explicit pipeline re-run. + +Retry policy (automatic retries, max attempts, backoff) is **deferred to ITL-526**, +which will design a shared retry mechanism for both `FunctionPod` and `SinkPod`. For +now, re-attempt occurs only when the pipeline is explicitly re-run by the user. + +Completion state is internalized — tracked in the pod's invocation log, visible to +observers, not emitted as pipeline data. + +**Canonical examples:** writing an audit row to a database, archiving a file to object +storage, posting a one-time notification. + +#### `TapPod` — always-fire, pass-through observation + +A `TapPod` reacts to every data packet passing through it, unconditionally. There is no +notion of completion, no memory of past encounters, no skip or retry logic. + +**Output contract: pass-through.** The input tag and data flow through unchanged. The +pod is a transparent interceptor — a tap on the stream. + +**No completion tracking.** Every pipeline run, every input triggers the effect. This +is correct and expected: a logger should log every packet every time, not once. + +**Canonical examples:** structured logging, metrics emission, real-time monitoring +feeds, debug tracing. + +#### Comparison + +| | `SinkPod` | `TapPod` | +|---|---|---| +| Output contract | Terminal | Pass-through | +| Completion tracking | Yes — success / failed / not-seen | No | +| Re-run behavior | Skip on success; retry on re-run if failed | Always fires | +| Invocation log | Required (completion state) | Optional (observability only) | +| Retry policy | Deferred to ITL-526 | N/A | + +**Rejected alternative — single class with `idempotent: bool`:** The semantic +difference between "always fire" and "fire once and track completion" is substantial +enough that a flag would obscure rather than clarify intent. A pod author should not +need to reason about idempotency to write a logger. ### 3. Caching / Re-Execution Semantics — Author-Declared Idempotency From 788b35e41898092b30b5d6cab0e45862c04eaf37 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:51:54 +0000 Subject: [PATCH 06/13] =?UTF-8?q?docs(spike):=20record=20Axes=203=20&=204?= =?UTF-8?q?=20=E2=80=94=20completion=20semantics=20and=20invocation=20hash?= =?UTF-8?q?=20(ITL-524)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Axis 3: SinkPod completion state machine (not-seen/success/failed), retry on explicit re-run only, retry policy deferred to ITL-526. TapPod always fires. Axis 4: Two-part invocation hash (pipeline_hash::full_input_packet_hash) using :: separator. TapPod adds verbatim pipeline_run_id as third component. Full input packet covers tag + data + source + system tag columns. InvocationHashConfig with encoding (hex/base64/binary) and component_length in bytes of raw digest. Table organisation mirrors FunctionNode: pipeline_hash scopes table, input packet hash is row key. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 169 +++++++++++------- 1 file changed, 107 insertions(+), 62 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index d4b3d01c..0e190d41 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -177,92 +177,137 @@ difference between "always fire" and "fire once and track completion" is substan enough that a flag would obscure rather than clarify intent. A pod author should not need to reason about idempotency to write a logger. -### 3. Caching / Re-Execution Semantics — Author-Declared Idempotency +### 3. Re-Execution and Completion Semantics -**Decision:** Controlled by `SideEffectConfig.idempotent: bool = False`. +**`TapPod`:** No caching or completion tracking. Every pipeline run, every input packet +triggers the effect unconditionally. There is no state to consult and no state to record +beyond optional observability logging. -```python -@dataclasses.dataclass(frozen=True) -class SideEffectConfig: - idempotent: bool = False - on_error: Literal["raise", "log"] = "raise" -``` +**`SinkPod`:** Completion-tracked per (pod topology, input packet). The framework +maintains a per-pod invocation table (see Axis 8) and consults it at the start of each +execution: -- `idempotent=False` (default): always execute the side effect, even if the same - `invocation_signature` was seen before. Correct for append-only operations (log - writes, metric points). A new `execution_id` is generated each time. -- `idempotent=True`: skip execution if `invocation_signature` already present in the - invocation log. Correct for upsert-style operations where re-running with the same - inputs should not produce a duplicate artifact. A `status="skipped"` row is recorded. +| Prior state | Action | +|---|---| +| Not yet seen | Execute the delivery | +| Previously succeeded | Skip silently | +| Previously failed | Re-attempt | -Regardless of `idempotent`, Orcapod **always persists an invocation log row** (with -appropriate status). This is required for the reverse-lookup guarantee — a hash found -in an artifact must be resolvable even if the pipeline has since been re-run. +Re-attempt occurs when the pipeline is **explicitly re-run by the user**. There is no +automatic background retry in v1. Full retry policy (max attempts, backoff, retryable +exception types) is deferred to ITL-526, which will design a shared mechanism for both +`FunctionPod` and `SinkPod`. -**Interaction with `FileHasher` / result records:** Side-effect pods do not participate -in the `ResultCache` / `FunctionNode` result table system. Their execution state is -tracked exclusively through the `_orcapod_side_effect_invocations` table. +`SinkPod` does **not** participate in the `ResultCache` / `FunctionNode` result table +system. Its execution state is tracked exclusively in its own invocation table. -### 4. Pipeline Execution Hash — Two Complementary Identifiers +### 4. Invocation Hash -**Decision:** Expose two hashes via `SideEffectContext`. +#### Raw identity components -#### `invocation_signature` (stable, deterministic) +The raw identity of any invocation is a **three-part compound** (two parts for +`SinkPod`, three for `TapPod`): -```python -import hashlib +1. **`pipeline_hash(pod)`** — the pipeline hash of the pod, encoding both the pod's own + function identity and the full upstream topology. This is the same `pipeline_hash()` + mechanism used by `FunctionNode` for DB path scoping: different pipeline topologies + using the same function produce different hashes, so completion records are namespaced + per topology. -invocation_signature = hashlib.sha256( - pod.content_hash().digest + input_data.content_hash().digest -).digest()[:16].hex() # first 16 bytes → 32 lowercase hex chars, 128 bits -``` +2. **`full_input_packet_hash`** — a hash of the entire input packet: tag columns, data + columns, source columns (`_source_*`), and system tag columns (`_tag_*`). This is + more inclusive than the `input_data.content_hash()` used by `ResultCache`, which + covers data columns only. Including tags and source/system columns means two packets + with identical payload but different originating provenance — different sources, + different upstream routes — produce different invocation hashes. This is correct: a + sink that writes an audit row needs to distinguish rows by their full provenance, not + just their data values. -This is computed from the pod's code identity and the content hash of the input data -packet — the same components that determine `record_id_hash` for function pods in -ITL-523 (which uses the output datagram UUID; for deterministic functions, these -converge to the same value). For side-effect pods there is no output datagram, so the -input-based formula is the canonical choice. +3. **`pipeline_run_id`** (`TapPod` only, when available) — a verbatim string + identifying the specific pipeline execution (e.g. `"run-2026-07-14-001"`). This is + user-controlled and included as-is without further hashing. When no run ID is + available (lazy/in-memory pipeline), the third component is omitted. -`pod.content_hash()` and `input_data.content_hash()` return `ContentHash` objects from -the existing Orcapod infrastructure. Only the `digest: bytes` field is used — the -method identifier is excluded so the formula is stable across hasher-version changes -(as long as the digest bytes are stable, which is guaranteed by each `ContentHash`'s -own versioning). +**Why `pipeline_hash` rather than `content_hash` for component 1:** +`content_hash(pod)` captures only the pod's function identity. `pipeline_hash(pod)` +captures function identity *plus* the full upstream topology, forming a Merkle chain. +Using `pipeline_hash` ensures that the same sink function used in two pipelines with +different upstream structures never shares a completion namespace — a delivery confirmed +in pipeline A does not suppress delivery in pipeline B. -Properties: -- **Deterministic:** same pod code + same inputs → same signature, always. -- **Stable across re-runs:** re-running the same pipeline with the same data produces - the same signature, even months later. -- **32 lowercase hex characters** (128 bits) — compact, embeddable, and practically - collision-free for any real-world invocation volume. +**Why `TapPod` includes the run ID:** +`SinkPod`'s hash must be run-agnostic so the framework can detect "already delivered" +across re-runs. `TapPod` fires every run intentionally and has no deduplication to +preserve — including the run ID makes every firing distinguishable, which is exactly +right for traceable observation. -`invocation_signature` is the primary key for idempotency checks and the primary -identifier embedded in external artifacts. +#### Internal record format -#### `execution_id` (unique per invocation) +Internally, the full compound is stored with `::` as the component separator (using +`::` avoids ambiguity with the `:` in `method:digest` form of `ContentHash.to_string()`): ``` -execution_id = str(uuid7()) # UUIDv7 — time-ordered, unique per actual execution +SinkPod: {pipeline_hash.to_string()}::{full_input_packet_hash.to_string()} +TapPod: {pipeline_hash.to_string()}::{full_input_packet_hash.to_string()}::{run_id} ``` -Properties: -- **Globally unique:** no two invocations share an `execution_id`, even with identical - inputs. -- **Ordered:** UUIDv7 embeds a timestamp, enabling chronological sorting of invocations - in the log. -- **36-character UUID string:** familiar format, works in every DB as a TEXT column. +Example (SinkPod): +``` +blake3_v1:a3f8c2d1e5b7f0c9...::blake3_v1:b7e491f0a2c3d8e1... +``` -Consistent with ITL-523's `record_id_hash`, which is `str(output_data.datagram_uuid)`, -a per-execution UUID. The two hashes serve different purposes: `invocation_signature` -for deterministic provenance, `execution_id` for per-run tracing. +The internal record always stores the full-fidelity compound. No output format +configuration is stored alongside it. -**Format when embedded in artifacts:** +#### User-facing output and `InvocationHashConfig` +When a pod author accesses `ctx.invocation_hash`, the compound is serialized according +to the pod's `InvocationHashConfig`. This serialized form is what gets embedded in +external artifacts (log fields, DB columns, message bodies) **and** what is used as +the lookup key in the invocation table — the same string, in both places. + +```python +@dataclasses.dataclass(frozen=True) +class InvocationHashConfig: + encoding: Literal["hex", "base64", "binary"] = "hex" + component_length: int | None = None # bytes of raw digest to use; None = full length ``` -orcapod-{invocation_signature} → orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f -``` -`ctx.format_id()` returns this canonical string. +`component_length` is specified in **bytes of raw digest** (encoding-independent): 8 bytes +produces 16 hex characters or 11 base64 characters depending on `encoding`. The `::` separator +is fixed. + +The format is self-describing: hex strings contain only `[0-9a-f]`, base64 contains +`[A-Za-z0-9+/=]`, binary is raw bytes. No configuration metadata needs to be stored +alongside the hash — the format is recoverable by inspection. + +Example outputs for `component_length=8`: + +| `encoding` | Example | +|---|---| +| `"hex"` (default) | `a3f8c2d1e5b7f0c9::b7e491f0a2c3d8e1` | +| `"base64"` | `o/jC0eW3::t+SR8KLD` | +| `"binary"` | `<8 raw bytes>::<8 raw bytes>` | + +`InvocationHashConfig` lives inside `SinkPodConfig` and `TapPodConfig` respectively, +so each pod can independently configure its output format. + +#### Lookup table organisation (mirrors `FunctionNode`) + +`SinkPod` records are stored in a table scoped by `pipeline_hash(pod)` — exactly the +same pattern `FunctionNode` uses for its result tables. Within that table, the row +lookup key is the `full_input_packet_hash`. The full compound is also stored as a column +for completeness. + +Reverse lookup from a user-provided hash: +1. Decode both components (format is self-describing). +2. Find tables whose scoping pipeline hash starts with component 1 (prefix match). +3. Within the matching table, find rows whose input packet hash starts with component 2. + +Shorter `component_length` values increase the chance of a prefix collision but the +lookup path remains valid — the user resolves any ambiguity by inspecting multiple +candidate rows. Collision probability at 8 bytes (64 bits) per component is negligible +for any realistic invocation volume. ### 5. Hash Surface — `SideEffectContext` Parameter Auto-Injection From 3b8b8c240311d6170d150d4f064e16c4bc413b59 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:58:56 +0000 Subject: [PATCH 07/13] =?UTF-8?q?docs(spike):=20record=20Axis=205=20?= =?UTF-8?q?=E2=80=94=20InvocationContext=20typed=20parameter=20injection?= =?UTF-8?q?=20(ITL-524)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Single InvocationContext shared by SinkPod and TapPod. Framework injects it when the user function declares a parameter typed InvocationContext — detection by type annotation only, no magic parameter names. Fields: invocation_hash, pod_name, pod_content_hash, pipeline_run_id. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 63 ++++++++++++------- 1 file changed, 39 insertions(+), 24 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index 0e190d41..df491d83 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -309,38 +309,53 @@ lookup path remains valid — the user resolves any ambiguity by inspecting mult candidate rows. Collision probability at 8 bytes (64 bits) per component is negligible for any realistic invocation volume. -### 5. Hash Surface — `SideEffectContext` Parameter Auto-Injection +### 5. Hash Surface — `InvocationContext` Parameter Injection -**Decision:** The framework auto-injects a `SideEffectContext` instance when the user -function declares a parameter typed `SideEffectContext`. +**Decision:** The framework injects an `InvocationContext` instance when the user +function declares a parameter typed `InvocationContext`. Detection is by type +annotation only — not by parameter name. No magic. ```python @dataclasses.dataclass(frozen=True) -class SideEffectContext: - invocation_signature: str # 32-char hex; stable across identical re-runs - execution_id: str # UUID string; unique per actual execution - pod_name: str # human-readable pod label - pod_content_hash: str # pod.content_hash().to_string() - pipeline_run_id: str | None # run-level context; None for lazy pipelines - - def format_id(self) -> str: - """Canonical artifact tag: 'orcapod-{invocation_signature}'.""" - return f"orcapod-{self.invocation_signature}" +class InvocationContext: + invocation_hash: str # serialized per pod's InvocationHashConfig + pod_name: str # human-readable pod label (pod.label) + pod_content_hash: str # pod.content_hash().to_string() — exact code version + pipeline_run_id: str | None # None for lazy/non-compiled pipelines ``` -Detection is by **type annotation** (`SideEffectContext`), not by parameter name. -If no `SideEffectContext` parameter is declared, the framework calls the function -without it (zero overhead for pods that don't need the hash). +`InvocationContext` is shared by both `SinkPod` and `TapPod`. The `invocation_hash` +field contains the serialized compound described in Axis 4 — two-part for `SinkPod`, +three-part (with verbatim `pipeline_run_id`) for `TapPod` when a run ID is available. +`pipeline_run_id` is also exposed as a first-class field so pod authors can store or +log it directly without parsing it back out of `invocation_hash`. + +If no `InvocationContext` parameter is present in the user function, the framework +calls the function without one. There is zero overhead — no context object is +constructed — for pods that don't need the hash. -**Why a context object, not a special parameter name?** -Type-based detection is explicit, IDE-discoverable, and carries no magic naming -convention. It mirrors how Python frameworks inject dependencies (FastAPI, pytest -fixtures with type resolution). It avoids the risk of a user accidentally declaring a -parameter named `ctx` for unrelated purposes. +**Usage:** + +```python +@sink_pod +def audit_write(data: PatientRecord, ctx: InvocationContext) -> None: + db.insert(ctx.invocation_hash, data.patient_id, ctx.pod_content_hash) + +@tap_pod +def log_row(data: PatientRecord, ctx: InvocationContext) -> None: + logger.info("[%s] Processing patient %s", ctx.invocation_hash, data.patient_id) + +@sink_pod +def simple_sink(data: PatientRecord) -> None: + # No ctx needed — valid, no overhead + external_api.push(data.patient_id) +``` -**Relationship to `InvocationIdentity` shared base:** -See Section 11 (Reconciliation with ITL-523). `SideEffectContext` embeds an -`InvocationIdentity` base — or is equivalent to one — enabling shared utilities. +**Why typed injection, not a magic parameter name:** +A parameter named `ctx` with any other type annotation (e.g. a user-defined context +object) must not be intercepted. Relying on name alone would make the mechanism fragile +and surprising. Type annotation detection is unambiguous — it fires if and only if the +declared type is exactly `InvocationContext`. ### 6. Failure Semantics — Author-Declared, Default Fail-Loud From 57906aaa58d00839e48c132b5faf321663ba0a82 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:20:22 +0000 Subject: [PATCH 08/13] docs(spike): record Axis 6 failure semantics decision (ITL-524) SinkPodConfig / TapPodConfig with on_error: Literal["raise","log"]="raise". SinkPod always records status="failed" on error regardless of on_error. Logs FunctionPod async/sync error inconsistency as DESIGN_ISSUES F15 and files ITL-527 for future alignment. Co-Authored-By: Claude Sonnet 4.6 --- DESIGN_ISSUES.md | 30 ++++++++++ ...6-07-14-itl-524-side-effect-pods-design.md | 57 +++++++++++++++---- 2 files changed, 75 insertions(+), 12 deletions(-) diff --git a/DESIGN_ISSUES.md b/DESIGN_ISSUES.md index 0ef2802f..9e8cd6ff 100644 --- a/DESIGN_ISSUES.md +++ b/DESIGN_ISSUES.md @@ -351,6 +351,36 @@ which column groups (meta, source, system_tags) are returned. --- +### F15 — `FunctionPod` has no pod-level error policy; sync and async paths are inconsistent +**Status:** open +**Severity:** high +**Issue:** ITL-527 + +`FunctionPod` has no `on_error` configuration. The two execution paths behave differently: + +- **Sync path** (`FunctionPodStream._iter_data_*`): no exception handling at all — exceptions + propagate unconditionally out of the iterator. +- **Async path** (`_FunctionPodBase.async_execute()`): exceptions are caught, the observer is + notified via `on_data_crash()`, and the item is **silently dropped from output** with no + indication to the caller and no way to configure this behaviour. + +The node-level `FunctionJobNode.execute()` adds `error_policy: Literal["continue", "fail_fast"]`, +but this is only available for DB-backed execution and uses different vocabulary from the rest +of the framework. + +`SinkPod` and `TapPod` (ITL-524/ITL-525) introduced a clean `on_error: Literal["raise", "log"]` +vocabulary at the pod level. `FunctionPod` should adopt the same model: pod-level `on_error` +config, consistent behaviour between sync and async paths, and the same `"raise"` / `"log"` +vocabulary. + +**Fix needed:** Add `on_error: Literal["raise", "log"] = "raise"` to `FunctionPodConfig` (or +equivalent). In the async path, replace the unconditional silent-drop with the configured +behaviour: `"raise"` propagates the exception; `"log"` logs at `WARNING` and drops the item +(current behaviour, but now explicit and configurable). Align `FunctionJobNode.error_policy` +to use the same vocabulary. + +--- + ## `src/orcapod/core/cached_function_pod.py` / `src/orcapod/core/data_function.py` ### CFP1 — Extract shared result caching logic from CachedDataFunction and CachedFunctionPod diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index df491d83..c454a1a5 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -357,21 +357,54 @@ object) must not be intercepted. Relying on name alone would make the mechanism and surprising. Type annotation detection is unambiguous — it fires if and only if the declared type is exactly `InvocationContext`. -### 6. Failure Semantics — Author-Declared, Default Fail-Loud +### 6. Failure Semantics -**Decision:** Controlled by `SideEffectConfig.on_error`: -- `"raise"` (default): exception propagates and the pipeline run fails. Consistent with - ITL-523's `HookConfig.on_error="raise"` default. -- `"log"`: exception is logged at `WARNING` level and swallowed; the input row is - passed through to downstream pods. Correct for non-critical side effects (metrics - emission) where a network hiccup should not abort the pipeline. +**Decision:** Each pod type has its own config class carrying `on_error`: -No retries in v1. Retry logic (with backoff, max attempts) is deferred. +```python +@dataclasses.dataclass(frozen=True) +class SinkPodConfig: + on_error: Literal["raise", "log"] = "raise" + hash_config: InvocationHashConfig = field(default_factory=InvocationHashConfig) + +@dataclasses.dataclass(frozen=True) +class TapPodConfig: + on_error: Literal["raise", "log"] = "raise" + hash_config: InvocationHashConfig = field(default_factory=InvocationHashConfig) +``` + +Separate config classes rather than a shared `SideEffectConfig`: `SinkPodConfig` will +gain `max_attempts` and `retry_on` from ITL-526 that are meaningless for `TapPod`. +Starting separate prevents a future config class with inapplicable fields. + +**`on_error` vocabulary** (consistent with ITL-523's `HookConfig`): +- `"raise"` (default): exception propagates; the pipeline row is aborted. +- `"log"`: exception logged at `WARNING` level; the input row continues downstream. -When `on_error="log"` and an exception is raised: -- A log row with `status="error"` and `error_message=str(exc)` is recorded in the - invocation log (so the failure is visible in the near-side provenance record). -- The input row is still emitted as output (pass-through continues). +No `"ignore"` option in v1. + +**`SinkPod` completion state on failure:** +Regardless of `on_error`, a failed delivery always records `status="failed"` in the +invocation log. `on_error` controls whether the exception propagates to the caller — +it does not affect completion tracking. A `"failed"` row remains eligible for +re-attempt on the next explicit pipeline re-run (Axis 3). + +| `on_error` | Pipeline behaviour | Completion state | +|---|---|---| +| `"raise"` (default) | Exception propagates; row aborted | `"failed"` | +| `"log"` | Exception logged; row continues downstream | `"failed"` | + +**`TapPod` on failure:** +No completion tracking. `on_error` controls pipeline propagation only: +- `"raise"`: exception propagates; row aborted; downstream sees nothing. +- `"log"`: exception logged at `WARNING`; input row emitted unchanged downstream. +An optional invocation log row with `status="error"` may be written for observability. + +**Relationship to `FunctionPod`:** +`FunctionPod` has no pod-level `on_error` config today. The sync path propagates +exceptions unconditionally; the async path silently drops failed items with no +configuration — an inconsistency. Aligning `FunctionPod` to adopt the same `on_error` +vocabulary is deferred; see DESIGN_ISSUES.md F15 and the corresponding Linear issue. ### 7. Ordering & DAG Placement From 055a7203aade94d4f47a068583f460f6fd58f9c9 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:45:18 +0000 Subject: [PATCH 09/13] docs(spike): refactor to unified SideEffectPod model (ITL-524) Replace SinkPod/TapPod as distinct types with a single SideEffectPod governed by track_completion and pass_on_success config axes. Update Axes 1-6 accordingly. @sink_pod/@tap_pod become convenience decorators. Hash formula now conditioned on track_completion, not type name. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 211 ++++++++---------- 1 file changed, 97 insertions(+), 114 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index c454a1a5..5f9136a2 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -111,102 +111,88 @@ core/ └── _FunctionPodBase (shared execution infrastructure) ├── FunctionPod (implements FunctionPodProtocol) └── SideEffectPod (implements SideEffectPodProtocol) -``` - -### 2. Pod Kinds and Output Contract - -**Decision:** There are two categorically distinct kinds of side-effect pod. They differ -enough in semantics, infrastructure requirements, and output contract to warrant -separate types rather than a single class with a flag. Whether they share a common -`SideEffectPod` umbrella type is deferred — it is not necessary to decide for the -initial implementation. - -#### `SinkPod` — completion-tracked, terminal delivery - -A `SinkPod` writes to an external source with a meaningful notion of success or failure -*per input*. The intent is exactly-once delivery: each (pod, input) pair should result -in exactly one successful write. - -**Output contract: terminal.** The pod ends the DAG branch. Delivering data outward is -the purpose; what happens next in the pipeline is a separate concern expressed as a -separate branch before this node. - -**Completion tracking.** The framework records completion state per (pod, input) in an -internal log: - -- **Not yet seen** → execute the delivery. -- **Previously succeeded** → skip silently on re-run. -- **Previously failed** → re-attempt on the next explicit pipeline re-run. - -Retry policy (automatic retries, max attempts, backoff) is **deferred to ITL-526**, -which will design a shared retry mechanism for both `FunctionPod` and `SinkPod`. For -now, re-attempt occurs only when the pipeline is explicitly re-run by the user. - -Completion state is internalized — tracked in the pod's invocation log, visible to -observers, not emitted as pipeline data. -**Canonical examples:** writing an audit row to a database, archiving a file to object -storage, posting a one-time notification. - -#### `TapPod` — always-fire, pass-through observation +Convenience decorators (both wrap SideEffectPod with preset config): +├── @sink_pod — track_completion=True, pass_on_success=True +└── @tap_pod — track_completion=False, pass_on_success=False +``` -A `TapPod` reacts to every data packet passing through it, unconditionally. There is no -notion of completion, no memory of past encounters, no skip or retry logic. +### 2. Pod Configuration Axes and Output Contract -**Output contract: pass-through.** The input tag and data flow through unchanged. The -pod is a transparent interceptor — a tap on the stream. +**Decision:** A single `SideEffectPod` type governed by two independent configuration +axes in `SideEffectPodConfig`. There are no separate `SinkPod` and `TapPod` types — +only convenience decorators that preset those axes. -**No completion tracking.** Every pipeline run, every input triggers the effect. This -is correct and expected: a logger should log every packet every time, not once. +**Axis A — Execution frequency** (`track_completion: bool = True`): +- `True`: the framework tracks per-(pod, input) completion state. Inputs that already + succeeded are skipped on re-run; failed inputs are re-attempted. +- `False`: no completion tracking. The pod fires unconditionally on every pipeline run. -**Canonical examples:** structured logging, metrics emission, real-time monitoring -feeds, debug tracing. +**Axis B — Pass-through filtering** (`pass_on_success: bool = True`): +- `True`: only rows where the delivery succeeded are emitted downstream. Failed rows + are dropped. +- `False`: all rows are emitted downstream regardless of delivery outcome. -#### Comparison +The four combinations and their canonical use cases: -| | `SinkPod` | `TapPod` | +| `track_completion` | `pass_on_success` | Canonical use case | |---|---|---| -| Output contract | Terminal | Pass-through | -| Completion tracking | Yes — success / failed / not-seen | No | -| Re-run behavior | Skip on success; retry on re-run if failed | Always fires | -| Invocation log | Required (completion state) | Optional (observability only) | -| Retry policy | Deferred to ITL-526 | N/A | - -**Rejected alternative — single class with `idempotent: bool`:** The semantic -difference between "always fire" and "fire once and track completion" is substantial -enough that a flag would obscure rather than clarify intent. A pod author should not -need to reason about idempotency to write a logger. +| `True` | `True` | Audit DB write, exactly-once delivery that gates downstream | +| `True` | `False` | Deliver once; don't gate downstream on delivery success | +| `False` | `True` | Re-notify every run; downstream only sees acknowledged rows | +| `False` | `False` | Structured logging, metrics, trace everything | + +**Output contract: pass-through** (all configurations). `SideEffectPod` always returns +a stream. When `pass_on_success=False`, all rows are emitted; when `pass_on_success=True`, +only successfully-delivered rows appear downstream. + +**Convenience decorators:** +`@sink_pod` presets `track_completion=True, pass_on_success=True` — the most common +"deliver and gate" pattern. +`@tap_pod` presets `track_completion=False, pass_on_success=False` — the most common +"observe everything" pattern. +Both decorators wrap the same underlying `SideEffectPod` class. All four combinations +remain expressible via `SideEffectPodConfig` directly. + +**Rejected alternative — two distinct types (`SinkPod` / `TapPod`):** +The two axes are genuinely orthogonal; all four combinations have legitimate use cases. +Encoding just two of them as named types leaves the other two without a natural +expression and forces users to reason about type identity rather than their actual intent. +A single configurable type with convenience shortcuts is cleaner and more extensible. ### 3. Re-Execution and Completion Semantics -**`TapPod`:** No caching or completion tracking. Every pipeline run, every input packet -triggers the effect unconditionally. There is no state to consult and no state to record -beyond optional observability logging. +**When `track_completion=False`:** No completion state is maintained. Every pipeline +run, every input packet triggers the delivery unconditionally. -**`SinkPod`:** Completion-tracked per (pod topology, input packet). The framework -maintains a per-pod invocation table (see Axis 8) and consults it at the start of each -execution: +**When `track_completion=True`:** The framework maintains a per-pod invocation table +(see Axis 8) and consults it at the start of each execution: | Prior state | Action | |---|---| | Not yet seen | Execute the delivery | -| Previously succeeded | Skip silently | +| Previously succeeded | Skip delivery; emit row downstream without re-delivery | | Previously failed | Re-attempt | -Re-attempt occurs when the pipeline is **explicitly re-run by the user**. There is no +Emitting previously-succeeded rows on re-run (without re-delivery) ensures downstream +pods receive the full dataset when the pipeline is re-run to catch up on failed rows — +not just the newly-caught rows. + +Re-attempt occurs only when the pipeline is **explicitly re-run by the user**. No automatic background retry in v1. Full retry policy (max attempts, backoff, retryable exception types) is deferred to ITL-526, which will design a shared mechanism for both -`FunctionPod` and `SinkPod`. +`FunctionPod` and `SideEffectPod`. -`SinkPod` does **not** participate in the `ResultCache` / `FunctionNode` result table -system. Its execution state is tracked exclusively in its own invocation table. +`SideEffectPod` with `track_completion=True` does **not** participate in the +`ResultCache` / `FunctionNode` result table system. Completion state is tracked +exclusively in its own invocation table. ### 4. Invocation Hash #### Raw identity components -The raw identity of any invocation is a **three-part compound** (two parts for -`SinkPod`, three for `TapPod`): +The raw identity of any invocation is a **compound hash** whose number of components +depends on `track_completion`: 1. **`pipeline_hash(pod)`** — the pipeline hash of the pod, encoding both the pod's own function identity and the full upstream topology. This is the same `pipeline_hash()` @@ -220,26 +206,28 @@ The raw identity of any invocation is a **three-part compound** (two parts for covers data columns only. Including tags and source/system columns means two packets with identical payload but different originating provenance — different sources, different upstream routes — produce different invocation hashes. This is correct: a - sink that writes an audit row needs to distinguish rows by their full provenance, not + pod that writes an audit row needs to distinguish rows by their full provenance, not just their data values. -3. **`pipeline_run_id`** (`TapPod` only, when available) — a verbatim string - identifying the specific pipeline execution (e.g. `"run-2026-07-14-001"`). This is - user-controlled and included as-is without further hashing. When no run ID is - available (lazy/in-memory pipeline), the third component is omitted. +3. **`pipeline_run_id`** (only when `track_completion=False`, when available) — a + verbatim string identifying the specific pipeline execution (e.g. + `"run-2026-07-14-001"`). This is user-controlled and included as-is without further + hashing. When no run ID is available (lazy/in-memory pipeline), the third component + is omitted. **Why `pipeline_hash` rather than `content_hash` for component 1:** `content_hash(pod)` captures only the pod's function identity. `pipeline_hash(pod)` captures function identity *plus* the full upstream topology, forming a Merkle chain. -Using `pipeline_hash` ensures that the same sink function used in two pipelines with +Using `pipeline_hash` ensures that the same pod function used in two pipelines with different upstream structures never shares a completion namespace — a delivery confirmed in pipeline A does not suppress delivery in pipeline B. -**Why `TapPod` includes the run ID:** -`SinkPod`'s hash must be run-agnostic so the framework can detect "already delivered" -across re-runs. `TapPod` fires every run intentionally and has no deduplication to -preserve — including the run ID makes every firing distinguishable, which is exactly -right for traceable observation. +**Why `track_completion=False` includes the run ID:** +When `track_completion=True`, the hash must be run-agnostic so the framework can +detect "already delivered" across re-runs — including the run ID would make the hash +unique per run and break deduplication. When `track_completion=False`, the pod fires +every run intentionally and has no deduplication to preserve — including the run ID +makes every firing distinguishable, which is exactly right for traceable observation. #### Internal record format @@ -247,11 +235,11 @@ Internally, the full compound is stored with `::` as the component separator (us `::` avoids ambiguity with the `:` in `method:digest` form of `ContentHash.to_string()`): ``` -SinkPod: {pipeline_hash.to_string()}::{full_input_packet_hash.to_string()} -TapPod: {pipeline_hash.to_string()}::{full_input_packet_hash.to_string()}::{run_id} +track_completion=True: {pipeline_hash.to_string()}::{full_input_packet_hash.to_string()} +track_completion=False: {pipeline_hash.to_string()}::{full_input_packet_hash.to_string()}::{run_id} ``` -Example (SinkPod): +Example (track_completion=True): ``` blake3_v1:a3f8c2d1e5b7f0c9...::blake3_v1:b7e491f0a2c3d8e1... ``` @@ -324,11 +312,12 @@ class InvocationContext: pipeline_run_id: str | None # None for lazy/non-compiled pipelines ``` -`InvocationContext` is shared by both `SinkPod` and `TapPod`. The `invocation_hash` -field contains the serialized compound described in Axis 4 — two-part for `SinkPod`, -three-part (with verbatim `pipeline_run_id`) for `TapPod` when a run ID is available. -`pipeline_run_id` is also exposed as a first-class field so pod authors can store or -log it directly without parsing it back out of `invocation_hash`. +`InvocationContext` is shared across all `SideEffectPod` configurations. The +`invocation_hash` field contains the serialized compound described in Axis 4 — +two-part when `track_completion=True`, three-part (with verbatim `pipeline_run_id`) +when `track_completion=False` and a run ID is available. `pipeline_run_id` is also +exposed as a first-class field so pod authors can store or log it directly without +parsing it back out of `invocation_hash`. If no `InvocationContext` parameter is present in the user function, the framework calls the function without one. There is zero overhead — no context object is @@ -359,52 +348,46 @@ declared type is exactly `InvocationContext`. ### 6. Failure Semantics -**Decision:** Each pod type has its own config class carrying `on_error`: +**Decision:** A single unified `SideEffectPodConfig` carries all configuration axes: ```python @dataclasses.dataclass(frozen=True) -class SinkPodConfig: - on_error: Literal["raise", "log"] = "raise" - hash_config: InvocationHashConfig = field(default_factory=InvocationHashConfig) - -@dataclasses.dataclass(frozen=True) -class TapPodConfig: +class SideEffectPodConfig: + track_completion: bool = True + pass_on_success: bool = True on_error: Literal["raise", "log"] = "raise" hash_config: InvocationHashConfig = field(default_factory=InvocationHashConfig) ``` -Separate config classes rather than a shared `SideEffectConfig`: `SinkPodConfig` will -gain `max_attempts` and `retry_on` from ITL-526 that are meaningless for `TapPod`. -Starting separate prevents a future config class with inapplicable fields. +When ITL-526 adds retry fields (`max_attempts`, `retry_on`), they are added to +`SideEffectPodConfig` — they are only meaningful when `track_completion=True`, and +that relationship is enforced at construction or with documentation. **`on_error` vocabulary** (consistent with ITL-523's `HookConfig`): - `"raise"` (default): exception propagates; the pipeline row is aborted. -- `"log"`: exception logged at `WARNING` level; the input row continues downstream. +- `"log"`: exception logged at `WARNING` level; delivery outcome determines whether + the row passes downstream (see interaction with `pass_on_success` below). No `"ignore"` option in v1. -**`SinkPod` completion state on failure:** -Regardless of `on_error`, a failed delivery always records `status="failed"` in the -invocation log. `on_error` controls whether the exception propagates to the caller — -it does not affect completion tracking. A `"failed"` row remains eligible for -re-attempt on the next explicit pipeline re-run (Axis 3). +**Interaction between `on_error` and `pass_on_success`:** -| `on_error` | Pipeline behaviour | Completion state | +| `on_error` | `pass_on_success` | Delivery fails → | |---|---|---| -| `"raise"` (default) | Exception propagates; row aborted | `"failed"` | -| `"log"` | Exception logged; row continues downstream | `"failed"` | +| `"raise"` | either | Exception propagates; row aborted | +| `"log"` | `True` | Exception logged; row **dropped** (only successes pass) | +| `"log"` | `False` | Exception logged; row **emitted** downstream regardless | -**`TapPod` on failure:** -No completion tracking. `on_error` controls pipeline propagation only: -- `"raise"`: exception propagates; row aborted; downstream sees nothing. -- `"log"`: exception logged at `WARNING`; input row emitted unchanged downstream. -An optional invocation log row with `status="error"` may be written for observability. +**Completion state on failure (when `track_completion=True`):** +Regardless of `on_error`, a failed delivery always records `status="failed"` in the +invocation log. `on_error` controls pipeline propagation, not the completion record. +A `"failed"` row remains eligible for re-attempt on the next explicit re-run (Axis 3). **Relationship to `FunctionPod`:** `FunctionPod` has no pod-level `on_error` config today. The sync path propagates exceptions unconditionally; the async path silently drops failed items with no configuration — an inconsistency. Aligning `FunctionPod` to adopt the same `on_error` -vocabulary is deferred; see DESIGN_ISSUES.md F15 and the corresponding Linear issue. +vocabulary is deferred; see DESIGN_ISSUES.md F15 and ITL-527. ### 7. Ordering & DAG Placement From caaf666f5a83356ec231bbd97c33af88913462db Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 06:17:52 +0000 Subject: [PATCH 10/13] =?UTF-8?q?docs(spike):=20rename=20pass=5Fon=5Fsucce?= =?UTF-8?q?ss=E2=86=92drop=5Fon=5Ffailure;=20lock=20Axis=207=20(ITL-524)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rename config flag to drop_on_failure throughout spec. Record Axis 7 decisions: synchronous per-row delivery in v1, max_concurrency model, no global barrier. File ITL-528 for future fire-and-forget delivery mode. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 78 +++++++++++-------- 1 file changed, 45 insertions(+), 33 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index 5f9136a2..f93f3ba5 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -113,8 +113,8 @@ core/ └── SideEffectPod (implements SideEffectPodProtocol) Convenience decorators (both wrap SideEffectPod with preset config): -├── @sink_pod — track_completion=True, pass_on_success=True -└── @tap_pod — track_completion=False, pass_on_success=False +├── @sink_pod — track_completion=True, drop_on_failure=True +└── @tap_pod — track_completion=False, drop_on_failure=False ``` ### 2. Pod Configuration Axes and Output Contract @@ -128,14 +128,14 @@ only convenience decorators that preset those axes. succeeded are skipped on re-run; failed inputs are re-attempted. - `False`: no completion tracking. The pod fires unconditionally on every pipeline run. -**Axis B — Pass-through filtering** (`pass_on_success: bool = True`): -- `True`: only rows where the delivery succeeded are emitted downstream. Failed rows - are dropped. -- `False`: all rows are emitted downstream regardless of delivery outcome. +**Axis B — Drop-on-failure control** (`drop_on_failure: bool = True`): +- `True`: rows where delivery failed are dropped; only successfully-delivered rows flow + downstream. +- `False`: all rows flow downstream regardless of delivery outcome. The four combinations and their canonical use cases: -| `track_completion` | `pass_on_success` | Canonical use case | +| `track_completion` | `drop_on_failure` | Canonical use case | |---|---|---| | `True` | `True` | Audit DB write, exactly-once delivery that gates downstream | | `True` | `False` | Deliver once; don't gate downstream on delivery success | @@ -143,13 +143,14 @@ The four combinations and their canonical use cases: | `False` | `False` | Structured logging, metrics, trace everything | **Output contract: pass-through** (all configurations). `SideEffectPod` always returns -a stream. When `pass_on_success=False`, all rows are emitted; when `pass_on_success=True`, -only successfully-delivered rows appear downstream. +a stream. When `drop_on_failure=True`, failed rows are dropped and only +successfully-delivered rows appear downstream. When `drop_on_failure=False`, all rows +flow through regardless of delivery outcome. **Convenience decorators:** -`@sink_pod` presets `track_completion=True, pass_on_success=True` — the most common +`@sink_pod` presets `track_completion=True, drop_on_failure=True` — the most common "deliver and gate" pattern. -`@tap_pod` presets `track_completion=False, pass_on_success=False` — the most common +`@tap_pod` presets `track_completion=False, drop_on_failure=False` — the most common "observe everything" pattern. Both decorators wrap the same underlying `SideEffectPod` class. All four combinations remain expressible via `SideEffectPodConfig` directly. @@ -354,7 +355,7 @@ declared type is exactly `InvocationContext`. @dataclasses.dataclass(frozen=True) class SideEffectPodConfig: track_completion: bool = True - pass_on_success: bool = True + drop_on_failure: bool = True on_error: Literal["raise", "log"] = "raise" hash_config: InvocationHashConfig = field(default_factory=InvocationHashConfig) ``` @@ -366,16 +367,16 @@ that relationship is enforced at construction or with documentation. **`on_error` vocabulary** (consistent with ITL-523's `HookConfig`): - `"raise"` (default): exception propagates; the pipeline row is aborted. - `"log"`: exception logged at `WARNING` level; delivery outcome determines whether - the row passes downstream (see interaction with `pass_on_success` below). + the row passes downstream (see interaction with `drop_on_failure` below). No `"ignore"` option in v1. -**Interaction between `on_error` and `pass_on_success`:** +**Interaction between `on_error` and `drop_on_failure`:** -| `on_error` | `pass_on_success` | Delivery fails → | +| `on_error` | `drop_on_failure` | Delivery fails → | |---|---|---| | `"raise"` | either | Exception propagates; row aborted | -| `"log"` | `True` | Exception logged; row **dropped** (only successes pass) | +| `"log"` | `True` | Exception logged; failed row **dropped** | | `"log"` | `False` | Exception logged; row **emitted** downstream regardless | **Completion state on failure (when `track_completion=True`):** @@ -391,23 +392,34 @@ vocabulary is deferred; see DESIGN_ISSUES.md F15 and ITL-527. ### 7. Ordering & DAG Placement -Side-effect pods are **synchronous barriers** in the DAG: - -1. Consume input row. -2. Execute side effect (call user function with `(data, ctx)` or `(data,)` as applicable). -3. On success: emit the same row as output. Record `status="success"` in invocation log. -4. On error with `on_error="raise"`: propagate exception; do not emit output. -5. On error with `on_error="log"`: log + record; emit input row as output anyway. - -For the **async channel execution path**, `async_execute()` must process each input -row, complete the side effect, and write to the output channel before returning. -Concurrent per-row execution within a single side-effect pod is permitted when -`PodConfig.max_concurrency > 1` and the pod is so configured — subject to the same -concurrency model as `FunctionPod`. - -**No fire-and-forget in v1.** All side effects complete (or fail) synchronously with -respect to the pipeline's progress on that row. Downstream pods cannot begin processing -a row until the side effect for that row has finished. +`SideEffectPod` is a per-data-packet operation — structurally identical to `FunctionPod` +in its relationship to the DAG. It can appear anywhere in the pipeline (beginning, middle, +or end), chains naturally with other pods, and follows the same concurrency model. + +**Row-by-row execution flow:** + +1. Receive `(tag, data)` from upstream. +2. If `track_completion=True`: check invocation log — skip delivery for previously-succeeded inputs (emit without re-delivery); re-attempt for previously-failed inputs. +3. Call user function (with `InvocationContext` injection if declared). +4. On success: write `status="success"` to invocation log; emit row downstream. +5. On error with `on_error="raise"`: propagate exception; row aborted. +6. On error with `on_error="log"` + `drop_on_failure=True`: log + write `status="failed"`; drop row. +7. On error with `on_error="log"` + `drop_on_failure=False`: log + write `status="failed"`; emit row downstream. + +**No fire-and-forget in v1.** Delivery completes (or fails) before the row is emitted +or dropped. This applies to all configurations including `drop_on_failure=False`. +`drop_on_failure=False` is the natural candidate for fire-and-forget in the future +(since downstream is not gated on delivery outcome), but that optimisation is deferred +to ITL-528. See also ITL-526 for retry policy. + +**Concurrency across rows:** `SideEffectPod` respects `max_concurrency` exactly as +`FunctionPod` does — multiple rows can be processed concurrently when `max_concurrency > 1`. +Invocation log writes are per-row and atomic; `track_completion=True` completion checks +use the same row-level idempotency guarantees as `FunctionNode`. + +**No global barrier.** `SideEffectPod` does not wait for all rows to complete before +emitting any. It processes and emits row-by-row (or in concurrent batches), consistent +with the streaming model. ### 8. Observability — `_orcapod_side_effect_invocations` Table From f305b1522004fd34096940d17f7c4d3a7e37438f Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 06:22:13 +0000 Subject: [PATCH 11/13] docs(spike): lock Axes 8-9 with per-pipeline-hash invocation tables (ITL-524) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Invocation tables scoped by pipeline_hash(pod), mirroring FunctionNode. Reverse lookup: parse invocation hash → route to table → look up by full_input_packet_hash. Update column names, status values, lookup steps. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 125 +++++++++++------- 1 file changed, 76 insertions(+), 49 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index f93f3ba5..68723cb2 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -421,82 +421,109 @@ use the same row-level idempotency guarantees as `FunctionNode`. emitting any. It processes and emits row-by-row (or in concurrent batches), consistent with the streaming model. -### 8. Observability — `_orcapod_side_effect_invocations` Table +### 8. Observability — Per-Pipeline-Hash Invocation Tables -**Decision:** Orcapod persists every side-effect invocation to a dedicated table in the -pipeline database. +**Decision:** Each `SideEffectPod`'s invocation log is stored in a table scoped by +`pipeline_hash(pod)`, mirroring how `FunctionNode` organises its result tables. + +**Table path:** +``` +//side_effect_invocations +``` + +Created on first invocation if it does not exist. + +**Schema:** | Column | Type | Notes | -|--------|------|-------| -| `execution_id` | TEXT | Primary key — unique per actual execution | -| `invocation_signature` | TEXT | Indexed — stable per (pod, inputs) pair | -| `pod_name` | TEXT | Human-readable pod label | +|---|---|---| +| `full_input_packet_hash` | TEXT | Primary lookup key — second component of the invocation hash | +| `invocation_hash` | TEXT | Full serialized hash per pod's `InvocationHashConfig` | | `pod_content_hash` | TEXT | `pod.content_hash().to_string()` — exact code version | -| `input_hash` | TEXT | `input_data.content_hash().to_string()` | -| `pipeline_run_id` | TEXT NULLABLE | `None` for lazy/non-compiled pipelines | -| `executed_at` | TIMESTAMP | UTC wall-clock time of execution | -| `status` | TEXT | `"success"` / `"error"` / `"skipped"` | -| `error_message` | TEXT NULLABLE | Set when `status = "error"` | +| `pipeline_run_id` | TEXT NULL | User-supplied run ID; `None` for lazy pipelines | +| `executed_at` | TIMESTAMP | UTC wall-clock time | +| `status` | TEXT | `"success"` / `"failed"` / `"skipped"` | +| `error_message` | TEXT NULL | Set when `status="failed"` | -**Schema location:** A shared table in the pipeline database (not scoped per-node), -allowing cross-pipeline lookup by `invocation_signature` alone. Table is created on -first invocation if it does not exist. +**`status` values:** +- `"success"`: delivery completed without exception. +- `"failed"`: delivery raised an exception (regardless of `on_error` setting). +- `"skipped"`: `track_completion=True` and this input was previously succeeded; delivery skipped, row re-emitted without re-delivery. -This table is the **near side of the reverse-lookup chain**. Without it, a signature -extracted from an external artifact has nothing to resolve against. +**When the log is written:** +- `track_completion=True`: always written. The framework reads this table at the start of each execution to determine skip/retry behaviour. +- `track_completion=False`: written by default. No completion state is required, but the log enables observability (what fired, when, with what outcome). A future opt-out may be added if log volume becomes a concern. -**For lazy/ephemeral pipelines** (no `PipelineJob`, no DB-backed nodes): the invocation -log still records the invocation, but `pipeline_run_id` is `None` and `input_hash` can -be used only to identify the data structure — the actual data rows are not persisted -and cannot be retrieved from the framework. +**This table is the near side of the reverse-lookup chain.** Without it, an invocation hash extracted from an external artifact has nothing to resolve against inside Orcapod. + +**For lazy/ephemeral pipelines** (no DB-backed nodes): the invocation log is still written if a DB is configured, but `pipeline_run_id` is `None` and the upstream data rows themselves are not persisted — only the hash identifies the data structure. ### 9. Reverse Lookup Path -**Given:** `invocation_signature = "d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f"` found in an -external artifact (e.g., a Postgres audit row or a log line). +**Given:** an invocation hash embedded in an external artifact — e.g., +`blake3_v1:a3f8c2d1e5b7f0c9::blake3_v1:b7e491f0a2c3d8e1` found in a Postgres audit row, +a log line, or a Slack message body. -**Step 1 — Signature → invocation record:** +The hash is self-routing: the `::` separator partitions it into its components, and the +first component is always the `pipeline_hash(pod)` value identifying which invocation +table to open. + +**Step 1 — Parse the invocation hash:** +Split on `::` to extract components: +- Component 1: `pipeline_hash` value (routes to the invocation table). +- Component 2: `full_input_packet_hash` (row lookup key within the table). +- Component 3 (if present): `pipeline_run_id` (for `track_completion=False` pods only). + +**Step 2 — Locate the invocation table:** +``` +/{pipeline_hash_component}/side_effect_invocations +``` +If `component_length` truncation was used, prefix-match on the pipeline hash component +to find the matching table path. + +**Step 3 — Look up the invocation record:** ```sql SELECT * -FROM _orcapod_side_effect_invocations -WHERE invocation_signature = 'd4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f' +FROM side_effect_invocations +WHERE full_input_packet_hash = '{input_packet_hash_component}' ORDER BY executed_at ``` -Returns: `{pod_name, pod_content_hash, input_hash, pipeline_run_id, executed_at, status}`. +Returns: `{invocation_hash, pod_content_hash, pipeline_run_id, executed_at, status, error_message}`. -Multiple rows may exist if `idempotent=False` and the pod ran multiple times with the -same inputs (each run has a distinct `execution_id`). +Multiple rows exist for `track_completion=False` pods (the pod fired on multiple runs +with the same input; each firing has a distinct `executed_at`). -**Step 2 — Input hash → input data:** -Query the pipeline's function node result tables for rows where -`_input_data_hash = input_hash`. This locates the exact `(Tag, Data)` packet that +**Step 4 — Input packet hash → upstream data:** +`full_input_packet_hash` covers the full input packet (tag, data, source columns, +system tag columns). Query the upstream `FunctionNode` result tables to find the +`(Tag, Data)` row whose content hash matches. This locates the exact input packet that triggered the side effect. -*(Requires a DB-backed pipeline — `FunctionJobNode` or `OperatorJobNode`. For lazy -pipelines, the input hash identifies the data structure but rows cannot be retrieved.)* +*(Requires a DB-backed pipeline. For lazy pipelines, the hash identifies the data +structure but the rows are not persisted and cannot be retrieved.)* -**Step 3 — Input data → upstream lineage:** -Follow `_tag_source_id` and `_tag_record_id` system tag columns on the retrieved input -row. These columns encode the full upstream provenance chain, back to the original -source and row number. +**Step 5 — Input data → upstream lineage:** +Follow system tag columns on the retrieved input row back through upstream nodes to +the original source rows. -**Step 4 — Pod code version:** -`pod_content_hash` identifies the exact pod function that ran. If the codebase is -version-controlled, this hash can be resolved to a specific commit and function definition. +**Step 6 — Pod code version:** +`pod_content_hash` identifies the exact function that ran. Resolve against the +codebase's version history to recover the exact function definition at that version. **Full reverse-lookup chain:** ``` -external artifact (orcapod-d4a8f3b2...) - → _orcapod_side_effect_invocations[invocation_signature] - → input_hash → function node result table → (Tag, Data) packet - → _tag_source_id / _tag_record_id → upstream nodes → root source rows - + pod_content_hash → exact pod code version +external artifact (contains invocation_hash) + → parse → pipeline_hash component + → /{pipeline_hash}/side_effect_invocations[full_input_packet_hash] + → upstream FunctionNode result tables → (Tag, Data) packet + → system tag columns → upstream nodes → root source rows + + pod_content_hash → exact pod code version ``` **Minimum persistence requirements:** -- `_orcapod_side_effect_invocations` table must be durable (not ephemeral). -- Input data rows must be persisted in DB-backed nodes for step 2 to work. -- For full lineage, all intermediate nodes must use DB-backed execution. +- The pod's per-pipeline-hash invocation table must be durable. +- Input data rows must be persisted in DB-backed upstream nodes for step 4 to work. +- For full lineage to source rows, all intermediate nodes must use DB-backed execution. ### 10. Serialization — Prescribe Canonical Format, Provide Helpers From c5a26a0eab5f0ccdf1765bec2b9040f312587b67 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 06:27:28 +0000 Subject: [PATCH 12/13] docs(spike): lock Axis 10 serialization conventions (ITL-524) format_id() accepts optional InvocationHashConfig override for per-call encoding/length control. Documents canonical orcapod- prefix convention, raw hash usage, and dropped InvocationIdentity base type. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 89 +++++++++++++------ 1 file changed, 61 insertions(+), 28 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index 68723cb2..d0a2cb1f 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -525,41 +525,74 @@ external artifact (contains invocation_hash) - Input data rows must be persisted in DB-backed upstream nodes for step 4 to work. - For full lineage to source rows, all intermediate nodes must use DB-backed execution. -### 10. Serialization — Prescribe Canonical Format, Provide Helpers +### 10. Serialization Conventions -**Decision:** Define a canonical artifact tag format and expose a `format_id()` helper. -Deviation is allowed but not recommended. +**Decision:** Prescribe a canonical `orcapod-{hash}` artifact tag format. Expose +`ctx.format_id()` on `InvocationContext` to produce it, accepting an optional +`InvocationHashConfig` override so each call site can independently control encoding +and component length. -**Canonical format:** `orcapod-{invocation_signature}` where `invocation_signature` is -32 lowercase hex characters (128 bits). +**`InvocationContext.format_id()`:** -**Examples:** - -Log line (structured logging): -``` -{"level": "info", "msg": "Patient processed", "orcapod_id": "orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f", "patient_id": "P-12345"} -``` - -Database column: -```sql -INSERT INTO audit_log (orcapod_record_id, patient_id, processed_at) -VALUES ('d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f', 'P-12345', NOW()) +```python +def format_id(self, config: InvocationHashConfig | None = None) -> str: + """Return 'orcapod-{hash}' with optional format override. + + Args: + config: Optional ``InvocationHashConfig`` controlling ``encoding`` + (``"hex"``, ``"base64"``, ``"binary"``) and ``component_length`` + (bytes of raw digest per component; ``None`` = full length). + When ``None``, the pod's own ``InvocationHashConfig`` is used, + producing the same serialization as ``ctx.invocation_hash``. + + Returns: + ``'orcapod-{serialized_hash}'`` ready for embedding in log fields, + DB columns, message bodies, or any plain-text artifact. + """ ``` -*(DB column stores the raw signature; the `orcapod-` prefix is for human-facing contexts.)* -Slack notification body: -``` -Pipeline run complete for patient P-12345. -Trace: orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f -``` +`InvocationContext` stores the raw hash components internally and re-serializes on +demand, so `format_id(config=...)` works without the user needing to re-construct +anything. -`SideEffectContext` provides: -- `ctx.format_id()` → `"orcapod-d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f"` (full canonical tag) -- `ctx.invocation_signature` → `"d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f"` (raw hex for DB columns) +**Usage examples:** -Pod authors who deviate from the canonical format do so at their own risk. The reverse -lookup machinery works as long as `invocation_signature` appears in the invocation log — -but human discoverability and tooling support will be reduced for non-canonical embeddings. +```python +@side_effect_pod +def audit_write(data: PatientRecord, ctx: InvocationContext) -> None: + # Pod's own InvocationHashConfig (full hex by default): + db.insert(ctx.format_id(), data.patient_id) + # → "orcapod-blake3_v1:a3f8c2d1e5b7f0c9...::blake3_v1:b7e491f0a2c3d8e1..." + + # 8-byte truncated hex — compact enough for a Slack message: + short = ctx.format_id(InvocationHashConfig(encoding="hex", component_length=8)) + slack.send(f"Patient {data.patient_id} processed. Trace: {short}") + # → "orcapod-a3f8c2d1e5b7f0c9::b7e491f0a2c3d8e1" + + # base64 — more compact for the same bit count: + b64 = ctx.format_id(InvocationHashConfig(encoding="base64", component_length=16)) + logger.info("invocation_id=%s", b64) + # → "orcapod-o/jC0eW38fCp::t+SR8KLDOOE=" + + # Raw hash without prefix — for a DB column where the field name implies origin: + db.insert_raw_hash(ctx.invocation_hash, data.patient_id) +``` + +**Canonical format:** `orcapod-{invocation_hash}` using the pod's configured +`InvocationHashConfig`. The `orcapod-` prefix makes artifacts machine-discoverable with +a single grep pattern across all log files, DB dumps, and message bodies — the prefix +unambiguously identifies the origin as an Orcapod pipeline provenance token. + +**Using `ctx.invocation_hash` directly (no prefix):** +Valid for DB columns where the field name already implies the origin (e.g., +`orcapod_record_id`). The reverse-lookup machinery (Axis 9) requires only the raw +hash string — the `orcapod-` prefix is for human discoverability, not lookup routing. + +**No shared `InvocationIdentity` base type:** +An earlier design draft proposed a shared base type for post-run hooks and side-effect +pods. With `format_id()` living directly on `InvocationContext`, no shared base is +needed. Post-run hooks (ITL-523) may adopt the `orcapod-` prefix convention +independently if desired. --- From c31bc2faa1e6f334c54440cf85fc60dcddb81845 Mon Sep 17 00:00:00 2001 From: "agent-kurodo[bot]" <268466204+agent-kurodo[bot]@users.noreply.github.com> Date: Tue, 14 Jul 2026 06:31:23 +0000 Subject: [PATCH 13/13] docs(spike): rewrite trailing sections with unified SideEffectPod API (ITL-524) Update Reconciliation, Worked Example, Public API Summary, Tests, Scope, Dependencies, and task breakdown. Replace all SideEffectContext/SideEffectConfig/ InvocationIdentity/idempotent references with InvocationContext/SideEffectPodConfig/ track_completion/drop_on_failure. Expand tests to T18. Co-Authored-By: Claude Sonnet 4.6 --- ...6-07-14-itl-524-side-effect-pods-design.md | 368 ++++++++++-------- 1 file changed, 198 insertions(+), 170 deletions(-) diff --git a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md index d0a2cb1f..2754e35a 100644 --- a/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md +++ b/superpowers/specs/2026-07-14-itl-524-side-effect-pods-design.md @@ -602,12 +602,13 @@ independently if desired. |-----------|------------------------|--------------------------| | DAG presence | No — passive observer | Yes — first-class node | | Triggers on | After a function pod runs | IS the pipeline step | -| Gates downstream | No | Yes | +| Gates downstream | No | Yes (when `drop_on_failure=True`) | | Composable | No | Yes | -| Hash name | `record_id_hash` | `invocation_signature` | -| Hash formula | `str(output_data.datagram_uuid)` (UUIDv7 per execution) | `content_hash(pod ∥ inputs).to_hex()` (deterministic) | -| Hash stability | Stable for cache hits; unique for fresh computations | Always deterministic | +| Hash name | `record_id_hash` | `invocation_hash` | +| Hash formula | `str(output_data.datagram_uuid)` (UUIDv7 per execution) | `pipeline_hash :: full_input_packet_hash [:: run_id]` (deterministic) | +| Hash stability | Stable for cache hits; unique for fresh computations | Always deterministic (when `track_completion=True`) | | `on_error` vocabulary | `"raise"` / `"log"` | Same: `"raise"` / `"log"` | +| Context type | `PodContext` in `PostRunPayload` | `InvocationContext` | **Hash formula divergence — justified:** @@ -617,61 +618,15 @@ This is appropriate for function pods because the "result record ID" IS the outp datagram identity. Side-effect pods have no output datagram, so the equivalent must be derived from -inputs. Using `content_hash(pod ∥ inputs)` is more fundamental: it is deterministic -regardless of caching, enabling idempotency checks across re-runs. For a deterministic -function pod, the two formulas converge: `content_hash(pod ∥ inputs)` ≡ -`output_data.datagram_uuid` whenever the output is content-addressed from the same -inputs. +inputs. Using `pipeline_hash :: full_input_packet_hash` is deterministic regardless of +caching, enabling completion-tracking across re-runs. For a deterministic function pod, +the two formulas converge whenever the output is content-addressed from the same inputs. -**Shared `InvocationIdentity` base type:** - -To avoid duplicating the canonical `format_id()` logic and enable shared tooling -(e.g., a logging helper that works for both hook payloads and side-effect contexts), -extract a common type from `src/orcapod/hooks.py`: - -```python -@dataclasses.dataclass(frozen=True) -class InvocationIdentity: - """Shared identity carrier for post-run hooks and side-effect pods. - - Attributes: - record_id: The invocation identifier. For post-run hooks this is - ``str(output_data.datagram_uuid)``; for side-effect pods this is - ``content_hash(pod || inputs).to_hex()``. Both are embeddable - plain-text strings. - pod_name: Human-readable pod label. - pipeline_run_id: Run context; ``None`` for lazy pipelines. - """ - record_id: str - pod_name: str - pipeline_run_id: str | None - - def format_id(self) -> str: - """Canonical artifact tag: 'orcapod-{record_id}'.""" - return f"orcapod-{self.record_id}" -``` - -`SideEffectContext` may directly embed the `InvocationIdentity` fields or inherit from it -(implementation decision); it adds `execution_id` and `pod_content_hash`: - -```python -@dataclasses.dataclass(frozen=True) -class SideEffectContext: - # InvocationIdentity fields: - invocation_signature: str # = record_id for side-effect pods - pod_name: str - pipeline_run_id: str | None - - # SideEffectPod-specific fields: - execution_id: str # UUIDv7 unique per execution - pod_content_hash: str # pod.content_hash().to_string() - - def format_id(self) -> str: ... -``` - -Post-run hook `PodContext` gains `InvocationIdentity` fields or a reference in the -updated `PostRunPayload` (minor ITL-523 extension, not in scope for this ticket but -called out for the implementer). +**No shared base type:** +An earlier draft proposed extracting a shared `InvocationIdentity` base type from +`hooks.py`. `format_id(config=None)` lives directly on `InvocationContext` (Axis 10), +so no shared base is needed. Post-run hooks (ITL-523) may adopt the `orcapod-` prefix +convention independently if desired — no coordinated change to `hooks.py` is required. --- @@ -685,21 +640,22 @@ back to the exact pipeline run and input data. ```python import orcapod as op -from orcapod import SideEffectContext, SideEffectConfig +from orcapod import InvocationContext, SideEffectPodConfig from datetime import datetime, timezone -@op.side_effect_pod(config=SideEffectConfig(idempotent=True, on_error="log")) +# @sink_pod presets track_completion=True, drop_on_failure=True. +# on_error="log" so a failed DB write logs and drops the row rather than aborting. +@op.sink_pod(config=SideEffectPodConfig(on_error="log")) def audit_patient_processing( data: PatientRecord, - ctx: SideEffectContext, + ctx: InvocationContext, ) -> None: """Write a compliance audit row to the external database. - Uses ctx.invocation_signature as the stable provenance key so that - re-running the same pipeline with the same inputs does NOT produce a - duplicate audit row (idempotent=True skips execution if the signature - is already in the invocation log, and ON CONFLICT DO NOTHING handles - any race between instances). + ctx.invocation_hash is deterministic per (pod topology, input packet): + re-running the pipeline with the same inputs produces the same hash, so + ON CONFLICT DO NOTHING provides a DB-level idempotency guard even if + the pipeline retries a row that was previously delivered. """ audit_db.execute( """ @@ -709,7 +665,7 @@ def audit_patient_processing( ON CONFLICT (orcapod_record_id) DO NOTHING """, ( - ctx.invocation_signature, + ctx.invocation_hash, data.patient_id, datetime.now(timezone.utc), ctx.pipeline_run_id, @@ -723,7 +679,7 @@ def audit_patient_processing( patient_stream = op.csv_source("patients_2026.csv") validated_stream = validate_patient_pod(patient_stream) enriched_stream = enrich_patient_pod(validated_stream) -# Side-effect node inserted mid-pipeline; downstream sees the same data. +# Audit pod: only successfully-audited rows reach aggregate_results_pod. audited_stream = audit_patient_processing(enriched_stream) results_stream = aggregate_results_pod(audited_stream) ``` @@ -734,64 +690,63 @@ Row in the external `patient_audit` table: | orcapod_record_id | patient_id | processed_at | pipeline_run | |---|---|---|---| -| `d4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f` | P-12345 | 2026-07-14 02:30:45 UTC | run-2026-07-14-001 | +| `blake3_v1:a3f8c2d1::blake3_v1:b7e491f0` | P-12345 | 2026-07-14 02:30:45 UTC | run-2026-07-14-001 | ### Reverse-lookup walk-through **Problem:** Six months later, an auditor finds the row for `P-12345` and asks: "What input data produced this audit record, and what pipeline logic ran?" -**Step 1 — Signature → invocation record:** +**Step 1 — Parse the invocation hash:** +`blake3_v1:a3f8c2d1::blake3_v1:b7e491f0` → split on `::`: +- Component 1 (pipeline hash): `blake3_v1:a3f8c2d1` +- Component 2 (input packet hash): `blake3_v1:b7e491f0` + +**Step 2 — Locate the invocation table and look up the record:** +``` +table: /blake3_v1:a3f8c2d1.../side_effect_invocations +``` ```sql -SELECT pod_name, pod_content_hash, input_hash, pipeline_run_id, executed_at, status -FROM _orcapod_side_effect_invocations -WHERE invocation_signature = 'd4a8f3b2c1e90a7b5f3e8c2a9b6d4a1f' +SELECT * FROM side_effect_invocations +WHERE full_input_packet_hash = 'blake3_v1:b7e491f0...' ``` ``` -pod_name: "audit_patient_processing" -pod_content_hash: "object_v0.1:c9f1a3b2..." -input_hash: "object_v0.1:8f3a4b2c..." +invocation_hash: "blake3_v1:a3f8c2d1...::blake3_v1:b7e491f0..." +pod_content_hash: "blake3_v1:c9f1a3b2..." pipeline_run_id: "run-2026-07-14-001" executed_at: 2026-07-14T02:30:45Z status: "success" ``` -**Step 2 — Input hash → input data:** +**Step 3 — Input packet hash → input data:** ```sql --- Query enrich_patient_pod result table (shared pipeline hash table). --- _input_data_hash is system_constants.INPUT_DATA_HASH_COL = "_input_data_hash" +-- Query enrich_patient_pod result table using the full_input_packet_hash. SELECT * FROM enrich_patient_pod_results -WHERE _input_data_hash = 'object_v0.1:8f3a4b2c...' +WHERE _input_data_hash = 'blake3_v1:b7e491f0...' ``` ``` patient_id: "P-12345" birth_year: 1985 diagnosis: "T2D" risk_score: 0.73 -_tag_source_id::...::abc... "patients_2026_csv::row47" +_tag_source_id::... "patients_2026_csv::row47" ``` -**Step 3 — System tags → upstream lineage:** -The `_tag_source_id` column resolves to the original source: +**Step 4 — System tags → upstream lineage:** ``` Source: patients_2026.csv Row: 47 -Column: patient_id → "P-12345" ``` **Full lineage:** ``` -patient_audit[orcapod_record_id=d4a8f3b2...] - ← audit_patient_processing (invocation_signature=d4a8f3b2..., pod_content_hash=c9f1a3b2...) +patient_audit[orcapod_record_id=blake3_v1:a3f8c2d1::blake3_v1:b7e491f0] + ← audit_patient_processing (invocation_hash, pod_content_hash=blake3_v1:c9f1a3b2...) ← enrich_patient_pod(PatientRecord{patient_id=P-12345}) ← validate_patient_pod(PatientRecord{patient_id=P-12345}) ← patients_2026.csv : row 47 ``` -**Pod code version:** `pod_content_hash = "object_v0.1:c9f1a3b2..."` can be resolved -to the exact Python function definition (git commit, code hash) via Orcapod's pod -registry or by running `content_hash()` against the current function definition. - --- ## Public API Summary @@ -800,30 +755,44 @@ registry or by running `content_hash()` against the current function definition. # src/orcapod/side_effects.py @dataclasses.dataclass(frozen=True) -class SideEffectConfig: - idempotent: bool = False +class InvocationHashConfig: + encoding: Literal["hex", "base64", "binary"] = "hex" + component_length: int | None = None # bytes of raw digest per component; None = full + + +@dataclasses.dataclass(frozen=True) +class SideEffectPodConfig: + track_completion: bool = True # True = fire-once with completion tracking + drop_on_failure: bool = True # True = drop failed rows; False = always pass through on_error: Literal["raise", "log"] = "raise" + hash_config: InvocationHashConfig = field(default_factory=InvocationHashConfig) @dataclasses.dataclass(frozen=True) -class SideEffectContext: - invocation_signature: str # 32-char hex; deterministic per (pod, inputs) - execution_id: str # UUID string; unique per actual execution - pod_name: str # human-readable pod label - pod_content_hash: str # pod.content_hash().to_string() - pipeline_run_id: str | None # None for lazy/non-compiled pipelines +class InvocationContext: + invocation_hash: str # serialized per pod's InvocationHashConfig + pod_name: str # human-readable pod label (pod.label) + pod_content_hash: str # pod.content_hash().to_string() — exact code version + pipeline_run_id: str | None # None for lazy/non-compiled pipelines - def format_id(self) -> str: - """Return 'orcapod-{invocation_signature}'.""" - return f"orcapod-{self.invocation_signature}" + def format_id(self, config: InvocationHashConfig | None = None) -> str: + """Return 'orcapod-{hash}' with optional format override. + + Args: + config: Optional ``InvocationHashConfig`` overriding encoding and + component_length for this specific call. When ``None``, uses + the pod's own ``InvocationHashConfig``. + """ + ... class SideEffectPod(_FunctionPodBase): """A pipeline node whose primary purpose is a side effect. - The wrapped function returns None. The input stream is passed through - unchanged as output. Every invocation is recorded in the pipeline - database's _orcapod_side_effect_invocations table. + Takes an input stream and returns a filtered output stream. When + ``drop_on_failure=True``, only rows where delivery succeeded are emitted + downstream. Every invocation is recorded in a per-pipeline-hash invocation + table at ``//side_effect_invocations``. """ ... @@ -831,27 +800,60 @@ class SideEffectPod(_FunctionPodBase): def side_effect_pod( fn: SideEffectFn | None = None, *, - config: SideEffectConfig | None = None, + config: SideEffectPodConfig | None = None, ) -> SideEffectPod | Callable[[SideEffectFn], SideEffectPod]: - """Decorator that wraps a function as a SideEffectPod. + """Wrap a function as a SideEffectPod with explicit config. Examples: @side_effect_pod - def emit_metric(data: MyData, ctx: SideEffectContext) -> None: + def emit_metric(data: MyData, ctx: InvocationContext) -> None: metrics.emit("my.metric", data.value, tags={"id": ctx.format_id()}) - @side_effect_pod(config=SideEffectConfig(idempotent=True)) - def write_audit_row(data: MyData, ctx: SideEffectContext) -> None: - audit_db.execute("INSERT ...", (ctx.invocation_signature, ...)) + @side_effect_pod(config=SideEffectPodConfig(track_completion=False, drop_on_failure=False)) + def trace_row(data: MyData, ctx: InvocationContext) -> None: + logger.debug("[%s] row seen", ctx.invocation_hash) + """ + ... + + +def sink_pod( + fn: SideEffectFn | None = None, + *, + config: SideEffectPodConfig | None = None, +) -> SideEffectPod | Callable[[SideEffectFn], SideEffectPod]: + """Shortcut: track_completion=True, drop_on_failure=True. + + Examples: + @sink_pod(config=SideEffectPodConfig(on_error="log")) + def write_audit_row(data: MyData, ctx: InvocationContext) -> None: + audit_db.execute("INSERT ...", (ctx.invocation_hash, ...)) + """ + ... + + +def tap_pod( + fn: SideEffectFn | None = None, + *, + config: SideEffectPodConfig | None = None, +) -> SideEffectPod | Callable[[SideEffectFn], SideEffectPod]: + """Shortcut: track_completion=False, drop_on_failure=False. + + Examples: + @tap_pod + def log_row(data: MyData, ctx: InvocationContext) -> None: + logger.info("[%s] Processing %s", ctx.invocation_hash, data.id) """ ... ``` Re-exported from `orcapod.__init__`: - `SideEffectPod` -- `SideEffectConfig` -- `SideEffectContext` +- `SideEffectPodConfig` +- `InvocationHashConfig` +- `InvocationContext` - `side_effect_pod` +- `sink_pod` +- `tap_pod` --- @@ -861,90 +863,116 @@ New test file: `tests/test_core/side_effect_pod/test_side_effect_pod.py` | # | Scenario | Assertion | |---|---|---| -| T1 | Basic execution — pod runs, output equals input | Pass-through: input (Tag, Data) emitted unchanged | -| T2 | `ctx` auto-injection — function with `ctx: SideEffectContext` | `ctx.invocation_signature` is 32 hex chars; `ctx.format_id()` returns `"orcapod-{sig}"` | -| T3 | No-ctx function — function without `ctx` param | Pod runs without error; `SideEffectContext` not constructed | -| T4 | Invocation log written — DB-backed pipeline | `_orcapod_side_effect_invocations` has one row; `status="success"` | -| T5 | `idempotent=True` — same inputs re-run | Second invocation skipped; `status="skipped"` row added; side-effect function called once only | -| T6 | `idempotent=False` (default) — same inputs re-run | Side-effect function called both times; two rows in invocation log (different `execution_id`) | -| T7 | `on_error="raise"` — function raises | Exception propagates; no output row; `status="error"` in log | -| T8 | `on_error="log"` — function raises | Exception logged; input row emitted unchanged; `status="error"` in log | -| T9 | Multiple inputs — N input rows | N log rows; pass-through of all N rows | -| T10 | `invocation_signature` determinism | Re-running pod with identical inputs + identical code → identical `invocation_signature` | -| T11 | `execution_id` uniqueness | Two runs with identical inputs → same `invocation_signature`, different `execution_id` | -| T12 | Async channel execution | Pod runs correctly through `async_execute()` path; log rows written | -| T13 | Parallel execution — `max_concurrency > 1` | All invocations complete; all log rows written; no data loss | -| T14 | Pipeline composition — side-effect pod mid-pipeline | Downstream pod receives same data as upstream; side effect runs | -| T15 | `@side_effect_pod` decorator — functional form | `@side_effect_pod` without args and `@side_effect_pod(config=...)` both produce `SideEffectPod` | +| T1 | Basic pass-through — `drop_on_failure=False`, success | Output stream contains all input rows unchanged | +| T2 | `drop_on_failure=True` (default), success | Successfully-delivered rows emitted; no rows dropped | +| T3 | `ctx: InvocationContext` auto-injection | `ctx.invocation_hash` is a non-empty string; `ctx.format_id()` returns `"orcapod-{hash}"` | +| T4 | No-`ctx` function | Pod runs without error; `InvocationContext` not constructed | +| T5 | Invocation log written — DB-backed pipeline | Row written to `/side_effect_invocations`; `status="success"` | +| T6 | `track_completion=True` — same inputs re-run | Second run: delivery skipped, `status="skipped"` row added; function called once only; row still emitted downstream | +| T7 | `track_completion=False` — same inputs re-run | Both runs: function called; two `status="success"` log rows | +| T8 | `on_error="raise"` — function raises | Exception propagates; row aborted; `status="failed"` in log | +| T9 | `on_error="log"` + `drop_on_failure=True` — function raises | Exception logged; row **dropped** from output; `status="failed"` in log | +| T10 | `on_error="log"` + `drop_on_failure=False` — function raises | Exception logged; row **emitted** downstream; `status="failed"` in log | +| T11 | `invocation_hash` determinism | Re-running pod with identical inputs + code → identical `invocation_hash` | +| T12 | `invocation_hash` format override | `ctx.format_id(InvocationHashConfig(encoding="base64", component_length=8))` produces valid base64 compound | +| T13 | Async channel execution | Pod runs correctly through `async_execute()` path; log rows written | +| T14 | Parallel execution — `max_concurrency > 1` | All invocations complete; all log rows written; no data loss | +| T15 | Pipeline composition mid-pipeline | Downstream pod receives correct filtered output; side effect runs | +| T16 | `@sink_pod` shortcut | Produces `SideEffectPod` with `track_completion=True, drop_on_failure=True` | +| T17 | `@tap_pod` shortcut | Produces `SideEffectPod` with `track_completion=False, drop_on_failure=False` | +| T18 | `@side_effect_pod(config=...)` with all four combinations | Each combination behaves per spec | --- ## Scope & Boundaries **In scope:** -- `SideEffectPod` class, `SideEffectConfig`, `SideEffectContext`, `side_effect_pod` - decorator. -- `_orcapod_side_effect_invocations` table: creation, insertion, idempotency check. -- `SideEffectContext` auto-injection by type annotation. -- Pass-through output stream. +- `SideEffectPod` class, `SideEffectPodConfig`, `InvocationHashConfig`, + `InvocationContext`, `side_effect_pod` / `sink_pod` / `tap_pod` decorators. +- Per-pipeline-hash invocation tables: creation, insertion, completion-state lookup. +- `InvocationContext` auto-injection by type annotation. +- Filtered pass-through output stream (`drop_on_failure` axis). +- Completion tracking and skip-on-success logic (`track_completion` axis). - Sync and async execution paths. -- `InvocationIdentity` shared base type extracted into `hooks.py`. - All re-exports from `orcapod.__init__`. -- Tests covering T1–T15. -- Documentation update: new `docs/concepts/side-effect-pods.md`. +- Tests covering T1–T18. +- Documentation: new `docs/concepts/side-effect-pods.md`. **Out of scope:** - Actual implementation (this is a design spike). -- Retry logic. +- Retry policy (ITL-526). +- Fire-and-forget delivery mode (ITL-528). +- `FunctionPod` error handling alignment (ITL-527). - External adapter helpers (Slack notifier, structured-logging wrapper). -- Run history UI or searchable index service (downstream of this primitive; see PLT-1950). -- Pre-run or on-error hooks for side-effect pods. +- Run history UI or searchable index service (PLT-1950). - Remote/cross-process side-effect execution. -- `idempotency_key` customization (v1 uses `invocation_signature` only). --- ## Dependencies & Risks **Dependencies:** -- ITL-523 (post-run hook, In Progress) — `hooks.py` is the insertion point for the - shared `InvocationIdentity` base type. Coordinate to avoid conflicting changes to - `_FunctionPodBase`. -- PLT-1950 (EDI provenance) — the `_orcapod_side_effect_invocations` table is a - natural data source for provenance stamping. Schema should be coordinated. +- ITL-523 (post-run hook, In Progress) — coordinate to avoid conflicting changes to + `_FunctionPodBase`. No shared base type needed; `InvocationContext` is independent. +- ITL-526 (retry policy) — deferred; will add fields to `SideEffectPodConfig`. +- ITL-527 (`FunctionPod` error handling alignment) — deferred; will reuse the same + `on_error` vocabulary defined here. +- ITL-528 (fire-and-forget mode) — deferred; will add `async_delivery` to + `SideEffectPodConfig`. +- PLT-1950 (EDI provenance) — the per-pipeline-hash invocation tables are a natural + data source for provenance stamping. Schema should be coordinated when that work begins. **Risks:** -- **Framework creep:** Side-effect pods that "almost" produce output. Enforced by - the type signature (`-> None` required) and the pass-through contract. -- **Overly long hash:** 32 hex chars (128 bits) is short enough for all common embedding - contexts. Test with real log pipelines before finalizing. -- **Invocation log growth:** A high-volume pipeline calling a side-effect pod per row - will produce large invocation logs. Add a note in docs about pruning / archiving. -- **Idempotency false-positive:** If pod code changes between runs, the same inputs - produce the same `invocation_signature` (code change is not reflected). This is - correct behavior (the signature is input + code hash, so code changes produce - different signatures), but implementers should verify the `pod_content_hash` column - includes code identity, not just name. +- **Invocation log growth:** A high-volume pipeline with `track_completion=False` will + write one log row per row per run. Document pruning / archiving guidance. +- **Hash truncation collision:** Shorter `component_length` values (via `InvocationHashConfig`) + increase prefix-match ambiguity in reverse lookup. Document the trade-off; 8 bytes + (64 bits) per component is negligible collision probability for realistic volumes. +- **`track_completion=True` + code change:** `pipeline_hash(pod)` captures the full + upstream topology and function identity, so a code change does produce a different + pipeline hash and a different invocation table — previously-succeeded rows in the old + table are not seen, and delivery re-runs. This is correct but should be documented + clearly so pod authors are not surprised. --- ## Follow-Up Implementation Issue -See ITL-525: *Implement `SideEffectPod` with invocation hash and invocation log -(ITL-524 follow-up).* +See ITL-525: *Implement `SideEffectPod` (ITL-524 follow-up).* Task breakdown for ITL-525: -1. **`InvocationIdentity` base type** — add to `src/orcapod/hooks.py`; update - `PostRunPayload` to embed it (minor ITL-523 coordination). -2. **`src/orcapod/side_effects.py`** — new module: `SideEffectConfig`, - `SideEffectContext`, `SideEffectPod`, `side_effect_pod` decorator. -3. **`SideEffectPod` core execution** — `process_data()` / `async_process_data()` - pass-through implementation; `SideEffectContext` injection; invocation log writes. -4. **Invocation log** — `_orcapod_side_effect_invocations` table creation and insertion - in the pipeline DB; idempotency check for `idempotent=True`. -5. **`orcapod.__init__` re-exports** — `SideEffectPod`, `SideEffectConfig`, - `SideEffectContext`, `side_effect_pod`. -6. **Tests** — `tests/test_core/side_effect_pod/test_side_effect_pod.py` covering T1–T15. -7. **Documentation** — `docs/concepts/side-effect-pods.md` with overview, usage - examples, reverse-lookup walk-through, and idempotency guidance. +1. **`src/orcapod/side_effects.py`** — new module: + - `InvocationHashConfig` dataclass + - `SideEffectPodConfig` dataclass (`track_completion`, `drop_on_failure`, `on_error`, + `hash_config`) + - `InvocationContext` dataclass with `format_id(config=None)` method + - `SideEffectPod` class (extends `_FunctionPodBase`) + - `side_effect_pod`, `sink_pod`, `tap_pod` decorators + +2. **`SideEffectPod` core execution** — `process_data()` / `async_process_data()`: + - `InvocationContext` injection (detect by type annotation, construct from raw + hash components) + - Row filtering per `drop_on_failure` + - `on_error` handling: `"raise"` propagates; `"log"` logs + drops-or-emits per + `drop_on_failure` + +3. **Invocation table** — per-pipeline-hash table at + `//side_effect_invocations`: + - Table creation on first invocation + - Row insertion with all schema columns + - Completion-state lookup for `track_completion=True` (skip / re-attempt logic) + - `"skipped"` row for re-emitted previously-succeeded inputs + +4. **Hash computation** — `full_input_packet_hash` covering tag + data + source + + system tag columns; invocation hash compound construction and serialization per + `InvocationHashConfig` + +5. **`orcapod.__init__` re-exports** — `SideEffectPod`, `SideEffectPodConfig`, + `InvocationHashConfig`, `InvocationContext`, `side_effect_pod`, `sink_pod`, `tap_pod` + +6. **Tests** — `tests/test_core/side_effect_pod/test_side_effect_pod.py` covering T1–T18 + +7. **Documentation** — `docs/concepts/side-effect-pods.md` with overview, config axes, + usage examples (all four `track_completion` × `drop_on_failure` combinations), + reverse-lookup walk-through, and log growth guidance