Allow multimodal SDG to skip exhausted validation failures

docs/plugins/data-designer-retrieval-sdg/multimodal-reference.md

@@ -20,6 +20,7 @@ image paths inside source records resolve relative to their JSONL file.
 | `concurrency` | `8` | Model-call concurrency, from 1 to 128. |
 | `batch_size` | `30` | Request batch size, from 1 to 128. |
 | `missing_response_attempts` | `3` | Total attempts for missing or invalid rows in normally completed batches, from 1 to 3. |
+| `failed_record_policy` | `error` | Use `skip` to record exhausted model-validation failures and continue with usable records. |
 | `seed` | `42` | Seed for instruction sampling, selection, and query-group splitting. Model output can still vary. |
 | `resume` | `false` | Reuse a compatible run and cached responses. |
 | `ratios` | `{train: 0.8, validation: 0.0, evaluation: 0.2}` | Query-group split fractions, each from 0 to 1 and summing to 1. |
@@ -41,6 +42,58 @@ provider retry behavior. Batch size and retry settings do not define a total
 API-call, cost, or elapsed-time budget. Refer to
 [Resume an Interrupted Run](multimodal.md#resume-an-interrupted-run) for recovery.
 
+### Continue After Failed Records
+
+For a new run, add this to the existing YAML configuration:
+
+```yaml
+failed_record_policy: skip
+```
+
+The default `error` policy stops when required structured responses remain
+unavailable. With `skip`, the structured column records a diagnostic after Data
+Designer's native validation retries are exhausted. This adds no retry loop and
+does not change the native retry budget. `missing_response_attempts` keeps its
+separate meaning; a recorded failure is terminal and is not retried by that loop
+or by resume. Unexplained missing rows, duplicate identities, provider errors,
+and integrity errors are never converted into skipped records.
+
+| Failed stage | Disposition |
+| --- | --- |
+| TOC confirmation | Preserve the source and omit unverified TOC projections. |
+| Visual or document enrichment | Preserve original text/images and omit the failed enrichment. |
+| Summary or summary combination | Omit the unavailable summary from generation inputs. |
+| Summary judgment | Exclude the summary; a failed judgment cannot become a pass. |
+| Query generation | Exclude the context's requested query slots. |
+| Query metadata, self-sufficiency, relevance, or localization | Exclude the affected query; successful responses remain cached. |
+
+The original corpus and quality thresholds are preserved. If no queries pass,
+export still fails. Inspect `generation_failures.json` in the run directory and
+portable bundle for stage, request ID, source/context/query identity where
+applicable, validation detail, native attempt limit, and the last invalid
+response when available. The portable ledger is included in the manifest's
+checksums. `report.json.generation_failures` counts failures by stage, failed
+requested slots, and excluded query IDs. Failed query-generation slots contribute
+to `requested_slots`, separately from model abstentions and quality rejections.
+Earlier planning failures have no known query-slot count.
+Queries that fail a later judgment or localization still count in
+`funnel.generated` and the `retention_fraction` denominator, once per query.
+For example, three generated queries with one localization failure and two
+accepted queries report three generated and a retention fraction of `2/3`.
+
+The same policy is supported by `RetrievalStructuredColumnConfig`. In skip mode,
+non-null output cells are JSON strings containing either a validated response
+or a `_retrieval_sdg_failure` diagnostic. The multimodal workflow decodes these
+before its typed consumers. Direct column users must handle the diagnostic
+before treating a cell as a model response. Failed records have `None` in any
+configured trace and reasoning-content columns, including when every record in
+a batch fails, so downstream expressions can still reference those columns.
+
+Resume preserves both cached successes and recorded failures. Changing the
+policy or upgrading the implementation changes the frozen run identity; use a
+new output directory. Failure rates and exclusions can affect sample composition,
+so include them when comparing generation yield.
+
 ## Source Records
 
 Each line in `sources_file` is one JSON object describing an independently

docs/plugins/data-designer-retrieval-sdg/multimodal.md

@@ -264,6 +264,11 @@ reviewed relevance labels when comparing retrieval models.
 
 ## Resume an Interrupted Run
 
+For runs that should continue past persistently malformed model responses,
+set `failed_record_policy: skip` when creating a new run. See
+[Continue After Failed Records](multimodal-reference.md#continue-after-failed-records)
+for stage behavior and failure diagnostics. The default remains `error`.
+
 Preserve the run directory and inspect the error and saved artifacts. After
 resolving an external issue, set `resume: true` in the same configuration and run
 the command again. Keep the inputs, remaining configuration, package code, and

plugins/data-designer-retrieval-sdg/docs/multimodal-reference.md

@@ -20,6 +20,7 @@ image paths inside source records resolve relative to their JSONL file.
 | `concurrency` | `8` | Model-call concurrency, from 1 to 128. |
 | `batch_size` | `30` | Request batch size, from 1 to 128. |
 | `missing_response_attempts` | `3` | Total attempts for missing or invalid rows in normally completed batches, from 1 to 3. |
+| `failed_record_policy` | `error` | Use `skip` to record exhausted model-validation failures and continue with usable records. |
 | `seed` | `42` | Seed for instruction sampling, selection, and query-group splitting. Model output can still vary. |
 | `resume` | `false` | Reuse a compatible run and cached responses. |
 | `ratios` | `{train: 0.8, validation: 0.0, evaluation: 0.2}` | Query-group split fractions, each from 0 to 1 and summing to 1. |
@@ -41,6 +42,58 @@ provider retry behavior. Batch size and retry settings do not define a total
 API-call, cost, or elapsed-time budget. Refer to
 [Resume an Interrupted Run](multimodal.md#resume-an-interrupted-run) for recovery.
 
+### Continue After Failed Records
+
+For a new run, add this to the existing YAML configuration:
+
+```yaml
+failed_record_policy: skip
+```
+
+The default `error` policy stops when required structured responses remain
+unavailable. With `skip`, the structured column records a diagnostic after Data
+Designer's native validation retries are exhausted. This adds no retry loop and
+does not change the native retry budget. `missing_response_attempts` keeps its
+separate meaning; a recorded failure is terminal and is not retried by that loop
+or by resume. Unexplained missing rows, duplicate identities, provider errors,
+and integrity errors are never converted into skipped records.
+
+| Failed stage | Disposition |
+| --- | --- |
+| TOC confirmation | Preserve the source and omit unverified TOC projections. |
+| Visual or document enrichment | Preserve original text/images and omit the failed enrichment. |
+| Summary or summary combination | Omit the unavailable summary from generation inputs. |
+| Summary judgment | Exclude the summary; a failed judgment cannot become a pass. |
+| Query generation | Exclude the context's requested query slots. |
+| Query metadata, self-sufficiency, relevance, or localization | Exclude the affected query; successful responses remain cached. |
+
+The original corpus and quality thresholds are preserved. If no queries pass,
+export still fails. Inspect `generation_failures.json` in the run directory and
+portable bundle for stage, request ID, source/context/query identity where
+applicable, validation detail, native attempt limit, and the last invalid
+response when available. The portable ledger is included in the manifest's
+checksums. `report.json.generation_failures` counts failures by stage, failed
+requested slots, and excluded query IDs. Failed query-generation slots contribute
+to `requested_slots`, separately from model abstentions and quality rejections.
+Earlier planning failures have no known query-slot count.
+Queries that fail a later judgment or localization still count in
+`funnel.generated` and the `retention_fraction` denominator, once per query.
+For example, three generated queries with one localization failure and two
+accepted queries report three generated and a retention fraction of `2/3`.
+
+The same policy is supported by `RetrievalStructuredColumnConfig`. In skip mode,
+non-null output cells are JSON strings containing either a validated response
+or a `_retrieval_sdg_failure` diagnostic. The multimodal workflow decodes these
+before its typed consumers. Direct column users must handle the diagnostic
+before treating a cell as a model response. Failed records have `None` in any
+configured trace and reasoning-content columns, including when every record in
+a batch fails, so downstream expressions can still reference those columns.
+
+Resume preserves both cached successes and recorded failures. Changing the
+policy or upgrading the implementation changes the frozen run identity; use a
+new output directory. Failure rates and exclusions can affect sample composition,
+so include them when comparing generation yield.
+
 ## Source Records
 
 Each line in `sources_file` is one JSON object describing an independently

plugins/data-designer-retrieval-sdg/docs/multimodal.md

@@ -264,6 +264,11 @@ reviewed relevance labels when comparing retrieval models.
 
 ## Resume an Interrupted Run
 
+For runs that should continue past persistently malformed model responses,
+set `failed_record_policy: skip` when creating a new run. See
+[Continue After Failed Records](multimodal-reference.md#continue-after-failed-records)
+for stage behavior and failure diagnostics. The default remains `error`.
+
 Preserve the run directory and inspect the error and saved artifacts. After
 resolving an external issue, set `resume: true` in the same configuration and run
 the command again. Keep the inputs, remaining configuration, package code, and

plugins/data-designer-retrieval-sdg/src/data_designer_retrieval_sdg/multimodal/config.py

@@ -12,7 +12,9 @@ class RetrievalStructuredColumnConfig(LLMStructuredColumnConfig):
     """Generate schema-validated retrieval JSON, with or without a Markdown fence.
 
     All model, image, prompt, skip, and trace settings retain their native
-    Data Designer meanings. Only response decoding uses the plugin parser.
+    Data Designer meanings. The optional skip policy records exhausted model
+    validation failures as tagged diagnostics instead of dropping the row.
     """
 
     column_type: Literal["retrieval-structured"] = "retrieval-structured"
+    failed_record_policy: Literal["error", "skip"] = "error"

plugins/data-designer-retrieval-sdg/src/data_designer_retrieval_sdg/multimodal/export.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import errno
+import json
 import tempfile
 from collections import Counter
 from pathlib import Path
@@ -308,7 +309,24 @@ def write_multimodal_bundle(
     }
     from data_designer_retrieval_sdg.multimodal.reporting import generation_report
 
-    counts.update(generation_report(sources, candidates, outcomes))
+    failed_query_ids = set()
+    if config.failed_record_policy == "skip":
+        failures = json.loads((config.output_dir / "generation_failures.json").read_text())
+        write_json(root / "generation_failures.json", failures)
+        failed_query_ids = {row["query_id"] for row in failures["records"] if "query_id" in row}
+        failed_slots = sum(
+            len(row["instructions"]) for row in outcomes if row.get("selection_reason") == "generation_failed"
+        )
+        counts["requested_slots"] += failed_slots
+        counts["generation_failures"] = {
+            "policy": "skip",
+            "failed_records": len(failures["records"]),
+            "failed_requested_slots": failed_slots,
+            "excluded_query_ids": len(failed_query_ids),
+            "by_stage": dict(Counter(row["stage"] for row in failures["records"])),
+            "diagnostics": "generation_failures.json",
+        }
+    counts.update(generation_report(sources, candidates, outcomes, failed_query_ids))
     write_json(root / "report.json", counts)
     write_json(
         root / "run_manifest.json",

plugins/data-designer-retrieval-sdg/src/data_designer_retrieval_sdg/multimodal/inference.py

@@ -22,9 +22,10 @@
 
 from data_designer_retrieval_sdg.multimodal.bounds import check_request
 from data_designer_retrieval_sdg.multimodal.config import RetrievalStructuredColumnConfig
-from data_designer_retrieval_sdg.multimodal.models import GenerationContext, MultimodalSDGConfig
+from data_designer_retrieval_sdg.multimodal.models import FailedGeneration, GenerationContext, MultimodalSDGConfig
 from data_designer_retrieval_sdg.multimodal.source_models import RetrievalSource
 from data_designer_retrieval_sdg.multimodal.storage import digest, fingerprint, write_json
+from data_designer_retrieval_sdg.multimodal.structured import FAILURE_KEY
 
 
 @dataclass(frozen=True)
@@ -60,6 +61,73 @@ def source_request(
     )
 
 
+def request_identity(context: GenerationContext, **details) -> dict:
+    """Attach source membership and optional query/slot details to a failure record.
+
+    Args:
+        context: Original context owning the request.
+        **details: Additional query, slot, or instruction provenance.
+
+    Returns:
+        JSON-compatible identity retained with any generation failure.
+    """
+    return {"context_id": context.context_id, "source_unit_ids": context.unit_ids, **details}
+
+
+def generate_records(
+    inference: DataDesignerInference, requests: list[Request], stage: str, records: list[dict]
+) -> list[BaseModel]:
+    """Generate aligned responses and retain contextual diagnostics for failed ones.
+
+    Args:
+        inference: Run-owned response cache and native Data Designer transport.
+        requests: Ordered requests, including any identical requests shared by records.
+        stage: Workflow stage responsible for these requests.
+        records: One context/source/query identity per request.
+
+    Returns:
+        Validated responses or explicit failures in the original record order.
+        Consumers decide the stage-specific disposition without shifting alignment.
+
+    Raises:
+        ValueError: Request, identity, or response counts disagree.
+    """
+    if len(requests) != len(records):
+        raise ValueError("Every generation request requires a record identity")
+    responses = inference.generate(requests)
+    for request, record, response in zip(requests, records, responses, strict=True):
+        if isinstance(response, FailedGeneration):
+            key = inference.request_key(request)
+            identity = fingerprint([stage, key, record])
+            inference.failures[identity] = {
+                **record,
+                "stage": stage,
+                "request_id": key,
+                "failure": response.model_dump(mode="json"),
+            }
+    return responses
+
+
+def parse_failure(value: object, policy: str) -> FailedGeneration | None:
+    """Accept only a well-formed diagnostic under the explicit skip policy.
+
+    Args:
+        value: Decoded native response or cached payload.
+        policy: Configured error/skip policy.
+
+    Returns:
+        The typed failure, or None for an ordinary response.
+
+    Raises:
+        ValueError: A failure marker is malformed or the skip policy is disabled.
+    """
+    if not isinstance(value, dict) or FAILURE_KEY not in value:
+        return None
+    if policy != "skip" or set(value) != {FAILURE_KEY}:
+        raise ValueError("Saved failure requires the skip policy and a valid diagnostic")
+    return FailedGeneration.model_validate(value[FAILURE_KEY])
+
+
 class DataDesignerInference:
     """Run schema-homogeneous batches with immutable per-response caches.
 
@@ -71,6 +139,7 @@ class DataDesignerInference:
     def __init__(self, root: Path, config: MultimodalSDGConfig):
         self.root = root
         self.config = config
+        self.failures: dict[str, dict] = {}
 
     def request_key(self, request: Request) -> str:
         """Bind responses to exact prompts, schema, pixels and model settings."""
@@ -97,15 +166,18 @@ def cached(self, request: Request) -> BaseModel | None:
         value = json.loads(path.read_text())
         if value["request_id"] != key or fingerprint(value["response"]) != value["response_sha256"]:
             raise ValueError(f"Response cache integrity failure: {key}")
-        return request.schema.model_validate(value["response"])
+        failure = parse_failure(value["response"], self.config.failed_record_policy)
+        return failure if failure is not None else request.schema.model_validate(value["response"])
 
     def generate(self, requests: list[Request]) -> list[BaseModel]:
         """Retry only missing rows within the configured finite attempt budget.
 
         Missing outputs fail the stage after preserving successful responses and
         attempt evidence. Exceptions stop immediately; only normally completed
         batches with missing/invalid rows get another attempt. Native DD's own
-        schema corrections remain bounded inside each attempt.
+        schema corrections remain bounded inside each attempt. In skip mode,
+        explicit exhausted validation failures are cached terminal results;
+        unexplained missing rows still fail the stage.
         """
         groups: dict[str, dict[str, Request]] = {}
         for request in requests:
@@ -169,6 +241,7 @@ def run_batch(self, items: list[tuple[str, Request]]) -> None:
                 model_alias="retrieval",
                 prompt="{{ request_text }}",
                 output_format=request.schema,
+                failed_record_policy=self.config.failed_record_policy,
                 multi_modal_context=[dd.ImageContext(column_name="request_images")] if request.images else None,
             )
         )
@@ -226,6 +299,7 @@ def collect_responses(self, items: list[tuple[str, Request]], dataset: Path, att
                     raise ValueError("Unexpected or duplicate Data Designer response identity")
                 seen.add(key)
         found: set[str] = set()
+        failed: list[str] = []
         invalid: list[str] = []
         for batch in batches:
             for row in pq.read_table(batch).to_pylist():
@@ -235,15 +309,20 @@ def collect_responses(self, items: list[tuple[str, Request]], dataset: Path, att
                 raw = row["response"]
                 request = expected[key]
                 try:
-                    response = (
-                        request.schema.model_validate_json(raw)
-                        if isinstance(raw, str)
-                        else request.schema.model_validate(raw)
-                    )
+                    value = json.loads(raw) if isinstance(raw, str) else raw
                 except ValueError:
                     invalid.append(key)
                     continue
-                payload = response.model_dump(mode="json")
+                failure = parse_failure(value, self.config.failed_record_policy)
+                if failure is not None:
+                    payload = {FAILURE_KEY: failure.model_dump(mode="json")}
+                    failed.append(key)
+                else:
+                    try:
+                        payload = request.schema.model_validate(value).model_dump(mode="json")
+                    except ValueError:
+                        invalid.append(key)
+                        continue
                 write_json(
                     self.root / "responses" / f"{key}.json",
                     {
@@ -257,8 +336,9 @@ def collect_responses(self, items: list[tuple[str, Request]], dataset: Path, att
             attempt / "outcome.json",
             {
                 "requested": len(items),
-                "returned": len(found),
+                "returned": len(found) - len(failed),
                 "missing": sorted(expected.keys() - found),
                 "invalid": invalid,
+                **({"failed": failed} if self.config.failed_record_policy == "skip" else {}),
             },
         )