diff --git a/application/single_app/config.py b/application/single_app/config.py
index 9a05b6817..55eb38109 100644
--- a/application/single_app/config.py
+++ b/application/single_app/config.py
@@ -101,7 +101,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.308"
+VERSION = "0.261.309"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform
diff --git a/application/single_app/functions_document_analysis.py b/application/single_app/functions_document_analysis.py
index 7bd22284a..6ab59228b 100644
--- a/application/single_app/functions_document_analysis.py
+++ b/application/single_app/functions_document_analysis.py
@@ -14,11 +14,12 @@
 
 from content_screening.access import PROVENANCE_FIELD, assert_evidence_available, guard_model_callable
 from content_screening.contracts import ScreeningError
-from functions_appinsights import log_event
+from functions_appinsights import log_event, workflow_log_context
 from functions_debug import debug_print
 from functions_document_analysis_results import (
     EVIDENCE_MATCH_TIERS,
     EVIDENCE_MISS_REASONS,
+    analysis_response_shape_errors,
     apply_document_analysis_options,
     build_analysis_source,
     build_analysis_work_unit,
@@ -28,6 +29,7 @@
     get_unassigned_analysis_chunks,
     index_analysis_source_manifest,
     normalize_analysis_options,
+    validate_analysis_response_correction,
 )
 from functions_generated_file_exports import get_requested_structured_artifact_format
 from functions_orchestration_execution_policy import generated_file_publication_allowed
@@ -483,10 +485,12 @@ def _build_window_analysis_prompt(
             '- "status": "supported" for a finding supported by this slice, including a documented uncertainty '
             'as the finding itself; otherwise "unresolved".\n'
             '- "caveats": optional qualifications that do not stop the values being final, such as an unstated '
-            'unit, currency, period or entity.\n'
-            '- "issues": only problems that prevent a requested value in this finding being concluded from this '
+            'unit, currency, period or entity. Use a JSON array of strings ([] when empty), never a string or object.\n'
+            '- "issues": an optional JSON array of strings (use [] when empty, never a string or object), '
+            'containing only problems that prevent a requested value in this finding being concluded from this '
             'slice, such as contradictory values. Values absent from this slice are not issues. Put ordinary '
             'recommendations and follow-up discussion in "values", not "issues".\n'
+            'Top-level "issues" and "notes" must also be JSON arrays of strings when supplied. '
             'Top-level "issues" are only task requirements that this slice\'s own content leaves unresolved, never '
             'sources analyzed separately; put other observations about the slice in "notes". '
             'Formatting of the final report and exports is handled separately. Put requested export fields inside '
@@ -1176,6 +1180,10 @@ def prepare(window):
         check_cancelled(cancel_requested, 'narrative', request_correlation_id=request_correlation_id)
         unit = window['analysis_work_unit']
         cached = checkpoints.load_unit(unit) if checkpoints is not None else None
+        repair_response = cached if cached is not None and cached.get('needs_response_correction') else None
+        window['_analysis_repair_response'] = repair_response
+        if repair_response is not None:
+            cached = None
         window['_cached_analysis_unit'] = cached
         if checkpoints is not None and cached is None:
             window['_analysis_claim'] = checkpoints.claim_unit(unit)
@@ -1184,13 +1192,14 @@ def prepare(window):
             'window_range': _serialize_window_range(window), 'attempt_number': 1,
             'analysis_result_version': 'analyze-final-v1', 'work_unit_id': unit['work_unit_id'],
             'assigned_document_ids': [unit['document_id']],
+            'json_output': True,
         }
         window['_analysis_invoker'] = (
             invoke_factory(deepcopy(metadata)) if invoke_factory and cached is None else invoke
         )
         if not callable(window['_analysis_invoker']):
             raise ValueError('The isolated analysis invocation factory did not return a callable.')
-        if concurrency > 1 and cached is None:
+        if concurrency > 1 and cached is None and repair_response is None:
             prompt_text = _build_window_analysis_prompt(
                 prompt, document, window, metadata['window_range'],
                 result_version='analyze-final-v1', analysis_options=options,
@@ -1717,6 +1726,9 @@ def iter_document_runs():
             candidate_result = cached_unit['candidate_result'] if cached_unit is not None else None
             prompt_text = ''
             last_error = ''
+            correction_text = None
+            correction_errors = []
+            retained_response = None
             max_attempts = targets.get('max_retries_per_window', DEFAULT_MAX_RETRIES_PER_WINDOW) + 1
             for attempt_number in range(1, 1 if cached_unit is not None else max_attempts + 1):
                 raise_if_mixed_source_cancelled(
@@ -1738,6 +1750,7 @@ def iter_document_runs():
 
                 try:
                     analysis_text = ''
+                    shape_errors = []
                     prompt_text = _build_window_analysis_prompt(
                         normalized_analysis_prompt,
                         document_payload.get('document', {}),
@@ -1747,6 +1760,18 @@ def iter_document_runs():
                             'result_version': result_version, 'analysis_options': normalized_options,
                         } if use_final_records else {}),
                     )
+                    if correction_text is not None:
+                        prompt_text += (
+                            '\n\n<ResponseShapeCorrection>\n'
+                            'Correct only the malformed fields listed below in the previous JSON response. '
+                            'Keep the findings in their original order, with the same values, evidence, '
+                            'supported fields and uncertainty. Preserve every qualification verbatim; '
+                            'a text qualification belongs in a one-item array, not an empty array. '
+                            'Do not resolve unsupported findings, invent evidence or discard invalid content. '
+                            'Return the entire corrected JSON object, without commentary.\n'
+                            f'Fields: {json.dumps(correction_errors, ensure_ascii=True)}\n'
+                            f'Previous response:\n{correction_text}\n</ResponseShapeCorrection>'
+                        )
                     metadata = {
                         'document_id': document_id, 'document_name': document_name,
                         'window_range': window_range, 'attempt_number': attempt_number,
@@ -1754,9 +1779,16 @@ def iter_document_runs():
                             'analysis_result_version': result_version,
                             'work_unit_id': window_payload['analysis_work_unit']['work_unit_id'],
                             'assigned_document_ids': [document_id],
+                            'json_output': True,
+                            **({
+                                'analysis_response_correction': True,
+                                'complete_saved_analysis_input': True,
+                            } if correction_text is not None else {}),
                         } if use_final_records else {}),
                     }
-                    if use_final_records and attempt_number == 1 and window_payload.get('_first_attempt_future') is not None:
+                    if use_final_records and attempt_number == 1 and window_payload.get('_analysis_repair_response') is not None:
+                        analysis_text = window_payload['_analysis_repair_response']['analysis_text']
+                    elif use_final_records and attempt_number == 1 and window_payload.get('_first_attempt_future') is not None:
                         analysis_text = _await_analysis_invocation(
                             window_payload['_first_attempt_future'], cancel_requested, request_correlation_id,
                             raise_if_mixed_source_cancelled,
@@ -1779,9 +1811,29 @@ def iter_document_runs():
                     if use_final_records:
                         collection_started = time.perf_counter()
                         try:
+                            if correction_text is not None:
+                                validate_analysis_response_correction(
+                                    correction_text, analysis_text, correction_errors,
+                                )
+                            shape_errors = analysis_response_shape_errors(analysis_text)
                             candidate_result = collect_analysis_window_candidates(
                                 analysis_text, analysis_source, window_payload['analysis_work_unit'], window_payload,
                             )
+                            if shape_errors and attempt_number < max_attempts:
+                                correction_text, correction_errors = analysis_text, shape_errors
+                                retained_response = (analysis_text, candidate_result)
+                                log_event(
+                                    '[DOCUMENT_ANALYSIS] Correcting malformed response fields.',
+                                    extra={
+                                        **workflow_log_context(
+                                            conversation_id=conversation_id,
+                                            step_id=window_payload['analysis_work_unit']['work_unit_id'],
+                                        ),
+                                        'validation_code': 'analysis_response_shape',
+                                        'attempt_count': attempt_number, 'field_count': len(shape_errors),
+                                    }, level=logging.WARNING,
+                                )
+                                raise ValueError('The analysis response has malformed fields.')
                         finally:
                             analysis_metrics['durations_ms']['local_consolidation'] += (time.perf_counter() - collection_started) * 1000
                     break
@@ -1800,6 +1852,8 @@ def iter_document_runs():
                     ):
                         _abort_analysis_work(exc, active_windows, analysis_metrics, metrics_lock, work_unit_checkpoints)
                         raise
+                    if use_final_records and shape_errors and correction_text is None and analysis_text:
+                        correction_text, correction_errors = analysis_text, shape_errors
                     last_error = 'The document window could not be analyzed. Please retry.'
                     if use_final_records:
                         window_payload['analysis_work_unit']['failure_code'] = (
@@ -1857,6 +1911,24 @@ def iter_document_runs():
                     if attempt_number >= max_attempts:
                         break
 
+            if retained_response is not None and not analysis_text:
+                analysis_text, candidate_result = retained_response
+            if correction_text is not None:
+                correction_status = (
+                    'failed' if not analysis_text else 'partial' if analysis_text == correction_text else 'corrected'
+                )
+                log_event(
+                    '[DOCUMENT_ANALYSIS] Response correction finished.',
+                    extra={
+                        **workflow_log_context(
+                            conversation_id=conversation_id,
+                            step_id=window_payload['analysis_work_unit']['work_unit_id'],
+                        ),
+                        'validation_code': 'analysis_response_shape',
+                        'status': correction_status,
+                        'attempt_count': attempt_number,
+                    }, level=logging.INFO if correction_status == 'corrected' else logging.WARNING,
+                )
             if analysis_text:
                 if use_final_records:
                     work_unit = window_payload['analysis_work_unit']
diff --git a/application/single_app/functions_document_analysis_checkpoints.py b/application/single_app/functions_document_analysis_checkpoints.py
index 85a1cf972..30a7d5b61 100644
--- a/application/single_app/functions_document_analysis_checkpoints.py
+++ b/application/single_app/functions_document_analysis_checkpoints.py
@@ -19,6 +19,7 @@
     analysis_source_snapshot,
     authorize_analysis_sources,
 )
+from functions_document_analysis_results import ANALYSIS_RESPONSE_SHAPE_CODES, analysis_has_response_shape_issues
 from functions_workflow_result_store import (
     AnalysisWorkUnitConflictError,
     WorkflowResultIntegrityError,
@@ -170,6 +171,12 @@ def load_unit(self, unit):
                 or not isinstance(payload.get('analysis_text'), str)
             ):
                 raise WorkflowResultIntegrityError('A completed analysis work unit is invalid.')
+            if binding != self.binding and any(
+                issue.get('code') in ANALYSIS_RESPONSE_SHAPE_CODES
+                for candidate in payload['candidate_result'].get('candidates') or []
+                for issue in candidate.get('issues') or []
+            ):
+                payload['needs_response_correction'] = True
             return payload
         return None
 
@@ -206,6 +213,8 @@ def load_final_result(self):
                 coverage = (result.get('analysis_validation') or {}).get('coverage') or {}
                 if coverage.get('failed_work_units') or coverage.get('pending_work_units'):
                     continue
+                if binding != self.binding and analysis_has_response_shape_issues(result):
+                    continue
                 return saved
         return None
 
diff --git a/application/single_app/functions_document_analysis_results.py b/application/single_app/functions_document_analysis_results.py
index b05cd1341..abdb31f25 100644
--- a/application/single_app/functions_document_analysis_results.py
+++ b/application/single_app/functions_document_analysis_results.py
@@ -22,6 +22,9 @@
 EVIDENCE_MATCHER_VERSION = 'evidence-matcher-v2'
 EVIDENCE_MATCH_TIERS = ('exact', 'normalized', 'normalized_casefold')
 EVIDENCE_MISS_REASONS = ('missing_quote', 'missing_location', 'ambiguous', 'not_in_cited_location', 'not_in_window')
+ANALYSIS_RESPONSE_SHAPE_CODES = frozenset({
+    'invalid_identity', 'invalid_values', 'invalid_status', 'invalid_issues', 'invalid_caveats',
+})
 _EVIDENCE_SELECTORS = ('chunk_sequence', 'page_number', 'chunk_id')
 _EVIDENCE_MARKUP = re.compile(r'<!--.*?-->|</?([A-Za-z][A-Za-z0-9:-]*)(?:[\s/][^<>]*)?>', re.DOTALL)
 _EVIDENCE_LINE = re.compile(r'[^\n]+')
@@ -280,6 +283,98 @@ def _unique_json_keys(pairs):
     return value
 
 
+def _analysis_response(analysis_text):
+    cleaned = str(analysis_text or '').strip()
+    fence = re.fullmatch(r'```(?:json)?\s*(.*?)\s*```', cleaned, flags=re.DOTALL | re.IGNORECASE)
+    if fence:
+        cleaned = fence.group(1)
+    return json.loads(cleaned, parse_constant=_reject_json_constant, object_pairs_hook=_unique_json_keys)
+
+
+def analysis_response_shape_errors(analysis_text):
+    """Describe malformed fields without copying their values into diagnostics."""
+    payload = _analysis_response(analysis_text)
+    if not isinstance(payload, dict) or not isinstance(payload.get('findings'), list):
+        raise ValueError('The analysis response must contain a findings list.')
+    errors = []
+    for field in ('issues', 'notes'):
+        value = payload.get(field, [])
+        if field == 'notes' and value is None:
+            continue
+        if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
+            errors.append({'field': field, 'code': f'invalid_{field}'})
+    for index, finding in enumerate(payload['findings']):
+        if not isinstance(finding, dict):
+            errors.append({'finding_index': index, 'field': 'finding', 'code': 'invalid_values'})
+            continue
+        for field, valid in (
+            ('finding_key', isinstance(finding.get('finding_key'), str) and bool(finding['finding_key'].strip())),
+            ('values', isinstance(finding.get('values'), dict) and bool(finding['values'])),
+            ('status', finding.get('status') in ('supported', 'unresolved')),
+            *(
+                (field, (
+                    field == 'caveats' and finding.get(field) is None
+                    or isinstance(finding.get(field, []), list)
+                    and all(isinstance(item, str) for item in finding.get(field, []))
+                ))
+                for field in ('issues', 'caveats')
+            ),
+        ):
+            if not valid:
+                errors.append({
+                    'finding_index': index, 'field': field,
+                    'code': 'invalid_identity' if field == 'finding_key' else f'invalid_{field}',
+                })
+    return errors
+
+
+def validate_analysis_response_correction(original_text, corrected_text, errors):
+    """A shape correction cannot rewrite findings or remove recorded qualifications."""
+    original = _analysis_response(original_text)
+    corrected = _analysis_response(corrected_text)
+    if analysis_response_shape_errors(corrected_text):
+        raise ValueError('The corrected analysis response still has malformed fields.')
+    top_fields = {item['field'] for item in errors if 'finding_index' not in item}
+    original_fields = {key: value for key, value in original.items() if key not in top_fields | {'findings'}}
+    corrected_fields = {key: value for key, value in corrected.items() if key not in top_fields | {'findings'}}
+    if _identity('response', original_fields) != _identity('response', corrected_fields):
+        raise ValueError('The analysis correction changed unrelated fields.')
+    if len(original['findings']) != len(corrected['findings']):
+        raise ValueError('The analysis correction changed the assigned findings.')
+    pairs = [(None, original, corrected), *(
+        (index, left, right) for index, (left, right) in enumerate(
+            zip(original['findings'], corrected['findings'])
+        )
+    )]
+    for index, left, right in pairs:
+        fields = {item['field'] for item in errors if item.get('finding_index') == index}
+        if not isinstance(left, dict) or not isinstance(right, dict):
+            raise ValueError('The analysis correction cannot discard malformed finding content.')
+        if index is not None and _identity('finding', {
+            key: value for key, value in left.items() if key not in fields
+        }) != _identity('finding', {key: value for key, value in right.items() if key not in fields}):
+            raise ValueError('The analysis correction changed supported content.')
+        if fields & {'values', 'finding_key', 'finding'}:
+            raise ValueError('The analysis correction cannot synthesize malformed finding content.')
+        if 'status' in fields and right['status'] != 'unresolved':
+            raise ValueError('The analysis correction cannot promote an unknown support status.')
+        for field in fields & {'issues', 'caveats', 'notes'}:
+            value = left.get(field)
+            texts = [value] if isinstance(value, str) else value
+            if not isinstance(texts, list) or any(not isinstance(item, str) for item in texts):
+                raise ValueError('The analysis correction cannot discard malformed qualifications.')
+            if sorted(texts) != sorted(right[field]):
+                raise ValueError('The analysis correction changed recorded qualifications.')
+
+
+def analysis_has_response_shape_issues(result):
+    validation = (result or {}).get('analysis_validation') or {}
+    return any(
+        isinstance(issue, dict) and issue.get('code') in ANALYSIS_RESPONSE_SHAPE_CODES
+        for issue in validation.get('issues') or []
+    )
+
+
 def _issue(code, message, **context):
     return {'code': code, 'message': message, **context}
 
@@ -537,11 +632,7 @@ def occurrence(index, tier):
 
 def collect_analysis_window_candidates(analysis_text, source, work_unit, window_payload):
     """Check an extraction response against its assigned original source window."""
-    cleaned = str(analysis_text or '').strip()
-    fence = re.fullmatch(r'```(?:json)?\s*(.*?)\s*```', cleaned, flags=re.DOTALL | re.IGNORECASE)
-    if fence:
-        cleaned = fence.group(1)
-    payload = json.loads(cleaned, parse_constant=_reject_json_constant, object_pairs_hook=_unique_json_keys)
+    payload = _analysis_response(analysis_text)
     if not isinstance(payload, dict) or not isinstance(payload.get('findings'), list):
         raise ValueError('The analysis response must contain a findings list.')
     # Also catches exponent overflow (for example 1e999), not only JSON NaN literals.
@@ -570,7 +661,9 @@ def collect_analysis_window_candidates(analysis_text, source, work_unit, window_
         if not isinstance(values, dict) or not values:
             issues.append(_issue('invalid_values', 'A finding did not supply an object of requested values.'))
             values = {}
-        if finding.get('status') != 'supported':
+        if finding.get('status') not in ('supported', 'unresolved'):
+            issues.append(_issue('invalid_status', 'A finding supplied an invalid support status.'))
+        elif finding.get('status') != 'supported':
             issues.append(_issue('unresolved_finding', 'The source slice did not support a final conclusion.'))
         finding_issues = finding.get('issues', [])
         if not isinstance(finding_issues, list) or any(not isinstance(item, str) for item in finding_issues):
diff --git a/application/single_app/functions_orchestration_analysis_results.py b/application/single_app/functions_orchestration_analysis_results.py
index 1fc330b8d..f924ab1b2 100644
--- a/application/single_app/functions_orchestration_analysis_results.py
+++ b/application/single_app/functions_orchestration_analysis_results.py
@@ -8,6 +8,7 @@
 from copy import deepcopy
 
 from functions_analysis_access import analysis_source_snapshot
+from functions_document_analysis_results import ANALYSIS_RESPONSE_SHAPE_CODES
 from functions_orchestration_result_contracts import (
     Completeness,
     Coverage,
@@ -58,6 +59,12 @@ def _native_completeness(reader):
         if type(check) is not dict or type(check.get("name")) is not str or type(check.get("status")) is not str:
             raise ResultContractError("result_analysis_validation_invalid")
         checks.append(f'{check["name"]}:{check["status"]}')
+    checks.append('analysis_response_schema:' + (
+        'repairable' if any(
+            issue.get('code') in ANALYSIS_RESPONSE_SHAPE_CODES for issue in validation.get('issues') or []
+            if isinstance(issue, dict)
+        ) else 'passed'
+    ))
     limitations = list(validation.get("limitations") or [])
     if status == "partial":
         limitations.append("Only accepted native findings are retained; unresolved or unfinished work is not complete.")
diff --git a/application/single_app/functions_orchestration_recovery.py b/application/single_app/functions_orchestration_recovery.py
index c0c7a7b5f..42175d33a 100644
--- a/application/single_app/functions_orchestration_recovery.py
+++ b/application/single_app/functions_orchestration_recovery.py
@@ -80,6 +80,35 @@ def _retained_statuses(record):
     return {'completed', 'partial'}
 
 
+def _partial_dependency_recovery(record):
+    """Complete-only consumers must recover their partial producer, not replay it unchanged."""
+    saved_steps = {step['step_id']: step for step in _execution_steps(record)}
+    rerun, blocked = set(), set()
+    for step in record['plan'].get('steps') or []:
+        if not step.get('enabled', True):
+            continue
+        for spec in step_input_specs(step):
+            producer_id = spec.binding.step_id
+            saved = saved_steps.get(producer_id) or {}
+            if spec.allow_partial or saved.get('status') != 'partial':
+                continue
+            task = TaskResult.from_dict(saved['task_result']) if saved.get('task_result') is not None else None
+            if task is None:
+                raise CheckpointError('checkpoint_invalid')
+            completeness = task.output(spec.binding.output_name).completeness
+            if completeness.status != 'partial':
+                continue
+            if (
+                task.producer.capability_id == 'document_analyze'
+                and 'analysis_response_schema:passed' in completeness.checks
+                and completeness.coverage.expected == completeness.coverage.completed
+            ):
+                blocked.add(producer_id)
+            else:
+                rerun.add(producer_id)
+    return rerun, blocked
+
+
 def _stopped_without_effects(step):
     """Whether a failed step is known to have changed nothing outside the plan.
 
@@ -124,7 +153,7 @@ def _retained_producer_steps(record):
     ]
 
 
-def _reuse_invalidated_by_rerun(record, retained, *, new_attempt=True):
+def _reuse_invalidated_by_rerun(record, retained, *, new_attempt=True, partial_producers=None):
     """Retained dependency steps that a resume must run again.
 
     In a new attempt a render step always runs again. Its file belongs to the attempt
@@ -169,7 +198,11 @@ def _reuse_invalidated_by_rerun(record, retained, *, new_attempt=True):
         step['step_id']: {spec.binding.step_id for spec in step_input_specs(step) if spec.binding.step_id is not None}
         for step in steps if step['step_id'] in retained
     }
-    invalidated = renders | reads | redraws
+    if not new_attempt:
+        partial_producers = set()
+    elif partial_producers is None:
+        partial_producers, _ = _partial_dependency_recovery(record)
+    invalidated = renders | reads | redraws | partial_producers
     while True:
         added = {
             step_id for step_id, producers in consumed.items()
@@ -483,9 +516,10 @@ def recovery_projection(record):
         step['step_id'] for step in steps if step.get('status') in _retained_statuses(record)
     ]
     try:
+        partial_producers, blocked_partial_producers = _partial_dependency_recovery(record)
         stale = _reuse_invalidated_by_rerun(record, reused)
-    except (PlanValidationError, ResultContractError):
-        stale, invalid = set(), True
+    except (CheckpointError, PlanValidationError, ResultContractError):
+        stale, partial_producers, blocked_partial_producers, invalid = set(), set(), set(), True
     reused = [step_id for step_id in reused if step_id not in stale]
     retry = [
         step['step_id'] for step in record.get('plan', {}).get('steps') or []
@@ -529,6 +563,12 @@ def recovery_projection(record):
         reason, message = 'checkpoint_unavailable', build_failure('checkpoint_unavailable')['message']
     elif repeats_refusal:
         reason, message = 'retry_would_repeat', build_failure('retry_would_repeat')['message']
+    elif blocked_partial_producers and not any(
+        step.get('status') == 'failed'
+        and (step.get('failure') or {}).get('code') != 'input_partial_not_accepted'
+        for step in steps
+    ):
+        reason, message = 'input_partial_not_recoverable', build_failure('input_partial_not_recoverable')['message']
     return {
         'eligible': reason is None, 'reason_code': reason, 'message': message,
         'expected_version': record.get('recovery_version'),
@@ -1145,7 +1185,9 @@ def validate_resume(record, context, settings, authorize, *, source_run_id=None,
     # it renders its own files instead of reusing the superseded attempt's.
     stale = set() if allow_waiting else _reuse_invalidated_by_rerun(record, {
         step_id for step_id, saved in source_steps.items() if saved.get('status') in _retained_statuses(source)
-    }, new_attempt=source_run_id is not None or bool(record.get('retry_of_run_id')))
+    }, new_attempt=source_run_id is not None or bool(record.get('retry_of_run_id')),
+        partial_producers=_partial_dependency_recovery(source)[0],
+    )
     payloads = {}
     state_fields = STATE_FIELDS + OPTIONAL_STATE_FIELDS + DEPENDENCY_STATE_FIELDS
     initial_state = {key: deepcopy(getattr(context, key, None)) for key in state_fields}
diff --git a/application/single_app/functions_orchestration_schema.py b/application/single_app/functions_orchestration_schema.py
index 61cbd2d35..9409a4c00 100644
--- a/application/single_app/functions_orchestration_schema.py
+++ b/application/single_app/functions_orchestration_schema.py
@@ -1799,6 +1799,10 @@ def summarize_plan(plan):
     'input_partial_not_accepted': (
         'An earlier step returned only a partial result, and this step requires complete results, so it did not run.'
     ),
+    'input_partial_not_recoverable': (
+        'The saved analysis has unresolved findings despite complete source coverage. Retrying the same '
+        'dependent steps cannot complete this request. Review the unresolved findings and revise the request or sources.'
+    ),
     'dependency_unavailable': 'A required dependency did not complete. This operation was not executed.',
     'file_publication_not_allowed': 'Gathering and reasoning cannot create downloadable files.',
     'checkpoint_unavailable': 'Progress could not be saved or verified. This attempt cannot safely resume.',
diff --git a/application/single_app/functions_workflow_result_store.py b/application/single_app/functions_workflow_result_store.py
index 02b67b929..5d80a188e 100644
--- a/application/single_app/functions_workflow_result_store.py
+++ b/application/single_app/functions_workflow_result_store.py
@@ -1142,6 +1142,15 @@ def prepare_analysis_attempt(self, identity, *, token, resume_from=None):
             self._write_analysis_lifecycle(row)
         except CosmosResourceExistsError:
             guard = self._analysis_guard(identity, required=True, writable=True, token=token)
+            if (
+                identity["scope_type"] == "orchestration" and self._orchestration_execution is not None
+                and resume_from is not None and guard.get("resume_from") is None
+                and guard.get("prepared") and guard.get("request_registered") is False
+                and guard.get("request_digest") is None
+            ):
+                # Bootstrap prepared this producer's generic fence before Analyze bound its lineage.
+                self._write_analysis_lifecycle(row, previous=guard)
+                guard = self._analysis_guard(identity, required=True, writable=True, token=token)
             _require_fields(guard, {"resume_from": resume_from})
         return {"binding": dict(identity)}
 
diff --git a/docs/explanation/features/ORCHESTRATION_CHECKPOINT_RECOVERY.md b/docs/explanation/features/ORCHESTRATION_CHECKPOINT_RECOVERY.md
index 0d0ee21e4..f127dc813 100644
--- a/docs/explanation/features/ORCHESTRATION_CHECKPOINT_RECOVERY.md
+++ b/docs/explanation/features/ORCHESTRATION_CHECKPOINT_RECOVERY.md
@@ -1,10 +1,18 @@
 # Orchestration Checkpoint Recovery
 
-**Version: 0.261.303**
+**Version: 0.261.309**
 
 Implemented in version: **0.261.105**, recorded in
 `application/single_app/config.py`.
 
+Partial-producer recovery implemented in version: **0.261.309**. A manual retry
+revisits malformed Analyze outputs required by complete-only consumers, repairing
+affected saved windows without repeating completed extraction work. Genuine
+unresolved findings with complete coverage do not offer an unchanged retry that
+would simply fail drafting again. Same-attempt continuation preserves its existing
+immutable receipts. See the
+[document derivation reliability fix](../fixes/ORCHESTRATION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md).
+
 Since **0.261.303**, a settings save made by a background task no longer
 invalidates saved progress, and an attempt that ended while a step was waiting
 can be retried. See the
diff --git a/docs/explanation/features/ORCHESTRATION_GATHER_REASON_RENDER.md b/docs/explanation/features/ORCHESTRATION_GATHER_REASON_RENDER.md
index 6d156ba07..0f5dda369 100644
--- a/docs/explanation/features/ORCHESTRATION_GATHER_REASON_RENDER.md
+++ b/docs/explanation/features/ORCHESTRATION_GATHER_REASON_RENDER.md
@@ -1,6 +1,14 @@
 # Gather / Reason / Render orchestration
 
-**Version: 0.261.140**
+**Version: 0.261.309**
+
+Document-derived handoff reliability implemented in version: **0.261.309**,
+recorded in `application/single_app/config.py`. Native Analyze requests explicit
+JSON field shapes and can correct malformed metadata within its existing window
+retry limit. Complete-only drafting remains blocked by genuine uncertainty,
+missing coverage, or invalid evidence. Manual recovery revisits affected partial
+producers instead of replaying the same refused consumer input. See the
+[document derivation reliability fix](../fixes/ORCHESTRATION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md).
 
 Foundation implemented in version: **0.261.125**; shared export source bindings
 implemented in version: **0.261.126**; application integration implemented in
diff --git a/docs/explanation/fixes/ORCHESTRATION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md b/docs/explanation/fixes/ORCHESTRATION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md
new file mode 100644
index 000000000..4743272db
--- /dev/null
+++ b/docs/explanation/fixes/ORCHESTRATION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md
@@ -0,0 +1,104 @@
+# Orchestration document derivation reliability fix
+
+**Version: 0.261.309**
+
+Fixed in version: **0.261.309**, recorded in
+`application/single_app/config.py`.
+
+## Issue and root cause
+
+A document-derived report could stop before drafting even when Analyze had read
+every source page. The native collector correctly refused malformed metadata, but
+the window retry loop only retried exceptions: findings-level validation issues
+became a partial result without a corrective model attempt. Recovery then reused
+that partial producer and retried a complete-only consumer, repeating the refusal.
+
+The investigated Azure incident processed all 20 pages in four windows and
+retained 22 of 28 candidate findings. Six candidates had `invalid_caveats`
+diagnostics; telemetry did not record the exact malformed field values or types.
+Compose rejected the partial input, so the Word renderer was never reached.
+This fix addresses the shared handoff, not that document or output format.
+
+## Bounded response correction
+
+`functions_document_analysis.py` gives explicit array-of-string requirements for
+qualifications and requests JSON mode through analysis invocation metadata.
+For parseable malformed responses, it supplies the original source window,
+previous response, and application-owned field paths to a corrective invocation.
+Sequential and concurrent windows use the existing per-window retry allowance;
+there is no additional unbounded repair loop.
+
+`functions_document_analysis_results.py` validates the correction before collecting
+source-grounded candidates. Findings cannot be reordered, dropped, or rewritten.
+Valid values and evidence remain unchanged, qualification text must be preserved
+verbatim, and malformed values or identities cannot be synthesized. Unknown
+support statuses can become only unresolved, not supported. Evidence still has
+to match the assigned original source window.
+
+If correction fails, accepted original candidates remain available as partial
+results. Missing coverage, unsupported claims, invalid evidence, and recorded
+issues still prevent complete-only composition. Cancellation and access loss
+abort work rather than publishing the retained fallback.
+
+## Recovery and compatibility
+
+`functions_document_analysis_checkpoints.py` recognizes saved shape diagnostics
+and repairs affected ancestor windows in a linked new attempt. Unaffected windows
+are reused; completed payloads from the original attempt are never rewritten.
+Source snapshots, authorization, request identity, and durable execution fences
+remain enforced.
+
+`functions_orchestration_analysis_results.py` records an analysis schema check in
+output completeness. `functions_orchestration_recovery.py` follows named
+complete-only dependencies, schedules affected partial producers, and invalidates
+their dependent retained outputs. Analyze results with complete coverage and
+schema-passed but unresolved findings require review or revised instructions;
+an unchanged retry is not advertised when it would only repeat that refusal.
+Older results without the schema check remain eligible for producer recovery.
+
+Same-attempt continuation retains its immutable partial receipt. A linked manual
+retry can attach Analyze parent lineage to a bootstrap-created generic guard only
+before request registration and behind the real orchestration execution fence.
+`functions_workflow_result_store.py` keeps token, lease, and conditional-write
+checks on that transition. `functions_orchestration_schema.py` supplies the safe
+unresolved-input recovery explanation.
+
+Ordinary search/RAG chat and the existing Compose correction loop are unchanged.
+Explicit native Analyze actions in chat and workflows share the collector and
+bounded response correction. This change does not enable partial consumption,
+add a setting, change a route, or require a new Azure resource.
+
+## Diagnostics
+
+Correction events report field counts, attempt counts, and corrected/partial/failed
+outcomes with hashed conversation and work-unit correlation. They do not include
+the private response or source content. Existing analysis validation and coverage
+diagnostics still distinguish malformed metadata from genuine incompleteness.
+
+## Validation
+
+`functional_tests/test_orchestration_document_derivation_reliability.py` runs real
+collectors, checkpoints, orchestration, and renderers with offline external I/O.
+It covers malformed qualifications, preservation of uncertainty and accepted
+candidates, bounded failed correction, concurrent execution, cancellation, access
+loss, and selective saved-window repair across chat, workflow, and orchestration.
+
+End-to-end cases use both explicitly selected and search-discovered sources and
+verify actual Markdown and Word file contents and durable publication. A linked
+retry verifies that malformed saved windows are corrected before Compose and
+that genuine unresolved findings do not offer a pointless unchanged retry.
+Existing evidence, result-store, execution-fence, dependency, Compose, rendering,
+and workflow suites provide compatibility coverage.
+
+Local validation results:
+
+| Coverage | Result |
+| --- | --- |
+| Derivation, analysis evidence/recovery, execution fences, orchestration dependencies/recovery, Compose, and durable workflow suites | 573 tests passed |
+| Structured and Office renderers, rendering access failures, check cadence, waiting, and resume suites | 579 tests and 82 subtests passed |
+| `python functional_tests/test_docs_app_surface_coverage.py` | 7 checks passed; generated inventory remained current |
+| `python functional_tests/test_docs_site_quality.py` | 6 checks passed |
+| `git diff --check` | Passed |
+
+These are local offline checks, not a production replay. Deployment and a
+controlled production retry remain separate administrative actions.
diff --git a/functional_tests/test_orchestration_document_derivation_reliability.py b/functional_tests/test_orchestration_document_derivation_reliability.py
new file mode 100644
index 000000000..bc4d41b48
--- /dev/null
+++ b/functional_tests/test_orchestration_document_derivation_reliability.py
@@ -0,0 +1,387 @@
+# test_orchestration_document_derivation_reliability.py
+"""Source-grounded schema correction and producer-aware document generation recovery.
+
+Version: 0.261.309
+Implemented in: 0.261.309
+
+Real collectors, producers, checkpoints, orchestration and renderers run offline.
+Malformed metadata must not erase uncertainty or rewrite accepted source findings.
+"""
+
+import importlib
+import io
+import json
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from docx import Document
+
+from test_document_analysis_final_results import run_analysis
+from test_document_analysis_work_recovery import make_checkpoints, run_saved
+from test_orchestration_deliverables import _claim_retry, _retry_request
+from test_orchestration_harness_execution import harness, initialized_application  # noqa: F401
+from test_orchestration_reason_render_pipeline import download_outputs, narrative_io  # noqa: F401
+from test_support.document_analysis import (
+    FixtureAnalysisClient,
+    document_analysis_runtime,
+    extract_fixture_findings,
+    original_document,
+)
+from test_support.orchestration_harness_execution import compose_step, decoded_frames, input_binding, render_step
+from test_support.orchestration_research import _definitions
+
+
+QUALIFICATION = 'The period is not stated.'
+
+
+def correction_reply(prompt):
+    if '<ResponseShapeCorrection>' in prompt:
+        raw = prompt.split('Previous response:\n', 1)[1].rsplit('\n</ResponseShapeCorrection>', 1)[0]
+        payload = json.loads(raw)
+        for container in [payload, *payload['findings']]:
+            for field in ('issues', 'caveats', 'notes'):
+                if isinstance(container.get(field), str):
+                    container[field] = [container[field]]
+        return payload
+    return json.loads(extract_fixture_findings(prompt))
+
+
+@pytest.mark.parametrize('field', ['caveats', 'issues', 'notes'])
+def test_malformed_text_metadata_is_corrected_without_erasing_uncertainty(field, monkeypatch):
+    documents = {'one': original_document('one', ['The service owner is Mira.'])}
+
+    def reply(prompt):
+        payload = correction_reply(prompt)
+        if '<ResponseShapeCorrection>' not in prompt:
+            target = payload if field == 'notes' else payload['findings'][0]
+            target[field] = QUALIFICATION
+        return json.dumps(payload)
+
+    with document_analysis_runtime(documents) as runtime:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        result, client = run_analysis(
+            runtime, documents, FixtureAnalysisClient(reply), max_retries_per_window=1,
+        )
+    assert len(client.calls) == 2
+    assert all(call['metadata']['json_output'] for call in client.calls)
+    assert client.calls[1]['metadata']['analysis_response_correction']
+    assert result['analysis_validation']['coverage']['completed_work_units'] == 1
+    assert result['analysis_validation']['status'] == ('partial' if field == 'issues' else 'valid')
+    if field == 'caveats':
+        assert result['authoritative_result']['value'][0]['caveats'] == [QUALIFICATION]
+    elif field == 'issues':
+        assert result['authoritative_result']['value'] == []
+        assert QUALIFICATION in json.dumps(result['analysis_diagnostics'])
+    else:
+        assert QUALIFICATION in result['analysis_reply']
+
+
+@pytest.mark.parametrize('change', ['values', 'evidence', 'status', 'drop_finding', 'drop_caveat', 'extra_call'])
+def test_unusable_correction_retains_original_partial_result_and_is_bounded(change, monkeypatch):
+    documents = {'one': original_document('one', ['The service owner is Mira.'])}
+
+    def reply(prompt):
+        payload = correction_reply(prompt)
+        if '<ResponseShapeCorrection>' not in prompt or change == 'extra_call':
+            payload['findings'][0]['caveats'] = QUALIFICATION
+        elif change == 'drop_finding':
+            payload['findings'] = []
+        elif change == 'drop_caveat':
+            payload['findings'][0]['caveats'] = []
+        elif change == 'values':
+            payload['findings'][0]['values']['owner'] = 'Invented owner'
+        elif change == 'evidence':
+            payload['findings'][0]['evidence'][0]['quote'] = 'Invented quote'
+        else:
+            payload['findings'][0]['status'] = 'unresolved'
+        return json.dumps(payload)
+
+    with document_analysis_runtime(documents) as runtime:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        result, client = run_analysis(
+            runtime, documents, FixtureAnalysisClient(reply), max_retries_per_window=1,
+        )
+    assert len(client.calls) == 2
+    assert result['coverage']['retries'] == 1
+    assert result['analysis_validation']['status'] == 'partial'
+    assert result['analysis_validation']['coverage']['completed_work_units'] == 1
+    assert result['analysis_validation']['issues'][0]['code'] == 'invalid_caveats'
+    assert result['analysis_diagnostics']['candidates'][0]['values'] == {'owner': 'Mira'}
+
+
+@pytest.mark.parametrize('kind', ['chat', 'workflow', 'orchestration'])
+def test_retry_repairs_only_malformed_saved_window_without_reextracting_good_windows(kind, monkeypatch):
+    documents = {'one': original_document('one', [
+        'The service owner is Mira.', 'Reviews take place every quarter.',
+    ])}
+
+    def first_reply(prompt):
+        payload = json.loads(extract_fixture_findings(prompt))
+        if '[Page 1, Chunk 1]' in prompt:
+            payload['findings'][0]['caveats'] = QUALIFICATION
+        return json.dumps(payload)
+
+    with document_analysis_runtime(documents) as runtime:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        first = make_checkpoints(documents, kind=kind)
+        initial, original_client = run_saved(
+            runtime, documents, first, client=FixtureAnalysisClient(first_reply),
+        )
+        assert initial['analysis_validation']['status'] == 'partial'
+        second = make_checkpoints(
+            documents, kind=kind, store=first.store, attempt='assistant-2', previous='assistant-1',
+        )
+        result, repair_client = run_saved(
+            runtime, documents, second,
+            client=FixtureAnalysisClient(lambda prompt: json.dumps(correction_reply(prompt))),
+            max_retries_per_window=1,
+        )
+    assert len(original_client.calls) == 2
+    assert len(repair_client.calls) == 1
+    assert '<ResponseShapeCorrection>' in repair_client.calls[0]['prompt']
+    assert result['analysis_validation']['status'] == 'valid'
+    assert result['authoritative_result']['value'][0]['values'] == {
+        'owner': 'Mira', 'review_frequency': 'quarterly',
+    }
+    assert result['authoritative_result']['value'][0]['caveats'] == [QUALIFICATION]
+    assert result['analysis_metrics']['recovery']['reused_windows'] == 1
+
+
+@pytest.mark.parametrize('supported', [False, True])
+def test_unknown_support_status_cannot_be_promoted_by_schema_correction(supported, monkeypatch):
+    documents = {'one': original_document('one', ['The service owner is Mira.'])}
+
+    def reply(prompt):
+        payload = correction_reply(prompt)
+        payload['findings'][0]['status'] = (
+            'supported' if supported else 'unresolved'
+        ) if '<ResponseShapeCorrection>' in prompt else 'unknown'
+        return json.dumps(payload)
+
+    with document_analysis_runtime(documents) as runtime:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        result, client = run_analysis(
+            runtime, documents, FixtureAnalysisClient(reply), max_retries_per_window=1,
+        )
+    assert len(client.calls) == 2
+    assert result['analysis_validation']['status'] == 'partial'
+    assert result['authoritative_result']['value'] == []
+    assert result['analysis_validation']['issues'][0]['code'] == (
+        'invalid_status' if supported else 'unresolved_finding'
+    )
+
+
+@pytest.mark.parametrize('malformed', ['non_object', 'values'])
+def test_malformed_findings_preserve_other_accepted_candidates_without_inventing_data(malformed, monkeypatch):
+    documents = {'one': original_document('one', ['The service owner is Mira.'])}
+
+    def reply(prompt):
+        payload = correction_reply(prompt)
+        if '<ResponseShapeCorrection>' not in prompt:
+            payload['findings'].append(
+                'Uninterpreted source content' if malformed == 'non_object' else {
+                    **payload['findings'][0], 'finding_key': 'other', 'values': 'Unknown',
+                }
+            )
+        else:
+            payload['findings'][1] = {
+                **payload['findings'][0], 'finding_key': 'other', 'values': {'owner': 'Invented'},
+            }
+        return json.dumps(payload)
+
+    with document_analysis_runtime(documents) as runtime:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        result, client = run_analysis(
+            runtime, documents, FixtureAnalysisClient(reply), max_retries_per_window=1,
+        )
+    assert len(client.calls) == 2
+    assert result['analysis_validation']['status'] == 'partial'
+    assert result['analysis_validation']['coverage']['completed_work_units'] == 1
+    assert result['authoritative_result']['value'][0]['values'] == {'owner': 'Mira'}
+    assert 'Invented' not in json.dumps(result['authoritative_result'])
+
+
+def test_concurrent_windows_use_bounded_correction_and_safe_correlated_events(monkeypatch):
+    documents = {'one': original_document('one', [
+        'The service owner is Mira.', 'Reviews take place every quarter.',
+    ])}
+    events = []
+    calls = []
+
+    def factory(metadata):
+        def invoke(prompt, **kwargs):
+            calls.append(kwargs['metadata'])
+            payload = correction_reply(prompt)
+            if '<ResponseShapeCorrection>' not in prompt:
+                payload['findings'][0]['caveats'] = QUALIFICATION
+            return json.dumps(payload)
+        return invoke
+
+    with document_analysis_runtime(documents) as runtime, ThreadPoolExecutor(max_workers=2) as executor:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        monkeypatch.setattr(runtime.producer, 'log_event', lambda message, **kwargs: events.append((message, kwargs)))
+        monkeypatch.setattr(
+            runtime.producer, 'workflow_log_context',
+            _definitions('functions_appinsights.py', names={'workflow_log_context'})['workflow_log_context'],
+        )
+        result, _ = run_analysis(
+            runtime, documents, max_retries_per_window=1, max_window_concurrency=2,
+            invoke_prompt_factory=factory, executor=executor, conversation_id='private-conversation',
+        )
+    assert result['analysis_validation']['status'] == 'valid'
+    assert len(calls) == 4
+    assert all(call['json_output'] for call in calls)
+    assert sum(bool(call.get('analysis_response_correction')) for call in calls) == 2
+    correction_events = [kwargs['extra'] for message, kwargs in events if 'Response correction finished' in message]
+    assert len(correction_events) == 2
+    assert all(event['status'] == 'corrected' and event['conversation_id_hash'] for event in correction_events)
+    assert QUALIFICATION not in json.dumps(correction_events)
+    assert 'private-conversation' not in json.dumps(correction_events)
+
+
+@pytest.mark.parametrize('interruption', ['cancel', 'access'])
+def test_correction_does_not_publish_after_cancellation_or_access_loss(interruption, monkeypatch):
+    documents = {'one': original_document('one', ['The service owner is Mira.'])}
+    cancelled = [False]
+
+    def reply(prompt):
+        payload = correction_reply(prompt)
+        if '<ResponseShapeCorrection>' not in prompt:
+            payload['findings'][0]['caveats'] = QUALIFICATION
+        elif interruption == 'access':
+            raise PermissionError('Current source access is unavailable.')
+        else:
+            cancelled[0] = True
+        return json.dumps(payload)
+
+    with document_analysis_runtime(documents) as runtime:
+        monkeypatch.setattr(runtime.producer, '_wait_analysis_retry', lambda *args: None)
+        error = PermissionError if interruption == 'access' else runtime.cancellation.MixedSourceCancellationError
+        with pytest.raises(error):
+            run_analysis(
+                runtime, documents, FixtureAnalysisClient(reply), max_retries_per_window=1,
+                cancel_requested=lambda: cancelled[0],
+            )
+
+
+def analysis_step():
+    return {
+        'step_id': 'analyze', 'capability_id': 'document_analyze',
+        'arguments': {
+            'document_ids': ['document-1'], 'doc_scope': 'personal',
+            'analysis_prompt': 'Read every source page and retain its findings.',
+        },
+        'outputs': [{'name': 'findings', 'kind': 'records-v1'}, {'name': 'coverage', 'kind': 'structured-v1'}],
+    }
+
+
+def document_steps(output_format):
+    draft = compose_step(inputs={
+        'findings': {'binding': input_binding('analyze', 'findings'), 'allow_partial': False},
+    })
+    profile = 'prepared_report_v1' if output_format == 'docx' else 'prepared_text_v1'
+    return [analysis_step(), draft, render_step('report', output_format, profile=profile)]
+
+
+@pytest.mark.parametrize('output_format', ['md', 'docx'])
+@pytest.mark.parametrize('search_sources', [False, True])
+def test_repaired_findings_reach_actual_composition_and_durable_document(
+    harness, narrative_io, output_format, search_sources, monkeypatch,
+):
+    def reply():
+        prompt = harness.model_calls[-1]['messages'][-1]['content']
+        if '<DocumentSlice>' in prompt:
+            payload = json.loads(narrative_io['reply']())
+            if '<ResponseShapeCorrection>' in prompt:
+                raw = prompt.split('Previous response:\n', 1)[1].rsplit('\n</ResponseShapeCorrection>', 1)[0]
+                payload = json.loads(raw)
+                payload['findings'][0]['caveats'] = [QUALIFICATION]
+            elif '[Page 1, Chunk 1]' in prompt:
+                payload['findings'][0]['caveats'] = QUALIFICATION
+            return json.dumps(payload)
+        return '# Derived document\n\nComplete finding 006. The period is not stated.'
+
+    steps = document_steps(output_format)
+    if search_sources:
+        search = importlib.import_module('functions_search')
+        monkeypatch.setattr(search, 'hybrid_search', lambda *args, **kwargs: [{
+            'document_id': 'document-1', 'chunk_id': 'document-1-chunk-1',
+            'chunk_sequence': 1, 'page_number': 1,
+            'chunk_text': 'Item 000 contains original source detail.', 'file_name': 'document-1.txt',
+        }])
+        steps[0]['arguments'].pop('document_ids')
+        steps[0]['inputs'] = {'sources': {'binding': input_binding('search', 'sources')}}
+        steps.insert(0, {
+            'step_id': 'search', 'capability_id': 'document_search',
+            'arguments': {'query': 'Find the source report.', 'doc_scope': 'personal'},
+        })
+    harness.create(
+        steps, replies=[reply] * 10,
+        seeds={'document_ids': ['document-1'], 'doc_scope': 'personal'},
+        original_seeds={'document_ids': ['document-1'], 'doc_scope': 'personal'},
+    )
+    execution = harness.prepare()
+    frames = execution.execute()
+    done = decoded_frames(frames)[-1]
+    assert done['status'] == 'completed', done
+    outputs, artifacts, payloads = download_outputs(harness)
+    assert len(outputs) == len(artifacts) == 1
+    content = payloads[f'report.{output_format}']
+    if output_format == 'docx':
+        text = '\n'.join(paragraph.text for paragraph in Document(io.BytesIO(content)).paragraphs)
+    else:
+        text = content.decode('utf-8')
+    assert 'Complete finding 006.' in text and QUALIFICATION in text
+    calls = harness.model_calls
+    assert all(call['response_format'] == {'type': 'json_object'} for call in calls[:-1])
+    assert len([call for call in calls if '<ResponseShapeCorrection>' in str(call['messages'])]) == 1
+
+
+@pytest.mark.parametrize('malformed', [True, False])
+def test_recovery_revisits_malformed_producer_but_does_not_offer_pointless_uncertainty_retry(
+    harness, narrative_io, malformed,
+):
+    def reply():
+        payload = json.loads(narrative_io['reply']())
+        if malformed:
+            payload['findings'][0]['caveats'] = QUALIFICATION
+        else:
+            payload['findings'][0]['issues'] = ['The source has contradictory requested values.']
+        return json.dumps(payload)
+
+    harness.create(
+        document_steps('md'), replies=[reply] * 10,
+        seeds={'document_ids': ['document-1'], 'doc_scope': 'personal'},
+        original_seeds={'document_ids': ['document-1'], 'doc_scope': 'personal'},
+    )
+    execution = harness.prepare()
+    frames = execution.execute()
+    done = decoded_frames(frames)[-1]
+    assert done['status'] != 'completed'
+    assert harness.blobs.file_uploads == 0
+    projection = harness.recovery.recovery_projection(harness.read())
+    if malformed:
+        assert projection['eligible'], projection
+        assert 'analyze' in projection['retry_step_ids']
+        assert 'analyze' not in projection['reused_step_ids']
+        child = _retry_request(harness, execution, 'repair-malformed-producer')
+
+        def repaired_reply():
+            prompt = harness.model_calls[-1]['messages'][-1]['content']
+            if '<ResponseShapeCorrection>' in prompt:
+                return json.dumps(correction_reply(prompt))
+            return '# Recovered document\n\nComplete finding 006.'
+
+        previous_calls = len(harness.model_calls)
+        harness.replies = [repaired_reply] * 10
+        resumed = _claim_retry(harness, child)
+        resumed_frames = resumed.execute()
+        resumed_done = decoded_frames(resumed_frames)[-1]
+        assert resumed_done['status'] == 'completed', resumed_done
+        new_calls = harness.model_calls[previous_calls:]
+        assert len(new_calls) == 3
+        assert all('<ResponseShapeCorrection>' in str(call['messages']) for call in new_calls[:-1])
+        assert harness.blobs.file_uploads == 1
+    else:
+        assert not projection['eligible'], projection
+        assert projection['reason_code'] == 'input_partial_not_recoverable'