Add v0.4 adaptive large-book scheduler safeguards

.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

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,

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

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`)

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,
         }
 
 

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

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,
+    )