From 157fce3074e9a4d0feabef7c5e2ad726e5bea4af Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 13:06:44 +0900 Subject: [PATCH 1/4] feat(reconcile): reconstruct bounded reconciliation on current main --- pg_llm_batch/reconciliation.py | 209 +++++++++++++++++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 pg_llm_batch/reconciliation.py diff --git a/pg_llm_batch/reconciliation.py b/pg_llm_batch/reconciliation.py new file mode 100644 index 000000000..fe147dbdd --- /dev/null +++ b/pg_llm_batch/reconciliation.py @@ -0,0 +1,209 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) ContextualWisdomLab. +"""Bounded scheduler-independent reconciliation through validated Batch API clients.""" + +from __future__ import annotations + +from dataclasses import dataclass +from itertools import islice +from typing import Any, Iterable, Mapping, Protocol + +from .db import validate_endpoint_alias, validate_remote_resource_id +from .exceptions import GatewayError, ValidationError + +MAX_RECONCILIATION_JOBS = 100 +MAX_RECONCILIATION_CANDIDATES = 400 +_PUBLIC_BATCH_STATUSES = { + "validating": "validating", + "failed": "failed", + "in_progress": "in_progress", + "finalizing": "finalizing", + "completed": "completed", + "expired": "expired", + "cancelling": "cancelling", + "cancelled": "cancelled", +} +_ERROR_TYPES = { + GatewayError: "GatewayError", + ValidationError: "ValidationError", +} + + +class ReconciliationClient(Protocol): + """Minimal validated provider client surface consumed by one reconciliation pass.""" + + async def get_batch_status( + self, + batch_id: str, + endpoint_alias: str, + ) -> Mapping[str, Any]: + """Return one validated remote batch status snapshot.""" + ... + + async def download_results( + self, + batch_id: str, + endpoint_alias: str, + ) -> Mapping[str, Any]: + """Retrieve one terminal batch through the existing bounded client path.""" + ... + + +@dataclass(frozen=True) +class ReconciliationCandidate: + """One host-selected provider lifecycle identity eligible for reconciliation.""" + + endpoint_alias: str + remote_batch_id: str + + +@dataclass(frozen=True) +class ReconciliationOutcome: + """Payload-free bounded evidence for one attempted reconciliation candidate.""" + + outcome: str + batch_status: str | None = None + error_type: str | None = None + + +@dataclass(frozen=True) +class ReconciliationReport: + """Aggregate payload-free evidence from one finite reconciliation pass.""" + + processed_count: int + retrieved_count: int + failed_count: int + outcomes: tuple[ReconciliationOutcome, ...] + + +def _validate_work_budget(max_jobs: Any) -> int: + """Return one strict positive bounded per-pass provider-work budget.""" + if ( + type(max_jobs) is not int + or max_jobs < 1 + or max_jobs > MAX_RECONCILIATION_JOBS + ): + raise ValidationError( + field="max_jobs", + value="", + reason="must be an integer within the supported reconciliation budget", + message="Reconciliation work budget is invalid", + ) + return max_jobs + + +def _validate_candidate(candidate: ReconciliationCandidate) -> ReconciliationCandidate: + """Validate one selected provider identity without reflecting rejected content.""" + try: + endpoint_alias = validate_endpoint_alias(candidate.endpoint_alias) + remote_batch_id = validate_remote_resource_id( + candidate.remote_batch_id, + "remote_batch_id", + ) + except (AttributeError, ValidationError): + raise ValidationError( + field="reconciliation_candidate", + value="", + reason="must contain a valid endpoint alias and remote batch identifier", + message="Reconciliation candidate identity is invalid", + ) from None + return ReconciliationCandidate(endpoint_alias, remote_batch_id) + + +def _select_candidates( + candidates: Iterable[ReconciliationCandidate], + *, + max_jobs: int, +) -> tuple[ReconciliationCandidate, ...]: + """Return unique work or fail closed when the bounded candidate scan saturates.""" + selected: list[ReconciliationCandidate] = [] + seen: set[tuple[str, str]] = set() + candidate_iterator = iter(candidates) + for candidate in islice(candidate_iterator, MAX_RECONCILIATION_CANDIDATES): + validated = _validate_candidate(candidate) + identity = (validated.endpoint_alias, validated.remote_batch_id) + if identity in seen: + continue + seen.add(identity) + selected.append(validated) + if len(selected) == max_jobs: + return tuple(selected) + + try: + next(candidate_iterator) + except StopIteration: + return tuple(selected) + + raise ValidationError( + field="reconciliation_candidates", + value="", + reason="candidate scan exceeded the bounded reconciliation limit", + message="Reconciliation candidate scan is saturated", + ) + + +def _bounded_error_type(error: Exception) -> str: + """Map one ordinary failure to a finite type vocabulary without dynamic names.""" + return _ERROR_TYPES.get(type(error), "_OTHER") + + +async def reconcile_batch_candidates( + client: ReconciliationClient, + candidates: Iterable[ReconciliationCandidate], + *, + max_jobs: int, +) -> ReconciliationReport: + """Poll and retrieve a finite set of provider jobs without retaining payloads. + + The host owns candidate discovery, scheduling, tenant authorization, and any + cross-process lease. This primitive validates the selected identities before + the first provider operation, executes only through the supplied validated + Batch API client surface, and returns bounded status/error categories rather + than provider content or dynamic exception diagnostics. + """ + work_budget = _validate_work_budget(max_jobs) + selected = _select_candidates(candidates, max_jobs=work_budget) + outcomes: list[ReconciliationOutcome] = [] + retrieved_count = 0 + failed_count = 0 + + for candidate in selected: + try: + status = await client.get_batch_status( + candidate.remote_batch_id, + candidate.endpoint_alias, + ) + batch_status = _PUBLIC_BATCH_STATUSES.get(status.get("status"), "_OTHER") + if status.get("is_complete") is True: + retrieval = await client.download_results( + candidate.remote_batch_id, + candidate.endpoint_alias, + ) + retrieval_succeeded = retrieval.get("success") is True + outcome = {True: "retrieved", False: "deferred"}[ + retrieval_succeeded + ] + retrieved_count += int(retrieval_succeeded) + else: + outcome = "polled" + outcomes.append( + ReconciliationOutcome( + outcome=outcome, + batch_status=batch_status, + ) + ) + except Exception as error: + failed_count += 1 + outcomes.append( + ReconciliationOutcome( + outcome="failed", + error_type=_bounded_error_type(error), + ) + ) + + return ReconciliationReport( + processed_count=len(outcomes), + retrieved_count=retrieved_count, + failed_count=failed_count, + outcomes=tuple(outcomes), + ) From fcf486136fe72b98b3ef44f2eaab43f957feb552 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 13:07:04 +0900 Subject: [PATCH 2/4] test(reconcile): restore exact bounded reconciliation contract --- tests/test_provider_reconciliation_worker.py | 190 +++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 tests/test_provider_reconciliation_worker.py diff --git a/tests/test_provider_reconciliation_worker.py b/tests/test_provider_reconciliation_worker.py new file mode 100644 index 000000000..e26bec882 --- /dev/null +++ b/tests/test_provider_reconciliation_worker.py @@ -0,0 +1,190 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Test-first contract for bounded scheduler-independent provider reconciliation.""" + +from __future__ import annotations + +from dataclasses import asdict + +import pytest + +from pg_llm_batch.reconciliation import ( + MAX_RECONCILIATION_JOBS, + ReconciliationCandidate, + reconcile_batch_candidates, +) + + +class FakeClient: + """Minimal async Batch API seam used to observe reconciliation behavior.""" + + def __init__(self, statuses, *, failure_batch_id: str | None = None) -> None: + self.statuses = statuses + self.failure_batch_id = failure_batch_id + self.status_calls: list[tuple[str, str]] = [] + self.download_calls: list[tuple[str, str]] = [] + + async def get_batch_status(self, batch_id: str, endpoint_alias: str): + """Return one configured status or raise a confidential synthetic failure.""" + self.status_calls.append((batch_id, endpoint_alias)) + if batch_id == self.failure_batch_id: + raise SecretNamedProviderError("provider-secret-sentinel") + return dict(self.statuses[batch_id]) + + async def download_results(self, batch_id: str, endpoint_alias: str): + """Return content-bearing data that must not escape the worker report.""" + self.download_calls.append((batch_id, endpoint_alias)) + return { + "success": True, + "responses": [{"private": "provider-payload-sentinel"}], + "errors": [], + "response_count": 1, + "error_count": 0, + } + + +class SecretNamedProviderError(Exception): + """Synthetic caller/provider exception whose type and message are private.""" + + +@pytest.mark.asyncio +async def test_reconciliation_bounds_unique_work_and_discards_provider_payloads() -> None: + """One pass processes only the bounded unique prefix and returns no payloads.""" + client = FakeClient( + { + "batch-a": { + "status": "completed", + "is_complete": True, + "output_file_id": "file-a", + }, + "batch-b": {"status": "in_progress", "is_complete": False}, + "batch-c": {"status": "in_progress", "is_complete": False}, + } + ) + candidates = [ + ReconciliationCandidate("default", "batch-a"), + ReconciliationCandidate("default", "batch-a"), + ReconciliationCandidate("backup", "batch-b"), + ReconciliationCandidate("backup", "batch-c"), + ] + + report = await reconcile_batch_candidates(client, candidates, max_jobs=2) + + assert client.status_calls == [("batch-a", "default"), ("batch-b", "backup")] + assert client.download_calls == [("batch-a", "default")] + assert report.processed_count == 2 + assert report.retrieved_count == 1 + assert report.failed_count == 0 + assert [outcome.outcome for outcome in report.outcomes] == ["retrieved", "polled"] + assert [outcome.batch_status for outcome in report.outcomes] == [ + "completed", + "in_progress", + ] + assert "provider-payload-sentinel" not in repr(report) + assert all("response" not in asdict(outcome) for outcome in report.outcomes) + + +@pytest.mark.asyncio +async def test_reconciliation_validates_selected_candidates_before_provider_io() -> None: + """Malformed selected identity must fail before any provider operation begins.""" + client = FakeClient({"batch-a": {"status": "in_progress", "is_complete": False}}) + candidates = [ + ReconciliationCandidate("default", "batch-a"), + ReconciliationCandidate("default", "bad id provider-secret-sentinel"), + ] + + with pytest.raises(Exception) as exc_info: + await reconcile_batch_candidates(client, candidates, max_jobs=2) + + assert client.status_calls == [] + assert client.download_calls == [] + assert "provider-secret-sentinel" not in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_reconciliation_isolates_one_failure_with_finite_error_evidence() -> None: + """One provider failure does not block later work or export dynamic diagnostics.""" + client = FakeClient( + { + "batch-b": { + "status": "completed", + "is_complete": True, + "error_file_id": "file-error-b", + } + }, + failure_batch_id="batch-a", + ) + candidates = [ + ReconciliationCandidate("default", "batch-a"), + ReconciliationCandidate("backup", "batch-b"), + ] + + report = await reconcile_batch_candidates(client, candidates, max_jobs=2) + + assert client.status_calls == [("batch-a", "default"), ("batch-b", "backup")] + assert client.download_calls == [("batch-b", "backup")] + assert report.processed_count == 2 + assert report.retrieved_count == 1 + assert report.failed_count == 1 + assert report.outcomes[0].outcome == "failed" + assert report.outcomes[0].error_type == "_OTHER" + assert report.outcomes[1].outcome == "retrieved" + assert "SecretNamedProviderError" not in repr(report) + assert "provider-secret-sentinel" not in repr(report) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("max_jobs", [True, False, 0, -1, MAX_RECONCILIATION_JOBS + 1]) +async def test_reconciliation_rejects_invalid_work_budget_before_provider_io(max_jobs) -> None: + """The finite per-run work budget is a strict local authority boundary.""" + client = FakeClient({"batch-a": {"status": "in_progress", "is_complete": False}}) + + with pytest.raises(Exception): + await reconcile_batch_candidates( + client, + [ReconciliationCandidate("default", "batch-a")], + max_jobs=max_jobs, + ) + + assert client.status_calls == [] + assert client.download_calls == [] + + +@pytest.mark.asyncio +async def test_reconciliation_allows_candidate_source_to_exhaust_before_work_budget() -> None: + """A short finite candidate source returns a consistent partial-work report.""" + client = FakeClient({"batch-a": {"status": "in_progress", "is_complete": False}}) + + report = await reconcile_batch_candidates( + client, + [ReconciliationCandidate("default", "batch-a")], + max_jobs=2, + ) + + assert client.status_calls == [("batch-a", "default")] + assert client.download_calls == [] + assert report.processed_count == 1 + assert report.retrieved_count == 0 + assert report.failed_count == 0 + assert [outcome.outcome for outcome in report.outcomes] == ["polled"] + + +@pytest.mark.asyncio +async def test_reconciliation_fails_closed_when_candidate_scan_truncates_unique_work() -> None: + """Candidate scan saturation cannot silently hide later unique provider work.""" + client = FakeClient( + { + "batch-a": {"status": "in_progress", "is_complete": False}, + "batch-b": {"status": "in_progress", "is_complete": False}, + } + ) + candidates = [ReconciliationCandidate("default", "batch-a")] * 400 + [ + ReconciliationCandidate("backup", "batch-b") + ] + + with pytest.raises(Exception) as exc_info: + await reconcile_batch_candidates(client, candidates, max_jobs=2) + + assert client.status_calls == [] + assert client.download_calls == [] + assert "batch-a" not in str(exc_info.value) + assert "batch-b" not in str(exc_info.value) From 5bb9048d00bf875592e471a82d071a3221a80549 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 14:02:15 +0900 Subject: [PATCH 3/4] test(reconcile): cover finite outcome vocabulary --- tests/test_provider_reconciliation_worker.py | 85 ++++++++++++++++++-- 1 file changed, 80 insertions(+), 5 deletions(-) diff --git a/tests/test_provider_reconciliation_worker.py b/tests/test_provider_reconciliation_worker.py index e26bec882..f35c7ded9 100644 --- a/tests/test_provider_reconciliation_worker.py +++ b/tests/test_provider_reconciliation_worker.py @@ -7,7 +7,9 @@ import pytest +from pg_llm_batch.exceptions import GatewayError, ValidationError from pg_llm_batch.reconciliation import ( + MAX_RECONCILIATION_CANDIDATES, MAX_RECONCILIATION_JOBS, ReconciliationCandidate, reconcile_batch_candidates, @@ -17,15 +19,26 @@ class FakeClient: """Minimal async Batch API seam used to observe reconciliation behavior.""" - def __init__(self, statuses, *, failure_batch_id: str | None = None) -> None: + def __init__( + self, + statuses, + *, + failure_batch_id: str | None = None, + status_errors=None, + download_success: bool = True, + ) -> None: self.statuses = statuses self.failure_batch_id = failure_batch_id + self.status_errors = status_errors or {} + self.download_success = download_success self.status_calls: list[tuple[str, str]] = [] self.download_calls: list[tuple[str, str]] = [] async def get_batch_status(self, batch_id: str, endpoint_alias: str): """Return one configured status or raise a confidential synthetic failure.""" self.status_calls.append((batch_id, endpoint_alias)) + if batch_id in self.status_errors: + raise self.status_errors[batch_id] if batch_id == self.failure_batch_id: raise SecretNamedProviderError("provider-secret-sentinel") return dict(self.statuses[batch_id]) @@ -34,7 +47,7 @@ async def download_results(self, batch_id: str, endpoint_alias: str): """Return content-bearing data that must not escape the worker report.""" self.download_calls.append((batch_id, endpoint_alias)) return { - "success": True, + "success": self.download_success, "responses": [{"private": "provider-payload-sentinel"}], "errors": [], "response_count": 1, @@ -132,6 +145,68 @@ async def test_reconciliation_isolates_one_failure_with_finite_error_evidence() assert "provider-secret-sentinel" not in repr(report) +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("error", "expected_error_type"), + [ + (GatewayError("provider-secret-sentinel"), "GatewayError"), + ( + ValidationError( + field="provider", + value="", + reason="provider-secret-sentinel", + ), + "ValidationError", + ), + ], +) +async def test_reconciliation_maps_domain_failures_to_finite_error_types( + error: Exception, + expected_error_type: str, +) -> None: + """Known domain failures expose only their finite public error categories.""" + client = FakeClient({}, status_errors={"batch-a": error}) + + report = await reconcile_batch_candidates( + client, + [ReconciliationCandidate("default", "batch-a")], + max_jobs=1, + ) + + assert report.processed_count == 1 + assert report.retrieved_count == 0 + assert report.failed_count == 1 + assert report.outcomes[0].outcome == "failed" + assert report.outcomes[0].error_type == expected_error_type + assert report.outcomes[0].batch_status is None + assert "provider-secret-sentinel" not in repr(report) + + +@pytest.mark.asyncio +async def test_reconciliation_defers_unsuccessful_download_and_bounds_unknown_status() -> None: + """Incomplete retrieval and unknown provider status remain bounded report evidence.""" + client = FakeClient( + {"batch-a": {"status": "provider-secret-sentinel", "is_complete": True}}, + download_success=False, + ) + + report = await reconcile_batch_candidates( + client, + [ReconciliationCandidate("default", "batch-a")], + max_jobs=1, + ) + + assert client.status_calls == [("batch-a", "default")] + assert client.download_calls == [("batch-a", "default")] + assert report.processed_count == 1 + assert report.retrieved_count == 0 + assert report.failed_count == 0 + assert report.outcomes[0].outcome == "deferred" + assert report.outcomes[0].batch_status == "_OTHER" + assert report.outcomes[0].error_type is None + assert "provider-secret-sentinel" not in repr(report) + + @pytest.mark.asyncio @pytest.mark.parametrize("max_jobs", [True, False, 0, -1, MAX_RECONCILIATION_JOBS + 1]) async def test_reconciliation_rejects_invalid_work_budget_before_provider_io(max_jobs) -> None: @@ -177,9 +252,9 @@ async def test_reconciliation_fails_closed_when_candidate_scan_truncates_unique_ "batch-b": {"status": "in_progress", "is_complete": False}, } ) - candidates = [ReconciliationCandidate("default", "batch-a")] * 400 + [ - ReconciliationCandidate("backup", "batch-b") - ] + candidates = [ + ReconciliationCandidate("default", "batch-a") + ] * MAX_RECONCILIATION_CANDIDATES + [ReconciliationCandidate("backup", "batch-b")] with pytest.raises(Exception) as exc_info: await reconcile_batch_candidates(client, candidates, max_jobs=2) From a30d923aa39f6df9c4d9018dd86acc170ab43833 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Fri, 14 Aug 2026 14:02:47 +0900 Subject: [PATCH 4/4] refactor(reconcile): simplify retrieval outcome mapping --- pg_llm_batch/reconciliation.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pg_llm_batch/reconciliation.py b/pg_llm_batch/reconciliation.py index fe147dbdd..04140418f 100644 --- a/pg_llm_batch/reconciliation.py +++ b/pg_llm_batch/reconciliation.py @@ -180,9 +180,7 @@ async def reconcile_batch_candidates( candidate.endpoint_alias, ) retrieval_succeeded = retrieval.get("success") is True - outcome = {True: "retrieved", False: "deferred"}[ - retrieval_succeeded - ] + outcome = "retrieved" if retrieval_succeeded else "deferred" retrieved_count += int(retrieval_succeeded) else: outcome = "polled"