Fix orchestration review contracts and context-only recovery

application/single_app/config.py

@@ -103,7 +103,7 @@
 EXECUTOR_MAX_WORKERS = 30
 CHAT_STREAM_EXECUTOR_MAX_WORKERS = max(8, int(os.getenv('SIMPLECHAT_CHAT_STREAM_WORKERS', '32') or '32'))
 SESSION_TYPE = 'filesystem'
-VERSION = "0.261.417"
+VERSION = "0.261.418"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 # Opt-out for deployments where App Service Easy Auth is active but the platform

application/single_app/functions_orchestration_composition.py

@@ -61,7 +61,7 @@
     build_step_result, failure_from_exception, safe_failure, step_input_specs, validate_inline_output_schema,
 )
 from functions_orchestration_visuals import (
-    build_answer_visual_guidance, build_existing_charts_note, collect_run_charts,
+    any_visual, build_answer_visual_guidance, build_existing_charts_note, collect_run_charts,
     image_proposals_available, place_chart_blocks,
 )
 from functions_orchestration_workflow_context import resolve_turn_time_zone, workflow_answer_time_line
@@ -490,14 +490,16 @@
                    'value': read_complete_input(reader)}
             for name, reader in readers.items() if name != reference.output_name
         }
+        legacy_evidence = dict(evidence)
+        evidence['user_provided_content'] = context.user_request
         store = getattr(context, 'completeness_store', None)
         binding = getattr(context, 'completeness_binding', None)
         saved = store.load_completeness(step['step_id'], binding=binding) if store is not None else None
         invoke = context.invoke_prompt
         # Context budgeting is an execution dependency, not a bootstrap dependency.
         from functions_workflow_context import WorkflowContextBudgetError, calculate_workflow_context_budget
 
-        def invoke_check(messages, stage):
+        def invoke_check(messages, stage, *, response_format=None):
             recheck()
             audit = calculate_workflow_context_budget(
                 messages, getattr(invoke, 'model_metadata', None) or context.gpt_model or '',
@@ -507,6 +509,7 @@
                 raise WorkflowContextBudgetError(audit)
             reply = invoke(messages, stage=stage, metadata={
                 'run_id': context.run_id, 'step_id': step['step_id'], 'json_output': True,
+                **({'response_format': response_format} if response_format is not None else {}),
             })
             recheck()
             return reply
@@ -533,6 +536,7 @@
             evidence=evidence, policy=policy, invoke=invoke_check,
             validate=lambda value: [], checkpoint=checkpoint, saved_state=saved, recheck=recheck,
             owner_step_id=step['step_id'],
+            evidence_basis='mixed' if legacy_evidence else 'request', legacy_evidence=legacy_evidence,
             on_error=lambda error_type: log_event(
                 '[ORCHESTRATION_EXECUTOR] Retained final-answer verification unavailable.',
                 level=logging.WARNING, extra={'error_type': error_type, 'step_id': step['step_id']},
@@ -785,6 +789,11 @@
         check_policy = completeness_policy(
             settings, 'chat_orchestration',
             semantic_review=step['arguments'].get('semantic_review', 'auto'),
+            allow_semantic_error_skip=(
+                step['step_id'] in getattr(context, 'completeness_context_only_step_ids', ())
+                and not readers and not unavailable and not images and not any_visual(visuals) and not charts
+                and callable(getattr(context, 'revalidate_conversation_context', None))
+            ),
         )
         completeness_store = getattr(context, 'completeness_store', None)
         completeness_binding = getattr(context, 'completeness_binding', None)
@@ -798,6 +807,8 @@
                 saved_completeness = deepcopy(retained_check)
                 # This is an explicit new checking attempt, never a replay of an ambiguous correction.
                 if saved_completeness.get('status') in ('verification_unavailable', 'incomplete'):
+                    if saved_completeness['status'] == 'verification_unavailable':
+                        saved_completeness['retry_semantic_error'] = saved_completeness.get('reason_code')
                     saved_completeness.update(
                         status='checking', reason_code='', report_repairs=0,
                         deterministic_status='pending', semantic_status='pending',
@@ -826,7 +837,7 @@
                 visual_rules = _compose_visual_rules(step, prepared)
                 return [visual_rules] if visual_rules else []
 
-            def invoke_check(check_messages, stage):
+            def invoke_check(check_messages, stage, *, response_format=None):
                 recheck()
                 check_audit = calculate_workflow_context_budget(
                     check_messages, getattr(invoke, 'model_metadata', None) or context.gpt_model or '',
@@ -837,17 +848,31 @@
                     raise WorkflowContextBudgetError(check_audit)
                 reply = invoke(check_messages, stage=stage, metadata={
                     **metadata, 'json_output': stage == 'completeness_check' or not plain_text,
+                    **({'response_format': response_format} if response_format is not None else {}),
                 })
                 recheck()
                 return normalized_candidate(reply) if stage == 'completeness_correct' else reply
 
             def save_completeness(state):
                 state['model_binding'] = step.get('model_binding') or {'model_id': str(context.gpt_model or '')}
                 state['owner_step_id'] = step['step_id']
+                if state['status'] == 'complete' and state.get('semantic_skip_reason'):
+                    context.revalidate_conversation_context()
                 if completeness_store is not None:
                     completeness_store.save_completeness(
                         step['step_id'], state, binding=completeness_binding,
                     )
+                if state['status'] == 'complete' and state.get('semantic_skip_reason'):
+                    log_event(
+                        '[ORCHESTRATION_EXECUTOR] Optional answer review skipped after a checker error.',
+                        level=logging.INFO, extra={
+                            **workflow_log_context(
+                                run_id=context.run_id, conversation_id=getattr(context, 'conversation_id', None),
+                                step_id=step['step_id'],
+                            ),
+                            'semantic_outcome': 'skipped', 'reason_code': state['semantic_skip_reason'],
+                        },
+                    )
                 if callable(emit):
                     emit({
                         'type': 'step', 'step_id': step['step_id'], 'phase': 'running',
@@ -856,6 +881,7 @@
                             else 'Checking the prepared answer.' if state['status'] not in ('complete', 'incomplete', 'verification_unavailable')
                             else 'Answer prepared; checking could not finish.' if state['status'] == 'verification_unavailable'
                             else 'Answer needs an adjustment.' if state['status'] == 'incomplete'
+                            else 'Prepared answer; required checks completed.' if state.get('semantic_skip_reason')
                             else 'Prepared answer checks completed.'
                         ),
                         'task_completeness': public_completeness_report(state),
@@ -876,6 +902,8 @@
                 **({'saved_memory': memory['context_messages']} if memory.get('context_messages') else {}),
                 **({'conversation_history': history} if history else {}),
             }
+            legacy_evidence = dict(completeness_evidence)
+            completeness_evidence['user_provided_content'] = context.user_request
             response, completeness_state = check_and_correct(
                 response,
                 instruction={
@@ -898,6 +926,13 @@
                 policy=check_policy, invoke=invoke_check, validate=validate_candidate,
                 checkpoint=save_completeness, saved_state=saved_completeness, recheck=recheck,
                 owner_step_id=step['step_id'],
+                evidence_basis=(
+                    'mixed' if inputs or missing or images else
+                    'user_message' if step['arguments'].get('content_intent') in (
+                        'transform_supplied_text', 'draft_supplied_text',
+                    ) else 'request'
+                ),
+                legacy_evidence=legacy_evidence,
                 on_error=lambda error_type: log_event(
                     '[ORCHESTRATION_EXECUTOR] Task completeness verification unavailable.',
                     level=logging.WARNING, extra={'error_type': error_type, 'step_id': step['step_id']},

application/single_app/functions_orchestration_execution.py

@@ -189,8 +189,10 @@
     WorkflowContextBudgetError,
     calculate_workflow_context_budget,
 )
+from functions_task_completeness import CompletenessError
 from model_endpoint_clients import (
-    ModelEndpointBehavior, extract_chat_completion_response_text, is_response_format_rejection,
+    ModelEndpointBehavior, create_completion_with_response_format,
+    extract_chat_completion_response_text, is_transient_model_error,
 )
 
 
@@ -321,27 +323,24 @@ def guarded(*args, **kwargs):
     return guarded
 
 
-def _create_answer_completion(model, messages, *, json_output, stage):
-    """Request JSON mode when asked, resending without it if the endpoint refuses the option."""
+def _create_answer_completion(model, messages, *, json_output, stage, response_format=None):
+    """Preserve the review schema, with bounded endpoint-format compatibility."""
     options = {
         "messages": messages, "temperature": ANSWER_TEMPERATURE,
         "max_tokens": ANSWER_MAX_TOKENS, "use_model_response_length": True,
     }
-    if not json_output:
-        return model.create_completion(**options)
-    try:
-        return model.create_completion(**options, response_format={"type": "json_object"})
-    except Exception as exc:
-        if not is_response_format_rejection(exc):
-            raise
-        # JSON-mode support differs by endpoint and API version; the prompt already asks
-        # for one JSON object, and the caller validates the reply either way.
+    def adjusted(details):
         log_event(
-            "[ORCHESTRATION] Retrying the answer model without a JSON response format.",
+            "[ORCHESTRATION] Retrying the answer model with a compatible response format.",
             level=logging.INFO,
-            extra={"stage": stage, "reason": "json_format_retry", "error_type": type(exc).__name__},
+            extra={"stage": stage, "reason": "json_format_retry", **details},
         )
-        return model.create_completion(**options)
+
+    return create_completion_with_response_format(
+        model.create_completion, options,
+        response_format=response_format or ({"type": "json_object"} if json_output else None),
+        on_adjustment=adjusted,
+    )
 
 
 def build_harness_invoke_prompt(model, *, token_usage, revalidate):
@@ -378,14 +377,20 @@ def invoke_prompt(prompt_text, stage="window_analysis", metadata=None):
         try:
             response = _create_answer_completion(
                 model, messages, json_output=(metadata or {}).get("json_output") is True, stage=stage,
+                response_format=(metadata or {}).get("response_format"),
             )
-        except (OpenAIError, AzureError) as exc:
-            log_event(
-                "[ORCHESTRATION] The answer model request failed.",
-                level=logging.WARNING,
-                extra={"stage": stage, "error_type": type(exc).__name__},
-            )
-            raise HarnessExecutionError("model_failed") from exc
+        except Exception as exc:
+            transient_check = stage == "completeness_check" and is_transient_model_error(exc)
+            if transient_check or isinstance(exc, (OpenAIError, AzureError)):
+                log_event(
+                    "[ORCHESTRATION] The answer model request failed.",
+                    level=logging.WARNING,
+                    extra={"stage": stage, "error_type": type(exc).__name__},
+                )
+                if transient_check:
+                    raise CompletenessError("completeness_checker_unavailable") from exc
+                raise HarnessExecutionError("model_failed") from exc
+            raise
         usage = getattr(response, "usage", None)
         for field in _USAGE_FIELDS:
             value = getattr(usage, field, None)
@@ -394,12 +399,16 @@ def invoke_prompt(prompt_text, stage="window_analysis", metadata=None):
         revalidate()
         assert_current_request_sources_available()
         if not response or not response.choices:
+            if stage == "completeness_check":
+                raise CompletenessError("completeness_report_invalid")
             raise HarnessExecutionError("model_failed")
         choice = response.choices[0]
         if choice.finish_reason == "content_filter" or getattr(choice.message, "refusal", None):
             raise HarnessExecutionError("model_failed")
         text = extract_chat_completion_response_text(response)
         if not text.strip():
+            if stage == "completeness_check":
+                raise CompletenessError("completeness_report_invalid")
             raise HarnessExecutionError("model_failed")
         return text
 

application/single_app/functions_orchestration_executor.py

@@ -82,6 +82,7 @@
 from functions_orchestration_results import ResultUnavailableError
 from functions_workflow_collections import CollectionSizeError
 from functions_workflow_result_store import WorkflowResultIntegrityError, WorkflowResultStorageUnavailableError
+from functions_task_completeness import context_only_review_step_ids
 from functions_orchestration_schema import (
     PLAN_STATUS_CANCELLED,
     PLAN_STATUS_COMPLETED,
@@ -1463,6 +1464,7 @@ def _execute_dependency_plan(
     context.completeness_binding = getattr(checkpoints, 'binding', None)
     context.completeness_final_step = (plan.get('final_response') or {}).get('step_id')
     context.completeness_final_output = (plan.get('final_response') or {}).get('output_name')
+    context.completeness_context_only_step_ids = context_only_review_step_ids(steps)
     context.completeness_plan_scope = [
         {'enabled': step['enabled'], 'instruction': step.get('arguments', {}).get('instruction', step.get('title', ''))}
         for step in steps

application/single_app/functions_orchestration_registry.py

@@ -1608,7 +1608,12 @@ def _document_transform_descriptor(capability_id, label, instruction):
             'narrative documents use response_policy: available_information: answer supported '
             'questions and explain unavailable or conflicting information normally. A data gap '
             'is not an execution failure. Use complete_inputs for exact datasets or an explicitly '
-            'complete-only transformation; never invent missing values.'
+            'complete-only transformation; never invent missing values. Set content_intent to '
+            'transform_supplied_text for rewriting, shortening, summarizing, translating or '
+            'reformatting supplied in-chat text, or draft_supplied_text for a simple prose draft '
+            'using only the user-provided account. Neither permits added knowledge, calculations, '
+            'recommendations or analysis. Other work is substantive. This classification describes '
+            'the actual task, independently of the selected model and model-routing category.'
         ),
         'settings_gates': (),
         'settings_gates_any': (),
@@ -1627,6 +1632,15 @@ def _document_transform_descriptor(capability_id, label, instruction):
                     'type': 'string', 'enum': ['auto', 'on', 'off'],
                     'description': 'Optional semantic review; required output validation always runs.',
                 },
+                'content_intent': {
+                    'type': 'string',
+                    'enum': ['transform_supplied_text', 'draft_supplied_text', 'substantive'],
+                    'description': (
+                        'Classify the actual content task. Only simple prose based on supplied '
+                        'in-chat text may recover from an optional Auto checker error; '
+                        'research, analysis, calculations and recommendations are substantive.'
+                    ),
+                },
                 'visuals': {
                     'type': 'array', 'items': {'type': 'string', 'enum': list(VISUAL_KINDS)},
                     'uniqueItems': True, 'maxItems': len(VISUAL_KINDS),