diff --git a/.gitignore b/.gitignore index 68afc7721..87f1d925e 100644 --- a/.gitignore +++ b/.gitignore @@ -44,6 +44,9 @@ docs/papers/ # Generated replay-suite validation artefacts docs/generated/validation_reports/replay_suite_* +# Local live-turn diagnostic evidence +.fastpath_evidence/ + # Ignore directories and sensitive files ignore/ token.* diff --git a/src/backend/workflows/llm_step_executor.py b/src/backend/workflows/llm_step_executor.py index cfcbacbaf..e70441558 100644 --- a/src/backend/workflows/llm_step_executor.py +++ b/src/backend/workflows/llm_step_executor.py @@ -29,6 +29,14 @@ ) from .llm_call_telemetry import stamp_llm_call_timestamps from .prompt_metadata_resolution import resolve_model_prompt_variant +from .recovery_prompt_compaction import ( + RECOVERY_CONTEXT_FIELD_CHAR_BUDGET, + RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET, + TOTAL_TRUNCATION_MARKER, + compact_recovery_context_field, + is_recovery_decision_state, + render_compacted_recovery_field_text, +) from .turn_expected_outcome_contract import TurnExpectedOutcomeContract from .definitions import ( CHAT_ASSISTANT_WORKFLOW_ID, @@ -943,13 +951,37 @@ def _resolve_prompt_render( return None, None, {}, None -def _compose_llm_prompt( +def _append_recovery_total_truncation_marker( + *, + sections: list[str], + remaining_budget: int, + diagnostics: dict[str, Any], +) -> int: + if remaining_budget <= 0: + diagnostics["total_truncation_marker_rendered"] = False + return remaining_budget + marker = ( + TOTAL_TRUNCATION_MARKER + if len(TOTAL_TRUNCATION_MARKER) <= remaining_budget + else TOTAL_TRUNCATION_MARKER[:remaining_budget] + ) + if not marker: + diagnostics["total_truncation_marker_rendered"] = False + return remaining_budget + sections.append(marker) + diagnostics["total_truncation_marker_rendered"] = True + return remaining_budget - len(marker) + + +def _compose_llm_prompt_with_diagnostics( *, base_prompt: str, llm_policy: Mapping[str, Any], context: Mapping[str, Any], -) -> str: + workflow_state_id: str | None = None, +) -> tuple[str, dict[str, Any]]: sections: list[str] = [] + prompt_context_diagnostics: dict[str, Any] = {} prefix_text = _context_string( llm_policy.get("prompt_prefix_text") or llm_policy.get("system_preamble") @@ -969,20 +1001,155 @@ def _compose_llm_prompt( if isinstance(raw_context_fields, Sequence) and not isinstance( raw_context_fields, (str, bytes, bytearray) ): - for item in raw_context_fields: + context_field_items = list(raw_context_fields) + # JVNAUTOSCI-2514: the recovery-decision step inlines the full turn + # state; project each field to bounded runtime evidence and cap the + # total so the prompt cannot balloon to tens of thousands of tokens. + is_recovery = is_recovery_decision_state(workflow_state_id) + remaining_recovery_budget = RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET + recovery_diagnostics: dict[str, Any] | None = None + if is_recovery: + recovery_diagnostics = { + "schema_version": "recovery_context_compaction.v1", + "enabled": True, + "workflow_state_id": _context_string(workflow_state_id), + "field_char_budget": RECOVERY_CONTEXT_FIELD_CHAR_BUDGET, + "total_context_char_budget": RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET, + "candidate_context_field_count": len(context_field_items), + "rendered_context_field_count": 0, + "skipped_empty_context_field_count": 0, + "omitted_context_field_count": 0, + "rendered_context_chars": 0, + "total_context_budget_exhausted": False, + "total_truncation_marker_rendered": False, + "fields": [], + } + + for index, item in enumerate(context_field_items): if not isinstance(item, Mapping): continue context_key = _context_string(item.get("context_key")) if not context_key: continue value = context.get(context_key) + label = _context_string(item.get("label")) or context_key.replace("_", " ") + if is_recovery: + assert recovery_diagnostics is not None + if remaining_recovery_budget <= 0: + recovery_diagnostics["total_context_budget_exhausted"] = True + recovery_diagnostics["omitted_context_field_count"] = ( + len(context_field_items) - index + ) + break + projection = compact_recovery_context_field(value) + value_text = _serialise_prompt_context_value(projection.value) + field_diagnostics = dict(projection.diagnostics) + field_diagnostics.update( + { + "context_key": context_key, + "label": label, + "input_value_kind": type(value).__name__, + } + ) + if not value_text: + field_diagnostics["skipped_empty"] = True + recovery_diagnostics["skipped_empty_context_field_count"] = ( + int( + recovery_diagnostics.get( + "skipped_empty_context_field_count" + ) + or 0 + ) + + 1 + ) + recovery_diagnostics["fields"].append(field_diagnostics) + continue + header = f"{label}:\n" + body_budget = min( + RECOVERY_CONTEXT_FIELD_CHAR_BUDGET, + max(0, remaining_recovery_budget - len(header)), + ) + if body_budget <= 0: + recovery_diagnostics["total_context_budget_exhausted"] = True + recovery_diagnostics["omitted_context_field_count"] = ( + len(context_field_items) - index + ) + break + serialised_compacted_chars = len(value_text) + value_text = render_compacted_recovery_field_text( + value_text, + char_budget=body_budget, + ) + section = f"{header}{value_text}" + remaining_recovery_budget -= len(section) + field_diagnostics.update( + { + "skipped_empty": False, + "serialised_compacted_chars": serialised_compacted_chars, + "rendered_body_chars": len(value_text), + "rendered_section_chars": len(section), + "field_char_budget_used": body_budget, + "body_truncated_by_char_budget": ( + serialised_compacted_chars > body_budget + ), + } + ) + recovery_diagnostics["fields"].append(field_diagnostics) + recovery_diagnostics["rendered_context_field_count"] = ( + int(recovery_diagnostics.get("rendered_context_field_count") or 0) + + 1 + ) + recovery_diagnostics["rendered_context_chars"] = int( + recovery_diagnostics.get("rendered_context_chars") or 0 + ) + len(section) + sections.append(section) + continue value_text = _serialise_prompt_context_value(value) if not value_text: continue - label = _context_string(item.get("label")) or context_key.replace("_", " ") sections.append(f"{label}:\n{value_text}") - return "\n\n".join(section for section in sections if section).strip() + if recovery_diagnostics is not None: + if recovery_diagnostics["omitted_context_field_count"]: + remaining_recovery_budget = _append_recovery_total_truncation_marker( + sections=sections, + remaining_budget=remaining_recovery_budget, + diagnostics=recovery_diagnostics, + ) + recovery_diagnostics["context_char_budget_remaining"] = max( + 0, + remaining_recovery_budget, + ) + prompt_context_diagnostics["schema_version"] = ( + "llm_prompt_context_diagnostics.v1" + ) + prompt_context_diagnostics["workflow_state_id"] = _context_string( + workflow_state_id + ) + prompt_context_diagnostics["recovery_context_compaction"] = ( + recovery_diagnostics + ) + + rendered_prompt = "\n\n".join(section for section in sections if section).strip() + if prompt_context_diagnostics: + prompt_context_diagnostics["rendered_prompt_chars"] = len(rendered_prompt) + return rendered_prompt, prompt_context_diagnostics + + +def _compose_llm_prompt( + *, + base_prompt: str, + llm_policy: Mapping[str, Any], + context: Mapping[str, Any], + workflow_state_id: str | None = None, +) -> str: + rendered_prompt, _diagnostics = _compose_llm_prompt_with_diagnostics( + base_prompt=base_prompt, + llm_policy=llm_policy, + context=context, + workflow_state_id=workflow_state_id, + ) + return rendered_prompt def _validation_output_format(validation_policy: Mapping[str, Any]) -> str: @@ -1621,6 +1788,7 @@ def _build_result( required_tool_obligation_ledger: Mapping[str, Any] | None = None, max_tool_invocations: int | None = None, prompt_variant_selection: Mapping[str, Any] | None = None, + prompt_context_diagnostics: Mapping[str, Any] | None = None, ) -> WorkflowActionResult: validated_outputs, validation_summary = _apply_validation_policy( request=request, @@ -1672,6 +1840,10 @@ def _build_result( llm_step_envelope["max_tool_invocations"] = int(max_tool_invocations) if isinstance(prompt_variant_selection, Mapping) and prompt_variant_selection: llm_step_envelope["prompt_variant_selection"] = dict(prompt_variant_selection) + if isinstance(prompt_context_diagnostics, Mapping) and prompt_context_diagnostics: + llm_step_envelope["prompt_context_diagnostics"] = dict( + prompt_context_diagnostics + ) outputs = { "final_response": response_text, @@ -1700,6 +1872,8 @@ def _build_result( outputs["max_tool_invocations"] = int(max_tool_invocations) if isinstance(prompt_variant_selection, Mapping) and prompt_variant_selection: outputs["prompt_variant_selection"] = dict(prompt_variant_selection) + if isinstance(prompt_context_diagnostics, Mapping) and prompt_context_diagnostics: + outputs["prompt_context_diagnostics"] = dict(prompt_context_diagnostics) outputs.update(validated_outputs) validation_status = _context_string(validation_summary.get("status")).lower() if validation_status == "failed": @@ -1725,6 +1899,7 @@ def _build_timeout_failure_result( llm_calls: Sequence[Mapping[str, Any]], aux_llm_calls: Sequence[Mapping[str, Any]], prompt_variant_selection: Mapping[str, Any] | None = None, + prompt_context_diagnostics: Mapping[str, Any] | None = None, ) -> WorkflowActionResult: llm_step_envelope = { "execution_mode": "llm", @@ -1754,13 +1929,20 @@ def _build_timeout_failure_result( "base_prompt_concept_id" ) llm_step_envelope["prompt_variant_selection"] = dict(prompt_variant_selection) + if isinstance(prompt_context_diagnostics, Mapping) and prompt_context_diagnostics: + llm_step_envelope["prompt_context_diagnostics"] = dict( + prompt_context_diagnostics + ) + outputs: dict[str, Any] = { + "llm_step_envelope": llm_step_envelope, + "llm_calls": list(llm_calls), + "aux_llm_calls": list(aux_llm_calls), + } + if isinstance(prompt_context_diagnostics, Mapping) and prompt_context_diagnostics: + outputs["prompt_context_diagnostics"] = dict(prompt_context_diagnostics) return WorkflowActionResult( status="failed", - outputs={ - "llm_step_envelope": llm_step_envelope, - "llm_calls": list(llm_calls), - "aux_llm_calls": list(aux_llm_calls), - }, + outputs=outputs, error=f"workflow_llm_step_timeout:{timeout_detail}", ) @@ -1776,6 +1958,7 @@ def _run_direct_llm_step( llm_policy_map: Mapping[str, Any], validation_policy_map: Mapping[str, Any], prompt_variant_selection: Mapping[str, Any] | None = None, + prompt_context_diagnostics: Mapping[str, Any] | None = None, ) -> WorkflowActionResult: llm_client = request.environment.llm_client if llm_client is None or not hasattr(llm_client, "generate"): @@ -1805,6 +1988,8 @@ def _run_direct_llm_step( duration_ms=duration_ms, note="generic_llm_step_direct", ) + if isinstance(prompt_context_diagnostics, Mapping) and prompt_context_diagnostics: + llm_call_entry["prompt_context_diagnostics"] = dict(prompt_context_diagnostics) _record_workflow_llm_duration_for_entry( request, llm_call_entry, @@ -1830,6 +2015,7 @@ def _run_direct_llm_step( llm_calls=request.data.get("llm_calls") or [], aux_llm_calls=request.data.get("aux_llm_calls") or [], prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) @@ -1844,6 +2030,7 @@ def _run_gateway_llm_step_no_tools( llm_policy_map: Mapping[str, Any], validation_policy_map: Mapping[str, Any], prompt_variant_selection: Mapping[str, Any] | None = None, + prompt_context_diagnostics: Mapping[str, Any] | None = None, ) -> WorkflowActionResult: orchestrator, policy_state, registry_snapshot, user_concept_id, org_concept_id = ( _build_gateway_runtime(request) @@ -1922,6 +2109,11 @@ def _record_llm_call( entry["error_class"] = error_class.strip() if isinstance(failure_kind, str) and failure_kind.strip(): entry["failure_kind"] = failure_kind.strip() + if ( + isinstance(prompt_context_diagnostics, Mapping) + and prompt_context_diagnostics + ): + entry["prompt_context_diagnostics"] = dict(prompt_context_diagnostics) stamp_llm_call_timestamps(entry, duration_ms=duration_ms) llm_calls.append(entry) _record_workflow_llm_duration_for_entry( @@ -1970,6 +2162,7 @@ def _record_llm_call( llm_calls=llm_calls, aux_llm_calls=aux_llm_calls, prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) return _build_result( @@ -1987,6 +2180,7 @@ def _record_llm_call( llm_calls=llm_calls, aux_llm_calls=aux_llm_calls, prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) @@ -2055,10 +2249,11 @@ def execute_llm_step(request: WorkflowActionRequest) -> WorkflowActionResult: if prompt_variant_resolution.match_reason != "base_prompt": prompt_source = f"{prompt_source or 'prompt'}:model_variant" - rendered_prompt = _compose_llm_prompt( + rendered_prompt, prompt_context_diagnostics = _compose_llm_prompt_with_diagnostics( base_prompt=base_prompt_text, llm_policy=llm_policy_map, context=request.data, + workflow_state_id=request.workflow_state_id, ) if not rendered_prompt: return WorkflowActionResult( @@ -2109,6 +2304,7 @@ def execute_llm_step(request: WorkflowActionRequest) -> WorkflowActionResult: llm_policy_map=llm_policy_map, validation_policy_map=validation_policy_map, prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) if _tool_mode(llm_policy_map) != "allowed": @@ -2122,6 +2318,7 @@ def execute_llm_step(request: WorkflowActionRequest) -> WorkflowActionResult: llm_policy_map=llm_policy_map, validation_policy_map=validation_policy_map, prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) allowed_tools_raw = llm_policy_map.get("allowed_tools") @@ -2301,6 +2498,11 @@ def _record_llm_call( entry["error_class"] = error_class.strip() if isinstance(failure_kind, str) and failure_kind.strip(): entry["failure_kind"] = failure_kind.strip() + if ( + isinstance(prompt_context_diagnostics, Mapping) + and prompt_context_diagnostics + ): + entry["prompt_context_diagnostics"] = dict(prompt_context_diagnostics) stamp_llm_call_timestamps(entry, duration_ms=duration_ms) llm_calls.append(entry) _record_workflow_llm_duration_for_entry( @@ -2516,6 +2718,7 @@ def _build_simple_error_result( llm_calls=llm_calls, aux_llm_calls=aux_llm_calls, prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) orchestrator_result = shared_data.get("orchestrator_result") @@ -2588,6 +2791,7 @@ def _build_simple_error_result( required_tool_obligation_ledger=final_required_tool_obligation_ledger, max_tool_invocations=max_tool_invocations, prompt_variant_selection=prompt_variant_selection, + prompt_context_diagnostics=prompt_context_diagnostics, ) diff --git a/src/backend/workflows/recovery_prompt_compaction.py b/src/backend/workflows/recovery_prompt_compaction.py new file mode 100644 index 000000000..a10bf1b2f --- /dev/null +++ b/src/backend/workflows/recovery_prompt_compaction.py @@ -0,0 +1,207 @@ +"""Compaction of turn-execution recovery-decision LLM context. + +JVNAUTOSCI-2514. The conversation-turn ``recovery_decision`` step inlines the +whole accumulated turn state -- the selected-workflow trace, the workflow +discovery result, completion-gate evidence, prior recovery attempts, and a long +tail of guidance fields -- into a single system prompt via the step's +``llm_policy.context_fields``. Most of that payload is "just in case" bulk: the +selected-workflow trace alone embeds a second full copy of the discovery +result, and the discovery candidates carry verbose descriptions and routing +internals that the recovery decision never needs. For one real incident +(request ``2ea8b384``) the composed prompt reached 150,704 chars (~37K tokens) +to drive a 206-char decision; prefilling that on the local ``qwen3:8b`` model is +the dominant cost of every recovery call. + +This module is runtime prompt-shaping support, not recovery policy. It drops +duplicated/internal telemetry sub-payloads that are available elsewhere or are +execution bookkeeping, truncates oversized values, and enforces hard per-field +and total character budgets so the recovery prompt can never balloon again, +regardless of how large the live turn state grows. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + +RECOVERY_DECISION_STATE_ID = "recovery_decision" + +# Per-field and total character budgets for the recovery-decision prompt's +# context sections. The base instruction prompt (~6 KB) and response contract +# are composed separately and are never truncated by these budgets; only the +# inlined evidence fields are bounded here. +RECOVERY_CONTEXT_FIELD_CHAR_BUDGET = 4000 +RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET = 20000 + +# Recursive compaction thresholds applied within each context field value. +_MAX_STRING_CHARS = 600 +_MAX_LIST_ITEMS = 6 +_MAX_DEPTH = 6 + +_STRING_TRUNCATION_SUFFIX = "...[+{count} chars omitted]" +_LIST_TRUNCATION_ITEM = "...[+{count} more items]" +_DEPTH_TRUNCATION_MARKER = "...[truncated: nested too deep]" +FIELD_TRUNCATION_MARKER = "\n...[recovery evidence truncated to fit prompt budget]" +TOTAL_TRUNCATION_MARKER = "[remaining recovery evidence omitted to fit prompt budget]" + +# Keys dropped anywhere inside a recovery context value. These are duplicate +# payloads already supplied by dedicated context fields, selector bookkeeping, or +# low-level timing/cache telemetry. The support surface deliberately preserves +# payloads that are not listed here, including failed tools, failure codes, +# unresolved preconditions, missing/required tools, execution summaries, and +# selected workflow IDs. +_DROP_KEYS = frozenset( + { + # Whole-payload duplicates already supplied as their own context fields. + "workflow_discovery_result", + "expected_outcome_contract", + "expected_outcome_contract_state", + # Selector provenance / fast-path bookkeeping: not needed to choose the + # next recovery action. + "selector_context_lineage", + "selector_represented_fast_path", + "selector_selection_metadata", + "selector_fast_path_policy", + "selector_fast_path_policy_diagnostics", + # Per-candidate routing internals; the decision uses ids + verdicts. + "routing_index_metadata", + "routing_readiness_detail", + "routing_readiness_diagnostics", + "routing_profile", + "executability_detail", + # Caches and fine-grained timing telemetry. + "workflow_discovery_cache", + "stage_timings", + } +) + + +@dataclass(frozen=True) +class RecoveryContextCompactionResult: + value: Any + diagnostics: dict[str, Any] + + +def is_recovery_decision_state(workflow_state_id: Any) -> bool: + """Return True when the workflow state is the recovery-decision step.""" + + return str(workflow_state_id or "").strip().lower() == RECOVERY_DECISION_STATE_ID + + +def _new_stats() -> dict[str, Any]: + return { + "changed": False, + "dropped_key_count": 0, + "dropped_keys": {}, + "truncated_string_count": 0, + "truncated_string_omitted_chars": 0, + "truncated_list_count": 0, + "truncated_list_omitted_items": 0, + "depth_truncation_count": 0, + } + + +def _record_dropped_key(stats: dict[str, Any], key: str) -> None: + stats["changed"] = True + stats["dropped_key_count"] = int(stats.get("dropped_key_count") or 0) + 1 + dropped_keys = stats.get("dropped_keys") + if not isinstance(dropped_keys, dict): + dropped_keys = {} + stats["dropped_keys"] = dropped_keys + dropped_keys[key] = int(dropped_keys.get(key) or 0) + 1 + + +def _compact_value(value: Any, *, depth: int, stats: dict[str, Any]) -> Any: + if depth > _MAX_DEPTH: + stats["changed"] = True + stats["depth_truncation_count"] = ( + int(stats.get("depth_truncation_count") or 0) + 1 + ) + return _DEPTH_TRUNCATION_MARKER + if isinstance(value, Mapping): + compacted: dict[str, Any] = {} + for raw_key, raw_value in value.items(): + key = str(raw_key) + if key in _DROP_KEYS: + _record_dropped_key(stats, key) + continue + compacted[key] = _compact_value( + raw_value, + depth=depth + 1, + stats=stats, + ) + return compacted + if isinstance(value, (list, tuple)): + items = [ + _compact_value(item, depth=depth + 1, stats=stats) + for item in list(value)[:_MAX_LIST_ITEMS] + ] + extra = len(value) - _MAX_LIST_ITEMS + if extra > 0: + stats["changed"] = True + stats["truncated_list_count"] = ( + int(stats.get("truncated_list_count") or 0) + 1 + ) + stats["truncated_list_omitted_items"] = ( + int(stats.get("truncated_list_omitted_items") or 0) + extra + ) + items.append(_LIST_TRUNCATION_ITEM.format(count=extra)) + return items + if isinstance(value, str): + if len(value) > _MAX_STRING_CHARS: + omitted = len(value) - _MAX_STRING_CHARS + stats["changed"] = True + stats["truncated_string_count"] = ( + int(stats.get("truncated_string_count") or 0) + 1 + ) + stats["truncated_string_omitted_chars"] = ( + int(stats.get("truncated_string_omitted_chars") or 0) + omitted + ) + return value[ + :_MAX_STRING_CHARS + ].rstrip() + _STRING_TRUNCATION_SUFFIX.format(count=omitted) + return value + return value + + +def compact_recovery_context_value(value: Any) -> Any: + """Return a structurally compacted copy of a recovery context value. + + Drops duplicated/internal sub-payloads, truncates long strings, and caps + long lists. Returns the same scalar for non-container inputs. + """ + + return compact_recovery_context_field(value).value + + +def compact_recovery_context_field(value: Any) -> RecoveryContextCompactionResult: + """Return compacted recovery context plus projection diagnostics.""" + + stats = _new_stats() + compacted = _compact_value(value, depth=0, stats=stats) + stats.update( + { + "schema_version": "recovery_context_field_compaction.v1", + "max_string_chars": _MAX_STRING_CHARS, + "max_list_items": _MAX_LIST_ITEMS, + "max_depth": _MAX_DEPTH, + } + ) + return RecoveryContextCompactionResult(value=compacted, diagnostics=stats) + + +def render_compacted_recovery_field_text( + serialised_text: str, + *, + char_budget: int, +) -> str: + """Apply the per-field character cap to an already-serialised section body.""" + + if char_budget <= 0: + return "" + if len(serialised_text) <= char_budget: + return serialised_text + if char_budget <= len(FIELD_TRUNCATION_MARKER): + return FIELD_TRUNCATION_MARKER[:char_budget] + body_budget = char_budget - len(FIELD_TRUNCATION_MARKER) + return serialised_text[:body_budget].rstrip() + FIELD_TRUNCATION_MARKER diff --git a/tests/backend/test_llm_step_executor.py b/tests/backend/test_llm_step_executor.py index 96a8b18e3..d52902c0f 100644 --- a/tests/backend/test_llm_step_executor.py +++ b/tests/backend/test_llm_step_executor.py @@ -15,8 +15,14 @@ from src.backend.workflows.conversation_turn_llm_timeout import ( DEFAULT_CONVERSATION_TURN_LLM_TIMEOUT_SEC, ) -from src.backend.workflows.llm_step_executor import execute_llm_step -from src.backend.workflows.llm_step_executor import _compose_llm_prompt +from src.backend.workflows.llm_step_executor import ( + _compose_llm_prompt, + _compose_llm_prompt_with_diagnostics, + execute_llm_step, +) +from src.backend.workflows.recovery_prompt_compaction import ( + RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET, +) from src.backend.workflows import prompt_metadata_resolution as pmr @@ -90,6 +96,120 @@ def test_compose_llm_prompt_includes_workflow_experience_guidance_labels() -> No assert "Inspect telemetry before asking the user." in prompt +def test_compose_llm_prompt_compacts_recovery_context_with_diagnostics() -> None: + huge_description = "candidate detail " * 800 + huge_internal_payload = "internal routing payload " * 1200 + prompt, diagnostics = _compose_llm_prompt_with_diagnostics( + base_prompt="Return the recovery JSON.", + llm_policy={ + "context_fields": [ + { + "context_key": "selected_workflow_trace", + "label": "Selected Workflow Outcome", + }, + { + "context_key": "workflow_discovery_result", + "label": "Workflow Discovery Result", + }, + { + "context_key": "completion_gate_evidence_payload", + "label": "Completion Gate Evidence", + }, + { + "context_key": "completion_gate_loop_attempts", + "label": "Recovery Loop Attempts", + }, + { + "context_key": "extra_payload", + "label": "Extra Payload", + }, + ] + }, + context={ + "selected_workflow_trace": { + "workflow_id": "#V#tool_calling_workflow", + "failure_detail": "gmail_list_messages failed: invalid_grant", + "workflow_discovery_result": { + "duplicated_bulk": huge_internal_payload, + }, + "stage_timings": {"prefill": huge_internal_payload}, + }, + "workflow_discovery_result": { + "candidates": [ + { + "workflow_id": f"#V#candidate_{index}", + "description": huge_description, + "routing_index_metadata": {"bulk": huge_internal_payload}, + } + for index in range(12) + ] + }, + "completion_gate_evidence_payload": { + "blocking_failure_codes": ["invalid_grant"], + "unresolved_preconditions": [ + { + "status_reason": ( + "gmail_list_messages returned invalid_grant for the " + "authenticated Gmail profile" + ), + } + ], + "long_internal_note": huge_internal_payload, + }, + "completion_gate_loop_attempts": 4, + "extra_payload": { + f"payload_{index}": huge_internal_payload for index in range(12) + }, + }, + workflow_state_id="recovery_decision", + ) + + assert len(prompt) <= RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET + 200 + assert "gmail_list_messages failed: invalid_grant" in prompt + assert "Recovery Loop Attempts:\n4" in prompt + assert "duplicated_bulk" not in prompt + assert '"stage_timings"' not in prompt + + recovery = diagnostics["recovery_context_compaction"] + assert recovery["enabled"] is True + assert recovery["rendered_context_chars"] <= RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET + assert recovery["rendered_context_field_count"] >= 4 + field_diagnostics = {field["context_key"]: field for field in recovery["fields"]} + assert ( + field_diagnostics["selected_workflow_trace"]["dropped_keys"][ + "workflow_discovery_result" + ] + == 1 + ) + assert ( + field_diagnostics["workflow_discovery_result"]["truncated_list_omitted_items"] + == 6 + ) + assert field_diagnostics["extra_payload"]["body_truncated_by_char_budget"] is True + + +def test_compose_llm_prompt_leaves_non_recovery_context_uncompacted() -> None: + long_value = "full context " * 700 + prompt, diagnostics = _compose_llm_prompt_with_diagnostics( + base_prompt="Use the selector policy.", + llm_policy={ + "context_fields": [ + {"context_key": "workflow_discovery_result", "label": "Discovery"} + ] + }, + context={ + "workflow_discovery_result": { + "workflow_discovery_result": {"nested": long_value}, + "description": long_value, + } + }, + workflow_state_id="selector_decision", + ) + + assert long_value in prompt + assert diagnostics == {} + + def test_execute_llm_step_parses_json_value_output() -> None: request = _build_request( llm_response='{"meeting_type":"project_meeting","title":"Roadmap sync"}' @@ -130,6 +250,72 @@ def test_execute_llm_step_recovers_embedded_json_value_output() -> None: assert result.outputs["validated_json_parse_mode"] == "embedded_json" +def test_execute_llm_step_records_recovery_prompt_compaction_diagnostics() -> None: + llm_client = MagicMock() + llm_client.generate.return_value = ( + '{"turn_next_action":{"action_type":"respond_with_follow_up",' + '"target_workflow_id":null,' + '"response_text":"Gmail retrieval failed; please reconnect Gmail.",' + '"tool_calls":null},' + '"reasoning":"The Gmail tool failed with invalid_grant."}' + ) + bulky_trace = { + "workflow_id": "#V#tool_calling_workflow", + "failure_detail": "gmail_list_messages failed: invalid_grant", + "workflow_discovery_result": {"bulk": "duplicated discovery " * 2000}, + "stage_timings": {"prefill": "timing detail " * 2000}, + } + request = WorkflowActionRequest( + action_id="llm.action", + inputs={}, + environment=WorkflowEnvironment(llm_client=llm_client, model="qwen3:8b"), + data={ + "selected_workflow_trace": bulky_trace, + "completion_gate_evidence_payload": { + "blocking_failure_codes": ["invalid_grant"], + "unresolved_preconditions": [ + {"status_reason": "gmail_list_messages failed: invalid_grant"} + ], + }, + "completion_gate_loop_attempts": 4, + }, + prompt_contract={"prompt_text": "Return recovery JSON only."}, + llm_policy={ + "context_fields": [ + { + "context_key": "selected_workflow_trace", + "label": "Selected Workflow Outcome", + }, + { + "context_key": "completion_gate_evidence_payload", + "label": "Completion Gate Evidence", + }, + { + "context_key": "completion_gate_loop_attempts", + "label": "Recovery Loop Attempts", + }, + ] + }, + validation_policy={"output_format": "json_value"}, + workflow_state_id="recovery_decision", + ) + + result = execute_llm_step(request) + + assert result.status == "success" + rendered_prompt = llm_client.generate.call_args.args[0] + assert "gmail_list_messages failed: invalid_grant" in rendered_prompt + assert "duplicated discovery" not in rendered_prompt + diagnostics = result.outputs["prompt_context_diagnostics"] + recovery = diagnostics["recovery_context_compaction"] + assert recovery["enabled"] is True + assert recovery["workflow_state_id"] == "recovery_decision" + assert recovery["rendered_context_chars"] <= RECOVERY_CONTEXT_TOTAL_CHAR_BUDGET + envelope = result.outputs["llm_step_envelope"] + assert envelope["prompt_context_diagnostics"] == diagnostics + assert result.outputs["llm_calls"][0]["prompt_context_diagnostics"] == diagnostics + + def test_execute_llm_step_records_duration_baseline_for_direct_workflow_step( monkeypatch, ) -> None: