diff --git a/.env.example b/.env.example index 93ed800..6f11136 100644 --- a/.env.example +++ b/.env.example @@ -10,7 +10,7 @@ PARAGRAPH_SUMMARY_SERVICE_URL=http://127.0.0.1:8001 # Paragraph Summary Service Configuration (if running locally) SUMMARY_SERVICE_PROVIDER=mock SUMMARY_SERVICE_ENABLE_PROVIDER_CALLS=false -SUMMARY_SERVICE_MODEL=gemini-2.5-flash +SUMMARY_SERVICE_MODEL=gemini-3.1-flash-lite SUMMARY_TEMPLATE_VERSION=paragraph_one_sentence_v1 SUMMARY_LANE_COUNT=10 diff --git a/backend/tests/test_remote_summary.py b/backend/tests/test_remote_summary.py index 7db932a..6f431d1 100644 --- a/backend/tests/test_remote_summary.py +++ b/backend/tests/test_remote_summary.py @@ -74,7 +74,7 @@ def get_job_artifact(self, job_id): "summary_text": "" if failed else f"Summary for {record['record_id']}", "summary_style": "one_sentence", "provider": "error" if failed else self.provider, - "model": "gemini-2.5-flash" if self.provider == "gemini" else "mock-deterministic-v1", + "model": "gemini-3.1-flash-lite" if self.provider == "gemini" else "mock-deterministic-v1", "template_version": "v1", "status": "failed" if failed else "completed", "error_code": "provider_rate_limited" if failed else None, diff --git a/docs/GEMINI_PROVIDER_VALIDATION.md b/docs/GEMINI_PROVIDER_VALIDATION.md index f405746..9ba2a59 100644 --- a/docs/GEMINI_PROVIDER_VALIDATION.md +++ b/docs/GEMINI_PROVIDER_VALIDATION.md @@ -4,7 +4,7 @@ DeepReader v0.6 validates the existing PDF-to-paragraph summary pipeline against the real Gemini API. It does not add a new retrieval or QA model: Gemini only creates one-sentence paragraph summaries inside `paragraph-summary-service`. DeepReader still persists the original source text, validates imported artifacts against document ID, stable ID, and source hash, and uses original source text for citation-grade QA evidence. -> **Model-name note**: The persistent service default is `gemini-2.5-flash` (see `.env.example` and `config.py`). The Makefile canary targets and some validation scripts override this to `gemini-3.1-flash-lite` as a manual canary override. The `.env.local` example below shows the persistent default. +> **Model-name note**: The persistent service default is `gemini-3.1-flash-lite` (see `.env.example` and `config.py`). The `.env.local` example below shows the persistent default. The v0.5 behavior remains the default. `SUMMARY_SERVICE_PROVIDER=mock` requires no API key and makes no external request. Gemini requests are possible only when the provider is explicitly `gemini` and `SUMMARY_SERVICE_ENABLE_PROVIDER_CALLS=true`. @@ -31,7 +31,7 @@ PARAGRAPH_SUMMARY_SERVICE_URL=http://127.0.0.1:8001 SUMMARY_SERVICE_PROVIDER=gemini SUMMARY_SERVICE_ENABLE_PROVIDER_CALLS=true -SUMMARY_SERVICE_MODEL=gemini-2.5-flash +SUMMARY_SERVICE_MODEL=gemini-3.1-flash-lite SUMMARY_LANE_COUNT=1 SUMMARY_LANE_RPM=4 diff --git a/docs/validation-log.md b/docs/validation-log.md index 33923a4..00508dc 100644 --- a/docs/validation-log.md +++ b/docs/validation-log.md @@ -1,5 +1,47 @@ # DeepReader Validation Log +## Tag: v0.4-adaptive-large-book-scheduler (2026-07-04) + +* **Commits:** `12d34a4` Add dynamic provider call cap planning → `2730938` Wire dynamic provider call cap → `5e7fe22` Add large-book lane tuning → `df8d0c2` Consolidate provider call cap planning → `b992578` Add cleanup plan router → `5a3cddb` Extract artifact validator helper → `7862817` Document artifact validator validation +* **Backend Tests:** Full paragraph-summary-service suite: 154 passed, 1 warning in 7.94s (offline, mock provider only) +* **Warning:** Pre-existing `StarletteDeprecationWarning` from `fastapi/testclient.py` (httpx/starlette.testclient), unrelated to v0.4 work. +* **Completed Scope:** + * Dynamic provider-call cap planning (`app/scheduler/cap_planner.py`): pure `estimate_provider_call_cap` / `resolve_effective_provider_call_cap` / `plan_provider_call_cap` helpers sized off the actual packed batch count, not a record-count estimate. + * Dispatcher wiring: `_run_job_background` resolves the effective cap through `plan_provider_call_cap` and records it (`dynamic_provider_call_cap_enabled`, `dynamic_provider_call_cap_value`, `hard_max_provider_calls_per_job`) in `job.stats`. + * Provider-call cap source-label consolidation: a single `ProviderCallCapPlan` dataclass bundles the enforced cap and its `provider_call_cap_source` telemetry label (`static` / `dynamic` / `*_hard_ceiling`) so the two can never drift apart. + * Large-book lane tuning (`app/scheduler/lane_tuning.py`): when `SUMMARY_LARGE_BOOK_MODE` is enabled and the record count meets `SUMMARY_LARGE_BOOK_RECORD_THRESHOLD`, adaptive-RPM upshift is pinned off (`adaptive_rpm_max` = base lane RPM) while rate-limit downshift and streak telemetry are preserved. + * Cleanup plan router (`app/scheduler/cleanup_router.py`): classifies failed/skipped artifact lines into cleanup classes and reports a counts-only plan (`job.stats["cleanup_plan"]`) — no record IDs, stable IDs, or source text included. + * Artifact validator extraction (`app/scheduler/artifact_validator.py`): per-record integrity/content checks pulled out of `dispatcher._validate_results` into a pure, independently tested module; behavior-preserving. + * All new config knobs (`SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_*`, `SUMMARY_HARD_MAX_PROVIDER_CALLS_PER_JOB`, `SUMMARY_LARGE_BOOK_*`) default off/unset and are surfaced in `safe_summary()`. +* **Reviewed:** Final branch review completed against `main` (full diff, dispatcher/lane wiring, `job.stats` shape, cleanup-plan counts-only guarantee, config validation). No blockers found; no code changes were required. +* **Rollout / tuning note:** Dynamic provider-call caps are opt-in and default-off. The default `safety_factor=2.0` fits the large-book case the feature targets, but at low batch counts it can sit below the theoretical worst-case per-batch retry/rate-limit claim consumption. Operators enabling this outside large-book mode should raise `SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_SAFETY_FACTOR` and/or `SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_MARGIN`, or scope it to large-book jobs where the higher batch count makes the 2x budget statistically ample. +* **Explicitly Deferred / Not Run:** + * No live Gemini/provider calls were made; validation is offline/static/unit only (mock provider, pure-function unit tests, and dispatcher integration tests). + * No services were started. + * OpenStax was not touched. + * `.env.local` / secrets were not inspected or exposed. + +## 2026-07-04 — Extract artifact validator helper (v0.4 adaptive large-book scheduler) + +Commit: 5a3cddb `Extract artifact validator helper` + +Behavior: +- Extracted per-record provider-result checks out of `dispatcher.py` into a new pure module, `app/scheduler/artifact_validator.py`. +- `check_record_integrity` covers source-hash/stable-id mismatches and unrecognized result statuses. +- `check_completed_summary` covers empty summaries and the Gemini one-sentence schema check. +- `dispatcher._validate_results` now calls these helpers instead of inlining the checks; control flow and error codes/messages are unchanged. +- Refactor only; behavior-preserving. + +Validated: +- Full paragraph-summary-service suite: 154 passed, 1 warning in 8.03s. +- Warning is a pre-existing `StarletteDeprecationWarning` from `fastapi/testclient.py` (httpx/starlette.testclient), unrelated to this change. +- New `tests/test_artifact_validator.py` added covering the extracted helpers directly. + +Notes: +- No API, schema, or persistent config changes. +- No live Gemini/provider calls made. +- OpenStax was not touched. + ## Tag: v0.8-demo-assets-polish (2026-07-02) * **Commit:** `c496a0b` (points to `c496a0b Align screenshot guide and README with demo screenshot assets`) diff --git a/services/paragraph-summary-service/app/config.py b/services/paragraph-summary-service/app/config.py index 670ed2a..d601059 100644 --- a/services/paragraph-summary-service/app/config.py +++ b/services/paragraph-summary-service/app/config.py @@ -22,7 +22,7 @@ class Settings(BaseModel): default_factory=lambda: _env_bool("SUMMARY_SERVICE_ENABLE_PROVIDER_CALLS") ) summary_service_model: str = Field( - default_factory=lambda: os.getenv("SUMMARY_SERVICE_MODEL", "gemini-2.5-flash").strip() + default_factory=lambda: os.getenv("SUMMARY_SERVICE_MODEL", "gemini-3.1-flash-lite").strip() ) summary_template_version: str = Field( default_factory=lambda: os.getenv( @@ -58,6 +58,20 @@ class Settings(BaseModel): summary_max_input_tokens_per_job: int = Field( default_factory=lambda: int(os.getenv("SUMMARY_MAX_INPUT_TOKENS_PER_JOB") or "0") ) + summary_dynamic_provider_call_cap_enabled: bool = Field( + default_factory=lambda: _env_bool("SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_ENABLED") + ) + summary_dynamic_provider_call_cap_safety_factor: float = Field( + default_factory=lambda: float( + os.getenv("SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_SAFETY_FACTOR", "2.0") + ) + ) + summary_dynamic_provider_call_cap_margin: int = Field( + default_factory=lambda: int(os.getenv("SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_MARGIN", "0")) + ) + summary_hard_max_provider_calls_per_job: int = Field( + default_factory=lambda: int(os.getenv("SUMMARY_HARD_MAX_PROVIDER_CALLS_PER_JOB") or "0") + ) summary_provider_rate_limit_cooldown_seconds: float = Field( default_factory=lambda: float( os.getenv("SUMMARY_PROVIDER_RATE_LIMIT_COOLDOWN_SECONDS", "60") @@ -81,6 +95,12 @@ class Settings(BaseModel): summary_mock_provider_delay_ms: int = Field( default_factory=lambda: int(os.getenv("SUMMARY_MOCK_PROVIDER_DELAY_MS", "0")) ) + summary_large_book_mode: bool = Field( + default_factory=lambda: _env_bool("SUMMARY_LARGE_BOOK_MODE") + ) + summary_large_book_record_threshold: int = Field( + default_factory=lambda: int(os.getenv("SUMMARY_LARGE_BOOK_RECORD_THRESHOLD", "1000")) + ) def lane_credential_env_names(self) -> list[str]: """Return the configured lane variable names without reading key values.""" @@ -103,6 +123,12 @@ def safe_summary(self) -> dict[str, str | int | float | bool]: "batch_max_records": self.summary_batch_max_records, "max_provider_calls_per_job": self.summary_max_provider_calls_per_job, "max_input_tokens_per_job": self.summary_max_input_tokens_per_job, + "dynamic_provider_call_cap_enabled": self.summary_dynamic_provider_call_cap_enabled, + "dynamic_provider_call_cap_safety_factor": ( + self.summary_dynamic_provider_call_cap_safety_factor + ), + "dynamic_provider_call_cap_margin": self.summary_dynamic_provider_call_cap_margin, + "hard_max_provider_calls_per_job": self.summary_hard_max_provider_calls_per_job, "provider_rate_limit_cooldown_seconds": ( self.summary_provider_rate_limit_cooldown_seconds ), @@ -111,6 +137,8 @@ def safe_summary(self) -> dict[str, str | int | float | bool]: "adaptive_rpm_success_threshold": self.summary_adaptive_rpm_success_threshold, "adaptive_rpm_max": self.summary_adaptive_rpm_max, "adaptive_rpm_backoff_factor": self.summary_adaptive_rpm_backoff_factor, + "large_book_mode": self.summary_large_book_mode, + "large_book_record_threshold": self.summary_large_book_record_threshold, } diff --git a/services/paragraph-summary-service/app/scheduler/artifact_validator.py b/services/paragraph-summary-service/app/scheduler/artifact_validator.py new file mode 100644 index 0000000..b69988d --- /dev/null +++ b/services/paragraph-summary-service/app/scheduler/artifact_validator.py @@ -0,0 +1,55 @@ +"""Pure per-record checks for provider summary results. + +These functions judge a single expected record against its matching +provider result. Batch-level checks that only make sense across the whole +response collection (malformed collections, unknown/missing/duplicate +record IDs) stay in the dispatcher, since they have no single-record +equivalent. +""" + +from __future__ import annotations + +import re + +from app.records.schema import InputRecord, SummaryArtifactLine + +RecordCheckFailure = tuple[str, str] + + +def is_reasonable_one_sentence(summary_text: str) -> bool: + """Return whether ``summary_text`` reads as a single plain-text sentence.""" + + text = summary_text.strip() + if not text or "\n" in text or re.match(r"^(?:[-*#]|\d+[.)]\s)", text): + return False + sentence_starts = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9])", text) + return len([part for part in sentence_starts if part.strip()]) == 1 + + +def check_record_integrity( + record: InputRecord, result: SummaryArtifactLine +) -> RecordCheckFailure | None: + """Return an ``(error_code, message)`` failure if ``result`` doesn't match + ``record``'s identity or carries an unrecognized status, else ``None``.""" + + if result.source_hash != record.source_hash: + return ("source_hash_mismatch", "Provider returned a mismatched source hash") + if result.stable_id != record.stable_id: + return ("unknown_record_id", "Provider returned a mismatched stable ID") + if result.status not in {"completed", "skipped", "failed"}: + return ("schema_validation_failed", "Provider returned an invalid result status") + return None + + +def check_completed_summary(result: SummaryArtifactLine) -> RecordCheckFailure | None: + """Return an ``(error_code, message)`` failure if a completed result's + summary text fails content checks, else ``None``.""" + + if not result.summary_text.strip(): + return ("schema_validation_failed", "Provider returned an empty summary") + if result.provider == "gemini" and not is_reasonable_one_sentence(result.summary_text): + return ( + "schema_validation_failed", + "Provider summary did not satisfy the one-sentence schema", + ) + return None diff --git a/services/paragraph-summary-service/app/scheduler/cap_planner.py b/services/paragraph-summary-service/app/scheduler/cap_planner.py new file mode 100644 index 0000000..de187bb --- /dev/null +++ b/services/paragraph-summary-service/app/scheduler/cap_planner.py @@ -0,0 +1,114 @@ +"""Pure helpers for planning the per-job provider-call cap. + +These functions take the actual packed batch count (not a record-count +estimate) so that token-bound packing producing more batches than a simple +``records / max_records`` division would predict never silently +under-provisions the cap. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import math + + +def estimate_provider_call_cap( + batch_count: int, + safety_factor: float = 2.0, + margin: int = 0, +) -> int: + """Return a provider-call cap sized for retries against ``batch_count`` batches. + + ``safety_factor`` must cover the worst-case number of claims a single + batch can consume across the dispatcher's retry and rate-limit-retry + loops, not just packing slack. + """ + + if batch_count < 0: + raise ValueError("batch_count cannot be negative") + if safety_factor < 1.0: + raise ValueError("safety_factor must be at least 1.0") + if margin < 0: + raise ValueError("margin cannot be negative") + + return math.ceil(batch_count * safety_factor) + margin + + +def resolve_effective_provider_call_cap( + *, + dynamic_cap_enabled: bool, + static_cap: int, + dynamic_cap: int | None, + hard_ceiling: int, +) -> int: + """Resolve the cap actually enforced for a job. + + When ``dynamic_cap_enabled`` is true, the dynamic cap *replaces* the + static default outright — it is never combined with ``min(static, dynamic)``, + since the static default is the value that caused the cap-exhaustion + incident this planner exists to prevent. ``hard_ceiling`` of 0 (or any + non-positive value) means "no explicit ceiling configured" and is a no-op. + """ + + if dynamic_cap_enabled: + if dynamic_cap is None: + raise ValueError("dynamic_cap is required when dynamic_cap_enabled is True") + effective_cap = dynamic_cap + else: + effective_cap = static_cap + + if hard_ceiling > 0: + effective_cap = min(effective_cap, hard_ceiling) + + return effective_cap + + +@dataclass(frozen=True) +class ProviderCallCapPlan: + """The enforced provider-call cap paired with its telemetry source label. + + Bundling both together guarantees the cap the dispatcher enforces and the + ``provider_call_cap_source`` it reports can never drift apart. + """ + + effective_cap: int + dynamic_cap: int | None + source: str + + +def plan_provider_call_cap( + *, + dynamic_cap_enabled: bool, + static_cap: int, + dynamic_cap: int | None, + hard_ceiling: int, +) -> ProviderCallCapPlan: + """Plan the enforced cap and its source label from a single derivation. + + Delegates the enforced-cap arithmetic to + :func:`resolve_effective_provider_call_cap` so that resolver remains the + single authority for the value actually enforced, then derives the + telemetry source label from the same inputs. The ``*_hard_ceiling`` labels + apply only when a positive ``hard_ceiling`` sits strictly below the + pre-ceiling cap; a ceiling equal to the cap is not treated as a clamp. + """ + + effective_cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=dynamic_cap_enabled, + static_cap=static_cap, + dynamic_cap=dynamic_cap, + hard_ceiling=hard_ceiling, + ) + + pre_ceiling_cap = dynamic_cap if dynamic_cap_enabled else static_cap + clamped_by_hard_ceiling = hard_ceiling > 0 and hard_ceiling < pre_ceiling_cap + if dynamic_cap_enabled: + source = "dynamic_hard_ceiling" if clamped_by_hard_ceiling else "dynamic" + else: + source = "static_hard_ceiling" if clamped_by_hard_ceiling else "static" + + return ProviderCallCapPlan( + effective_cap=effective_cap, + dynamic_cap=dynamic_cap, + source=source, + ) diff --git a/services/paragraph-summary-service/app/scheduler/cleanup_router.py b/services/paragraph-summary-service/app/scheduler/cleanup_router.py new file mode 100644 index 0000000..540b0ad --- /dev/null +++ b/services/paragraph-summary-service/app/scheduler/cleanup_router.py @@ -0,0 +1,85 @@ +"""Pure helpers for classifying failed/skipped artifact lines into cleanup classes. + +These functions only read ``SummaryArtifactLine`` values and return plain +data -- they never dispatch work, change job status, or touch the scheduler. +The resulting plan is counts-only: it groups error codes by cleanup class so +an operator (or a future executable cleanup mode) knows what kind of recovery +is needed, without ever surfacing record IDs, stable IDs, or source text. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from app.records.schema import SummaryArtifactLine + +RATE_LIMIT_CLEANUP = "rate_limit_cleanup" +SCHEMA_CLEANUP = "schema_cleanup" +SKIPPED_RECONCILIATION = "skipped_reconciliation" +INPUT_PLANNING = "input_planning" +BUDGET_RESUME = "budget_resume" +TERMINAL = "terminal" + +_CLEANUP_CLASS_BY_ERROR_CODE: dict[str, str] = { + "provider_rate_limited": RATE_LIMIT_CLEANUP, + "schema_validation_failed": SCHEMA_CLEANUP, + "response_parse_failed": SCHEMA_CLEANUP, + "missing_record": SCHEMA_CLEANUP, + "duplicate_record_id": SCHEMA_CLEANUP, + "unknown_record_id": SCHEMA_CLEANUP, + "source_hash_mismatch": SCHEMA_CLEANUP, + "max_input_tokens_exceeded": INPUT_PLANNING, + "max_provider_calls_exceeded": BUDGET_RESUME, + "job_cancelled": SKIPPED_RECONCILIATION, + "job_stopped": SKIPPED_RECONCILIATION, +} + + +def classify_error(error_code: str | None, status: str | None = None) -> str: + """Return the cleanup class for a failed/skipped artifact line. + + ``error_code`` is checked first against the known mapping (this also + covers ``job_cancelled``/``job_stopped``). If it is unmapped or absent, + a ``status`` of ``"skipped"`` still routes to reconciliation. Anything + else -- unknown error codes, or a failure with no code at all -- falls + through to ``terminal``. + """ + + if error_code is not None: + cleanup_class = _CLEANUP_CLASS_BY_ERROR_CODE.get(error_code) + if cleanup_class is not None: + return cleanup_class + + if status == "skipped": + return SKIPPED_RECONCILIATION + + return TERMINAL + + +def build_counts_only_cleanup_plan( + lines: list["SummaryArtifactLine"], +) -> dict[str, dict[str, object]]: + """Group non-completed artifact lines into a counts-only cleanup plan. + + Returns ``{cleanup_class: {"count": N, "error_codes": {code: count}}}``. + Completed lines need no cleanup and are excluded. No record IDs, stable + IDs, or source text are ever read from the lines or included in the + result -- only ``status`` and ``error_code``. + """ + + plan: dict[str, dict[str, object]] = {} + + for line in lines: + if line.status == "completed": + continue + + cleanup_class = classify_error(line.error_code, line.status) + entry = plan.setdefault(cleanup_class, {"count": 0, "error_codes": {}}) + entry["count"] += 1 + + code_key = line.error_code or "unknown" + error_codes = entry["error_codes"] + error_codes[code_key] = error_codes.get(code_key, 0) + 1 + + return plan diff --git a/services/paragraph-summary-service/app/scheduler/dispatcher.py b/services/paragraph-summary-service/app/scheduler/dispatcher.py index 5d81698..f876905 100644 --- a/services/paragraph-summary-service/app/scheduler/dispatcher.py +++ b/services/paragraph-summary-service/app/scheduler/dispatcher.py @@ -20,7 +20,18 @@ ) from app.providers.mock import MockProvider from app.records.schema import InputRecord, SummaryArtifactLine, SummaryRequest +from app.scheduler.artifact_validator import ( + check_completed_summary, + check_record_integrity, + is_reasonable_one_sentence as _is_reasonable_one_sentence, +) +from app.scheduler.cap_planner import ( + estimate_provider_call_cap, + plan_provider_call_cap, +) +from app.scheduler.cleanup_router import build_counts_only_cleanup_plan from app.scheduler.lane import QuotaLane +from app.scheduler.lane_tuning import large_book_lane_tuning from app.scheduler.token_packer import estimate_batch_input_tokens, pack_batches LOGGER = logging.getLogger(__name__) @@ -200,6 +211,11 @@ def __init__(self, job_id: str, document_id: str, total_records: int): "provider_availability_by_alias": {}, "lane_unavailability_reasons": {}, "effective_config": settings.safe_summary(), + "dynamic_provider_call_cap_enabled": settings.summary_dynamic_provider_call_cap_enabled, + "dynamic_provider_call_cap_value": None, + "hard_max_provider_calls_per_job": settings.summary_hard_max_provider_calls_per_job, + "provider_call_cap_source": "static", + "cleanup_plan": {}, } self._provider_call_lock = asyncio.Lock() self._run_gate = asyncio.Event() @@ -340,6 +356,22 @@ def validate_provider_configuration() -> dict[str, str]: raise ProviderConfigurationError( "SUMMARY_ADAPTIVE_RPM_BACKOFF_FACTOR must be greater than 0 and less than or equal to 1" ) + if settings.summary_dynamic_provider_call_cap_safety_factor < 1.0: + raise ProviderConfigurationError( + "SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_SAFETY_FACTOR must be at least 1.0" + ) + if settings.summary_dynamic_provider_call_cap_margin < 0: + raise ProviderConfigurationError( + "SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_MARGIN cannot be negative" + ) + if settings.summary_hard_max_provider_calls_per_job < 0: + raise ProviderConfigurationError( + "SUMMARY_HARD_MAX_PROVIDER_CALLS_PER_JOB cannot be negative" + ) + if settings.summary_large_book_record_threshold < 1: + raise ProviderConfigurationError( + "SUMMARY_LARGE_BOOK_RECORD_THRESHOLD must be at least 1" + ) if provider == "mock": return {} @@ -375,10 +407,12 @@ def validate_provider_configuration() -> dict[str, str]: def _build_lanes_and_providers( provider_name: str, credentials: dict[str, str], + total_records: int = 0, ) -> tuple[list[QuotaLane], dict[str, Any]]: time_scale = 0.001 if provider_name == "mock" else 1.0 lanes: list[QuotaLane] = [] providers: dict[str, Any] = {} + lane_tuning = large_book_lane_tuning(settings, total_records) identity_items: list[tuple[str | None, str | None]] if provider_name == "gemini": @@ -402,9 +436,9 @@ def _build_lanes_and_providers( settings.summary_provider_rate_limit_cooldown_seconds ), retry_backoff_base_seconds=settings.summary_retry_backoff_base_seconds, - adaptive_rpm_enabled=settings.summary_adaptive_rpm_enabled, - adaptive_rpm_success_threshold=settings.summary_adaptive_rpm_success_threshold, - adaptive_rpm_max=settings.summary_adaptive_rpm_max, + adaptive_rpm_enabled=lane_tuning.adaptive_rpm_enabled, + adaptive_rpm_success_threshold=lane_tuning.adaptive_rpm_success_threshold, + adaptive_rpm_max=lane_tuning.adaptive_rpm_max, adaptive_rpm_backoff_factor=settings.summary_adaptive_rpm_backoff_factor, ) lanes.append(lane) @@ -453,14 +487,6 @@ def _record_lane_availability( } -def _is_reasonable_one_sentence(summary_text: str) -> bool: - text = summary_text.strip() - if not text or "\n" in text or re.match(r"^(?:[-*#]|\d+[.)]\s)", text): - return False - sentence_starts = re.split(r"(?<=[.!?])\s+(?=[A-Z0-9])", text) - return len([part for part in sentence_starts if part.strip()]) == 1 - - def _validate_results( job: JobState, current_batch: list[InputRecord], @@ -514,16 +540,9 @@ def failure(error_code: str, message: str) -> FailureDetail: continue result = results_by_id[record_id] - if result.source_hash != record.source_hash: - errors[record_id] = failure("source_hash_mismatch", "Provider returned a mismatched source hash") - retry_records.append(record) - continue - if result.stable_id != record.stable_id: - errors[record_id] = failure("unknown_record_id", "Provider returned a mismatched stable ID") - retry_records.append(record) - continue - if result.status not in {"completed", "skipped", "failed"}: - errors[record_id] = failure("schema_validation_failed", "Provider returned an invalid result status") + integrity_failure = check_record_integrity(record, result) + if integrity_failure is not None: + errors[record_id] = failure(*integrity_failure) retry_records.append(record) continue if result.status == "failed": @@ -553,21 +572,12 @@ def failure(error_code: str, message: str) -> FailureDetail: ) retry_records.append(record) continue - if result.status == "completed" and not result.summary_text.strip(): - errors[record_id] = failure("schema_validation_failed", "Provider returned an empty summary") - retry_records.append(record) - continue - if ( - result.status == "completed" - and result.provider == "gemini" - and not _is_reasonable_one_sentence(result.summary_text) - ): - errors[record_id] = failure( - "schema_validation_failed", - "Provider summary did not satisfy the one-sentence schema", - ) - retry_records.append(record) - continue + if result.status == "completed": + summary_failure = check_completed_summary(result) + if summary_failure is not None: + errors[record_id] = failure(*summary_failure) + retry_records.append(record) + continue if lane is not None: result.lane_id = result.lane_id or lane.lane_id result.provider_alias = result.provider_alias or lane.provider_alias @@ -939,7 +949,9 @@ async def _run_job_background(job: JobState, request: SummaryRequest) -> None: ) return try: - lanes, providers = _build_lanes_and_providers(provider_name, credentials) + lanes, providers = _build_lanes_and_providers( + provider_name, credentials, len(request.records) + ) except Exception as exc: LOGGER.error("Provider initialization failed (%s)", type(exc).__name__) _fail_batches( @@ -994,9 +1006,31 @@ async def _run_job_background(job: JobState, request: SummaryRequest) -> None: len(batches), settings.summary_batch_max_records, ) - maximum_calls = ( - settings.summary_max_provider_calls_per_job if provider_name == "gemini" else None - ) + if provider_name == "gemini": + dynamic_cap_enabled = settings.summary_dynamic_provider_call_cap_enabled + dynamic_cap = ( + estimate_provider_call_cap( + len(batches), + safety_factor=settings.summary_dynamic_provider_call_cap_safety_factor, + margin=settings.summary_dynamic_provider_call_cap_margin, + ) + if dynamic_cap_enabled + else None + ) + hard_ceiling = settings.summary_hard_max_provider_calls_per_job + cap_plan = plan_provider_call_cap( + dynamic_cap_enabled=dynamic_cap_enabled, + static_cap=settings.summary_max_provider_calls_per_job, + dynamic_cap=dynamic_cap, + hard_ceiling=hard_ceiling, + ) + maximum_calls = cap_plan.effective_cap + job.stats["dynamic_provider_call_cap_enabled"] = dynamic_cap_enabled + job.stats["dynamic_provider_call_cap_value"] = cap_plan.dynamic_cap + job.stats["hard_max_provider_calls_per_job"] = hard_ceiling + job.stats["provider_call_cap_source"] = cap_plan.source + else: + maximum_calls = None async def acquire_best_lane() -> QuotaLane | None: if not lanes: @@ -1219,6 +1253,8 @@ async def worker() -> None: else: job.error = "One or more provider results failed" + job.stats["cleanup_plan"] = build_counts_only_cleanup_plan(job.artifact_lines) + _record_lane_availability(job, lanes, time.monotonic()) job.touch() diff --git a/services/paragraph-summary-service/app/scheduler/lane_tuning.py b/services/paragraph-summary-service/app/scheduler/lane_tuning.py new file mode 100644 index 0000000..5edb9f3 --- /dev/null +++ b/services/paragraph-summary-service/app/scheduler/lane_tuning.py @@ -0,0 +1,38 @@ +"""Pure helpers for tuning QuotaLane construction for large-book jobs. + +These functions only compute the adaptive-RPM parameters to pass into +``QuotaLane.__init__`` -- they do not touch ``QuotaLane`` itself, and they +never disable ``record_success``'s rate_limit_streak reset or success-streak +telemetry. In large-book mode, ``adaptive_rpm_max`` is pinned to the base +lane RPM so ``current_rpm`` (which starts at the base RPM) can never be +below it, making upshift impossible while leaving telemetry intact. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class EffectiveLaneTuning: + adaptive_rpm_enabled: bool + adaptive_rpm_success_threshold: int + adaptive_rpm_max: int + + +def large_book_lane_tuning(settings, total_records: int) -> EffectiveLaneTuning: + """Return the adaptive-RPM tuning to apply when constructing a QuotaLane.""" + + if ( + settings.summary_large_book_mode + and total_records >= settings.summary_large_book_record_threshold + ): + adaptive_rpm_max = settings.summary_lane_rpm + else: + adaptive_rpm_max = settings.summary_adaptive_rpm_max + + return EffectiveLaneTuning( + adaptive_rpm_enabled=settings.summary_adaptive_rpm_enabled, + adaptive_rpm_success_threshold=settings.summary_adaptive_rpm_success_threshold, + adaptive_rpm_max=adaptive_rpm_max, + ) diff --git a/services/paragraph-summary-service/scripts/gemini_smoke_test.py b/services/paragraph-summary-service/scripts/gemini_smoke_test.py index 992c723..6f19413 100644 --- a/services/paragraph-summary-service/scripts/gemini_smoke_test.py +++ b/services/paragraph-summary-service/scripts/gemini_smoke_test.py @@ -112,7 +112,7 @@ def main() -> None: raise RuntimeError("Artifact is missing one or more smoke-test records") print(f"provider: {provider}") - print(f"model: {os.getenv('SUMMARY_SERVICE_MODEL', 'gemini-2.5-flash')}") + print(f"model: {os.getenv('SUMMARY_SERVICE_MODEL', 'gemini-3.1-flash-lite')}") print(f"configured lane cap: {os.getenv('SUMMARY_LANE_COUNT', '10')}") print(f"active provider identities: {status_payload['stats']['provider_identity_count']}") print(f"number of records: {len(records)}") diff --git a/services/paragraph-summary-service/tests/test_artifact_validator.py b/services/paragraph-summary-service/tests/test_artifact_validator.py new file mode 100644 index 0000000..09969a9 --- /dev/null +++ b/services/paragraph-summary-service/tests/test_artifact_validator.py @@ -0,0 +1,96 @@ +from datetime import datetime, timezone + +from app.records.schema import InputRecord, SummaryArtifactLine +from app.scheduler.artifact_validator import ( + check_completed_summary, + check_record_integrity, + is_reasonable_one_sentence, +) + +NOW = datetime.now(timezone.utc).isoformat() + + +def _record(**overrides) -> InputRecord: + defaults = dict(record_id="r1", stable_id="s1", text="body", source_hash="hash1") + defaults.update(overrides) + return InputRecord(**defaults) + + +def _result(**overrides) -> SummaryArtifactLine: + defaults = dict( + document_id="doc1", + record_id="r1", + stable_id="s1", + source_hash="hash1", + summary_text="A single sentence summary.", + summary_style="one_sentence", + provider="mock", + model="test", + template_version="v1", + status="completed", + created_at=NOW, + ) + defaults.update(overrides) + return SummaryArtifactLine(**defaults) + + +# --- check_record_integrity ----------------------------------------------------- + + +def test_integrity_accepts_matching_record(): + assert check_record_integrity(_record(), _result()) is None + + +def test_integrity_rejects_source_hash_mismatch(): + failure = check_record_integrity(_record(), _result(source_hash="different")) + assert failure == ("source_hash_mismatch", "Provider returned a mismatched source hash") + + +def test_integrity_rejects_stable_id_mismatch(): + failure = check_record_integrity(_record(), _result(stable_id="other")) + assert failure == ("unknown_record_id", "Provider returned a mismatched stable ID") + + +def test_integrity_rejects_unknown_status(): + failure = check_record_integrity(_record(), _result(status="bogus")) + assert failure == ("schema_validation_failed", "Provider returned an invalid result status") + + +def test_integrity_accepts_skipped_and_failed_status(): + assert check_record_integrity(_record(), _result(status="skipped")) is None + assert check_record_integrity(_record(), _result(status="failed")) is None + + +# --- check_completed_summary ----------------------------------------------------- + + +def test_completed_summary_accepts_reasonable_sentence(): + assert check_completed_summary(_result()) is None + + +def test_completed_summary_rejects_empty_text(): + failure = check_completed_summary(_result(summary_text=" ")) + assert failure == ("schema_validation_failed", "Provider returned an empty summary") + + +def test_completed_summary_rejects_multi_sentence_gemini_output(): + text = "Locke defends empiricism. Rationalists disagree sharply with this view." + failure = check_completed_summary(_result(provider="gemini", summary_text=text)) + assert failure == ( + "schema_validation_failed", + "Provider summary did not satisfy the one-sentence schema", + ) + + +def test_completed_summary_one_sentence_rule_only_applies_to_gemini(): + text = "Locke defends empiricism. Rationalists disagree sharply with this view." + assert check_completed_summary(_result(provider="mock", summary_text=text)) is None + + +# --- is_reasonable_one_sentence --------------------------------------------------- + + +def test_is_reasonable_one_sentence_matches_dispatcher_alias(): + from app.scheduler.dispatcher import _is_reasonable_one_sentence + + assert is_reasonable_one_sentence is _is_reasonable_one_sentence diff --git a/services/paragraph-summary-service/tests/test_cap_planner.py b/services/paragraph-summary-service/tests/test_cap_planner.py new file mode 100644 index 0000000..598236e --- /dev/null +++ b/services/paragraph-summary-service/tests/test_cap_planner.py @@ -0,0 +1,244 @@ +import math + +import pytest + +from app.scheduler.cap_planner import ( + ProviderCallCapPlan, + estimate_provider_call_cap, + plan_provider_call_cap, + resolve_effective_provider_call_cap, +) + + +# --- estimate_provider_call_cap ------------------------------------------------- + + +def test_formula_matches_ceil_times_factor_plus_margin(): + assert estimate_provider_call_cap(506, safety_factor=2.0, margin=10) == math.ceil(506 * 2.0) + 10 + + +def test_default_safety_factor_is_at_least_two(): + # Documents the incident-driven default: it gives retry headroom above + # one claim per batch while keeping the cap bounded; operators can raise + # it for more retry-heavy runs. + assert estimate_provider_call_cap(506) == 1012 + + +def test_cap_is_never_below_batch_count_for_valid_inputs(): + for batch_count in (0, 1, 5, 506, 5051): + for safety_factor in (1.0, 1.5, 2.0, 3.0): + for margin in (0, 10, 200): + cap = estimate_provider_call_cap(batch_count, safety_factor, margin) + assert cap >= batch_count + + +def test_safety_factor_of_one_with_zero_margin_equals_batch_count(): + assert estimate_provider_call_cap(506, safety_factor=1.0, margin=0) == 506 + + +def test_zero_margin_default(): + assert estimate_provider_call_cap(10, safety_factor=2.0) == 20 + + +def test_single_batch_edge_case(): + assert estimate_provider_call_cap(1, safety_factor=2.0, margin=5) == 7 + + +def test_zero_batches_edge_case(): + assert estimate_provider_call_cap(0, safety_factor=2.0, margin=5) == 5 + + +def test_negative_batch_count_rejected(): + with pytest.raises(ValueError): + estimate_provider_call_cap(-1) + + +def test_safety_factor_below_one_rejected(): + with pytest.raises(ValueError): + estimate_provider_call_cap(10, safety_factor=0.9) + + +def test_negative_margin_rejected(): + with pytest.raises(ValueError): + estimate_provider_call_cap(10, margin=-1) + + +# --- resolve_effective_provider_call_cap ---------------------------------------- + + +def test_dynamic_disabled_uses_static_cap_unchanged(): + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=False, + static_cap=1000, + dynamic_cap=None, + hard_ceiling=0, + ) + assert cap == 1000 + + +def test_dynamic_enabled_replaces_static_default_even_when_lower(): + # Regression guard for the min(static, dynamic) bug: a dynamic cap must + # win outright, not be capped down by the static default that caused the + # original incident. + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=1000, + dynamic_cap=810, + hard_ceiling=0, + ) + assert cap == 810 + + +def test_dynamic_enabled_replaces_static_default_even_when_higher(): + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=0, + ) + assert cap == 1012 + + +def test_hard_ceiling_unset_is_a_noop(): + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=0, + ) + assert cap == 1012 + + +def test_hard_ceiling_below_dynamic_cap_wins(): + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=500, + ) + assert cap == 500 + + +def test_hard_ceiling_above_dynamic_cap_is_a_noop(): + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=2000, + ) + assert cap == 1012 + + +def test_hard_ceiling_applies_even_when_dynamic_disabled(): + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=False, + static_cap=1000, + dynamic_cap=None, + hard_ceiling=500, + ) + assert cap == 500 + + +def test_dynamic_enabled_without_dynamic_cap_value_raises(): + with pytest.raises(ValueError): + resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=1000, + dynamic_cap=None, + hard_ceiling=0, + ) + + +def test_enabled_path_never_yields_cap_lower_than_actual_batch_count(): + # End-to-end regression across the pair of functions: for a realistic + # batch count, the dynamic path must never under-provision relative to + # the number of batches that actually need dispatching. + batch_count = 506 + dynamic_cap = estimate_provider_call_cap(batch_count) + cap = resolve_effective_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=1000, + dynamic_cap=dynamic_cap, + hard_ceiling=0, + ) + assert cap >= batch_count + + +# --- plan_provider_call_cap ----------------------------------------------------- + + +def test_plan_dynamic_disabled_labels_static(): + plan = plan_provider_call_cap( + dynamic_cap_enabled=False, + static_cap=1000, + dynamic_cap=None, + hard_ceiling=0, + ) + assert plan == ProviderCallCapPlan(effective_cap=1000, dynamic_cap=None, source="static") + + +def test_plan_dynamic_enabled_labels_dynamic(): + plan = plan_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=1000, + dynamic_cap=810, + hard_ceiling=0, + ) + assert plan == ProviderCallCapPlan(effective_cap=810, dynamic_cap=810, source="dynamic") + + +def test_plan_dynamic_enabled_with_hard_ceiling_below_dynamic_labels_dynamic_hard_ceiling(): + plan = plan_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=500, + ) + assert plan.effective_cap == 500 + assert plan.dynamic_cap == 1012 + assert plan.source == "dynamic_hard_ceiling" + + +def test_plan_dynamic_disabled_with_hard_ceiling_below_static_labels_static_hard_ceiling(): + plan = plan_provider_call_cap( + dynamic_cap_enabled=False, + static_cap=1000, + dynamic_cap=None, + hard_ceiling=500, + ) + assert plan.effective_cap == 500 + assert plan.dynamic_cap is None + assert plan.source == "static_hard_ceiling" + + +def test_plan_hard_ceiling_equal_to_cap_is_not_labeled_as_clamped(): + # Existing semantics use a strict `<` comparison, so a ceiling equal to the + # cap enforces the same value but is not reported as a hard-ceiling clamp. + dynamic_plan = plan_provider_call_cap( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=1012, + ) + assert dynamic_plan.effective_cap == 1012 + assert dynamic_plan.source == "dynamic" + + static_plan = plan_provider_call_cap( + dynamic_cap_enabled=False, + static_cap=1000, + dynamic_cap=None, + hard_ceiling=1000, + ) + assert static_plan.effective_cap == 1000 + assert static_plan.source == "static" + + +def test_plan_effective_cap_matches_resolver(): + kwargs = dict( + dynamic_cap_enabled=True, + static_cap=100, + dynamic_cap=1012, + hard_ceiling=500, + ) + plan = plan_provider_call_cap(**kwargs) + assert plan.effective_cap == resolve_effective_provider_call_cap(**kwargs) diff --git a/services/paragraph-summary-service/tests/test_cleanup_router.py b/services/paragraph-summary-service/tests/test_cleanup_router.py new file mode 100644 index 0000000..17108a1 --- /dev/null +++ b/services/paragraph-summary-service/tests/test_cleanup_router.py @@ -0,0 +1,170 @@ +import pytest + +from app.records.schema import InputRecord, SummaryArtifactLine, SummaryRequest +from app.scheduler.cleanup_router import ( + BUDGET_RESUME, + INPUT_PLANNING, + RATE_LIMIT_CLEANUP, + SCHEMA_CLEANUP, + SKIPPED_RECONCILIATION, + TERMINAL, + build_counts_only_cleanup_plan, + classify_error, +) +from app.scheduler.dispatcher import JobState, _run_job_background + + +def _line(status: str, error_code: str | None, record_id: str = "r1") -> SummaryArtifactLine: + return SummaryArtifactLine( + document_id="doc-1", + record_id=record_id, + stable_id=f"stable-{record_id}", + source_hash="hash-should-not-leak", + summary_text="some sensitive summary text" if status == "completed" else "", + summary_style="one_sentence", + provider="mock", + model="mock-model", + template_version="v1", + status=status, + error_code=error_code, + message="message text that should not appear in the plan" if error_code else None, + created_at="now", + ) + + +# --- classify_error -------------------------------------------------------- + + +@pytest.mark.parametrize( + "error_code,expected", + [ + ("provider_rate_limited", RATE_LIMIT_CLEANUP), + ("schema_validation_failed", SCHEMA_CLEANUP), + ("response_parse_failed", SCHEMA_CLEANUP), + ("missing_record", SCHEMA_CLEANUP), + ("duplicate_record_id", SCHEMA_CLEANUP), + ("unknown_record_id", SCHEMA_CLEANUP), + ("source_hash_mismatch", SCHEMA_CLEANUP), + ("max_input_tokens_exceeded", INPUT_PLANNING), + ("max_provider_calls_exceeded", BUDGET_RESUME), + ("job_cancelled", SKIPPED_RECONCILIATION), + ("job_stopped", SKIPPED_RECONCILIATION), + ], +) +def test_classify_error_maps_each_known_code_to_its_cleanup_class(error_code, expected): + assert classify_error(error_code) == expected + + +def test_classify_error_unknown_code_is_terminal(): + assert classify_error("some_never_seen_code") == TERMINAL + + +def test_classify_error_none_with_no_status_is_terminal(): + assert classify_error(None) == TERMINAL + assert classify_error(None, status="failed") == TERMINAL + + +def test_classify_error_skipped_status_without_error_code_is_reconciliation(): + assert classify_error(None, status="skipped") == SKIPPED_RECONCILIATION + + +# --- build_counts_only_cleanup_plan ----------------------------------------- + + +def test_counts_only_plan_groups_error_codes_by_cleanup_class(): + lines = [ + _line("failed", "provider_rate_limited", "r1"), + _line("failed", "provider_rate_limited", "r2"), + _line("failed", "schema_validation_failed", "r3"), + _line("failed", "response_parse_failed", "r4"), + _line("failed", "max_provider_calls_exceeded", "r5"), + _line("skipped", "job_cancelled", "r6"), + _line("skipped", None, "r7"), + _line("failed", "totally_unmapped_code", "r8"), + _line("completed", None, "r9"), + ] + + plan = build_counts_only_cleanup_plan(lines) + + assert plan[RATE_LIMIT_CLEANUP] == { + "count": 2, + "error_codes": {"provider_rate_limited": 2}, + } + assert plan[SCHEMA_CLEANUP] == { + "count": 2, + "error_codes": {"schema_validation_failed": 1, "response_parse_failed": 1}, + } + assert plan[BUDGET_RESUME] == { + "count": 1, + "error_codes": {"max_provider_calls_exceeded": 1}, + } + assert plan[SKIPPED_RECONCILIATION] == { + "count": 2, + "error_codes": {"job_cancelled": 1, "unknown": 1}, + } + assert plan[TERMINAL] == { + "count": 1, + "error_codes": {"totally_unmapped_code": 1}, + } + # Completed lines need no cleanup and contribute nothing. + assert sum(entry["count"] for entry in plan.values()) == 8 + + +def test_counts_only_plan_is_empty_for_all_completed_lines(): + lines = [_line("completed", None, "r1"), _line("completed", None, "r2")] + + assert build_counts_only_cleanup_plan(lines) == {} + + +def test_plan_never_contains_record_ids_stable_ids_or_source_text(): + lines = [ + _line("failed", "provider_rate_limited", "leaked-record-id"), + _line("skipped", "job_cancelled", "another-leaked-id"), + ] + + plan = build_counts_only_cleanup_plan(lines) + + plan_repr = repr(plan) + assert "leaked-record-id" not in plan_repr + assert "another-leaked-id" not in plan_repr + assert "stable-" not in plan_repr + assert "hash-should-not-leak" not in plan_repr + assert "message text that should not appear" not in plan_repr + for entry in plan.values(): + assert set(entry.keys()) == {"count", "error_codes"} + + +# --- dispatcher wiring: cleanup_plan lands in job.stats --------------------- + + +@pytest.mark.asyncio +async def test_clean_job_produces_empty_cleanup_plan_in_job_stats(): + request = SummaryRequest( + document_id="doc-clean", + records=[ + InputRecord(record_id="r1", text="some real paragraph text", source_hash="h1"), + InputRecord(record_id="r2", text="another real paragraph", source_hash="h2"), + ], + ) + job = JobState("job-clean", request.document_id, len(request.records)) + + await _run_job_background(job, request) + + assert job.status == "completed" + assert job.stats["cleanup_plan"] == {} + + +@pytest.mark.asyncio +async def test_job_with_skipped_records_reports_them_under_reconciliation(): + request = SummaryRequest( + document_id="doc-with-skips", + records=[ + InputRecord(record_id="r1", text="", source_hash="h1"), + InputRecord(record_id="r2", text="a normal paragraph", source_hash="h2"), + ], + ) + job = JobState("job-with-skips", request.document_id, len(request.records)) + + await _run_job_background(job, request) + + assert job.stats["cleanup_plan"][SKIPPED_RECONCILIATION]["count"] == 1 diff --git a/services/paragraph-summary-service/tests/test_dynamic_call_cap_config.py b/services/paragraph-summary-service/tests/test_dynamic_call_cap_config.py new file mode 100644 index 0000000..b10297a --- /dev/null +++ b/services/paragraph-summary-service/tests/test_dynamic_call_cap_config.py @@ -0,0 +1,262 @@ +from datetime import datetime, timezone + +import pytest + +import app.scheduler.dispatcher as dispatcher_module +from app.config import Settings, settings +from app.records.schema import InputRecord, SummaryArtifactLine, SummaryRequest +from app.scheduler.dispatcher import ( + JobState, + ProviderConfigurationError, + _run_job_background, + validate_provider_configuration, +) +from app.scheduler.lane import QuotaLane + + +def _configure_valid_gemini(monkeypatch): + monkeypatch.setattr(settings, "summary_service_provider", "gemini") + monkeypatch.setattr(settings, "summary_service_enable_provider_calls", True) + monkeypatch.setattr(settings, "summary_lane_count", 1) + monkeypatch.setenv("GEMINI_API_KEY_LANE_01", "test-key") + + +class _AlwaysSucceedsProvider: + """Minimal provider stub that never fails, so batches never retry/requeue.""" + + def __init__(self): + self.calls = 0 + + async def summarize_batch(self, document_id, records, summary_style): + self.calls += 1 + now_str = datetime.now(timezone.utc).isoformat() + return [ + SummaryArtifactLine( + document_id=document_id, + record_id=r.record_id, + source_hash=r.source_hash, + summary_text="ok", + summary_style=summary_style, + provider="mock", + model="test", + template_version="v1", + status="completed", + created_at=now_str, + ) + for r in records + ] + + +async def _run_dispatcher_wiring_job(monkeypatch, *, record_count): + """Run _run_job_background end-to-end against a stubbed lane/provider pair. + + No real Gemini calls are made: _build_lanes_and_providers is monkeypatched + to return a single in-memory lane backed by _AlwaysSucceedsProvider. + """ + + _configure_valid_gemini(monkeypatch) + + lane = QuotaLane( + "lane_01", + rpm=600_000, + time_scale=0.001, + provider_alias="gemini_01", + jitter_ratio=0, + ) + provider = _AlwaysSucceedsProvider() + monkeypatch.setattr( + dispatcher_module, + "_build_lanes_and_providers", + lambda provider_name, credentials, total_records=0: ([lane], {lane.lane_id: provider}), + ) + + records = [ + InputRecord(record_id=f"r{i}", text="safe input", source_hash=f"h{i}") + for i in range(record_count) + ] + request = SummaryRequest(document_id="doc-dynamic-cap", records=records) + job = JobState("job-dynamic-cap", request.document_id, len(records)) + + await _run_job_background(job, request) + + return job, provider + + +def test_dynamic_call_cap_knobs_default_off_and_unset(monkeypatch): + for name in ( + "SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_ENABLED", + "SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_SAFETY_FACTOR", + "SUMMARY_DYNAMIC_PROVIDER_CALL_CAP_MARGIN", + "SUMMARY_HARD_MAX_PROVIDER_CALLS_PER_JOB", + ): + monkeypatch.delenv(name, raising=False) + + fresh_settings = Settings() + + assert fresh_settings.summary_dynamic_provider_call_cap_enabled is False + assert fresh_settings.summary_dynamic_provider_call_cap_safety_factor == 2.0 + assert fresh_settings.summary_dynamic_provider_call_cap_margin == 0 + assert fresh_settings.summary_hard_max_provider_calls_per_job == 0 + + +def test_safe_summary_surfaces_dynamic_call_cap_knobs(monkeypatch): + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", True) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 3.0) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", 25) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 500) + + summary = settings.safe_summary() + + assert summary["dynamic_provider_call_cap_enabled"] is True + assert summary["dynamic_provider_call_cap_safety_factor"] == 3.0 + assert summary["dynamic_provider_call_cap_margin"] == 25 + assert summary["hard_max_provider_calls_per_job"] == 500 + + +def test_validation_passes_with_valid_dynamic_call_cap_knobs(monkeypatch): + _configure_valid_gemini(monkeypatch) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", True) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 2.0) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", 0) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + + validate_provider_configuration() # Should not raise + + +def test_validation_rejects_safety_factor_below_one(monkeypatch): + _configure_valid_gemini(monkeypatch) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 0.5) + + with pytest.raises(ProviderConfigurationError, match="SAFETY_FACTOR"): + validate_provider_configuration() + + +def test_validation_rejects_negative_margin(monkeypatch): + _configure_valid_gemini(monkeypatch) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", -1) + + with pytest.raises(ProviderConfigurationError, match="MARGIN"): + validate_provider_configuration() + + +def test_validation_rejects_negative_hard_ceiling(monkeypatch): + _configure_valid_gemini(monkeypatch) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", -1) + + with pytest.raises(ProviderConfigurationError, match="HARD_MAX_PROVIDER_CALLS_PER_JOB"): + validate_provider_configuration() + + +def test_validation_allows_unset_zero_hard_ceiling(monkeypatch): + _configure_valid_gemini(monkeypatch) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + + validate_provider_configuration() # Should not raise + + +@pytest.mark.parametrize("threshold", [0, -1]) +def test_validation_rejects_non_positive_large_book_record_threshold(monkeypatch, threshold): + _configure_valid_gemini(monkeypatch) + monkeypatch.setattr(settings, "summary_large_book_record_threshold", threshold) + + with pytest.raises(ProviderConfigurationError, match="LARGE_BOOK_RECORD_THRESHOLD"): + validate_provider_configuration() + + +# --- dispatcher wiring: dynamic cap flows from actual batches to the job ------- + + +@pytest.mark.asyncio +async def test_dynamic_cap_disabled_preserves_static_cap_behavior(monkeypatch): + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", False) + monkeypatch.setattr(settings, "summary_max_provider_calls_per_job", 1000) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + monkeypatch.setattr(settings, "summary_batch_max_records", 2) + + job, provider = await _run_dispatcher_wiring_job(monkeypatch, record_count=5) + + assert job.stats["dynamic_provider_call_cap_enabled"] is False + assert job.stats["dynamic_provider_call_cap_value"] is None + assert job.stats["provider_call_cap_source"] == "static" + assert job.completed_records == 5 + assert provider.calls == 3 # ceil(5 / batch_max_records=2) + + +@pytest.mark.asyncio +async def test_dynamic_cap_enabled_uses_actual_batch_count_not_record_count(monkeypatch): + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", True) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 2.0) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", 0) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + monkeypatch.setattr(settings, "summary_batch_max_records", 2) + + job, _provider = await _run_dispatcher_wiring_job(monkeypatch, record_count=5) + + # 5 records / batch_max_records=2 pack into 3 batches, not 5 -- the dynamic + # cap must be derived from that batch count (ceil(3 * 2.0) = 6), never from + # a naive per-record estimate (ceil(5 * 2.0) = 10). + assert job.stats["total_batches"] == 3 + assert job.stats["dynamic_provider_call_cap_value"] == 6 + assert job.stats["provider_call_cap_source"] == "dynamic" + + +@pytest.mark.asyncio +async def test_hard_ceiling_clamps_only_when_explicitly_configured(monkeypatch): + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", True) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 2.0) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", 0) + monkeypatch.setattr(settings, "summary_batch_max_records", 2) + + # No hard ceiling configured: dynamic cap (6) passes through untouched. + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + job, _ = await _run_dispatcher_wiring_job(monkeypatch, record_count=5) + assert job.stats["dynamic_provider_call_cap_value"] == 6 + assert job.stats["hard_max_provider_calls_per_job"] == 0 + assert job.stats["provider_call_cap_source"] == "dynamic" + + # Explicit hard ceiling below the dynamic cap clamps it. + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 4) + job, _ = await _run_dispatcher_wiring_job(monkeypatch, record_count=5) + assert job.stats["dynamic_provider_call_cap_value"] == 6 + assert job.stats["hard_max_provider_calls_per_job"] == 4 + assert job.stats["provider_call_cap_source"] == "dynamic_hard_ceiling" + + +@pytest.mark.asyncio +async def test_dynamic_cap_not_clamped_by_static_max_provider_calls_per_job(monkeypatch): + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", True) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 2.0) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", 0) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + monkeypatch.setattr(settings, "summary_batch_max_records", 2) + # Static default is far below the dynamic cap; it must not clamp it. + monkeypatch.setattr(settings, "summary_max_provider_calls_per_job", 1) + + job, _provider = await _run_dispatcher_wiring_job(monkeypatch, record_count=5) + + assert job.stats["dynamic_provider_call_cap_value"] == 6 + assert job.stats["provider_call_cap_source"] == "dynamic" + assert job.completed_records == 5 + + +@pytest.mark.asyncio +async def test_resolved_cap_is_passed_through_to_process_batch_budget(monkeypatch): + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_enabled", True) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_safety_factor", 2.0) + monkeypatch.setattr(settings, "summary_dynamic_provider_call_cap_margin", 0) + monkeypatch.setattr(settings, "summary_hard_max_provider_calls_per_job", 0) + monkeypatch.setattr(settings, "summary_batch_max_records", 2) + + real_process_batch = dispatcher_module._process_batch + captured_caps: list[int | None] = [] + + async def _capturing_process_batch(*args, **kwargs): + captured_caps.append(kwargs.get("max_provider_calls")) + return await real_process_batch(*args, **kwargs) + + monkeypatch.setattr(dispatcher_module, "_process_batch", _capturing_process_batch) + + job, _provider = await _run_dispatcher_wiring_job(monkeypatch, record_count=5) + + assert job.stats["dynamic_provider_call_cap_value"] == 6 + assert captured_caps == [6, 6, 6] # one call per batch, all carrying the resolved cap diff --git a/services/paragraph-summary-service/tests/test_gemini_provider.py b/services/paragraph-summary-service/tests/test_gemini_provider.py index 5bd667c..2d8d879 100644 --- a/services/paragraph-summary-service/tests/test_gemini_provider.py +++ b/services/paragraph-summary-service/tests/test_gemini_provider.py @@ -40,7 +40,7 @@ def _configure_one_gemini_lane(monkeypatch, *, api_key="test-lane-key"): monkeypatch.delenv(name) monkeypatch.setattr(settings, "summary_service_provider", "gemini") monkeypatch.setattr(settings, "summary_service_enable_provider_calls", True) - monkeypatch.setattr(settings, "summary_service_model", "gemini-2.5-flash") + monkeypatch.setattr(settings, "summary_service_model", "gemini-3.1-flash-lite") monkeypatch.setattr(settings, "summary_lane_count", 1) monkeypatch.setattr(settings, "summary_lane_rpm", 1) monkeypatch.setattr(settings, "summary_max_parallel_lanes", 1) @@ -234,7 +234,7 @@ async def test_gemini_provider_parses_structured_response(): ) client = FakeClient(response) provider = GeminiProvider( - model_name="gemini-2.5-flash", + model_name="gemini-3.1-flash-lite", template_version="paragraph_one_sentence_v1", api_key="not-a-real-key", lane_id="lane_01", @@ -254,7 +254,7 @@ async def test_gemini_provider_parses_structured_response(): assert results[0].stable_id == record.stable_id assert results[0].source_hash == record.source_hash assert results[0].provider == "gemini" - assert results[0].model == "gemini-2.5-flash" + assert results[0].model == "gemini-3.1-flash-lite" assert results[0].usage["total_tokens"] == 51 assert client.models.calls[0]["config"]["response_mime_type"] == "application/json" assert "The pump requires steady inlet flow." in client.models.calls[0]["contents"] @@ -263,7 +263,7 @@ async def test_gemini_provider_parses_structured_response(): @pytest.mark.asyncio async def test_gemini_provider_rejects_malformed_response(): provider = GeminiProvider( - model_name="gemini-2.5-flash", + model_name="gemini-3.1-flash-lite", template_version="paragraph_one_sentence_v1", api_key="not-a-real-key", lane_id="lane_01", diff --git a/services/paragraph-summary-service/tests/test_lane_tuning.py b/services/paragraph-summary-service/tests/test_lane_tuning.py new file mode 100644 index 0000000..6aa078e --- /dev/null +++ b/services/paragraph-summary-service/tests/test_lane_tuning.py @@ -0,0 +1,93 @@ +from app.config import settings +from app.scheduler.lane import QuotaLane +from app.scheduler.lane_tuning import large_book_lane_tuning + + +def _configure(monkeypatch, *, large_book_mode, threshold=1000, lane_rpm=4, adaptive_rpm_max=3): + monkeypatch.setattr(settings, "summary_large_book_mode", large_book_mode) + monkeypatch.setattr(settings, "summary_large_book_record_threshold", threshold) + monkeypatch.setattr(settings, "summary_lane_rpm", lane_rpm) + monkeypatch.setattr(settings, "summary_adaptive_rpm_enabled", True) + monkeypatch.setattr(settings, "summary_adaptive_rpm_success_threshold", 5) + monkeypatch.setattr(settings, "summary_adaptive_rpm_max", adaptive_rpm_max) + + +def test_config_knobs_default_off_and_default_threshold(monkeypatch): + for name in ("SUMMARY_LARGE_BOOK_MODE", "SUMMARY_LARGE_BOOK_RECORD_THRESHOLD"): + monkeypatch.delenv(name, raising=False) + + from app.config import Settings + + fresh_settings = Settings() + + assert fresh_settings.summary_large_book_mode is False + assert fresh_settings.summary_large_book_record_threshold == 1000 + + +def test_disabled_mode_preserves_current_tuning(monkeypatch): + _configure(monkeypatch, large_book_mode=False, threshold=1000, lane_rpm=4, adaptive_rpm_max=3) + + tuning = large_book_lane_tuning(settings, total_records=5000) + + assert tuning.adaptive_rpm_enabled is True + assert tuning.adaptive_rpm_success_threshold == 5 + assert tuning.adaptive_rpm_max == 3 + + +def test_below_threshold_preserves_current_tuning(monkeypatch): + _configure(monkeypatch, large_book_mode=True, threshold=1000, lane_rpm=4, adaptive_rpm_max=3) + + tuning = large_book_lane_tuning(settings, total_records=999) + + assert tuning.adaptive_rpm_max == 3 + + +def test_above_threshold_prevents_upshift_by_pinning_max_to_base_rpm(monkeypatch): + _configure(monkeypatch, large_book_mode=True, threshold=1000, lane_rpm=4, adaptive_rpm_max=3) + + tuning = large_book_lane_tuning(settings, total_records=1000) + + # adaptive_rpm_max must equal the base lane RPM, not the (possibly higher + # or lower) configured ceiling -- current_rpm starts at base_rpm, so it + # can never fall below adaptive_rpm_max and upshift is impossible. + assert tuning.adaptive_rpm_max == 4 + assert tuning.adaptive_rpm_enabled is True + assert tuning.adaptive_rpm_success_threshold == 5 + + +def test_success_still_resets_rate_limit_streak_in_large_book_mode(monkeypatch): + _configure(monkeypatch, large_book_mode=True, threshold=1000, lane_rpm=4, adaptive_rpm_max=3) + tuning = large_book_lane_tuning(settings, total_records=5000) + + lane = QuotaLane( + "lane_01", + rpm=settings.summary_lane_rpm, + provider_alias="gemini_01", + adaptive_rpm_enabled=tuning.adaptive_rpm_enabled, + adaptive_rpm_success_threshold=tuning.adaptive_rpm_success_threshold, + adaptive_rpm_max=tuning.adaptive_rpm_max, + ) + lane.rate_limit_streak = 3 + + for _ in range(tuning.adaptive_rpm_success_threshold): + lane.record_success() + + assert lane.rate_limit_streak == 0 + # Telemetry (success_streak) still accrues, but never triggers an upshift: + # current_rpm starts at adaptive_rpm_max, so the upshift guard is never true. + assert lane.success_streak == tuning.adaptive_rpm_success_threshold + assert lane.adaptive_adjustment_count == 0 + assert lane.current_rpm == lane.base_rpm == tuning.adaptive_rpm_max + + +def test_flags_off_backward_compatibility_matches_pre_large_book_behavior(monkeypatch): + for name in ("SUMMARY_LARGE_BOOK_MODE", "SUMMARY_LARGE_BOOK_RECORD_THRESHOLD"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(settings, "summary_large_book_mode", False) + monkeypatch.setattr(settings, "summary_adaptive_rpm_max", 12) + + tuning_small = large_book_lane_tuning(settings, total_records=1) + tuning_huge = large_book_lane_tuning(settings, total_records=10_000_000) + + assert tuning_small.adaptive_rpm_max == 12 + assert tuning_huge.adaptive_rpm_max == 12 diff --git a/services/paragraph-summary-service/tests/test_one_sentence_heuristic.py b/services/paragraph-summary-service/tests/test_one_sentence_heuristic.py new file mode 100644 index 0000000..d00a569 --- /dev/null +++ b/services/paragraph-summary-service/tests/test_one_sentence_heuristic.py @@ -0,0 +1,61 @@ +"""Characterization tests for _is_reasonable_one_sentence. + +Diagnostic only: these quantify how the current one-sentence heuristic +behaves on realistic philosophy-summary text (abbreviations, decimals, +questions, quoted sentences) so that false-positive rate can be measured +before any validator or batch-downshift change is made. A test failing here +means the documented behavior changed, which is exactly what a future fix +should be checked against — it is not automatically a bug. + +The heuristic splits on `(?<=[.!?])\\s+(?=[A-Z0-9])`: a sentence boundary is +only detected when the terminator is followed by whitespace and then an +uppercase letter or digit. So "i.e. things" or "e.g. courage" (lowercase +continuation) are *not* falsely split, but "Dr. Smith" or "Cf. Kant" (a +capitalized word right after an abbreviation) *are* falsely split into two +"sentences" and rejected even though they are a single valid sentence. +""" + +from app.scheduler.dispatcher import _is_reasonable_one_sentence + + +def test_plain_one_sentence_is_accepted(): + assert _is_reasonable_one_sentence("Kant argues that reason has inherent limits.") + + +def test_abbreviation_ie_followed_by_lowercase_is_accepted(): + text = "Kant distinguishes phenomena from noumena, i.e. things as they appear versus things in themselves." + assert _is_reasonable_one_sentence(text) is True + + +def test_abbreviation_eg_followed_by_lowercase_is_accepted(): + text = "Some virtues, e.g. courage and temperance, require habituation according to Aristotle." + assert _is_reasonable_one_sentence(text) is True + + +def test_title_abbreviation_dr_before_capitalized_name_is_falsely_split(): + text = "Dr. Smith notes that Descartes' cogito is a foundational claim in modern epistemology." + assert _is_reasonable_one_sentence(text) is False + + +def test_decimal_number_is_not_falsely_split(): + text = "The chapter reports that 3.14 percent of respondents rejected moral realism outright." + assert _is_reasonable_one_sentence(text) is True + + +def test_question_form_summary_is_accepted(): + assert _is_reasonable_one_sentence("Does free will survive a deterministic universe?") + + +def test_quoted_sentence_with_internal_period_is_accepted(): + text = 'Hume\'s remark that "reason is the slave of the passions" reframes moral motivation.' + assert _is_reasonable_one_sentence(text) + + +def test_two_genuine_sentences_are_rejected(): + text = "Locke defends empiricism. Rationalists disagree sharply with this view." + assert _is_reasonable_one_sentence(text) is False + + +def test_citation_abbreviation_before_capitalized_name_is_falsely_split(): + text = "Cf. Kant's first Critique, i.e. the 1781 edition, vs. the 1787 B-edition revisions." + assert _is_reasonable_one_sentence(text) is False diff --git a/services/paragraph-summary-service/tests/test_scheduler_retry.py b/services/paragraph-summary-service/tests/test_scheduler_retry.py index 926ab70..1ae310c 100644 --- a/services/paragraph-summary-service/tests/test_scheduler_retry.py +++ b/services/paragraph-summary-service/tests/test_scheduler_retry.py @@ -113,7 +113,7 @@ async def summarize_batch(self, document_id, records, summary_style): monkeypatch.setattr( dispatcher_module, "_build_lanes_and_providers", - lambda provider_name, credentials: ([lane], {lane.lane_id: provider}), + lambda provider_name, credentials, total_records=0: ([lane], {lane.lane_id: provider}), ) request = SummaryRequest( document_id="doc-rate-limit-cap",