fix(durable): bound provider lifecycle fields (#170)

* test(durable): require bounded lifecycle fields

* test(durable): preserve sparse lifecycle compatibility

* fix(durable): validate persisted lifecycle vocabulary

* docs(durable): record verified lifecycle vocabulary

* test(durable): align NUL lifecycle contract

Author: seonghobae

docs/doctoring/durable-lifecycle-field-contract.md

@@ -0,0 +1,67 @@
+# Durable provider lifecycle field contract
+
+## Status
+
+ACTIVE-PR. This assurance note describes the bounded validation introduced by the
+current lifecycle-field change. It becomes IMPLEMENTED-ON-PROTECTED-MAIN only
+after the exact source is accepted and merged through repository governance.
+
+## Problem boundary
+
+Provider lifecycle observations are untrusted control-plane evidence. A durable
+projection must not persist arbitrary non-empty status or endpoint strings and
+then infer terminal semantics from those unverified values. At the same time,
+existing sparse observations remain compatible: an absent/empty status maps to
+the historical `unknown` state and an absent/empty endpoint remains `None`.
+
+The durable persistence boundary therefore validates every present non-empty
+status and endpoint before PostgreSQL acquisition. Unsupported values fail with
+fixed package-owned error categories and the rejected provider value is not
+copied into the error message. The HTTP client's broader endpoint grammar is a
+separate transport-compatibility boundary; durable evidence intentionally claims
+only the first-party Batch vocabulary verified here.
+
+## Verified vocabulary
+
+As checked on 2026-08-13, the OpenAI Batch API reference documents Batch support
+for `/v1/responses`, `/v1/chat/completions`, `/v1/embeddings`,
+`/v1/completions`, and `/v1/moderations`. The lifecycle vocabulary represented
+by the Batch object and cancellation flow is `validating`, `failed`,
+`in_progress`, `finalizing`, `completed`, `expired`, `cancelling`, and
+`cancelled`.
+
+`completed`, `failed`, `expired`, and `cancelled` are the package's durable
+terminal set. `cancelling` remains transitional and must not receive a terminal
+timestamp merely because cancellation has been requested.
+
+OpenAI-compatible providers may expose extensions, but pg-llm-batch does not
+silently promote an unverified extension into durable semantics. Supporting an
+additional endpoint or status requires a reviewed compatibility change and
+fresh regression/provider evidence.
+
+## Failure, rollback, and recovery
+
+This change adds no schema migration and does not rewrite historical lifecycle
+rows. Rejected new observations fail before database I/O. Rollback is therefore
+a source rollback of the validation change; existing PostgreSQL state is not
+transformed by that rollback.
+
+If an embedding provider adds a legitimate new lifecycle value, operators should
+not weaken validation locally. The recovery path is to verify the provider
+contract, add the value to the reviewed finite set with tests and documentation,
+and deploy that accepted package revision.
+
+## Verification expectations
+
+Acceptance requires regressions proving that unsupported, oversized,
+control-bearing, and non-string present values fail before PostgreSQL access;
+that sparse absence preserves the prior safe defaults; that every currently
+verified endpoint/status is accepted; and that terminal timestamps are assigned
+only to the reviewed terminal subset. Repository Python 3.10/3.12/3.14, exact
+owned-production statement/branch coverage, docstrings, packaging, security,
+SAST, PostgreSQL/container, and required central review gates remain mandatory.
+
+## Primary reference
+
+OpenAI. (2026). *Batch | OpenAI API reference*.
+https://platform.openai.com/docs/api-reference/batch/object

pg_llm_batch/db.py

@@ -38,6 +38,27 @@
 TENANT_SCOPE_PATTERN = re.compile(
     rf"[A-Za-z0-9][A-Za-z0-9._:-]{{0,{MAX_TENANT_SCOPE_CHARACTERS - 1}}}\Z"
 )
+SUPPORTED_REMOTE_BATCH_STATUSES = frozenset(
+    {
+        "validating",
+        "failed",
+        "in_progress",
+        "finalizing",
+        "completed",
+        "expired",
+        "cancelling",
+        "cancelled",
+    }
+)
+SUPPORTED_REMOTE_BATCH_ENDPOINTS = frozenset(
+    {
+        "/v1/responses",
+        "/v1/chat/completions",
+        "/v1/embeddings",
+        "/v1/completions",
+        "/v1/moderations",
+    }
+)
 REMOTE_TERMINAL_STATUSES = frozenset({"completed", "failed", "expired", "cancelled"})
 _REMOTE_BATCH_STATE_FIELDS = (
     "tenant_scope",
@@ -113,6 +134,26 @@ def normalize_optional_provider_text(value: Any) -> Optional[str]:
     )
 
 
+def _provider_batch_status(value: Any) -> str:
+    """Return one verified durable lifecycle status or the sparse safe default."""
+    if value is None or value == "":
+        return "unknown"
+    if type(value) is not str or value not in SUPPORTED_REMOTE_BATCH_STATUSES:
+        raise ValueError("batch_status is not a supported provider status")
+    return value
+
+
+def _provider_batch_endpoint(value: Any) -> Optional[str]:
+    """Return one verified durable batch endpoint or preserve sparse absence."""
+    if value is None or value == "":
+        return None
+    if type(value) is not str or value not in SUPPORTED_REMOTE_BATCH_ENDPOINTS:
+        raise ValueError(
+            "batch_endpoint is not a supported provider batch endpoint"
+        )
+    return value
+
+
 def _provider_count(value: Any) -> int:
     """Return a non-negative integer provider count or the safe default zero."""
     return value if type(value) is int and value >= 0 else 0
@@ -377,10 +418,8 @@ def _normalize_remote_batch_snapshot(
         provider_batch.get("error_file_id"),
         "error_file_id",
     )
-    batch_status = (
-        normalize_optional_provider_text(provider_batch.get("status"))
-        or "unknown"
-    )
+    batch_status = _provider_batch_status(provider_batch.get("status"))
+    batch_endpoint = _provider_batch_endpoint(provider_batch.get("endpoint"))
     counts_value = provider_batch.get("request_counts")
     request_counts = counts_value if isinstance(counts_value, Mapping) else {}
     raw_total_requests = request_counts.get("total")
@@ -405,9 +444,7 @@ def _normalize_remote_batch_snapshot(
         "remote_batch_id": remote_batch_id,
         "observation_order": observation_order,
         "input_file_id": input_file_id,
-        "batch_endpoint": normalize_optional_provider_text(
-            provider_batch.get("endpoint")
-        ),
+        "batch_endpoint": batch_endpoint,
         "batch_status": batch_status,
         "output_file_id": output_file_id,
         "error_file_id": error_file_id,

tests/test_durable_lifecycle_field_contract.py

@@ -0,0 +1,146 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Regression tests for durable provider lifecycle field validation."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+from pg_llm_batch import db
+
+
+class _NoDatabaseIO:
+    """Fail if invalid provider lifecycle fields reach PostgreSQL acquisition."""
+
+    def __init__(self) -> None:
+        self.connections: list[str] = []
+
+    def connect(self, dsn: str):
+        """Record an unexpected connection attempt and fail the test immediately."""
+        self.connections.append(dsn)
+        raise AssertionError("invalid lifecycle fields reached PostgreSQL")
+
+
+def _provider_batch(*, status: object, endpoint: object) -> dict[str, object]:
+    """Build one otherwise-valid provider lifecycle observation."""
+    return {
+        "id": "batch_contract_1",
+        "status": status,
+        "endpoint": endpoint,
+        "request_counts": {"total": 1, "completed": 0, "failed": 0},
+    }
+
+
+@pytest.mark.parametrize(
+    "status",
+    ["future_state", "COMPLETED", "x" * 65, "completed\x00secret", 7],
+)
+def test_persistence_rejects_unsupported_status_before_database_io(
+    monkeypatch: pytest.MonkeyPatch,
+    status: object,
+) -> None:
+    """Reject unsupported provider status evidence before PostgreSQL mutation."""
+    driver = _NoDatabaseIO()
+    monkeypatch.setattr(db, "psycopg", driver)
+
+    with pytest.raises(
+        ValueError,
+        match="batch_status is not a supported provider status",
+    ) as exc:
+        db.persist_remote_batch_state(
+            "postgresql://should-not-connect",
+            "default",
+            _provider_batch(status=status, endpoint="/v1/responses"),
+            1,
+            observed_at=datetime(2026, 8, 13, tzinfo=timezone.utc),
+        )
+
+    assert str(status) not in str(exc.value)
+    assert driver.connections == []
+
+
+@pytest.mark.parametrize(
+    "endpoint",
+    [
+        "/v1/future-endpoint",
+        "/v1/chat/completions?debug=1",
+        "/v1/../chat/completions",
+        "/v1/responses\x00secret",
+        7,
+    ],
+)
+def test_persistence_rejects_unsupported_endpoint_before_database_io(
+    monkeypatch: pytest.MonkeyPatch,
+    endpoint: object,
+) -> None:
+    """Reject unsupported provider endpoint evidence before PostgreSQL mutation."""
+    driver = _NoDatabaseIO()
+    monkeypatch.setattr(db, "psycopg", driver)
+
+    with pytest.raises(
+        ValueError,
+        match="batch_endpoint is not a supported provider batch endpoint",
+    ) as exc:
+        db.persist_remote_batch_state(
+            "postgresql://should-not-connect",
+            "default",
+            _provider_batch(status="validating", endpoint=endpoint),
+            1,
+            observed_at=datetime(2026, 8, 13, tzinfo=timezone.utc),
+        )
+
+    assert str(endpoint) not in str(exc.value)
+    assert driver.connections == []
+
+
+def test_sparse_status_and_endpoint_preserve_legacy_safe_defaults() -> None:
+    """Absent lifecycle fields retain the historical sparse-response contract."""
+    observed = datetime(2026, 8, 13, tzinfo=timezone.utc)
+    snapshot, _ = db._normalize_remote_batch_snapshot(
+        "standalone",
+        "default",
+        _provider_batch(status=None, endpoint=None),
+        1,
+        observed,
+    )
+    assert snapshot["batch_status"] == "unknown"
+    assert snapshot["batch_endpoint"] is None
+    assert snapshot["terminal_at"] is None
+
+
+def test_official_openai_statuses_and_endpoints_normalize_deterministically() -> None:
+    """Accept the currently documented OpenAI Batch status and endpoint sets."""
+    statuses = (
+        "validating",
+        "failed",
+        "in_progress",
+        "finalizing",
+        "completed",
+        "expired",
+        "cancelling",
+        "cancelled",
+    )
+    endpoints = (
+        "/v1/responses",
+        "/v1/chat/completions",
+        "/v1/embeddings",
+        "/v1/completions",
+        "/v1/moderations",
+    )
+    observed = datetime(2026, 8, 13, tzinfo=timezone.utc)
+
+    for index, status in enumerate(statuses, start=1):
+        endpoint = endpoints[(index - 1) % len(endpoints)]
+        snapshot, _ = db._normalize_remote_batch_snapshot(
+            "standalone",
+            "default",
+            _provider_batch(status=status, endpoint=endpoint),
+            index,
+            observed,
+        )
+        assert snapshot["batch_status"] == status
+        assert snapshot["batch_endpoint"] == endpoint
+        assert (snapshot["terminal_at"] is observed) is (
+            status in {"failed", "completed", "expired", "cancelled"}
+        )

tests/test_remote_batch_state_contracts.py

@@ -428,27 +428,27 @@ def recorder(
     assert chr(0) not in repr(recorded)
 
 
-def test_remote_field_contract_normalizes_nul_optional_text(
+def test_remote_field_contract_rejects_nul_lifecycle_text_before_database_access(
     monkeypatch: Any,
 ) -> None:
-    """NUL-bearing descriptive provider text cannot reach PostgreSQL columns."""
+    """NUL-bearing lifecycle status fails before PostgreSQL persistence."""
     driver = _Psycopg()
     monkeypatch.setattr(db, "psycopg", driver)
 
-    snapshot = db.persist_remote_batch_state(
-        "postgresql://example",
-        "primary",
-        {
-            "id": "batch-1",
-            "endpoint": f"/v1/responses{chr(0)}shadow",
-            "status": f"completed{chr(0)}shadow",
-        },
-        observation_order=6,
-    )
+    with pytest.raises(ValueError, match="batch_status"):
+        db.persist_remote_batch_state(
+            "postgresql://example",
+            "primary",
+            {
+                "id": "batch-1",
+                "endpoint": f"/v1/responses{chr(0)}shadow",
+                "status": f"completed{chr(0)}shadow",
+            },
+            observation_order=6,
+        )
 
-    assert snapshot["batch_endpoint"] is None
-    assert snapshot["batch_status"] == "unknown"
-    assert chr(0) not in repr(driver.executions[1][1])
+    assert driver.connections == []
+    assert driver.executions == []
 
 
 def test_remote_field_contract_adds_database_checks() -> None: