diff --git a/docs/engineering/von_workflow_language_manual.md b/docs/engineering/von_workflow_language_manual.md index f4666d578..4cff415ee 100644 --- a/docs/engineering/von_workflow_language_manual.md +++ b/docs/engineering/von_workflow_language_manual.md @@ -458,6 +458,9 @@ Current fields: - `role`: `execution` | `authoring` | `maintenance` - `authoring_intent_required`: bool - `prefer_existing_capability`: bool +- `direct_equivalent_capability_names`: optional ordered list of registered + capabilities that are represented as producing the same material outcome and + evidence as the workflow with lower orchestration cost Current override-policy rule: @@ -471,6 +474,22 @@ Current override-policy rule: semantic-fit and role checks, routing MUST preserve the non-custom fallback path and record an explicit decline reason in routing diagnostics. +Ordinary-turn capability selection treats direct calls, small compositions, +and represented workflows as plans on one bounded frontier. The adaptive model +owns semantic adequacy: it may prefer a workflow when composition, verification, +durability, governance, or recovery is material, but representedness alone is +not a quality signal. Actor-visible registered tools declared by a matched +workflow are exposed as plausible simpler component plans without claiming that +one component is equivalent to the whole workflow. + +Mechanical dominance is deliberately narrower. A read-only workflow may be +removed from the active frontier only when its authoritative routing profile +explicitly names an actor-visible read-only direct equivalent. The workflow +remains addressable by exact name and visible in the complete catalogue for +inspection. Do not use `direct_equivalent_capability_names` for an approximate +shortcut, a component that omits material evidence, or merely because a +workflow is slow; those cases remain semantic decisions and evaluation inputs. + Current fallback discipline: - generic fallback role inference MAY remain only for legacy workflows whose routing metadata has not yet been backfilled; diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py index 909a6d77c..02be05caf 100644 --- a/src/backend/services/adaptive_turn_service.py +++ b/src/backend/services/adaptive_turn_service.py @@ -66,6 +66,11 @@ _MODEL_TOOL_RESULT_BATCH_MAX_BYTES = 24_000 _MODEL_EVIDENCE_PREVIEW_MAX_CHARS = 240 _CAPABILITY_CATALOGUE_SCHEMA_VERSION = "adaptive_turn_capabilities.v1" +_CAPABILITY_PLAN_PROFILE_SCHEMA_VERSION = "capability_plan_profile.v1" +_CAPABILITY_EFFECT_PROFILE_SCHEMA_VERSION = "capability_effect_profile.v1" +_CAPABILITY_COST_PROFILE_SCHEMA_VERSION = "capability_cost_profile.v1" +_CAPABILITY_SELECTION_POLICY_SCHEMA_VERSION = "capability_selection_policy.v1" +_CAPABILITY_SELECTION_TRACE_SCHEMA_VERSION = "capability_selection_trace.v1" _CONVERSATION_SITUATION_START_TAG = "" _CONVERSATION_SITUATION_END_TAG = "" _CONVERSATION_SITUATION_PROTOCOL_RE = re.compile( @@ -282,6 +287,79 @@ def _compact_evidence_index( return compacted +def _compact_capability_selection_profiles( + capability: Mapping[str, Any], +) -> dict[str, Any]: + """Retain decision-bearing plan facts without repeating catalogue policy.""" + + compact: dict[str, Any] = {} + effect_profile = capability.get("effect_profile") + if isinstance(effect_profile, Mapping): + compact["effect_profile"] = { + key: effect_profile.get(key) + for key in ( + "schema_version", + "semantic_effect", + "semantic_effect_source", + "operational_state_effect", + "operational_state_effect_reason", + ) + if key in effect_profile + } + + plan_profile = capability.get("plan_profile") + if isinstance(plan_profile, Mapping): + compact_plan = { + key: plan_profile.get(key) + for key in ( + "schema_version", + "shape", + "evidence_surface_families", + ) + if key in plan_profile + } + if plan_profile.get("shape") == "represented_workflow": + for key in ( + "component_capability_names", + "direct_equivalent_capability_names", + ): + if key in plan_profile: + compact_plan[key] = plan_profile.get(key) + cost_profile = plan_profile.get("cost_profile") + if isinstance(cost_profile, Mapping): + compact_plan["cost_profile"] = dict(cost_profile) + compact["plan_profile"] = compact_plan + + selection = capability.get("selection") + if isinstance(selection, Mapping): + compact_selection = { + key: selection.get(key) + for key in ( + "frontier_status", + "dominated_by", + "dominance_reason", + ) + if key in selection + } + adequacy_evidence = selection.get("adequacy_evidence") + if isinstance(adequacy_evidence, Sequence) and not isinstance( + adequacy_evidence, + (str, bytes, bytearray), + ): + adequacy_sources = list( + dict.fromkeys( + str(item.get("source") or "").strip() + for item in adequacy_evidence + if isinstance(item, Mapping) + and str(item.get("source") or "").strip() + ) + ) + if adequacy_sources: + compact_selection["adequacy_sources"] = adequacy_sources + compact["selection"] = compact_selection + return compact + + def _capability_schema_reference( capability: Mapping[str, Any], *, @@ -294,6 +372,7 @@ def _capability_schema_reference( ( "name", "description", + "planner_hint", "query_match", "server_bound_arguments", "semantic_effect", @@ -310,6 +389,7 @@ def _capability_schema_reference( ) ) compact = {key: capability.get(key) for key in projected_keys if key in capability} + compact.update(_compact_capability_selection_profiles(capability)) name = str(capability.get("name") or "").strip() if not include_metadata: compact["capability_metadata_omitted_for_model_context"] = True @@ -351,6 +431,7 @@ def _capability_schema_focused_projection( ) if key in capability } + projected.update(_compact_capability_selection_profiles(capability)) projected["capability_metadata_omitted_for_model_context"] = True return projected @@ -392,7 +473,11 @@ def _bounded_capability_catalogue_output( "total", "delegated_total", "matched_total", + "frontier_total", + "dominated_total", "ranking", + "selection_policy", + "dominated_capabilities", "catalogue_scope", "offset", ) @@ -492,6 +577,12 @@ def assemble( bounded = compact_candidate continue + # Prefer another page over erasing decision-bearing metadata from an + # otherwise bounded capability. Minimal and name-only references are + # reserved for a capability that cannot fit by itself. + if included: + break + minimal = _capability_schema_reference( capability, include_metadata=False, @@ -1242,6 +1333,17 @@ def _scope_message( "without interpreting the user's intent; a natural-language query also " "retrieves actor-accessible executable represented workflows as bound " "capabilities.\n" + "- Treat a direct call, a small composition, and a represented workflow " + "as capability plans in one decision space. Choose the least costly plan " + "only after judging that it can produce the material work product and " + "evidence; representedness is not a quality score.\n" + "- Use each candidate's plan, effect, cost, and adequacy evidence. A " + "workflow is warranted when it adds needed composition, verification, " + "durability, governance, or recovery; otherwise prefer an adequate " + "observable and recoverable direct path.\n" + "- A matched workflow may expose visible component capabilities as " + "simpler alternatives. Their presence does not prove equivalence: inspect " + "the request and escalate when a direct result leaves a material gap.\n" f"- {_INVOKE_TOOL_NAME} invokes any named delegated capability.\n" f"- {_EVIDENCE_INDEX_TOOL_NAME} pages every evidence handle recorded " "for this turn.\n" @@ -1442,6 +1544,80 @@ def trusted_binding_is_present(binding_key: str) -> bool: return _registered_capability_names(gateway, requested) +def _registered_capability_plan_profile( + gateway: InternalMCPGateway, + *, + name: str, + definition: Any, +) -> tuple[dict[str, Any], dict[str, Any]]: + semantic_effect = bool( + definition.category == "write" and definition.ordinary_turn_effect + ) + effect_profile = { + "schema_version": _CAPABILITY_EFFECT_PROFILE_SCHEMA_VERSION, + "semantic_effect": semantic_effect, + "semantic_effect_source": "registered_method_definition", + "operational_state_effect": semantic_effect, + } + cost_profile: dict[str, Any] = { + "schema_version": _CAPABILITY_COST_PROFILE_SCHEMA_VERSION, + "basis": "declared_execution_shape", + "durable_runtime": False, + "declared_step_count": 1, + "declared_component_count": 1, + } + configured_timeout = gateway.get_method_timeout_sec(name) + if configured_timeout is not None: + cost_profile["maximum_handler_window_seconds"] = float(configured_timeout) + minimum_effect_window = ( + gateway.get_method_effect_admission_window_sec(name) + if semantic_effect + else None + ) + if minimum_effect_window is not None: + cost_profile["minimum_runtime_window_seconds"] = float(minimum_effect_window) + plan_profile = { + "schema_version": _CAPABILITY_PLAN_PROFILE_SCHEMA_VERSION, + "shape": "single_capability", + "component_capability_names": [name], + "direct_equivalent_capability_names": [], + "effect_profile": effect_profile, + "cost_profile": cost_profile, + } + return plan_profile, effect_profile + + +def _interleave_capability_frontier( + direct_candidates: Sequence[dict[str, Any]], + workflow_candidates: Sequence[dict[str, Any]], +) -> list[dict[str, Any]]: + """Keep both adequate plan shapes visible without choosing semantics.""" + + result: list[dict[str, Any]] = [] + direct_index = 0 + workflow_index = 0 + while direct_index < len(direct_candidates) or workflow_index < len( + workflow_candidates + ): + if direct_index < len(direct_candidates): + result.append(direct_candidates[direct_index]) + direct_index += 1 + if workflow_index < len(workflow_candidates): + result.append(workflow_candidates[workflow_index]) + workflow_index += 1 + return result + + +def _strip_capability_ranking_metadata( + capability: Mapping[str, Any], +) -> dict[str, Any]: + return { + key: value + for key, value in capability.items() + if not str(key).startswith("_ranking_") + } + + def _capability_catalogue( gateway: InternalMCPGateway, delegated_names: Sequence[str], @@ -1474,8 +1650,8 @@ def _capability_catalogue( query_tokens = { token for token in re.findall(r"[a-z0-9_]+", query) if len(token) > 1 } - ranked: list[tuple[int, str, dict[str, Any]]] = [] - matched_total = 0 + registered_candidates: list[dict[str, Any]] = [] + registered_by_name: dict[str, dict[str, Any]] = {} registered_delegated_total = 0 for name in delegated_names: definition = gateway.get_method_definition(name) @@ -1488,6 +1664,7 @@ def _capability_catalogue( or f"Use {name}." ) description = fallback_description + planner_hint: str | None = None surface_metadata: Any = None try: # These fields already govern planner/provenance displays elsewhere. @@ -1496,6 +1673,7 @@ def _capability_catalogue( from src.backend.services.tool_metadata_service import ( get_tool_description, get_tool_dispatch_surface_metadata, + get_tool_planner_hint, ) description = ( @@ -1505,6 +1683,7 @@ def _capability_catalogue( ) or fallback_description ) + planner_hint = get_tool_planner_hint(name) surface_metadata = get_tool_dispatch_surface_metadata(name) except Exception: # noqa: BLE001 # Represented metadata improves discovery but is not a new @@ -1522,19 +1701,34 @@ def _capability_catalogue( ) if str(value or "").strip() ) - searchable = ( - f"{name_key.replace('_', ' ')} {description.lower()} " - f"{positive_surface_terms}" + routing_searchable = ( + f"{name_key.replace('_', ' ')} " + f"{str(planner_hint or '').lower()} {positive_surface_terms}" ) + searchable = f"{routing_searchable} {description.lower()}" if query_tokens: - matched = sum(1 for token in query_tokens if token in searchable) - score = matched * 10 + (20 if query in searchable else 0) - query_match = matched > 0 + searchable_tokens = set(re.findall(r"[a-z0-9_]+", searchable)) + routing_searchable_tokens = set( + re.findall(r"[a-z0-9_]+", routing_searchable) + ) + description_searchable_tokens = set( + re.findall(r"[a-z0-9_]+", description.lower()) + ) + literal_match_count = sum( + 1 for token in query_tokens if token in searchable_tokens + ) + routing_match_tokens = sorted(query_tokens & routing_searchable_tokens) + description_literal_match_count = sum( + 1 for token in query_tokens if token in description_searchable_tokens + ) + literal_phrase_match = bool(query and query in searchable) + query_match = False else: - score = 0 + literal_match_count = 0 + routing_match_tokens = [] + description_literal_match_count = 0 + literal_phrase_match = False query_match = True - if query_match: - matched_total += 1 server_bound_arguments = sorted( { @@ -1548,16 +1742,52 @@ def _capability_catalogue( if str(argument_name).strip() } ) + plan_profile, effect_profile = _registered_capability_plan_profile( + gateway, + name=name, + definition=definition, + ) + adequacy_evidence: list[dict[str, Any]] = [] + if literal_phrase_match: + adequacy_evidence.append({"source": "literal_query_phrase"}) + if literal_match_count: + adequacy_evidence.append( + { + "source": "literal_query_terms", + "matched_term_count": literal_match_count, + "routing_term_count": len(routing_match_tokens), + "description_term_count": description_literal_match_count, + } + ) + if not query_tokens: + adequacy_evidence.append({"source": "unfiltered_catalogue_request"}) capability = { "name": name, "kind": "registered_tool", "description": description, + **({"planner_hint": planner_hint} if planner_hint else {}), "input_schema": _model_visible_input_schema(definition), "query_match": query_match, "server_bound_arguments": server_bound_arguments, + "semantic_effect": bool(effect_profile["semantic_effect"]), + "effect_profile": effect_profile, + "plan_profile": plan_profile, + "selection": { + "frontier_status": ( + "candidate" if query_match else "outside_query_frontier" + ), + "adequacy_evidence": adequacy_evidence, + "semantic_adequacy_owner": "adaptive_model", + }, + "_ranking_literal_match_count": literal_match_count, + "_ranking_literal_phrase_match": literal_phrase_match, + "_ranking_component_match_count": 0, + "_ranking_routing_match_tokens": routing_match_tokens, + "_ranking_description_literal_match_count": ( + description_literal_match_count + ), } - if definition.ordinary_turn_effect: - capability["semantic_effect"] = True + if bool(effect_profile["semantic_effect"]): minimum_effect_window = gateway.get_method_effect_admission_window_sec(name) if minimum_effect_window is not None: capability["minimum_effect_window_seconds"] = float( @@ -1573,13 +1803,67 @@ def _capability_catalogue( "external_surface": bool(surface_metadata.external_surface), } ) - ranked.append( - ( - -score, - name_key, - capability, - ) + plan_profile["evidence_surface_families"] = [ + surface_metadata.evidence_surface_family + ] + registered_candidates.append(capability) + registered_by_name[name_key] = capability + + # Treat lexical overlap as routing evidence only when it distinguishes a + # bounded portion of the delegated catalogue. This corpus-relative rule + # avoids language-specific stopword lists while preventing generic verbs + # and connective words from flooding the active frontier. + routing_term_frequency: dict[str, int] = {} + for candidate in registered_candidates: + for token in set(candidate.get("_ranking_routing_match_tokens") or []): + routing_term_frequency[token] = routing_term_frequency.get(token, 0) + 1 + maximum_informative_frequency = max( + 3, + int(len(registered_candidates) * 0.05), + ) + for candidate in registered_candidates: + routing_match_tokens = list( + candidate.get("_ranking_routing_match_tokens") or [] + ) + informative_routing_tokens = [ + token + for token in routing_match_tokens + if routing_term_frequency.get(token, 0) <= maximum_informative_frequency + ] + description_literal_match_count = int( + candidate.get("_ranking_description_literal_match_count") or 0 ) + query_match = bool( + not query_tokens + or candidate.get("_ranking_literal_phrase_match") + or informative_routing_tokens + or description_literal_match_count >= 2 + ) + candidate["query_match"] = query_match + candidate["_ranking_literal_match_count"] = ( + len(informative_routing_tokens) + description_literal_match_count + ) + candidate_selection = dict(candidate.get("selection") or {}) + candidate_selection["frontier_status"] = ( + "candidate" if query_match else "outside_query_frontier" + ) + adequacy_evidence = list(candidate_selection.get("adequacy_evidence") or []) + for evidence in adequacy_evidence: + if ( + isinstance(evidence, dict) + and evidence.get("source") == "literal_query_terms" + ): + evidence["informative_routing_term_count"] = len( + informative_routing_tokens + ) + evidence["common_routing_term_count"] = max( + 0, + len(routing_match_tokens) - len(informative_routing_tokens), + ) + candidate_selection["adequacy_evidence"] = adequacy_evidence + candidate["selection"] = candidate_selection + + workflow_candidates: list[dict[str, Any]] = [] represented_workflow_total = 0 for workflow_capability in workflow_capabilities: to_catalogue_entry = getattr( @@ -1608,39 +1892,302 @@ def _capability_catalogue( f"{description.lower()}" ) if query_tokens: - literal_matches = sum(1 for token in query_tokens if token in searchable) + searchable_tokens = set(re.findall(r"[a-z0-9_]+", searchable)) + literal_matches = sum( + 1 for token in query_tokens if token in searchable_tokens + ) + literal_phrase_match = bool(query and query in searchable) query_match = bool( literal_matches or float(capability.get("relevance_score") or 0.0) > 0.0 ) - semantic_score = int( - max( - 0.0, - min( - 1.0, - float(capability.get("relevance_score") or 0.0), - ), - ) - * 1000 - ) - score = semantic_score + (literal_matches * 10) else: query_match = True - score = int( - max( - 0.0, - min( - 1.0, - float(capability.get("relevance_score") or 0.0), - ), - ) - * 1000 - ) + literal_matches = 0 + literal_phrase_match = False capability["query_match"] = query_match - if query_match: - matched_total += 1 - ranked.append((-score, name_key, capability)) - ranked.sort(key=lambda item: (item[0], item[1])) - selected = [item[2] for item in ranked] + workflow_selection = capability.get("selection") + selection = ( + dict(workflow_selection) if isinstance(workflow_selection, Mapping) else {} + ) + adequacy_evidence: list[dict[str, Any]] = [] + if literal_phrase_match: + adequacy_evidence.append({"source": "literal_query_phrase"}) + if literal_matches: + adequacy_evidence.append( + { + "source": "literal_query_terms", + "matched_term_count": literal_matches, + } + ) + relevance_score = max( + 0.0, + min( + 1.0, + float(capability.get("relevance_score") or 0.0), + ), + ) + if relevance_score > 0.0: + adequacy_evidence.append( + { + "source": "represented_workflow_semantic_relevance", + "relevance_score": round(relevance_score, 4), + } + ) + if not query_tokens: + adequacy_evidence.append({"source": "unfiltered_catalogue_request"}) + selection.update( + { + "frontier_status": ( + "candidate" if query_match else "outside_query_frontier" + ), + "adequacy_evidence": adequacy_evidence, + "semantic_adequacy_owner": "adaptive_model", + } + ) + capability["selection"] = selection + capability["_ranking_literal_match_count"] = literal_matches + capability["_ranking_literal_phrase_match"] = literal_phrase_match + capability["_ranking_semantic_relevance"] = relevance_score + + plan_profile_value = capability.get("plan_profile") + plan_profile = ( + dict(plan_profile_value) if isinstance(plan_profile_value, Mapping) else {} + ) + raw_component_names = plan_profile.get("component_capability_names") + component_names = ( + [ + str(item).strip() + for item in raw_component_names + if isinstance(item, str) and str(item).strip() + ] + if isinstance(raw_component_names, Sequence) + and not isinstance(raw_component_names, (str, bytes, bytearray)) + else [] + ) + component_surfaces = sorted( + { + str( + registered_by_name[name.lower()].get("evidence_surface_family") + or "" + ).strip() + for name in component_names + if name.lower() in registered_by_name + and str( + registered_by_name[name.lower()].get("evidence_surface_family") + or "" + ).strip() + } + ) + if component_surfaces: + plan_profile["evidence_surface_families"] = component_surfaces + capability["plan_profile"] = plan_profile + workflow_candidates.append(capability) + + # A matched workflow's declared visible components are plausible simpler + # plans, not proven equivalents. Keep them on the same frontier and let the + # adaptive model judge whether one component or a small composition is + # adequate for the actual work product. + for workflow in workflow_candidates: + if workflow.get("query_match") is not True: + continue + plan_profile = workflow.get("plan_profile") + component_names = ( + plan_profile.get("component_capability_names") + if isinstance(plan_profile, Mapping) + else None + ) + if not isinstance(component_names, Sequence) or isinstance( + component_names, + (str, bytes, bytearray), + ): + continue + for component_name in component_names: + component = registered_by_name.get(str(component_name).strip().lower()) + if component is None: + continue + component["query_match"] = True + component["_ranking_component_match_count"] = ( + int(component.get("_ranking_component_match_count") or 0) + 1 + ) + component_selection = component.get("selection") + selection = ( + dict(component_selection) + if isinstance(component_selection, Mapping) + else {} + ) + adequacy_evidence = list(selection.get("adequacy_evidence") or []) + adequacy_evidence.append( + { + "source": "declared_component_of_matched_workflow", + "workflow_id": workflow.get("workflow_id"), + } + ) + selection.update( + { + "frontier_status": "candidate", + "adequacy_evidence": adequacy_evidence, + "semantic_adequacy_owner": "adaptive_model", + } + ) + component["selection"] = selection + + dominated_capabilities: list[dict[str, Any]] = [] + dominated_names: set[str] = set() + if not exact_names: + for workflow in workflow_candidates: + if workflow.get("query_match") is not True: + continue + plan_profile = workflow.get("plan_profile") + direct_equivalents = ( + plan_profile.get("direct_equivalent_capability_names") + if isinstance(plan_profile, Mapping) + else None + ) + if not isinstance(direct_equivalents, Sequence) or isinstance( + direct_equivalents, + (str, bytes, bytearray), + ): + continue + for equivalent_name in direct_equivalents: + direct = registered_by_name.get(str(equivalent_name).strip().lower()) + if direct is None: + continue + # The represented declaration supplies outcome equivalence. + # Mechanical pruning is still conservative: both plans must be + # known read-only, leaving only orchestration cost different. + if ( + direct.get("semantic_effect") is not False + or workflow.get("semantic_effect") is not False + ): + continue + workflow_name = str(workflow.get("name") or "").strip() + if not workflow_name: + continue + direct["query_match"] = True + direct_selection = dict(direct.get("selection") or {}) + direct_adequacy_evidence = list( + direct_selection.get("adequacy_evidence") or [] + ) + direct_adequacy_evidence.append( + { + "source": "represented_declared_direct_equivalence", + "workflow_id": workflow.get("workflow_id"), + } + ) + direct_selection.update( + { + "frontier_status": "candidate", + "adequacy_evidence": direct_adequacy_evidence, + "semantic_adequacy_owner": "adaptive_model", + } + ) + direct["selection"] = direct_selection + dominated_names.add(workflow_name.lower()) + workflow_selection = dict(workflow.get("selection") or {}) + workflow_selection.update( + { + "frontier_status": "declared_dominated", + "dominated_by": str(direct.get("name") or ""), + "dominance_reason": ( + "declared_direct_equivalence_with_lower_declared_" + "orchestration_cost" + ), + } + ) + workflow["selection"] = workflow_selection + dominated_capabilities.append( + { + "name": workflow_name, + "dominated_by": str(direct.get("name") or ""), + "reason": workflow_selection["dominance_reason"], + "equivalence_source": ("represented_workflow_routing_profile"), + } + ) + break + + def direct_sort_key(item: Mapping[str, Any]) -> tuple[Any, ...]: + return ( + -int(bool(item.get("_ranking_literal_phrase_match"))), + -int(item.get("_ranking_component_match_count") or 0), + -int(item.get("_ranking_literal_match_count") or 0), + str(item.get("name") or "").lower(), + ) + + def workflow_sort_key(item: Mapping[str, Any]) -> tuple[Any, ...]: + return ( + -int(bool(item.get("_ranking_literal_phrase_match"))), + -float(item.get("_ranking_semantic_relevance") or 0.0), + -int(item.get("_ranking_literal_match_count") or 0), + str(item.get("name") or "").lower(), + ) + + if exact_names: + selected_internal = sorted( + [*registered_candidates, *workflow_candidates], + key=lambda item: ( + 0 if item.get("kind") == "registered_tool" else 1, + str(item.get("name") or "").lower(), + ), + ) + else: + frontier_direct = sorted( + [item for item in registered_candidates if item.get("query_match") is True], + key=direct_sort_key, + ) + frontier_workflows = sorted( + [ + item + for item in workflow_candidates + if item.get("query_match") is True + and str(item.get("name") or "").lower() not in dominated_names + ], + key=workflow_sort_key, + ) + outside_frontier_direct = sorted( + [ + item + for item in registered_candidates + if item.get("query_match") is not True + ], + key=direct_sort_key, + ) + outside_frontier_workflows = sorted( + [ + item + for item in workflow_candidates + if item.get("query_match") is not True + ], + key=workflow_sort_key, + ) + dominated_workflows = sorted( + [ + item + for item in workflow_candidates + if str(item.get("name") or "").lower() in dominated_names + ], + key=workflow_sort_key, + ) + selected_internal = [ + *_interleave_capability_frontier( + frontier_direct, + frontier_workflows, + ), + *outside_frontier_direct, + *outside_frontier_workflows, + *dominated_workflows, + ] + + selected = [_strip_capability_ranking_metadata(item) for item in selected_internal] + matched_total = sum(1 for item in selected if item.get("query_match") is True) + frontier_total = sum( + 1 + for item in selected + if item.get("query_match") is True + and ( + not isinstance(item.get("selection"), Mapping) + or item["selection"].get("frontier_status") != "declared_dominated" + ) + ) page = selected[offset : offset + limit] next_offset = offset + len(page) return { @@ -1652,9 +2199,24 @@ def _capability_catalogue( "registered_tool_total": registered_delegated_total, "represented_workflow_total": represented_workflow_total, "matched_total": matched_total, - "ranking": ( - "represented_semantic_relevance_then_literal_query_match_then_name" - ), + "frontier_total": frontier_total, + "dominated_total": len(dominated_capabilities), + "ranking": "minimum_adequate_cost_sensitive_frontier", + "selection_policy": { + "schema_version": _CAPABILITY_SELECTION_POLICY_SCHEMA_VERSION, + "semantic_adequacy_owner": "adaptive_model", + "mechanical_dominance": "declared_equivalence_only", + "cost_rule": ( + "compare_declared_cost_only_after_material_outcome_and_" + "evidence_adequacy" + ), + "frontier_order": ( + "interleave_direct_and_workflow_candidates_with_lower_" + "orchestration_cost_first" + ), + "representedness_priority": False, + }, + "dominated_capabilities": dominated_capabilities, "catalogue_scope": ( "requested_exact_names" if exact_names @@ -2274,6 +2836,8 @@ def execute_adaptive_turn( delegated_lookup = {name.lower(): name for name in delegated_names} workflow_capabilities_by_name: dict[str, Any] = {} latest_workflow_discovery: dict[str, Any] | None = None + catalogued_capabilities_by_name: dict[str, dict[str, Any]] = {} + latest_capability_selection_policy: dict[str, Any] | None = None available_tools = _tool_definitions() final_synthesis_tools = [ tool @@ -3342,6 +3906,16 @@ def enter_answer_only(reason: str) -> None: max(1, requested_workflow_limit), ), timeout_seconds=remaining_discovery_seconds, + registered_capability_categories={ + name: str(definition.category).strip().lower() + for name in delegated_names + if ( + definition := gateway.get_method_definition( + name + ) + ) + is not None + }, ) ) except Exception as exc: # noqa: BLE001 @@ -3370,6 +3944,29 @@ def enter_answer_only(reason: str) -> None: ), workflow_discovery=latest_workflow_discovery, ) + raw_catalogued_capabilities = output.get("capabilities") + if isinstance( + raw_catalogued_capabilities, Sequence + ) and not isinstance( + raw_catalogued_capabilities, + (str, bytes, bytearray), + ): + for raw_capability in raw_catalogued_capabilities: + if not isinstance(raw_capability, Mapping): + continue + catalogued_name = str( + raw_capability.get("name") or "" + ).strip() + if catalogued_name: + catalogued_capabilities_by_name[ + catalogued_name.lower() + ] = dict(raw_capability) + raw_selection_policy = output.get("selection_policy") + latest_capability_selection_policy = ( + dict(raw_selection_policy) + if isinstance(raw_selection_policy, Mapping) + else None + ) else: output = _error_payload( "capability_gateway_unavailable", @@ -4281,6 +4878,71 @@ def contained( invocation[receipt_key] = receipt_value if transport_metadata: invocation["transport"] = transport_metadata + catalogued_capability = catalogued_capabilities_by_name.get( + canonical_name.lower() + ) + if isinstance(catalogued_capability, Mapping): + for profile_key in ( + "plan_profile", + "effect_profile", + "selection", + ): + profile_value = catalogued_capability.get(profile_key) + if isinstance(profile_value, Mapping): + invocation[profile_key] = dict(profile_value) + actual_cost = { + key: transport_metadata.get(key) + for key in ( + "duration_ms", + "queue_duration_ms", + "handler_duration_ms", + "transport_overhead_ms", + ) + if transport_metadata.get(key) is not None + } + aux_calls.append( + { + "type": "adaptive_turn_capability_selection", + "schema_version": (_CAPABILITY_SELECTION_TRACE_SCHEMA_VERSION), + "capability_name": canonical_name, + "capability_kind": contained_result.capability_kind, + "execution_method": execution_method_name, + "selection_policy": dict( + latest_capability_selection_policy or {} + ), + "plan_profile": ( + dict(invocation["plan_profile"]) + if isinstance( + invocation.get("plan_profile"), + Mapping, + ) + else None + ), + "effect_profile": ( + dict(invocation["effect_profile"]) + if isinstance( + invocation.get("effect_profile"), + Mapping, + ) + else None + ), + "selection": ( + dict(invocation["selection"]) + if isinstance( + invocation.get("selection"), + Mapping, + ) + else None + ), + "actual_cost": actual_cost, + "status": status, + **( + {"effect_status": effect_status} + if effect_status is not None + else {} + ), + } + ) tool_invocations.append(invocation) _emit( progress_tracker, diff --git a/src/backend/services/workflow_turn_capability_service.py b/src/backend/services/workflow_turn_capability_service.py index 592f36058..14e3eddb4 100644 --- a/src/backend/services/workflow_turn_capability_service.py +++ b/src/backend/services/workflow_turn_capability_service.py @@ -17,6 +17,9 @@ WORKFLOW_TURN_CAPABILITY_SCHEMA_VERSION = "workflow_turn_capability.v1" WORKFLOW_TURN_DISCOVERY_SCHEMA_VERSION = "workflow_turn_capability_discovery.v1" WORKFLOW_TURN_RECEIPT_SCHEMA_VERSION = "workflow_turn_effect_receipt.v1" +CAPABILITY_PLAN_PROFILE_SCHEMA_VERSION = "capability_plan_profile.v1" +CAPABILITY_EFFECT_PROFILE_SCHEMA_VERSION = "capability_effect_profile.v1" +CAPABILITY_COST_PROFILE_SCHEMA_VERSION = "capability_cost_profile.v1" _SERVER_PROVIDED_WORKFLOW_INPUT_KEYS = frozenset( { @@ -86,17 +89,122 @@ def _bounded_mapping_sequence( (str, bytes, bytearray), ): return [] - return [ - dict(item) - for item in list(value)[:limit] - if isinstance(item, Mapping) + return [dict(item) for item in list(value)[:limit] if isinstance(item, Mapping)] + + +def _bounded_text_tuple(value: Any, *, limit: int = 40) -> tuple[str, ...]: + if not isinstance(value, Sequence) or isinstance( + value, + (str, bytes, bytearray), + ): + return () + result: list[str] = [] + seen: set[str] = set() + for item in value: + text = _normalise_non_empty_text(item) + if text is None or text in seen: + continue + seen.add(text) + result.append(text) + if len(result) >= limit: + break + return tuple(result) + + +def _coerce_optional_non_negative_int(value: Any) -> int | None: + if isinstance(value, bool): + return None + try: + parsed = int(value) + except (TypeError, ValueError): + return None + return parsed if parsed >= 0 else None + + +def _workflow_plan_metadata( + raw_match: Mapping[str, Any], + *, + registered_capability_categories: Mapping[str, str] | None, +) -> dict[str, Any]: + """Project declared workflow shape without inventing semantic policy.""" + + routing_index_metadata = raw_match.get("routing_index_metadata") + index_metadata = ( + dict(routing_index_metadata) + if isinstance(routing_index_metadata, Mapping) + else {} + ) + routing_profile_value = raw_match.get("routing_profile") + routing_profile = ( + dict(routing_profile_value) + if isinstance(routing_profile_value, Mapping) + else {} + ) + visible_categories = { + str(name).strip(): str(category).strip().lower() + for name, category in (registered_capability_categories or {}).items() + if str(name).strip() and str(category).strip() + } + declared_components = _bounded_text_tuple( + index_metadata.get("required_tools"), + ) + visible_components = tuple( + name for name in declared_components if name in visible_categories + ) + compact_executability = index_metadata.get("compact_executability") + declared_step_count = ( + _coerce_optional_non_negative_int(compact_executability.get("step_count")) + if isinstance(compact_executability, Mapping) + else None + ) + direct_equivalents = tuple( + name + for name in _bounded_text_tuple( + routing_profile.get("direct_equivalent_capability_names"), + limit=12, + ) + if name in visible_categories + ) + + component_categories = [ + visible_categories[name] + for name in visible_components + if name in visible_categories ] + unresolved_component_count = max( + 0, + len(declared_components) - len(visible_components), + ) + if any(category == "write" for category in component_categories): + semantic_effect: bool | None = True + semantic_effect_source = "declared_registered_write_component" + elif ( + declared_components + and unresolved_component_count == 0 + and component_categories + and all(category == "read" for category in component_categories) + ): + semantic_effect = False + semantic_effect_source = "declared_registered_read_components" + else: + semantic_effect = None + semantic_effect_source = "insufficient_declared_component_evidence" + + return { + "component_capability_names": visible_components, + "declared_component_count": len(declared_components), + "unresolved_component_count": unresolved_component_count, + "declared_step_count": declared_step_count, + "direct_equivalent_capability_names": direct_equivalents, + "semantic_effect": semantic_effect, + "semantic_effect_source": semantic_effect_source, + } def _turn_capability_name(*, turn_id: str, workflow_id: str) -> str: - digest = hashlib.sha256( - f"{turn_id}\0{workflow_id}".encode("utf-8") - ).hexdigest()[:20] + digest = hashlib.sha256(f"{turn_id}\0{workflow_id}".encode("utf-8")).hexdigest()[ + :20 + ] return f"represented_workflow_{digest}" @@ -230,8 +338,45 @@ class WorkflowTurnCapability: input_schema: Mapping[str, Any] launch_input_contract_source: str | None = None discovery_metadata: Mapping[str, Any] = field(default_factory=dict) + component_capability_names: tuple[str, ...] = () + declared_component_count: int = 0 + unresolved_component_count: int = 0 + declared_step_count: int | None = None + direct_equivalent_capability_names: tuple[str, ...] = () + semantic_effect: bool | None = None + semantic_effect_source: str = "insufficient_declared_component_evidence" def to_catalogue_entry(self) -> dict[str, Any]: + effect_profile = { + "schema_version": CAPABILITY_EFFECT_PROFILE_SCHEMA_VERSION, + "semantic_effect": self.semantic_effect, + "semantic_effect_source": self.semantic_effect_source, + "operational_state_effect": True, + "operational_state_effect_reason": ( + "workflow invocation creates or advances a durable instance" + ), + } + cost_profile: dict[str, Any] = { + "schema_version": CAPABILITY_COST_PROFILE_SCHEMA_VERSION, + "basis": "declared_plan_structure", + "durable_runtime": True, + "declared_component_count": int(self.declared_component_count), + "visible_component_count": len(self.component_capability_names), + "unresolved_component_count": int(self.unresolved_component_count), + "minimum_runtime_window_seconds": 5.0, + } + if self.declared_step_count is not None: + cost_profile["declared_step_count"] = int(self.declared_step_count) + plan_profile = { + "schema_version": CAPABILITY_PLAN_PROFILE_SCHEMA_VERSION, + "shape": "represented_workflow", + "component_capability_names": list(self.component_capability_names), + "direct_equivalent_capability_names": list( + self.direct_equivalent_capability_names + ), + "effect_profile": effect_profile, + "cost_profile": cost_profile, + } return { "schema_version": WORKFLOW_TURN_CAPABILITY_SCHEMA_VERSION, "name": self.name, @@ -241,7 +386,9 @@ def to_catalogue_entry(self) -> dict[str, Any]: "description": self.description, "input_schema": dict(self.input_schema), "query_match": True, - "semantic_effect": True, + "semantic_effect": self.semantic_effect, + "effect_profile": effect_profile, + "plan_profile": plan_profile, "minimum_effect_window_seconds": 5.0, "server_bound_arguments": [ "workflow_id", @@ -265,6 +412,7 @@ def discover_turn_workflow_capabilities( turn_id: str, max_results: int = 5, timeout_seconds: float = 5.0, + registered_capability_categories: Mapping[str, str] | None = None, ) -> tuple[list[WorkflowTurnCapability], dict[str, Any]]: """Return bounded actor-accessible workflow affordances for a model query.""" @@ -339,6 +487,10 @@ def discover_turn_workflow_capabilities( ) routing_profile = raw_match.get("routing_profile") routing_index_metadata = raw_match.get("routing_index_metadata") + plan_metadata = _workflow_plan_metadata( + raw_match, + registered_capability_categories=(registered_capability_categories), + ) capabilities.append( WorkflowTurnCapability( name=_turn_capability_name( @@ -379,9 +531,12 @@ def discover_turn_workflow_capabilities( key: routing_index_metadata.get(key) for key in ( "authority_source", + "compact_executability", "description_source", "publication_lifecycle", + "required_tools_source", "routing_profile_source", + "workflow_action_ids_source", ) if key in routing_index_metadata } @@ -389,6 +544,21 @@ def discover_turn_workflow_capabilities( else None ), }, + component_capability_names=tuple( + plan_metadata["component_capability_names"] + ), + declared_component_count=int( + plan_metadata["declared_component_count"] + ), + unresolved_component_count=int( + plan_metadata["unresolved_component_count"] + ), + declared_step_count=plan_metadata["declared_step_count"], + direct_equivalent_capability_names=tuple( + plan_metadata["direct_equivalent_capability_names"] + ), + semantic_effect=plan_metadata["semantic_effect"], + semantic_effect_source=str(plan_metadata["semantic_effect_source"]), ) ) diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py index 8d417352f..3075be169 100644 --- a/tests/backend/test_adaptive_turn_service.py +++ b/tests/backend/test_adaptive_turn_service.py @@ -3020,6 +3020,304 @@ def fail_metadata(*_args, **_kwargs): ) +def test_capability_frontier_promotes_direct_components_without_hiding_workflow( + monkeypatch, +) -> None: + from src.backend.services import tool_metadata_service + from src.backend.services.workflow_turn_capability_service import ( + WorkflowTurnCapability, + ) + + catalogue = MethodCatalogue() + for name in ( + "fetch_concept", + "find_relations_with_argument", + "students_index", + ): + catalogue.register( + MethodDefinition( + name=name, + handler=lambda **_kwargs: {"success": True}, + input_schema=Schema( + optional={"concept_id": str}, + allow_unknown=False, + ), + category="read", + description=f"Use {name}.", + ) + ) + gateway = InternalMCPGateway( + catalogue=catalogue, + transport=InternalMCPTransport(read_timeout_sec=2.0), + enabled=True, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_description", + lambda _name, *, fallback_description=None: fallback_description, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_planner_hint", + lambda _name: None, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_dispatch_surface_metadata", + lambda _name: None, + ) + workflow = WorkflowTurnCapability( + name="represented_workflow_entity_lookup", + workflow_id="#V#entity_lookup_workflow", + display_name="Entity lookup workflow", + description="Retrieve represented relationships for an entity.", + relevance_score=0.93, + input_schema={"type": "object", "properties": {}}, + component_capability_names=( + "fetch_concept", + "find_relations_with_argument", + ), + declared_component_count=2, + declared_step_count=4, + semantic_effect=False, + semantic_effect_source="declared_registered_read_components", + ) + + result = _capability_catalogue( + gateway, + ("fetch_concept", "find_relations_with_argument", "students_index"), + {"query": "students supervised by me", "limit": 10}, + workflow_capabilities=(workflow,), + ) + + assert result["ranking"] == "minimum_adequate_cost_sensitive_frontier" + assert result["selection_policy"]["representedness_priority"] is False + assert [item["name"] for item in result["capabilities"]] == [ + "fetch_concept", + "represented_workflow_entity_lookup", + "find_relations_with_argument", + "students_index", + ] + direct = result["capabilities"][0] + represented = result["capabilities"][1] + assert direct["plan_profile"]["shape"] == "single_capability" + assert represented["plan_profile"]["shape"] == "represented_workflow" + assert represented["semantic_effect"] is False + assert represented["effect_profile"]["operational_state_effect"] is True + assert any( + evidence["source"] == "declared_component_of_matched_workflow" + for evidence in direct["selection"]["adequacy_evidence"] + ) + assert result["frontier_total"] == 4 + assert result["dominated_total"] == 0 + + +def test_capability_frontier_does_not_prefer_unrelated_direct_tool( + monkeypatch, +) -> None: + from src.backend.services import tool_metadata_service + from src.backend.services.workflow_turn_capability_service import ( + WorkflowTurnCapability, + ) + + catalogue = MethodCatalogue() + catalogue.register( + MethodDefinition( + name="unrelated_direct_read", + handler=lambda **_kwargs: {"success": True}, + input_schema=Schema(optional={"record_id": str}, allow_unknown=False), + category="read", + description=( + "Inspect an unrelated verified record. The shared adjective is " + "incidental rather than routing evidence." + ), + ) + ) + gateway = InternalMCPGateway( + catalogue=catalogue, + transport=InternalMCPTransport(read_timeout_sec=1.0), + enabled=True, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_description", + lambda _name, *, fallback_description=None: fallback_description, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_planner_hint", + lambda _name: None, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_dispatch_surface_metadata", + lambda _name: None, + ) + workflow = WorkflowTurnCapability( + name="represented_workflow_multi_step", + workflow_id="#V#multi_step_workflow", + display_name="Multi-step workflow", + description="Produce a verified multi-step research work product.", + relevance_score=0.95, + input_schema={"type": "object", "properties": {}}, + declared_step_count=6, + semantic_effect=False, + semantic_effect_source="declared_registered_read_components", + ) + + result = _capability_catalogue( + gateway, + ("unrelated_direct_read",), + {"query": "verified multi-step research work product", "limit": 10}, + workflow_capabilities=(workflow,), + ) + + assert [item["name"] for item in result["capabilities"]] == [ + "represented_workflow_multi_step", + "unrelated_direct_read", + ] + assert result["frontier_total"] == 1 + + +def test_capability_frontier_uses_catalogue_rarity_not_generic_routing_words( + monkeypatch, +) -> None: + from src.backend.services import tool_metadata_service + + catalogue = MethodCatalogue() + generic_names = tuple(f"list_inventory_{index}" for index in range(5)) + for name in (*generic_names, "students_relations"): + catalogue.register( + MethodDefinition( + name=name, + handler=lambda **_kwargs: {"success": True}, + input_schema=Schema(optional={}, allow_unknown=False), + category="read", + description="Inspect a bounded represented record.", + ) + ) + gateway = InternalMCPGateway( + catalogue=catalogue, + transport=InternalMCPTransport(read_timeout_sec=1.0), + enabled=True, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_description", + lambda _name, *, fallback_description=None: fallback_description, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_planner_hint", + lambda _name: None, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_dispatch_surface_metadata", + lambda _name: None, + ) + + result = _capability_catalogue( + gateway, + (*generic_names, "students_relations"), + {"query": "list students", "limit": 10}, + ) + + assert result["frontier_total"] == 1 + assert result["matched_total"] == 1 + assert result["capabilities"][0]["name"] == "students_relations" + by_name = {item["name"]: item for item in result["capabilities"]} + assert all( + by_name[name]["selection"]["frontier_status"] == "outside_query_frontier" + for name in generic_names + ) + generic_evidence = by_name[generic_names[0]]["selection"]["adequacy_evidence"] + assert generic_evidence[0]["common_routing_term_count"] == 1 + assert generic_evidence[0]["informative_routing_term_count"] == 0 + + +def test_declared_read_only_direct_equivalence_prunes_only_frontier( + monkeypatch, +) -> None: + from src.backend.services import tool_metadata_service + from src.backend.services.workflow_turn_capability_service import ( + WorkflowTurnCapability, + ) + + catalogue = MethodCatalogue() + catalogue.register( + MethodDefinition( + name="direct_lookup", + handler=lambda **_kwargs: {"success": True}, + input_schema=Schema(optional={"query": str}, allow_unknown=False), + category="read", + description="Look up the requested record.", + ) + ) + gateway = InternalMCPGateway( + catalogue=catalogue, + transport=InternalMCPTransport(read_timeout_sec=1.0), + enabled=True, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_description", + lambda _name, *, fallback_description=None: fallback_description, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_planner_hint", + lambda _name: None, + ) + monkeypatch.setattr( + tool_metadata_service, + "get_tool_dispatch_surface_metadata", + lambda _name: None, + ) + workflow = WorkflowTurnCapability( + name="represented_workflow_degenerate_lookup", + workflow_id="#V#degenerate_lookup_workflow", + display_name="Degenerate lookup workflow", + description="Look up the requested record through a durable workflow.", + relevance_score=0.9, + input_schema={"type": "object", "properties": {}}, + component_capability_names=("direct_lookup",), + declared_component_count=1, + declared_step_count=1, + direct_equivalent_capability_names=("direct_lookup",), + semantic_effect=False, + semantic_effect_source="declared_registered_read_components", + ) + + result = _capability_catalogue( + gateway, + ("direct_lookup",), + {"query": "look up the requested record", "limit": 10}, + workflow_capabilities=(workflow,), + ) + + assert [item["name"] for item in result["capabilities"]] == [ + "direct_lookup", + "represented_workflow_degenerate_lookup", + ] + assert result["total"] == 2 + assert result["frontier_total"] == 1 + assert result["dominated_total"] == 1 + assert result["dominated_capabilities"] == [ + { + "name": "represented_workflow_degenerate_lookup", + "dominated_by": "direct_lookup", + "reason": ( + "declared_direct_equivalence_with_lower_declared_orchestration_cost" + ), + "equivalence_source": "represented_workflow_routing_profile", + } + ] + assert result["capabilities"][1]["selection"]["frontier_status"] == ( + "declared_dominated" + ) + + def test_capability_page_budget_preserves_every_alternative_and_cursor( monkeypatch, ) -> None: @@ -3116,6 +3414,7 @@ def test_capability_page_budget_preserves_every_alternative_and_cursor( ) seen_names: list[str] = [] + seen_entries: list[dict[str, Any]] = [] schema_references: list[dict[str, Any]] = [] offset: int | None = 0 while offset is not None: @@ -3143,6 +3442,7 @@ def test_capability_page_budget_preserves_every_alternative_and_cursor( offset : offset + len(entries) ] seen_names.extend(item["name"] for item in entries) + seen_entries.extend(entries) schema_references.extend( item for item in entries @@ -3154,21 +3454,23 @@ def test_capability_page_budget_preserves_every_alternative_and_cursor( offset = next_offset assert seen_names == capability_names - assert schema_references - reference = schema_references[0] + reference = schema_references[0] if schema_references else seen_entries[0] assert reference["description"].startswith("Search research material.") assert reference["surface_family"] == "research" assert reference["evidence_surface_family"] == "research" assert reference["external_surface"] is False assert reference["server_bound_arguments"] == ["actor_id"] - assert reference["input_schema_hydration"] == { - "tool": "turn_capabilities", - "arguments": { - "names": [reference["name"]], - "limit": 1, - }, - "purpose": "dedicated_exact_name_schema_page", - } + if schema_references: + assert reference["input_schema_hydration"] == { + "tool": "turn_capabilities", + "arguments": { + "names": [reference["name"]], + "limit": 1, + }, + "purpose": "dedicated_exact_name_schema_page", + } + else: + assert "input_schema" in reference exact_page = _capability_catalogue( gateway, @@ -3354,23 +3656,22 @@ def test_bulky_capability_metadata_falls_back_without_hiding_the_schema( assert len(_json_bytes(bounded)) <= 24_000 assert bounded["next_offset"] is None assert bounded["capability_page_projection"]["returned"] == 1 - assert bounded["capability_page_projection"][ - "schema_reference_count" - ] == 1 + assert bounded["capability_page_projection"]["schema_reference_count"] == 1 compact = bounded["capabilities"][0] - assert compact == { - "name": name, - "capability_metadata_omitted_for_model_context": True, - "input_schema_omitted_for_model_context": True, - "input_schema_hydration": { - "tool": "turn_capabilities", - "arguments": { - "names": [name], - "limit": 1, - }, - "purpose": "dedicated_exact_name_schema_page", + assert compact["name"] == name + assert compact["capability_metadata_omitted_for_model_context"] is True + assert compact["input_schema_omitted_for_model_context"] is True + assert compact["input_schema_hydration"] == { + "tool": "turn_capabilities", + "arguments": { + "names": [name], + "limit": 1, }, + "purpose": "dedicated_exact_name_schema_page", } + assert compact["semantic_effect"] is False + assert compact["plan_profile"]["shape"] == "single_capability" + assert compact["selection"]["frontier_status"] == "candidate" exact = _capability_catalogue( gateway, @@ -5188,8 +5489,10 @@ def _execute_workflow(**kwargs: Any) -> dict[str, Any]: catalogue_result = client.calls[1]["tool_results"][0].output assert catalogue_result["represented_workflow_total"] == 1 - assert catalogue_result["capabilities"][0]["kind"] == ( - "represented_workflow" + assert catalogue_result["capabilities"][0]["kind"] == ("represented_workflow") + assert ( + catalogue_result["selection_policy"]["semantic_adequacy_owner"] + == "adaptive_model" ) assert seen_arguments["workflow_id"] == "#V#represented_test_workflow" assert seen_arguments["user_id"] == "#V#real_user" @@ -5218,6 +5521,15 @@ def _execute_workflow(**kwargs: Any) -> dict[str, Any]: assert invocation["changed"] is True assert invocation["instance_id"] == "workflow-instance-1" assert invocation["workflow_id"] == "#V#represented_test_workflow" + assert invocation["plan_profile"]["shape"] == "represented_workflow" + selection_trace = next( + item + for item in result.aux_llm_calls + if item.get("type") == "adaptive_turn_capability_selection" + ) + assert selection_trace["capability_name"] == workflow_capability.name + assert selection_trace["selection_policy"]["representedness_priority"] is False + assert selection_trace["plan_profile"]["shape"] == ("represented_workflow") assert result.response_text == "The represented work product was completed." diff --git a/tests/backend/test_workflow_turn_capability_service.py b/tests/backend/test_workflow_turn_capability_service.py index 5cd096362..b0fe4914d 100644 --- a/tests/backend/test_workflow_turn_capability_service.py +++ b/tests/backend/test_workflow_turn_capability_service.py @@ -37,6 +37,15 @@ def test_discovery_exposes_only_executable_routing_eligible_workflows( "is_executable": True, "routing_eligible": True, "match_source": "semantic", + "routing_index_metadata": { + "required_tools": [ + "fetch_concept", + "find_relations_with_argument", + ], + "compact_executability": { + "step_count": 3, + }, + }, }, { "concept_id": "#V#draft_workflow", @@ -88,17 +97,37 @@ def test_discovery_exposes_only_executable_routing_eligible_workflows( user_concept_id="#V#user", organisation_concept_id="#V#org", turn_id="turn-123", + registered_capability_categories={ + "fetch_concept": "read", + "find_relations_with_argument": "read", + }, ) - assert [item.workflow_id for item in capabilities] == [ - "#V#usable_workflow" - ] + assert [item.workflow_id for item in capabilities] == ["#V#usable_workflow"] capability = capabilities[0] assert capability.name.startswith("represented_workflow_") inputs_schema = capability.input_schema["properties"]["inputs"] assert list(inputs_schema["properties"]) == ["record_id"] assert inputs_schema["required"] == ["record_id"] assert "prompt" in capability.input_schema["x-von-server-provided-inputs"] + assert capability.semantic_effect is False + assert capability.semantic_effect_source == ("declared_registered_read_components") + catalogue_entry = capability.to_catalogue_entry() + assert catalogue_entry["semantic_effect"] is False + assert catalogue_entry["effect_profile"] == { + "schema_version": "capability_effect_profile.v1", + "semantic_effect": False, + "semantic_effect_source": "declared_registered_read_components", + "operational_state_effect": True, + "operational_state_effect_reason": ( + "workflow invocation creates or advances a durable instance" + ), + } + assert catalogue_entry["plan_profile"]["component_capability_names"] == [ + "fetch_concept", + "find_relations_with_argument", + ] + assert catalogue_entry["plan_profile"]["cost_profile"]["declared_step_count"] == 3 assert diagnostic["status"] == "completed" assert diagnostic["match_count"] == 1 assert diagnostic["candidate_count"] == 3 @@ -131,6 +160,71 @@ def test_discovery_requires_authenticated_actor_without_querying_index( assert diagnostic["status"] == "authenticated_actor_required" +def test_discovery_marks_declared_write_workflow_without_hiding_unknowns( + monkeypatch, +): + from src.backend.services.workflow_turn_capability_service import ( + discover_turn_workflow_capabilities, + ) + + monkeypatch.setattr( + "src.backend.services.workflow_discovery_service.discover_workflows_for_turn", + lambda *_args, **_kwargs: { + "matches": [ + { + "concept_id": "#V#write_workflow", + "name": "Write workflow", + "description": "Create a represented work product.", + "relevance_score": 0.9, + "is_executable": True, + "routing_eligible": True, + "routing_index_metadata": { + "required_tools": ["create_concepts"], + }, + }, + { + "concept_id": "#V#unknown_workflow", + "name": "Unknown workflow", + "description": "Run an unclassified represented process.", + "relevance_score": 0.8, + "is_executable": True, + "routing_eligible": True, + "routing_index_metadata": { + "required_tools": ["unavailable_private_operation"], + }, + }, + ], + "candidate_count": 2, + "errors": [], + }, + ) + monkeypatch.setattr( + "src.backend.workflows.vontology_loader.resolve_workflow_launch_input_contract", + lambda _workflow_id: (None, None), + ) + + capabilities, _diagnostic = discover_turn_workflow_capabilities( + "create or inspect a represented work product", + namespace="#V#user@org", + user_concept_id="#V#user", + organisation_concept_id="#V#org", + turn_id="turn-effects", + registered_capability_categories={ + "create_concepts": "write", + }, + ) + + by_id = {item.workflow_id: item for item in capabilities} + assert by_id["#V#write_workflow"].semantic_effect is True + assert by_id["#V#write_workflow"].semantic_effect_source == ( + "declared_registered_write_component" + ) + assert by_id["#V#unknown_workflow"].semantic_effect is None + unknown_entry = by_id["#V#unknown_workflow"].to_catalogue_entry() + assert unknown_entry["semantic_effect"] is None + assert unknown_entry["effect_profile"]["operational_state_effect"] is True + + def test_execution_arguments_bind_workflow_and_actor_server_side(): from src.backend.services.workflow_turn_capability_service import ( build_workflow_execution_arguments,