Skip to content

Converge adaptive effect outcomes without erasing useful answers #389

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
witbrock merged 1 commit into from
Aug 17, 2026
Merged

Converge adaptive effect outcomes without erasing useful answers #389

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 80 additions & 11 deletions src/backend/integrations/internal_mcp/catalogue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13267,7 +13267,9 @@ def _summarise_effect_observation_journal(
for phase_name in (
"dispatch_intent",
"turn_terminal",
"current_state_observation",
"late_terminal",
"canonical_reconciliation",
):
raw_phase = raw_entry.get(phase_name)
if not isinstance(raw_phase, Mapping):
Expand All @@ -13276,13 +13278,29 @@ def _summarise_effect_observation_journal(
phase_projection = {
key: raw_phase.get(key)
for key in (
"schema_version",
"phase",
"recorded_at_utc",
"dispatch_state",
"execution_id",
"outcome",
"status",
"verified",
"effect_status",
"changed",
"initial_effect_status",
"current_outcome_status",
"outcome_resolved",
"reconciliation_basis",
"observation_identity_sha256",
"method_name",
"read_method_name",
"read_call_id",
"receipt_id",
"intent_fingerprint",
"target_concept_ids",
"canonical_scope",
"evidence_id",
"output_schema_validation",
"output_schema_valid",
"payload_truncated",
Expand Down Expand Up @@ -13326,27 +13344,78 @@ def _summarise_effect_observation_journal(
if key in receipt
}
entry[phase_name] = phase_projection
entry["historical_phases"] = list(available_phases)
entry["available_phases"] = available_phases
entry["latest_phase"] = available_phases[-1] if available_phases else None
late_terminal = raw_entry.get("late_terminal")
turn_terminal = raw_entry.get("turn_terminal")
recorded_phases = [
phase_name
for phase_name in available_phases
if str(raw_entry.get(phase_name, {}).get("recorded_at_utc") or "")
]
entry["latest_phase"] = (
max(
recorded_phases,
key=lambda phase_name: str(
raw_entry.get(phase_name, {}).get("recorded_at_utc") or ""
),
)
if recorded_phases
else (available_phases[-1] if available_phases else None)
)
# Keep each immutable receipt visible above. Current outcome is only a
# derived actor-safe projection, with exact canonical read-back taking
# precedence over earlier transport and turn-terminal observations.
current_phase = next(
(
phase_name
for phase_name in (
"canonical_reconciliation",
"late_terminal",
"current_state_observation",
"turn_terminal",
)
if isinstance(raw_entry.get(phase_name), Mapping)
),
None,
)
current_raw = raw_entry.get(current_phase) if current_phase else None
current_projection = entry.get(current_phase) if current_phase else None
entry["current_outcome"] = (
dict(current_projection)
if isinstance(current_projection, Mapping)
else None
)
outcome_resolved = False
if isinstance(late_terminal, Mapping):
late_payload = late_terminal.get("payload")
if current_phase in {
"canonical_reconciliation",
"current_state_observation",
} and isinstance(
current_raw, Mapping
):
outcome_resolved = (
current_raw.get("outcome_resolved")
if isinstance(current_raw.get("outcome_resolved"), bool)
else bool(
current_raw.get("effect_status")
not in {None, "indeterminate", "unknown"}
and isinstance(current_raw.get("changed"), bool)
)
)
elif current_phase == "late_terminal" and isinstance(current_raw, Mapping):
late_payload = current_raw.get("payload")
outcome_resolved = bool(
late_terminal.get("outcome") == "late_success"
and late_terminal.get("effect_status")
current_raw.get("outcome") == "late_success"
and current_raw.get("effect_status")
not in {None, "indeterminate", "unknown"}
and isinstance(late_payload, Mapping)
and late_payload.get("mutation_outcome") != "unknown"
)
elif isinstance(turn_terminal, Mapping):
transport = turn_terminal.get("transport")
receipt = turn_terminal.get("receipt")
elif current_phase == "turn_terminal" and isinstance(current_raw, Mapping):
transport = current_raw.get("transport")
receipt = current_raw.get("receipt")
outcome_resolved = bool(
isinstance(transport, Mapping)
and transport.get("outcome") not in {None, "timed_out"}
and turn_terminal.get("effect_status")
and current_raw.get("effect_status")
not in {None, "indeterminate", "unknown"}
and isinstance(receipt, Mapping)
and receipt.get("mutation_outcome") != "unknown"
Expand Down
14 changes: 5 additions & 9 deletions src/backend/integrations/internal_mcp/orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@
WORKFLOW_RESULT_ENVELOPE_KEY,
WORKFLOW_STEP_RESULT_ENVELOPES_KEY,
derive_workflow_terminal_status,
normalise_workflow_termination_code,
)
from ...workflows.mcp_tool_bridge import (
apply_namespace_to_mcp_payload,
Expand Down Expand Up @@ -27490,14 +27491,6 @@ def _record_durable_instance_submission_event(
finalise_episode_fn = None
build_episode_key_fn = None

def _normalise_error_code(error_value: Any) -> str:
if not isinstance(error_value, str) or not error_value.strip():
return "terminated"
cleaned_error = error_value.strip()
if ":" in cleaned_error:
return cleaned_error.split(":", 1)[0].strip().lower() or "terminated"
return "terminated"

def _record_episode_telemetry(
*,
completed: bool,
Expand Down Expand Up @@ -27718,7 +27711,10 @@ def _record_episode_telemetry(
termination_code = "completed"
termination_detail = None
elif error_value:
termination_code = _normalise_error_code(error_value)
termination_code = normalise_workflow_termination_code(
error_value,
default="terminated",
)
termination_detail = explicit_failure_detail or error_value
elif _workflow_final_state_is_failure_like(final_state):
termination_code = "failed_terminal_state"
Expand Down
22 changes: 11 additions & 11 deletions src/backend/server/routes/von_routes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11674,12 +11674,12 @@ def _progress_update(info: dict[str, Any]) -> None:
updated_conversation_situation_text = getattr(
adaptive_turn_result, "conversation_situation", None
)
effect_finality_fallback = bool(
adaptive_turn_result.effect_finality_fallback
)
response_authority = str(
getattr(adaptive_turn_result, "response_authority", "model") or "model"
).strip()
canonical_outcome_response = response_authority == "canonical_outcome"
adaptive_partial_delivery = (
adaptive_terminal_status == "effect_partially_completed"
and not effect_finality_fallback
and isinstance(response_text, str)
and bool(response_text.strip())
)
Expand Down Expand Up @@ -11752,9 +11752,9 @@ def _progress_update(info: dict[str, Any]) -> None:
{
"screen": response_text,
"spoken": response_text,
"format": "effect_finality_fallback_v1",
"format": "effect_outcome_report_v1",
}
if presenter_mode_requested and effect_finality_fallback
if presenter_mode_requested and canonical_outcome_response
else _extract_presenter_channels(response_text)
)
current_turn_messages = [{"role": "user", "content": prompt_text}]
Expand Down Expand Up @@ -11786,7 +11786,7 @@ def _progress_update(info: dict[str, Any]) -> None:
default=True
)

if presenter_mode_requested and not effect_finality_fallback:
if presenter_mode_requested and not canonical_outcome_response:
screen_tag_present = _presenter_tag_present(response_text, "screen")
screen_backfill_screen_tag_present = screen_tag_present
required_screen_json_fence = _extract_required_screen_json_fence(
Expand Down Expand Up @@ -12230,9 +12230,9 @@ def _tool_messages_include_description_write(
if not presenter_mode_requested:
screen_backfill_status = "skipped"
screen_backfill_suppression_reason = "presenter_mode_disabled"
elif effect_finality_fallback:
elif canonical_outcome_response:
screen_backfill_status = "skipped"
screen_backfill_suppression_reason = "effect_finality_literal"
screen_backfill_suppression_reason = "canonical_outcome_report"
elif not needs_screen_backfill:
screen_backfill_status = "skipped"
screen_backfill_suppression_reason = "not_required"
Expand Down Expand Up @@ -12596,9 +12596,9 @@ def _coerce_spoken_text(text: object) -> str | None:
if not presenter_mode_requested:
spoken_backfill_status = "skipped"
spoken_backfill_suppression_reason = "presenter_mode_disabled"
elif effect_finality_fallback:
elif canonical_outcome_response:
spoken_backfill_status = "skipped"
spoken_backfill_suppression_reason = "effect_finality_literal"
spoken_backfill_suppression_reason = "canonical_outcome_report"
elif not needs_spoken_backfill:
spoken_backfill_status = "skipped"
spoken_backfill_suppression_reason = "not_required"
Expand Down
Loading
Loading