Skip to content

Make represented workflow Thinking progress meaningful #352

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 6, 2026
Merged

Make represented workflow Thinking progress meaningful #352

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
10 changes: 10 additions & 0 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 @@ -565,10 +565,19 @@ def _normalise_selected_workflow_execution_event(
"error",
"execution_mode",
"result_summary",
"effect_status",
"mutation_outcome",
"outcome_finality",
"failure_reason",
"next_action",
):
value = _progress_str(event.get(key)) or _progress_str(update.get(key))
if value:
normalised[key] = value
for key in ("changed", "semantic_effect"):
value = event.get(key)
if isinstance(value, bool):
normalised[key] = value
state_attempt = _progress_number(event.get("state_attempt"))
if state_attempt is not None:
normalised["state_attempt"] = int(max(0.0, state_attempt))
Expand Down Expand Up @@ -3839,6 +3848,7 @@ def _build_turn_execution_diagnostics(
"tool_pending_count",
"tool_call_start_count",
"tool_call_end_count",
"selected_workflow_execution",
"timing_spans",
)
latest_progress = (
Expand Down
281 changes: 250 additions & 31 deletions src/backend/services/adaptive_turn_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,9 @@
from src.backend.services.thinking_semantic_projection_service import (
build_semantic_operation_projection,
)
from src.backend.services.tool_evidence_projection_service import (
project_nested_workflow_progress_evidence,
)
from src.backend.services.turn_evidence_store import (
EVIDENCE_SLICE_SCHEMA_VERSION,
TrustedTurnScope,
Expand Down Expand Up @@ -3111,6 +3114,179 @@ def _emit(progress_tracker: Any, payload: Mapping[str, Any]) -> None:
emit(dict(payload))


def _bounded_workflow_progress_text(
value: Any,
*,
limit: int = 320,
) -> str | None:
if not isinstance(value, str):
return None
cleaned = " ".join(value.split()).strip()
if not cleaned:
return None
if len(cleaned) <= limit:
return cleaned
return f"{cleaned[: max(0, limit - 3)].rstrip()}..."


def _humanise_workflow_recovery_action(value: Any) -> str | None:
action_type = _bounded_workflow_progress_text(value, limit=120)
if not action_type:
return None
return action_type.replace("_", " ").strip().capitalize() or None


def _build_represented_workflow_execution_event(
*,
payload: Any,
workflow_id: str,
workflow_name: str | None,
) -> tuple[dict[str, Any], dict[str, Any] | None]:
"""Project one represented execution into the existing Thinking contract.

Workflow-authored progress facts remain the semantic authority. This bridge
only carries their bounded projection, plus typed execution and recovery
state already present in the durable workflow receipt.
"""

receipt = dict(payload) if isinstance(payload, Mapping) else {}
workflow_execution_raw = receipt.get("workflow_execution")
workflow_execution = (
dict(workflow_execution_raw)
if isinstance(workflow_execution_raw, Mapping)
else {}
)
workflow_instance_raw = receipt.get("workflow_instance")
workflow_instance = (
dict(workflow_instance_raw)
if isinstance(workflow_instance_raw, Mapping)
else {}
)
latest_step_raw = workflow_execution.get("latest_step_result_envelope")
latest_step = (
dict(latest_step_raw) if isinstance(latest_step_raw, Mapping) else {}
)
diagnostics_raw = latest_step.get("diagnostics")
diagnostics = (
dict(diagnostics_raw) if isinstance(diagnostics_raw, Mapping) else {}
)

progress_evidence = project_nested_workflow_progress_evidence(receipt)
progress_facts = (
list(progress_evidence.get("facts") or [])
if isinstance(progress_evidence, Mapping)
else []
)

effect_status = _bounded_workflow_progress_text(
receipt.get("effect_status"),
limit=80,
)
final_status = (
_bounded_workflow_progress_text(receipt.get("final_status"), limit=80)
or _bounded_workflow_progress_text(
workflow_execution.get("final_status"), limit=80
)
or _bounded_workflow_progress_text(
workflow_execution.get("current_status"), limit=80
)
or _bounded_workflow_progress_text(
workflow_instance.get("status"), limit=80
)
)
effect_status_key = (effect_status or "").lower()
final_status_key = (final_status or "").lower()
if effect_status_key == "succeeded":
event_status = "workflow_execution_complete"
elif effect_status_key == "partial":
event_status = "workflow_execution_waiting"
elif effect_status_key == "indeterminate":
event_status = "workflow_execution_indeterminate"
elif effect_status_key in {"failed", "not_started"}:
event_status = "workflow_execution_failed"
elif final_status_key in {"completed", "complete", "succeeded", "success"}:
event_status = "workflow_execution_complete"
elif final_status_key in {"pending", "queued", "running", "paused"}:
event_status = "workflow_execution_waiting"
else:
event_status = "workflow_execution_failed"

error = (
_bounded_workflow_progress_text(
workflow_execution.get("failure_reason"), limit=320
)
or _bounded_workflow_progress_text(
receipt.get("failure_reason"), limit=320
)
or _bounded_workflow_progress_text(diagnostics.get("error"), limit=320)
or _bounded_workflow_progress_text(
workflow_execution.get("error"), limit=320
)
or _bounded_workflow_progress_text(receipt.get("error_code"), limit=160)
)
recovery_affordances = receipt.get("recovery_affordances")
next_action = None
if isinstance(recovery_affordances, Sequence) and not isinstance(
recovery_affordances,
(str, bytes, bytearray),
):
for affordance in recovery_affordances:
if not isinstance(affordance, Mapping):
continue
next_action = _humanise_workflow_recovery_action(
affordance.get("action_type")
)
if next_action:
break

event: dict[str, Any] = {
"schema_version": "selected_workflow_execution_event.v1",
"status": event_status,
"event_kind": event_status,
"workflow_id": workflow_id,
"selected_workflow_id": workflow_id,
"selected_execution_mode": "adaptive_turn_capability",
}
optional_text = {
"selected_workflow_name": workflow_name,
"state_id": latest_step.get("state_id")
or workflow_execution.get("current_state"),
"action_id": latest_step.get("action_id"),
"action_status": latest_step.get("action_status"),
"action_outcome": latest_step.get("action_outcome"),
"final_state": workflow_execution.get("current_state")
or workflow_instance.get("current_state"),
"error": error,
"effect_status": effect_status,
"mutation_outcome": receipt.get("mutation_outcome"),
"outcome_finality": receipt.get("outcome_finality"),
"next_action": next_action,
}
for key, value in optional_text.items():
text_value = _bounded_workflow_progress_text(value)
if text_value:
event[key] = text_value
for key in ("changed", "semantic_effect"):
if isinstance(receipt.get(key), bool):
event[key] = receipt[key]
if isinstance(latest_step.get("state_attempt"), (int, float)) and not isinstance(
latest_step.get("state_attempt"), bool
):
event["state_attempt"] = max(0, int(latest_step["state_attempt"]))
if isinstance(diagnostics.get("duration_ms"), (int, float)) and not isinstance(
diagnostics.get("duration_ms"), bool
):
event["duration_ms"] = max(0, int(diagnostics["duration_ms"]))
if progress_facts:
event["progress_facts"] = progress_facts
progress_payload = (
dict(progress_evidence)
if isinstance(progress_evidence, Mapping)
else None
)
return event, progress_payload


def _check_cancellation(progress_tracker: Any) -> None:
if progress_tracker is None:
return
Expand Down Expand Up @@ -5113,21 +5289,31 @@ def contained(
lifecycle_status="running",
capability_display_name=item.capability_display_name,
)
_emit(
progress_tracker,
{
"status": "tool_call_start",
"event_kind": "tool_call_start",
"stage": "adaptive_research",
"phase": "adaptive_research",
"tool": canonical_name,
"subtask": semantic_operation["capability"]["label"],
"execution_method": execution_method_name,
"call_id": call.call_id,
"result_summary": semantic_operation["summary"],
"semantic_operation": semantic_operation,
},
)
start_progress: dict[str, Any] = {
"status": "tool_call_start",
"event_kind": "tool_call_start",
"stage": "adaptive_research",
"phase": "adaptive_research",
"tool": canonical_name,
"subtask": semantic_operation["capability"]["label"],
"execution_method": execution_method_name,
"call_id": call.call_id,
"result_summary": semantic_operation["summary"],
"semantic_operation": semantic_operation,
}
if item.capability_kind == "represented_workflow" and (
item.represented_workflow_id
):
start_progress["selected_workflow_execution_event"] = {
"schema_version": "selected_workflow_execution_event.v1",
"status": "workflow_execution_start",
"event_kind": "workflow_execution_start",
"workflow_id": item.represented_workflow_id,
"selected_workflow_id": item.represented_workflow_id,
"selected_workflow_name": item.capability_display_name,
"selected_execution_mode": "adaptive_turn_capability",
}
_emit(progress_tracker, start_progress)
with override_current_actor(
scope.user_concept_id,
scope.organisation_concept_id,
Expand Down Expand Up @@ -5437,6 +5623,20 @@ def contained(
else "ok"
)
)
selected_workflow_execution_event: dict[str, Any] | None = None
workflow_progress_evidence: dict[str, Any] | None = None
if (
contained_result.capability_kind == "represented_workflow"
and contained_result.represented_workflow_id
):
(
selected_workflow_execution_event,
workflow_progress_evidence,
) = _build_represented_workflow_execution_event(
payload=raw_payload,
workflow_id=contained_result.represented_workflow_id,
workflow_name=contained_result.capability_display_name,
)
envelope = evidence_store.record(
canonical_name,
call.call_id,
Expand Down Expand Up @@ -5471,6 +5671,10 @@ def contained(
status=effect_status or status,
)
envelope_payload = envelope.to_mapping()
if workflow_progress_evidence is not None:
envelope_payload["workflow_progress_evidence"] = dict(
workflow_progress_evidence
)
if isinstance(raw_payload, Mapping):
try:
from src.backend.services.tool_evidence_projection_service import (
Expand Down Expand Up @@ -5591,6 +5795,14 @@ def contained(
invocation["binding_diagnostics"] = dict(
contained_result.binding_diagnostics
)
if workflow_progress_evidence is not None:
invocation["workflow_progress_evidence"] = dict(
workflow_progress_evidence
)
if selected_workflow_execution_event is not None:
invocation["selected_workflow_execution_event"] = dict(
selected_workflow_execution_event
)
if is_effect:
invocation.update(
{
Expand Down Expand Up @@ -5711,22 +5923,29 @@ def contained(
success=status == "ok",
result=semantic_result,
)
_emit(
progress_tracker,
{
"status": "tool_completed",
"event_kind": "tool_call_end",
"stage": "adaptive_research",
"phase": "adaptive_research",
"tool": canonical_name,
"subtask": semantic_operation["capability"]["label"],
"execution_method": execution_method_name,
"call_id": call.call_id,
"success": status == "ok",
"result_summary": semantic_operation["summary"],
"semantic_operation": semantic_operation,
},
)
completed_progress: dict[str, Any] = {
"status": "tool_completed",
"event_kind": "tool_call_end",
"stage": "adaptive_research",
"phase": "adaptive_research",
"tool": canonical_name,
"subtask": semantic_operation["capability"]["label"],
"execution_method": execution_method_name,
"call_id": call.call_id,
"success": status == "ok",
"result_summary": semantic_operation["summary"],
"semantic_operation": semantic_operation,
}
if selected_workflow_execution_event is not None:
completed_progress["selected_workflow_execution_event"] = dict(
selected_workflow_execution_event
)
progress_facts = selected_workflow_execution_event.get(
"progress_facts"
)
if isinstance(progress_facts, list) and progress_facts:
completed_progress["progress_facts"] = list(progress_facts)
_emit(progress_tracker, completed_progress)

correlated_results = [
result for result in batch_results if isinstance(result, ToolResult)
Expand Down
Loading
Loading