From d4613594b81ffa87931f8aac646e39e91fc4e822 Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Fri, 9 Oct 2026 06:36:38 -0400 Subject: [PATCH] Fix orchestration document analysis handoffs and recovery Add bounded source-grounded response correction and rerun malformed partial producers without weakening completeness or evidence checks. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- application/single_app/config.py | 2 +- .../single_app/functions_document_analysis.py | 82 +++- ...functions_document_analysis_checkpoints.py | 9 + .../functions_document_analysis_results.py | 105 ++++- ...unctions_orchestration_analysis_results.py | 7 + .../functions_orchestration_recovery.py | 52 ++- .../functions_orchestration_schema.py | 4 + .../functions_workflow_result_store.py | 9 + .../ORCHESTRATION_CHECKPOINT_RECOVERY.md | 10 +- .../ORCHESTRATION_GATHER_REASON_RENDER.md | 10 +- ...ION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md | 104 +++++ ...tration_document_derivation_reliability.py | 387 ++++++++++++++++++ 12 files changed, 762 insertions(+), 19 deletions(-) create mode 100644 docs/explanation/fixes/ORCHESTRATION_DOCUMENT_DERIVATION_RELIABILITY_FIX.md create mode 100644 functional_tests/test_orchestration_document_derivation_reliability.py 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\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' + ) 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'|]*)?>', 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 '' in prompt: + raw = prompt.split('Previous response:\n', 1)[1].rsplit('\n', 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 '' 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 '' 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 '' 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 '' 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 '' 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 '' 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 '' 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 '' in prompt: + payload = json.loads(narrative_io['reply']()) + if '' in prompt: + raw = prompt.split('Previous response:\n', 1)[1].rsplit('\n', 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 '' 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 '' 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('' 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'