feat(iceberg): replay-safe immutable history ingestion

docs/guides/ingest-data.md

@@ -103,6 +103,60 @@ def countries(partition_date: str):
     return load_country_list()
 ```
 
+### Preserve immutable versions
+
+Use `merge_strategy="history"` when a source can replay immutable versions. Install `phlo-iceberg` as the table-store provider. Providers without atomic history support reject this mode before creating or writing a table.
+
+Declare both identity columns and select the payload to compare:
+
+```python
+import dlt
+import phlo
+from pandera.pandas import DataFrameModel
+
+
+class ExhibitVersionSchema(DataFrameModel):
+	exhibit_id: str
+	version_id: str
+	description: str
+
+
+@phlo.ingest.dlt(
+	table_name="exhibit_versions",
+	unique_key="exhibit_id",
+	group="museum",
+	validation_schema=ExhibitVersionSchema,
+	merge_strategy="history",
+	merge_config={
+		"entity_key": "exhibit_id",
+		"version_key": "version_id",
+		"payload_columns": ["description"],
+	},
+)
+def exhibit_versions(partition_date: str):
+	return dlt.resource(
+		[
+			{
+				"exhibit_id": "E-42",
+				"version_id": "V-1",
+				"description": "Bronze compass",
+			}
+		],
+		name="exhibit_versions",
+		write_disposition="append",
+	)
+```
+
+Materialise the same partition twice. The first run inserts one version. The second inserts zero rows and reports `rows_skipped=1`, despite new arrival metadata. Supply V-2 with a changed description to retain both versions. A later V-1 with a different description fails without changing the committed rows, schema, or policy properties.
+
+Keep DLT extraction append-only so Phlo receives every observation. Do not configure upstream replacement or deduplication that hides conflicting versions. `unique_key` remains a required decorator argument, but history identity comes from `entity_key` and `version_key` together. Different exhibits can each have a V-1.
+
+If your source already supplies a stable payload hash, replace `payload_columns` with `payload_hash_column="payload_hash"`. Never supply both. Include the hash column in your schema and every source row. Keep source arrival timestamps and `_phlo_` or `_dlt_` metadata out of comparison fields.
+
+For a rejected conflict, inspect the failed operation's `rows_conflicting` count and correct the source version or payload. Do not overwrite the stored version. For an unknown commit outcome, inspect the reconciliation evidence before explicitly replaying the batch. The failed operation does not claim an inserted count.
+
+Before changing identity or payload policy, stop all writers and perform a validated migration. Nonempty tables without a history policy cannot be adopted automatically. See the [history-mode reference](../reference/python-api.md#immutable-history-mode) for the concurrency guarantee, counts, and migration requirements.
+
 ### Choose a schema policy
 
 Iceberg writes now default to `schema_policy="strict"`. Extra source columns fail the batch instead of disappearing with a warning. This is an intentional compatibility change.

docs/learn/03-ingestion-with-dlt.md

@@ -18,9 +18,9 @@ Replayability is the practical benefit of these choices. A dated partition, stab
 
 The `phlo-dlt` package provides `@phlo.ingest.dlt`. The decorator turns a Python function into a Dagster asset and connects the returned dlt resource to table storage. `phlo-pandera` can validate each batch before publication.
 
-The tutorial asset uses a daily partition and `unique_key="event_id"`. The default `merge_strategy="merge"` upserts rows on that key. The other supported value is `merge_strategy="append"`, which inserts rows without using a merge key. Use append for an immutable event stream where each row is new. Set `partitioned=False` for reference data that has no natural partition.
+The tutorial asset uses a daily partition and `unique_key="event_id"`. The default `merge_strategy="merge"` upserts rows on that key. `merge_strategy="append"` inserts rows without using a merge key. Use append only when each row is new. For replayed immutable observations, `merge_strategy="history"` retains distinct entity/version pairs, skips identical versions, and rejects conflicting payloads. Set `partitioned=False` for reference data that has no natural partition.
 
-Phlo accepts `merge_strategy="append"` and `merge_strategy="merge"`. These are the only values the ingestion decorator accepts. `freshness_hours=(warning_hours, error_hours)` can add freshness checks, while `strict_validation` controls whether failed validation blocks the run.
+History mode requires explicit entity and version columns plus payload columns or a supplied payload hash. Arrival metadata does not participate in comparison. See [Preserve immutable versions](../guides/ingest-data.md#preserve-immutable-versions) for configuration and recovery. `freshness_hours=(warning_hours, error_hours)` can add freshness checks, while `strict_validation` controls whether failed validation blocks the run.
 
 The decorator does not decide what a row means for you. You still need to choose a key that identifies the source entity and a partition that matches the source's time semantics. If the source has immutable events, an event identifier can be retained as data while append preserves each event. If the source sends current entity state, a stable key and merge make replaying a partition safer.
 

docs/reference/python-api.md

@@ -44,7 +44,7 @@ The guarantee covers complete-set visibility on a local filesystem with atomic s
 | Parameter | Type | Default | Meaning |
 | --- | --- | --- | --- |
 | `table_name` | `str` | required | Destination table name. |
-| `unique_key` | `str` | required | Merge and deduplication key. |
+| `unique_key` | `str` | required | Merge and deduplication key. History uses explicit identity in `merge_config`. |
 | `group` | `str` | required | Asset group. |
 | `validation_schema` | `type[Any] \| None` | `None` | Pandera validation schema. |
 | `table_schema` | `Any \| None` | `None` | Explicit table-store schema. |
@@ -56,8 +56,8 @@ The guarantee covers complete-set visibility on a local filesystem with atomic s
 | `retry_delay_seconds` | `int` | `30` | Delay between retries. |
 | `validate` | `bool` | `True` | Enable validation. |
 | `strict_validation` | `bool` | `True` | Make validation failures blocking. |
-| `merge_strategy` | `Literal["append", "merge"]` | `"merge"` | Append rows or merge on `unique_key`. |
-| `merge_config` | `dict[str, Any] \| None` | `None` | Merge behaviour overrides. |
+| `merge_strategy` | `Literal["append", "merge", "history"]` | `"merge"` | Append rows, upsert on `unique_key`, or insert immutable versions. |
+| `merge_config` | `dict[str, Any] \| None` | `None` | Merge overrides or required [history policy](#immutable-history-mode). |
 | `schema_policy` | `Literal["strict", "additive", "drop_extra"]` | `"strict"` | Schema handling for opted-in table stores. See [Iceberg write schema policies](#iceberg-write-schema-policies). |
 | `add_metadata_columns` | `bool` | `True` | Add Phlo metadata columns. |
 | `owner` | `str \| None` | `None` | Owning team. |
@@ -80,6 +80,35 @@ def load_events(partition_date: str) -> object:
     return read_events(partition_date)
 ```
 
+### Immutable history mode
+
+`merge_strategy="history"` inserts only unseen `(entity_key, version_key)` pairs. It skips committed identical versions and identical duplicates within the batch. Any payload conflict rejects the entire batch before a data or metadata commit. Distinct late versions remain stored regardless of arrival order. An identical replay keeps the first committed row, including its original arrival metadata.
+
+| `merge_config` option | Requirement |
+| --- | --- |
+| `entity_key` | Explicit entity column name. |
+| `version_key` | Explicit immutable version column name, distinct from `entity_key`. |
+| `payload_columns` | Nonempty list of distinct comparison column names. Mutually exclusive with `payload_hash_column`. Order does not affect the policy. |
+| `payload_hash_column` | Column containing a supplied stable string or binary payload hash. Mutually exclusive with `payload_columns`. The project owns hashing and collision risk. |
+
+Every staged file must supply the selected columns. Identity and hash values cannot be null, empty, or non-finite. Identity values must be scalar. Payload values compare after safe schema alignment, with null equal to null and floating-point NaN equal to NaN, including nested values. Reserved `_phlo_` and `_dlt_` columns cannot define history identity or comparison. Other arrival metadata is excluded by selecting only payload fields. No identity generation or latest-version ordering is provided.
+
+All staged files from one ingestion write form one batch. The provider reads them into memory, so batch size must fit available memory. Iceberg pushes exact incoming `(entity_key, version_key)` pairs into its scan predicates, in groups of at most 1,000 distinct pairs. It projects only identity and comparison fields; unrelated entities sharing version IDs are not materialised for comparison. Schema-policy validation applies before the history write. Compatible additions, policy binding, and inserted rows publish in one transaction. A replay with no new versions publishes no schema or policy changes.
+
+`IcebergResource.history_parquet` accepts `table_name`, `data_paths`, a `HistoryPolicy` from `phlo.capabilities.history`, `override_ref`, `schema_policy`, and optional `evidence_context`. The storage helper is `phlo_iceberg.history.history_to_table`. Providers opt in with `TableStoreSupport.supports_history` and the optional `HistoryTableStore` protocol. Existing table-store providers need not implement a new mandatory method. Unsupported providers fail explicitly before table creation or writes.
+
+Successful writes report `rows_inserted`, `rows_skipped`, `rows_conflicting`, and `rows_deleted=0` through operation evidence and materialisation metadata. Counts describe incoming rows, not attempts. For an unseen pair repeated three times, one row is inserted and two are skipped. For a committed identical pair repeated three times, all three are skipped. On conflict, `HistoryConflictError.metrics` reports zero inserted rows and counts all incoming rows belonging to conflicting pairs. Nonconflicting replay or duplicate rows contribute to `rows_skipped`; unseen rows rejected with the batch are not counted as inserted or skipped.
+
+**Concurrency.** PyIceberg 0.11.1 append asserts the original snapshot head, including the absence of a snapshot on an empty table. Lookup is pinned to that head and uses the same loaded metadata as the transaction. A definite commit conflict reloads the table and repeats preparation, policy validation, lookup, comparison, and insertion. There are at most three whole-operation attempts. Append alone is never retried. Retry exhaustion fails explicitly.
+
+An ambiguous commit exception, including a lost transport response, triggers fresh policy and version readback. `HistoryCommitUnknownError` then fails the operation, even if the versions are now present. Evidence records readback under `reconciliation`, with `outcome="unknown"`. Successful readback partitions distinct incoming identity pairs into `versions_present` (matching committed payload), `versions_missing` (absent), and `versions_conflicting` (different committed payload). These counts sum to the number of distinct incoming pairs, not incoming rows. Two duplicate observations of one committed version therefore report `versions_present=1`. Conflicting readback has `state="conflicting"`; unavailable readback has `state="unavailable"` without version counts. This evidence does not invent inserted or skipped counts from an uncertain commit. A subsequent explicit replay performs the complete comparison again.
+
+**Policy metadata and migration.** Iceberg properties `phlo.history.policy` and `phlo.history.policy.sha256` contain the canonical policy and its SHA-256 fingerprint. The definition includes entity and version columns, comparison mode, payload fields, Iceberg field IDs, types, and policy format version. The first nonempty insertion binds these properties atomically. An unmarked nonempty table is rejected rather than adopted. Missing or incompatible policy metadata requires migration.
+
+Policy changes require quiesced writers and validated data/schema migration before rebinding the properties. A snapshot guard does not detect property-only changes. Renames, replaced field IDs, changed comparison types, and altered identity or payload selection cannot silently redefine existing history. Migration must validate uniqueness and payload consistency under the new policy. This API provides no live policy-edit or migration bypass.
+
+The uniqueness guarantee applies to cooperating history writers on the same table/ref. Ordinary append, merge, overwrite, rollback, external writers, and independent refs do not enforce history semantics. They must not mutate a table concurrently with history ingestion or migration.
+
 ### Iceberg write schema policies
 
 `IcebergResource.ensure_table`, `append_parquet`, `merge_parquet`, and `overwrite_parquet` accept the keyword-only `schema_policy` argument. The storage helpers `ensure_table`, `append_to_table`, `merge_to_table`, and `overwrite_table` accept the same argument. All default to `"strict"`.

packages/phlo-dlt/src/phlo_dlt/decorator.py

@@ -66,6 +66,7 @@ def load_users(partition_date: str):
     resolve_capability,
 )
 from phlo.capabilities.runtime import RuntimeContext, routing_from_context
+from phlo.capabilities.history import HistoryPolicy
 from phlo.contracts import SLA, Consumer, normalize_consumers, serialize_consumers, serialize_sla
 from phlo.exceptions import PhloConfigError
 from phlo.logging import log_event
@@ -169,20 +170,30 @@ def _validate_merge_config(
 ) -> None:
     """Validate merge strategy and merge configuration semantics.
 
-    `merge_strategy` must be `append` or `merge`; when provided, `merge_config`
+    `merge_strategy` must be `append`, `merge` or `history`; history requires
+    explicit entity/version and payload configuration. For other modes, `merge_config`
     must be a dict of supported keys only: `deduplication` (bool, requires a
     non-empty `unique_key` when true), `deduplication_method` (`first` or
     `last`), and `deduplication_order_by` (ordering column name used by
     `last`). Unknown keys or malformed values are rejected.
 
     Raises: PhloConfigError when the strategy or config values are invalid.
     """
-    if merge_strategy not in ("append", "merge"):
+    if merge_strategy not in ("append", "merge", "history"):
         raise PhloConfigError(
             message=f"Invalid merge_strategy: {merge_strategy}",
-            suggestions=["Use merge_strategy='append' or merge_strategy='merge'"],
+            suggestions=["Use merge_strategy='append', 'merge' or 'history'"],
         )
 
+    if merge_strategy == "history":
+        if merge_config is not None and not isinstance(merge_config, dict):
+            raise PhloConfigError(message="merge_config must be a dict")
+        try:
+            HistoryPolicy.from_config(merge_config)
+        except ValueError as exc:
+            raise PhloConfigError(message=str(exc)) from exc
+        return
+
     if merge_config is None:
         return
 
@@ -328,7 +339,7 @@ def phlo_ingestion(  # noqa: C901
     retry_delay_seconds: int = 30,
     validate: bool = True,
     strict_validation: bool = True,
-    merge_strategy: Literal["append", "merge"] = "merge",
+    merge_strategy: Literal["append", "merge", "history"] = "merge",
     merge_config: dict[str, Any] | None = None,
     add_metadata_columns: bool = True,
     owner: str | None = None,
@@ -374,9 +385,10 @@ def phlo_ingestion(  # noqa: C901
           ``"drop_extra"`` explicitly discards extra source columns. Providers
           without support retain their defaults and reject non-default policies.
         - `partition_spec` format depends on the provider (e.g., Iceberg transforms).
-        - `merge_strategy` selects insert-only ``"append"`` versus ``"merge"``
-          upserts on `unique_key`; `merge_config` overrides merge behaviour, e.g.
-          ``{"deduplication": True, "deduplication_method": "last"}``.
+        - `merge_strategy` selects ``"append"``, ``"merge"`` upserts on
+          `unique_key`, or immutable ``"history"``. History requires
+          `merge_config` with entity_key, version_key and exactly one of
+          payload_columns or payload_hash_column, excluding arrival metadata.
         - With `add_metadata_columns` enabled, Phlo injects `_phlo_row_id`,
           `_phlo_ingested_at`, `_phlo_partition_date`, and `_phlo_run_id`.
         - `freshness_hours` is a (warning_hours, error_hours) tuple used for SLA
@@ -885,6 +897,11 @@ def run(runtime: RuntimeContext) -> Iterator[RunResult]:  # noqa: C901
                         "partition_date": partition_date,
                         "rows_inserted": result.rows_inserted,
                         "rows_deleted": result.rows_deleted,
+                        **{
+                            key: result.metadata[key]
+                            for key in ("rows_skipped", "rows_conflicting")
+                            if key in result.metadata
+                        },
                         "unique_key": table_config.unique_key,
                         "table_name": table_config.full_table_name,
                         "table_store": table_store_name,